Merge branch 'dev/v5.0' into resolve-conflict-to-5.0

.github/workflows/run_cts_tests.yaml

@@ -89,7 +89,7 @@ jobs:
         if: matrix.connect_type == 'tls'
         run: |
           cat <<-EOF >> "$GITHUB_ENV"
-          EMQX_AUTH__MONGO__SSL=on
+          EMQX_AUTH__MONGO__SSL__ENABLE=on
           EMQX_AUTH__MONGO__SSL__CACERTFILE=/emqx/apps/emqx_auth_mongo/test/emqx_auth_mongo_SUITE_data/ca.pem
           EMQX_AUTH__MONGO__SSL__CERTFILE=/emqx/apps/emqx_auth_mongo/test/emqx_auth_mongo_SUITE_data/client-cert.pem
           EMQX_AUTH__MONGO__SSL__KEYFILE=/emqx/apps/emqx_auth_mongo/test/emqx_auth_mongo_SUITE_data/client-key.pem
@@ -101,7 +101,7 @@ jobs:
           MONGO_TAG: ${{ matrix.mongo_tag }}
         if: matrix.connect_type == 'tcp'
         run: |
-          echo EMQX_AUTH__MONGO__SSL=off >> "$GITHUB_ENV"
+          echo EMQX_AUTH__MONGO__SSL__ENABLE=off >> "$GITHUB_ENV"
       - name: setup
         if: matrix.network_type == 'ipv4'
         run: |
@@ -163,10 +163,10 @@ jobs:
         if: matrix.connect_type == 'tls'
         run: |
           cat <<-EOF >> "$GITHUB_ENV"
+            EMQX_AUTH__MYSQL__SSL__ENABLE=on
             EMQX_AUTH__MYSQL__USERNAME=ssluser
             EMQX_AUTH__MYSQL__PASSWORD=public
             EMQX_AUTH__MYSQL__DATABASE=mqtt
-            EMQX_AUTH__MYSQL__SSL=on
             EMQX_AUTH__MYSQL__SSL__CACERTFILE=/emqx/apps/emqx_auth_mysql/test/emqx_auth_mysql_SUITE_data/ca.pem
             EMQX_AUTH__MYSQL__SSL__CERTFILE=/emqx/apps/emqx_auth_mysql/test/emqx_auth_mysql_SUITE_data/client-cert.pem
             EMQX_AUTH__MYSQL__SSL__KEYFILE=/emqx/apps/emqx_auth_mysql/test/emqx_auth_mysql_SUITE_data/client-key.pem
@@ -182,7 +182,7 @@ jobs:
             EMQX_AUTH__MYSQL__USERNAME=root
             EMQX_AUTH__MYSQL__PASSWORD=public
             EMQX_AUTH__MYSQL__DATABASE=mqtt
-            EMQX_AUTH__MYSQL__SSL=off
+            EMQX_AUTH__MYSQL__SSL__ENABLE=off
           EOF
       - name: setup
         if: matrix.network_type == 'ipv4'
@@ -242,7 +242,7 @@ jobs:
         if: matrix.connect_type == 'tls'
         run: |
           cat <<-EOF >> "$GITHUB_ENV"
-          EMQX_AUTH__PGSQL__SSL=on
+          EMQX_AUTH__PGSQL__SSL__ENABLE=on
           EMQX_AUTH__PGSQL__SSL__CACERTFILE=/emqx/apps/emqx_auth_pgsql/test/emqx_auth_pgsql_SUITE_data/ca.pem
           EMQX_AUTH__PGSQL__SSL__CERTFILE=/emqx/apps/emqx_auth_pgsql/test/emqx_auth_pgsql_SUITE_data/client-cert.pem
           EMQX_AUTH__PGSQL__SSL__KEYFILE=/emqx/apps/emqx_auth_pgsql/test/emqx_auth_pgsql_SUITE_data/client-key.pem
@@ -254,7 +254,7 @@ jobs:
           PGSQL_TAG: ${{ matrix.pgsql_tag }}
         if: matrix.connect_type == 'tcp'
         run: |
-          echo EMQX_AUTH__PGSQL__SSL=off >> "$GITHUB_ENV"
+          echo EMQX_AUTH__PGSQL__SSL__ENABLE=off >> "$GITHUB_ENV"
       - name: setup
         if: matrix.network_type == 'ipv4'
         run: |
@@ -321,7 +321,7 @@ jobs:
         if: matrix.connect_type == 'tls'
         run: |
           cat <<-EOF >> "$GITHUB_ENV"
-          EMQX_AUTH__REDIS__SSL=on
+          EMQX_AUTH__REDIS__SSL__ENABLE=on
           EMQX_AUTH__REDIS__SSL__CACERTFILE=/emqx/apps/emqx_auth_redis/test/emqx_auth_redis_SUITE_data/certs/ca.crt
           EMQX_AUTH__REDIS__SSL__CERTFILE=/emqx/apps/emqx_auth_redis/test/emqx_auth_redis_SUITE_data/certs/redis.crt
           EMQX_AUTH__REDIS__SSL__KEYFILE=/emqx/apps/emqx_auth_redis/test/emqx_auth_redis_SUITE_data/certs/redis.key
@@ -333,7 +333,7 @@ jobs:
           REDIS_TAG: ${{ matrix.redis_tag }}
         if: matrix.connect_type == 'tcp'
         run: |
-          echo EMQX_AUTH__REDIS__SSL=off >> "$GITHUB_ENV"
+          echo EMQX_AUTH__REDIS__SSL__ENABLE=off >> "$GITHUB_ENV"
       - name: get server address
         run: |
           ipv4_address=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' redis)

apps/emqx_auth_http/etc/emqx_auth_http.conf

@@ -7,7 +7,7 @@
 ## Value: URL
 ##
 ## Examples: http://127.0.0.1:80/mqtt/auth, https://[::1]:80/mqtt/auth
-auth.http.auth_req.url = http://127.0.0.1:80/mqtt/auth
+auth.http.auth_req.url = "http://127.0.0.1:80/mqtt/auth"
 
 ## HTTP Request Method for Auth Request
 ##
@@ -18,7 +18,8 @@ auth.http.auth_req.method = post
 ## The possible values of the Content-Type header: application/x-www-form-urlencoded, application/json
 ##
 ## Examples: auth.http.auth_req.headers.accept = */*
-auth.http.auth_req.headers.content_type = application/x-www-form-urlencoded
+
+auth.http.auth_req.headers.content_type = "application/x-www-form-urlencoded"
 
 ## Parameters used to construct the request body or query string parameters
 ## When the request method is GET, these parameters will be converted into query string parameters
@@ -35,14 +36,14 @@ auth.http.auth_req.headers.content_type = application/x-www-form-urlencoded
 ##  - %d: subject of client TLS cert
 ##
 ## Value: <K1>=<V1>,<K2>=<V2>,...
-auth.http.auth_req.params = clientid=%c,username=%u,password=%P
+auth.http.auth_req.params = "clientid=%c,username=%u,password=%P"
 
 ## HTTP URL API path for SuperUser Request
 ##
 ## Value: URL
 ##
 ## Examples: http://127.0.0.1:80/mqtt/superuser, https://[::1]:80/mqtt/superuser
-auth.http.super_req.url = http://127.0.0.1:80/mqtt/superuser
+auth.http.super_req.url = "http://127.0.0.1:80/mqtt/superuser"
 
 ## HTTP Request Method for SuperUser Request
 ##
@@ -53,7 +54,7 @@ auth.http.super_req.method = post
 ## The possible values of the Content-Type header: application/x-www-form-urlencoded, application/json
 ##
 ## Examples: auth.http.super_req.headers.accept = */*
-auth.http.super_req.headers.content-type = application/x-www-form-urlencoded
+auth.http.super_req.headers.content-type = "application/x-www-form-urlencoded"
 
 ## Parameters used to construct the request body or query string parameters
 ## When the request method is GET, these parameters will be converted into query string parameters
@@ -70,7 +71,7 @@ auth.http.super_req.headers.content-type = application/x-www-form-urlencoded
 ##  - %d: subject of client TLS cert
 ##
 ## Value: <K1>=<V1>,<K2>=<V2>,...
-auth.http.super_req.params = clientid=%c,username=%u
+auth.http.super_req.params = "clientid=%c,username=%u"
 
 ## HTTP URL API path for ACL Request
 ## Comment out this config to disable ACL checks
@@ -78,7 +79,7 @@ auth.http.super_req.params = clientid=%c,username=%u
 ## Value: URL
 ##
 ## Examples: http://127.0.0.1:80/mqtt/acl, https://[::1]:80/mqtt/acl
-auth.http.acl_req.url = http://127.0.0.1:80/mqtt/acl
+auth.http.acl_req.url = "http://127.0.0.1:80/mqtt/acl"
 
 ## HTTP Request Method for ACL Request
 ##
@@ -89,7 +90,7 @@ auth.http.acl_req.method = post
 ## The possible values of the Content-Type header: application/x-www-form-urlencoded, application/json
 ##
 ## Examples: auth.http.acl_req.headers.accept = */*
-auth.http.acl_req.headers.content-type = application/x-www-form-urlencoded
+auth.http.acl_req.headers.content-type = "application/x-www-form-urlencoded"
 
 ## Parameters used to construct the request body or query string parameters
 ## When the request method is GET, these parameters will be converted into query string parameters
@@ -108,7 +109,7 @@ auth.http.acl_req.headers.content-type = application/x-www-form-urlencoded
 ##  - %t: topic
 ##
 ## Value: <K1>=<V1>,<K2>=<V2>,...
-auth.http.acl_req.params = access=%A,username=%u,clientid=%c,ipaddr=%a,topic=%t,mountpoint=%m
+auth.http.acl_req.params = "access=%A,username=%u,clientid=%c,ipaddr=%a,topic=%t,mountpoint=%m"
 
 ## Time-out time for the request.
 ##
@@ -143,17 +144,17 @@ auth.http.pool_size = 32
 ## are used during server authentication and when building the client certificate chain.
 ##
 ## Value: File
-## auth.http.ssl.cacertfile = {{ platform_etc_dir }}/certs/ca.pem
+## auth.http.ssl.cacertfile = "{{ platform_etc_dir }}/certs/ca.pem"
 
 ## The path to a file containing the client's certificate.
 ##
 ## Value: File
-## auth.http.ssl.certfile = {{ platform_etc_dir }}/certs/client-cert.pem
+## auth.http.ssl.certfile = "{{ platform_etc_dir }}/certs/client-cert.pem"
 
 ## Path to a file containing the client's private PEM-encoded key.
 ##
 ## Value: File
-## auth.http.ssl.keyfile = {{ platform_etc_dir }}/certs/client-key.pem
+## auth.http.ssl.keyfile = "{{ platform_etc_dir }}/certs/client-key.pem"
 
 ## In mode verify_none the default behavior is to allow all x509-path
 ## validation errors.

apps/emqx_auth_http/rebar.config

@@ -19,7 +19,7 @@
 {profiles,
  [{test,
    [{deps,
-     [{emqx_ct_helpers, {git, "https://github.com/emqx/emqx-ct-helpers", {tag, "1.2.2"}}},
+     [
       {emqtt, {git, "https://github.com/emqx/emqtt", {tag, "v1.2.2"}}}
      ]}
    ]}

apps/emqx_auth_jwt/etc/emqx_auth_jwt.conf

@@ -10,13 +10,13 @@ auth.jwt.secret = emqxsecret
 ## RSA or ECDSA public key file.
 ##
 ## Value: File
-#auth.jwt.pubkey = etc/certs/jwt_public_key.pem
+#auth.jwt.pubkey = "etc/certs/jwt_public_key.pem"
 
 ## The JWKs server address
 ##
 ## see: http://self-issued.info/docs/draft-ietf-jose-json-web-key.html
 ##
-#auth.jwt.jwks = https://127.0.0.1:8080/jwks
+#auth.jwt.jwks.endpoint = "https://127.0.0.1:8080/jwks"
 
 ## The JWKs refresh interval
 ##
@@ -32,7 +32,7 @@ auth.jwt.from = password
 ## Enable to verify claims fields
 ##
 ## Value: on | off
-auth.jwt.verify_claims = off
+auth.jwt.verify_claims.enable = off
 
 ## The checklist of claims to validate
 ##
@@ -46,4 +46,4 @@ auth.jwt.verify_claims = off
 ##
 ## For example, to verify that the username in the JWT payload is the same
 ## as the client (MQTT protocol) username
-#auth.jwt.verify_claims.username = %u
+#auth.jwt.verify_claims.username = "%u"

apps/emqx_auth_jwt/priv/emqx_auth_jwt.schema

@@ -4,7 +4,7 @@
   {datatype, string}
 ]}.
 
-{mapping, "auth.jwt.jwks", "emqx_auth_jwt.jwks", [
+{mapping, "auth.jwt.jwks.endpoint", "emqx_auth_jwt.jwks", [
   {datatype, string}
 ]}.
 
@@ -26,7 +26,7 @@
   {datatype, {enum, [raw, der]}}
 ]}.
 
-{mapping, "auth.jwt.verify_claims", "emqx_auth_jwt.verify_claims", [
+{mapping, "auth.jwt.verify_claims.enable", "emqx_auth_jwt.verify_claims", [
   {default, off},
   {datatype, flag}
 ]}.
@@ -36,7 +36,7 @@
 ]}.
 
 {translation, "emqx_auth_jwt.verify_claims", fun(Conf) ->
-    case cuttlefish:conf_get("auth.jwt.verify_claims", Conf) of
+    case cuttlefish:conf_get("auth.jwt.verify_claims.enable", Conf) of
         false -> cuttlefish:unset();
         true ->
             lists:foldr(

apps/emqx_auth_jwt/rebar.config

@@ -20,6 +20,6 @@
 
 {profiles,
  [{test,
-   [{deps, [{emqx_ct_helpers, {git, "http://github.com/emqx/emqx-ct-helpers", {tag, "1.2.2"}}}]}
+   [{deps, []}
    ]}
  ]}.

apps/emqx_auth_ldap/etc/emqx_auth_ldap.conf

@@ -5,7 +5,7 @@
 ## LDAP server list, seperated by ','.
 ##
 ## Value: String
-auth.ldap.servers = 127.0.0.1
+auth.ldap.servers = "127.0.0.1"
 
 ## LDAP server port.
 ##
@@ -20,7 +20,7 @@ auth.ldap.pool = 8
 ## LDAP Bind DN.
 ##
 ## Value: DN
-auth.ldap.bind_dn = cn=root,dc=emqx,dc=io
+auth.ldap.bind_dn = "cn=root,dc=emqx,dc=io"
 
 ## LDAP Bind Password.
 ##
@@ -37,7 +37,7 @@ auth.ldap.timeout = 30s
 ## Variables:
 ##
 ## Value: DN
-auth.ldap.device_dn = ou=device,dc=emqx,dc=io
+auth.ldap.device_dn = "ou=device,dc=emqx,dc=io"
 
 ## Specified ObjectClass
 ##
@@ -63,14 +63,14 @@ auth.ldap.password.attributetype = userPassword
 ## Whether to enable SSL.
 ##
 ## Value: true | false
-auth.ldap.ssl = false
+auth.ldap.ssl.enable = false
 
-#auth.ldap.ssl.certfile = etc/certs/cert.pem
+#auth.ldap.ssl.certfile = "etc/certs/cert.pem"
 
-#auth.ldap.ssl.keyfile = etc/certs/key.pem
+#auth.ldap.ssl.keyfile = "etc/certs/key.pem"
 
-#auth.ldap.ssl.cacertfile = etc/certs/cacert.pem
+#auth.ldap.ssl.cacertfile = "etc/certs/cacert.pem"
 
-#auth.ldap.ssl.verify = verify_peer
+#auth.ldap.ssl.verify = "verify_peer"
 
 #auth.ldap.ssl.server_name_indication = your_server_name

apps/emqx_auth_ldap/priv/emqx_auth_ldap.schema

@@ -31,7 +31,7 @@
   {datatype, {duration, ms}}
 ]}.
 
-{mapping, "auth.ldap.ssl", "emqx_auth_ldap.ldap", [
+{mapping, "auth.ldap.ssl.enable", "emqx_auth_ldap.ldap", [
   {default, false},
   {datatype, {enum, [true, false]}}
 ]}.
@@ -83,7 +83,7 @@
             {bind_password, BindPassword},
             {pool, Pool},
             {auto_reconnect, 2}],
-    case cuttlefish:conf_get("auth.ldap.ssl", Conf) of
+    case cuttlefish:conf_get("auth.ldap.ssl.enable", Conf) of
         true  -> [{ssl, true}, {sslopts, Filter(SslOpts())}|Opts];
         false -> [{ssl, false}|Opts]
     end

apps/emqx_auth_ldap/rebar.config

@@ -4,7 +4,7 @@
 
 {profiles,
  [{test,
-   [{deps, [{emqx_ct_helpers, {git, "https://github.com/emqx/emqx-ct-helpers", {tag, "1.2.2"}}}]}
+   [{deps, []}
    ]}
  ]}.
 

apps/emqx_auth_mnesia/etc/emqx_auth_mnesia.conf

@@ -10,12 +10,12 @@ auth.mnesia.password_hash = sha256
 ## Examples
 ##auth.client.1.clientid = id
 ##auth.client.1.password = passwd
-##auth.client.2.clientid = dev:devid
+##auth.client.2.clientid = "dev:devid"
 ##auth.client.2.password = passwd2
-##auth.client.3.clientid = app:appid
+##auth.client.3.clientid = "app:appid"
 ##auth.client.3.password = passwd3
-##auth.client.4.clientid = client~!@#$%^&*()_+
-##auth.client.4.password = passwd~!@#$%^&*()_+
+##auth.client.4.clientid = "client~!@#$%^&*()_+"
+##auth.client.4.password = "passwd~!@#$%^&*()_+"
 
 ##--------------------------------------------------------------------
 ## Username Authentication
@@ -26,5 +26,5 @@ auth.mnesia.password_hash = sha256
 ##auth.user.1.password = public
 ##auth.user.2.username = feng@emqtt.io
 ##auth.user.2.password = public
-##auth.user.3.username = name~!@#$%^&*()_+
-##auth.user.3.password = pwsswd~!@#$%^&*()_+
+##auth.user.3.username = "name~!@#$%^&*()_+"
+##auth.user.3.password = "pwsswd~!@#$%^&*()_+"

apps/emqx_auth_mongo/etc/emqx_auth_mongo.conf

@@ -16,8 +16,8 @@ auth.mongo.type = single
 ##
 ## Value: String
 ##
-## Examples: 127.0.0.1:27017,127.0.0.2:27017...
-auth.mongo.server = 127.0.0.1:27017
+## Examples: "127.0.0.1:27017,127.0.0.2:27017,..."
+auth.mongo.server = "127.0.0.1:27017"
 
 ## MongoDB pool size
 ##
@@ -53,7 +53,7 @@ auth.mongo.database = mqtt
 ## Whether to enable SSL connection.
 ##
 ## Value: on | off
-## auth.mongo.ssl = off
+## auth.mongo.ssl.enable = off
 
 ## SSL keyfile.
 ##
@@ -117,17 +117,17 @@ auth.mongo.topology.max_overflow = 0
 auth.mongo.auth_query.password_hash = sha256
 
 ## sha256 with salt suffix
-## auth.mongo.auth_query.password_hash = sha256,salt
+## auth.mongo.auth_query.password_hash = "sha256,salt"
 
 ## sha256 with salt prefix
-## auth.mongo.auth_query.password_hash = salt,sha256
+## auth.mongo.auth_query.password_hash = "salt,sha256"
 
 ## bcrypt with salt prefix
-## auth.mongo.auth_query.password_hash = salt,bcrypt
+## auth.mongo.auth_query.password_hash = "salt,bcrypt"
 
 ## pbkdf2 with macfun iterations dklen
 ## macfun: md4, md5, ripemd160, sha, sha224, sha256, sha384, sha512
-## auth.mongo.auth_query.password_hash = pbkdf2,sha256,1000,20
+## auth.mongo.auth_query.password_hash = "pbkdf2,sha256,1000,20"
 
 ## Authentication query.
 auth.mongo.auth_query.collection = mqtt_user
@@ -146,15 +146,15 @@ auth.mongo.auth_query.password_field = password
 ##  - %d: subject of client TLS cert
 ##
 ## auth.mongo.auth_query.selector = {Field}={Placeholder}
-auth.mongo.auth_query.selector = username=%u
+auth.mongo.auth_query.selector = "username=%u"
 
 ## -------------------------------------------------
 ## Super User Query
 ## -------------------------------------------------
 auth.mongo.super_query.collection = mqtt_user
 auth.mongo.super_query.super_field = is_superuser
-#auth.mongo.super_query.selector = username=%u, clientid=%c
-auth.mongo.super_query.selector = username=%u
+#auth.mongo.super_query.selector.1 = username=%u, clientid=%c
+auth.mongo.super_query.selector = "username=%u"
 
 ## ACL Selector.
 ##
@@ -165,8 +165,8 @@ auth.mongo.super_query.selector = username=%u
 ##
 ## With following 2 selectors configured:
 ##
-## auth.mongo.acl_query.selector.1 = username=%u
-## auth.mongo.acl_query.selector.2 = username=$all
+## auth.mongo.acl_query.selector.1 = "username=%u"
+## auth.mongo.acl_query.selector.2 = "username=$all"
 ##
 ## And if a client connected using username 'ilyas',
 ##   then the following mongo command will be used to
@@ -180,8 +180,8 @@ auth.mongo.super_query.selector = username=%u
 ##
 ## Examples:
 ##
-## auth.mongo.acl_query.selector.1 = username=%u,clientid=%c
-## auth.mongo.acl_query.selector.2 = username=$all
-## auth.mongo.acl_query.selector.3 = clientid=$all
+## auth.mongo.acl_query.selector.1 = "username=%u,clientid=%c"
+## auth.mongo.acl_query.selector.2 = "username=$all"
+## auth.mongo.acl_query.selector.3 = "clientid=$all"
 auth.mongo.acl_query.collection = mqtt_acl
-auth.mongo.acl_query.selector = username=%u
+auth.mongo.acl_query.selector = "username=%u"

apps/emqx_auth_mongo/priv/emqx_auth_mongo.schema

@@ -45,7 +45,7 @@
   {datatype, string}
 ]}.
 
-{mapping, "auth.mongo.ssl", "emqx_auth_mongo.server", [
+{mapping, "auth.mongo.ssl.enable", "emqx_auth_mongo.server", [
   {default, off},
   {datatype, {enum, [on, off, true, false]}} %% FIXME: ture/false is compatible with 4.0-4.2 version format, plan to delete in 5.0
 ]}.
@@ -130,8 +130,6 @@
     true -> [];
     false -> [{r_mode, R}]
   end,
-
-
   Filter  = fun(Opts) -> [{K, V} || {K, V} <- Opts, V =/= undefined] end,
   SslOpts = fun(Prefix) ->
                 Verify = case cuttlefish:conf_get(Prefix ++ ".verify", Conf, false) of
@@ -149,8 +147,14 @@
             end,
 
   %% FIXME: compatible with 4.0-4.2 version format, plan to delete in 5.0
-  Ssl = case cuttlefish:conf_get("auth.mongo.ssl", Conf) of
-          on -> [{ssl, true}, {ssl_opts, SslOpts("auth.mongo.ssl")}];
+  GenSsl = case cuttlefish:conf_get("auth.mongo.ssl.cacertfile", Conf, undefined) of
+               undefined -> [{ssl, true}, {ssl_opts, SslOpts("auth.mongo.ssl_opts")}];
+               _ -> [{ssl, true}, {ssl_opts, SslOpts("auth.mongo.ssl")}]
+           end,
+
+  %% FIXME: compatible with 4.0-4.2 version format, plan to delete in 5.0
+  Ssl = case cuttlefish:conf_get("auth.mongo.ssl.enable", Conf) of
+          on -> GenSsl;
           off -> [];
           true -> [{ssl, true}, {ssl_opts, SslOpts("auth.mongo.ssl_opts")}];
           false -> []

apps/emqx_auth_mysql/etc/emqx_auth_mysql.conf

@@ -7,7 +7,7 @@
 ## Value: Port | IP:Port
 ##
 ## Examples: 3306, 127.0.0.1:3306, localhost:3306
-auth.mysql.server = 127.0.0.1:3306
+auth.mysql.server = "127.0.0.1:3306"
 
 ## MySQL pool size.
 ##
@@ -50,7 +50,7 @@ auth.mysql.database = mqtt
 ##  - %C: common name of client TLS cert
 ##  - %d: subject of client TLS cert
 ##
-auth.mysql.auth_query = select password from mqtt_user where username = '%u' limit 1
+auth.mysql.auth_query = "select password from mqtt_user where username = '%u' limit 1"
 ## auth.mysql.auth_query = select password_hash as password from mqtt_user where username = '%u' limit 1
 
 ## Password hash.
@@ -59,17 +59,17 @@ auth.mysql.auth_query = select password from mqtt_user where username = '%u' lim
 auth.mysql.password_hash = sha256
 
 ## sha256 with salt prefix
-## auth.mysql.password_hash = salt,sha256
+## auth.mysql.password_hash = "salt,sha256"
 
 ## bcrypt with salt only prefix
-## auth.mysql.password_hash = salt,bcrypt
+## auth.mysql.password_hash = "salt,bcrypt"
 
 ## sha256 with salt suffix
-## auth.mysql.password_hash = sha256,salt
+## auth.mysql.password_hash = "sha256,salt"
 
 ## pbkdf2 with macfun iterations dklen
 ## macfun: md4, md5, ripemd160, sha, sha224, sha256, sha384, sha512
-## auth.mysql.password_hash = pbkdf2,sha256,1000,20
+## auth.mysql.password_hash = "pbkdf2,sha256,1000,20"
 
 ## Superuser query.
 ##
@@ -81,7 +81,7 @@ auth.mysql.password_hash = sha256
 ##  - %C: common name of client TLS cert
 ##  - %d: subject of client TLS cert
 ##
-auth.mysql.super_query = select is_superuser from mqtt_user where username = '%u' limit 1
+auth.mysql.super_query = "select is_superuser from mqtt_user where username = '%u' limit 1"
 
 ## ACL query.
 ##
@@ -93,12 +93,12 @@ auth.mysql.super_query = select is_superuser from mqtt_user where username = '%u
 ##  - %c: clientid
 ##
 ## Note: You can add the 'ORDER BY' statement to control the rules match order
-auth.mysql.acl_query = select allow, ipaddr, username, clientid, access, topic from mqtt_acl where ipaddr = '%a' or username = '%u' or username = '$all' or clientid = '%c'
+auth.mysql.acl_query = "select allow, ipaddr, username, clientid, access, topic from mqtt_acl where ipaddr = '%a' or username = '%u' or username = '$all' or clientid = '%c'"
 
 ## Mysql ssl configuration.
 ##
 ## Value: on | off
-#auth.mysql.ssl = off
+## auth.mysql.ssl.enable = off
 
 ## CA certificate.
 ##

apps/emqx_auth_mysql/priv/emqx_auth_mysql.schema

@@ -30,7 +30,7 @@
   {datatype, string}
 ]}.
 
-{mapping, "auth.mysql.ssl", "emqx_auth_mysql.server", [
+{mapping, "auth.mysql.ssl.enable", "emqx_auth_mysql.server", [
   {default, off},
   {datatype, flag}
 ]}.
@@ -94,7 +94,7 @@
              {keep_alive, true}],
   Filter  = fun(Opts) -> [{K, V} || {K, V} <- Opts, V =/= undefined] end,
   Options1 =
-      case cuttlefish:conf_get("auth.mysql.ssl", Conf) of
+      case cuttlefish:conf_get("auth.mysql.ssl.enable", Conf) of
             true ->
                 %% FIXME: compatible with 4.0-4.2 version format, plan to delete in 5.0
                 CA = cuttlefish:conf_get(

apps/emqx_auth_pgsql/README.md

@@ -49,7 +49,7 @@ auth.pgsql.encoding = utf8
 ## Whether to enable SSL connection.
 ##
 ## Value: true | false
-auth.pgsql.ssl = false
+auth.pgsql.ssl.enable = false
 
 ## SSL keyfile.
 ##

apps/emqx_auth_pgsql/etc/emqx_auth_pgsql.conf

@@ -6,8 +6,8 @@
 ##
 ## Value: Port | IP:Port
 ##
-## Examples: 5432, 127.0.0.1:5432, localhost:5432
-auth.pgsql.server = 127.0.0.1:5432
+## Examples: 5432, "127.0.0.1:5432", "localhost:5432"
+auth.pgsql.server = "127.0.0.1:5432"
 
 ## PostgreSQL pool size.
 ##
@@ -37,7 +37,7 @@ auth.pgsql.encoding = utf8
 ## Whether to enable SSL connection.
 ##
 ## Value: on | off
-auth.pgsql.ssl = off
+auth.pgsql.ssl.enable = off
 
 ## TLS version.
 ##
@@ -87,7 +87,7 @@ auth.pgsql.ssl = off
 ##  - %C: common name of client TLS cert
 ##  - %d: subject of client TLS cert
 ##
-auth.pgsql.auth_query = select password from mqtt_user where username = '%u' limit 1
+auth.pgsql.auth_query = "select password from mqtt_user where username = '%u' limit 1"
 
 ## Password hash.
 ##
@@ -95,17 +95,17 @@ auth.pgsql.auth_query = select password from mqtt_user where username = '%u' lim
 auth.pgsql.password_hash = sha256
 
 ## sha256 with salt prefix
-## auth.pgsql.password_hash = salt,sha256
+## auth.pgsql.password_hash = "salt,sha256"
 
 ## sha256 with salt suffix
-## auth.pgsql.password_hash = sha256,salt
+## auth.pgsql.password_hash = "sha256,salt"
 
 ## bcrypt with salt prefix
-## auth.pgsql.password_hash = salt,bcrypt
+## auth.pgsql.password_hash = "salt,bcrypt"
 
 ## pbkdf2 with macfun iterations dklen
 ## macfun: md4, md5, ripemd160, sha, sha224, sha256, sha384, sha512
-## auth.pgsql.password_hash = pbkdf2,sha256,1000,20
+## auth.pgsql.password_hash = "pbkdf2,sha256,1000,20"
 
 ## Superuser query.
 ##
@@ -117,7 +117,7 @@ auth.pgsql.password_hash = sha256
 ##  - %C: common name of client TLS cert
 ##  - %d: subject of client TLS cert
 ##
-auth.pgsql.super_query = select is_superuser from mqtt_user where username = '%u' limit 1
+auth.pgsql.super_query = "select is_superuser from mqtt_user where username = '%u' limit 1"
 
 ## ACL query. Comment this query, the ACL will be disabled.
 ##
@@ -129,4 +129,4 @@ auth.pgsql.super_query = select is_superuser from mqtt_user where username = '%u
 ##  - %c: clientid
 ##
 ## Note: You can add the 'ORDER BY' statement to control the rules match order
-auth.pgsql.acl_query = select allow, ipaddr, username, clientid, access, topic from mqtt_acl where ipaddr = '%a' or username = '%u' or username = '$all' or clientid = '%c'
+auth.pgsql.acl_query = "select allow, ipaddr, username, clientid, access, topic from mqtt_acl where ipaddr = '%a' or username = '%u' or username = '$all' or clientid = '%c'"

apps/emqx_auth_pgsql/priv/emqx_auth_pgsql.schema

@@ -30,7 +30,7 @@
   {datatype, atom}
 ]}.
 
-{mapping, "auth.pgsql.ssl", "emqx_auth_pgsql.server", [
+{mapping, "auth.pgsql.ssl.enable", "emqx_auth_pgsql.server", [
   {default, off},
   {datatype, {enum, [on, off, true, false]}} %% FIXME: true/fasle is compatible with 4.0-4.2 version format, plan to delete in 5.0
 ]}.
@@ -116,8 +116,14 @@
             end,
 
   %% FIXME: compatible with 4.0-4.2 version format, plan to delete in 5.0
-  Ssl = case cuttlefish:conf_get("auth.pgsql.ssl", Conf) of
-          on -> [{ssl, true}, {ssl_opts, SslOpts("auth.pgsql.ssl")}];
+  GenSsl = case cuttlefish:conf_get("auth.pgsql.ssl.cacertfile", Conf, undefined) of
+               undefined -> [{ssl, true}, {ssl_opts, SslOpts("auth.pgsql.ssl_opts")}];
+               _ -> [{ssl, true}, {ssl_opts, SslOpts("auth.pgsql.ssl")}]
+           end,
+
+  %% FIXME: compatible with 4.0-4.2 version format, plan to delete in 5.0
+  Ssl = case cuttlefish:conf_get("auth.pgsql.ssl.enable", Conf) of
+          on -> GenSsl;
           off -> [];
           true -> [{ssl, true}, {ssl_opts, SslOpts("auth.pgsql.ssl_opts")}];
           false -> []

apps/emqx_auth_redis/etc/emqx_auth_redis.conf

@@ -12,9 +12,9 @@ auth.redis.type = single
 ## Value: Port | IP:Port
 ##
 ## Single Redis Server: 127.0.0.1:6379, localhost:6379
-## Redis Sentinel: 127.0.0.1:26379,127.0.0.2:26379,127.0.0.3:26379
-## Redis Cluster: 127.0.0.1:6379,127.0.0.2:6379,127.0.0.3:6379
-auth.redis.server = 127.0.0.1:6379
+## Redis Sentinel: "127.0.0.1:26379,127.0.0.2:26379,127.0.0.3:26379"
+## Redis Cluster: "127.0.0.1:6379,127.0.0.2:6379,127.0.0.3:6379"
+auth.redis.server = "127.0.0.1:6379"
 
 ## Redis sentinel cluster name.
 ##
@@ -52,10 +52,10 @@ auth.redis.database = 0
 ##  - %d: subject of client TLS cert
 ##
 ## Examples:
-##  - HGET mqtt_user:%u password
-##  - HMGET mqtt_user:%u password
-##  - HMGET mqtt_user:%u password salt
-auth.redis.auth_cmd = HMGET mqtt_user:%u password
+##  - "HGET mqtt_user:%u password"
+##  - "HMGET mqtt_user:%u password"
+##  - "HMGET mqtt_user:%u password salt"
+auth.redis.auth_cmd = "HMGET mqtt_user:%u password"
 
 ## Password hash.
 ##
@@ -63,17 +63,17 @@ auth.redis.auth_cmd = HMGET mqtt_user:%u password
 auth.redis.password_hash = plain
 
 ## sha256 with salt prefix
-## auth.redis.password_hash = salt,sha256
+## auth.redis.password_hash = "salt,sha256"
 
 ## sha256 with salt suffix
-## auth.redis.password_hash = sha256,salt
+## auth.redis.password_hash = "sha256,salt"
 
 ## bcrypt with salt prefix
-## auth.redis.password_hash = salt,bcrypt
+## auth.redis.password_hash = "salt,bcrypt"
 
 ## pbkdf2 with macfun iterations dklen
 ## macfun: md4, md5, ripemd160, sha, sha224, sha256, sha384, sha512
-## auth.redis.password_hash = pbkdf2,sha256,1000,20
+## auth.redis.password_hash = "pbkdf2,sha256,1000,20"
 
 ## Superuser query command.
 ##
@@ -84,7 +84,7 @@ auth.redis.password_hash = plain
 ##  - %c: clientid
 ##  - %C: common name of client TLS cert
 ##  - %d: subject of client TLS cert
-auth.redis.super_cmd = HGET mqtt_user:%u is_superuser
+auth.redis.super_cmd = "HGET mqtt_user:%u is_superuser"
 
 ## ACL query command.
 ##
@@ -93,12 +93,12 @@ auth.redis.super_cmd = HGET mqtt_user:%u is_superuser
 ## Variables:
 ##  - %u: username
 ##  - %c: clientid
-auth.redis.acl_cmd = HGETALL mqtt_acl:%u
+auth.redis.acl_cmd = "HGETALL mqtt_acl:%u"
 
 ## Redis ssl configuration.
 ##
 ## Value: on | off
-#auth.redis.ssl = off
+# auth.redis.ssl.enable = off
 
 ## CA certificate.
 ##
@@ -108,12 +108,12 @@ auth.redis.acl_cmd = HGETALL mqtt_acl:%u
 ## Client ssl certificate.
 ##
 ## Value: File
-#auth.redis.ssl.certfile = path/to/your/certfile
+# auth.redis.ssl.certfile = path/to/your/certfile
 
 ## Client ssl keyfile.
 ##
 ## Value: File
-#auth.redis.ssl.keyfile = path/to/your/keyfile
+# auth.redis.ssl.keyfile = path/to/your/keyfile
 
 ## In mode verify_none the default behavior is to allow all x509-path
 ## validation errors.

apps/emqx_auth_redis/priv/emqx_auth_redis.schema

@@ -33,7 +33,7 @@
   hidden
 ]}.
 
-{mapping, "auth.redis.ssl", "emqx_auth_redis.options", [
+{mapping, "auth.redis.ssl.enable", "emqx_auth_redis.options", [
   {default, off},
   {datatype, flag}
 ]}.
@@ -75,7 +75,7 @@
 ]}.
 
 {translation, "emqx_auth_redis.options", fun(Conf) ->
-   Ssl = cuttlefish:conf_get("auth.redis.ssl", Conf, false),
+   Ssl = cuttlefish:conf_get("auth.redis.ssl.enable", Conf, false),
    Filter  = fun(Opts) -> [{K, V} || {K, V} <- Opts, V =/= undefined] end,
    case Ssl of
        true ->

apps/emqx_bridge_mqtt/README.md

@@ -53,13 +53,13 @@ The following is the basic configuration of RPC bridging. A simplest RPC bridgin
 
 ```
 ## Bridge Address: Use node name (nodename@host) for rpc bridging, and host:port for mqtt connection
-bridge.mqtt.emqx2.address = emqx2@192.168.1.2
+bridge.mqtt.emqx2.address = "emqx2@192.168.1.2"
 
 ## Forwarding topics of the message
-bridge.mqtt.emqx2.forwards = sensor1/#,sensor2/#
+bridge.mqtt.emqx2.forwards = "sensor1/#,sensor2/#"
 
 ## bridged mountpoint
-bridge.mqtt.emqx2.mountpoint = bridge/emqx2/${node}/
+bridge.mqtt.emqx2.mountpoint = "bridge/emqx2/${node}/"
 ```
 
 If the messages received by the local node emqx1 matches the topic `sersor1/#` or `sensor2/#`, these messages will be forwarded to the `sensor1/#` or `sensor2/#` topic of the remote node emqx2.
@@ -82,66 +82,66 @@ EMQ X MQTT bridging principle: Create an MQTT client on the EMQ X broker, and co
 
 ```
 ## Bridge Address: Use node name for rpc bridging, use host:port for mqtt connection
-bridge.mqtt.emqx2.address = 192.168.1.2:1883
+bridge.mqtt.emqx2.address = "192.168.1.2:1883"
 
 ## Bridged Protocol Version
 ## Enumeration value: mqttv3 | mqttv4 | mqttv5
-bridge.mqtt.emqx2.proto_ver = mqttv4
+bridge.mqtt.emqx2.proto_ver = "mqttv4"
 
 ## mqtt client's clientid
-bridge.mqtt.emqx2.clientid = bridge_emq
+bridge.mqtt.emqx2.clientid = "bridge_emq"
 
 ## mqtt client's clean_start field
 ## Note: Some MQTT Brokers need to set the clean_start value as `true`
 bridge.mqtt.emqx2.clean_start = true
 
 ##  mqtt client's username field
-bridge.mqtt.emqx2.username = user
+bridge.mqtt.emqx2.username = "user"
 
 ## mqtt client's password field
-bridge.mqtt.emqx2.password = passwd
+bridge.mqtt.emqx2.password = "passwd"
 
 ## Whether the mqtt client uses ssl to connect to a remote serve or not
 bridge.mqtt.emqx2.ssl = off
 
 ## CA Certificate of Client SSL Connection (PEM format)
-bridge.mqtt.emqx2.cacertfile = etc/certs/cacert.pem
+bridge.mqtt.emqx2.cacertfile = "etc/certs/cacert.pem"
 
 ## SSL certificate of Client SSL connection 
-bridge.mqtt.emqx2.certfile = etc/certs/client-cert.pem
+bridge.mqtt.emqx2.certfile = "etc/certs/client-cert.pem"
 
 ## Key file of Client SSL connection 
-bridge.mqtt.emqx2.keyfile = etc/certs/client-key.pem
+bridge.mqtt.emqx2.keyfile = "etc/certs/client-key.pem"
 
 ## SSL encryption
-bridge.mqtt.emqx2.ciphers = ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384
+bridge.mqtt.emqx2.ciphers = "ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384"
 
 ## TTLS PSK password
 ## Note 'listener.ssl.external.ciphers' and 'listener.ssl.external.psk_ciphers' cannot be configured at the same time
 ##
 ## See 'https://tools.ietf.org/html/rfc4279#section-2'.
-## bridge.mqtt.emqx2.psk_ciphers = PSK-AES128-CBC-SHA,PSK-AES256-CBC-SHA,PSK-3DES-EDE-CBC-SHA,PSK-RC4-SHA
+## bridge.mqtt.emqx2.psk_ciphers = "PSK-AES128-CBC-SHA,PSK-AES256-CBC-SHA,PSK-3DES-EDE-CBC-SHA,PSK-RC4-SHA"
 
 ## Client's heartbeat interval
 bridge.mqtt.emqx2.keepalive = 60s
 
 ## Supported TLS version
-bridge.mqtt.emqx2.tls_versions = tlsv1.3,tlsv1.2,tlsv1.1,tlsv1
+bridge.mqtt.emqx2.tls_versions = "tlsv1.3,tlsv1.2,tlsv1.1,tlsv1"
 
 ## Forwarding topics of the message
-bridge.mqtt.emqx2.forwards = sensor1/#,sensor2/#
+bridge.mqtt.emqx2.forwards = "sensor1/#,sensor2/#"
 
 ## Bridged mountpoint
-bridge.mqtt.emqx2.mountpoint = bridge/emqx2/${node}/
+bridge.mqtt.emqx2.mountpoint = "bridge/emqx2/${node}/"
 
 ## Subscription topic for bridging
-bridge.mqtt.emqx2.subscription.1.topic = cmd/topic1
+bridge.mqtt.emqx2.subscription.1.topic = "cmd/topic1"
 
 ## Subscription qos for bridging
 bridge.mqtt.emqx2.subscription.1.qos = 1
 
 ## Subscription topic for bridging
-bridge.mqtt.emqx2.subscription.2.topic = cmd/topic2
+bridge.mqtt.emqx2.subscription.2.topic = "cmd/topic2"
 
 ## Subscription qos for bridging
 bridge.mqtt.emqx2.subscription.2.qos = 1

apps/emqx_bridge_mqtt/docs/guide.rst

@@ -39,7 +39,7 @@ In EMQ X, bridge is configured by modifying ``etc/emqx.conf``. EMQ X distinguish
 .. code-block::
 
    ## Bridge address: node name for local bridge, host:port for remote.
-   bridge.mqtt.aws.address = 127.0.0.1:1883
+   bridge.mqtt.aws.address = "127.0.0.1:1883"
 
 This configuration declares a bridge named ``aws`` and specifies that it is bridged to the MQTT broker of 127.0.0.1:1883 by MQTT mode.
 
@@ -69,13 +69,13 @@ The following is the basic configuration of RPC bridging. A simplest RPC bridgin
 .. code-block::
 
    ## Bridge Address: Use node name (nodename@host) for rpc bridging, and host:port for mqtt connection
-   bridge.mqtt.emqx2.address = emqx2@192.168.1.2
+   bridge.mqtt.emqx2.address = "emqx2@192.168.1.2"
 
    ## Forwarding topics of the message
-   bridge.mqtt.emqx2.forwards = sensor1/#,sensor2/#
+   bridge.mqtt.emqx2.forwards = "sensor1/#,sensor2/#"
 
    ## bridged mountpoint
-   bridge.mqtt.emqx2.mountpoint = bridge/emqx2/${node}/
+   bridge.mqtt.emqx2.mountpoint = "bridge/emqx2/${node}/"
 
 If the messages received by the local node emqx1 matches the topic ``sersor1/#`` or ``sensor2/#``\ , these messages will be forwarded to the ``sensor1/#`` or ``sensor2/#`` topic of the remote node emqx2.
 
@@ -86,10 +86,10 @@ If the messages received by the local node emqx1 matches the topic ``sersor1/#``
 Limitations of RPC bridging:
 
 
-#. 
+#.
    The RPC bridge of emqx can only forward local messages to the remote node, and cannot synchronize the messages of the remote node to the local node;
 
-#. 
+#.
    RPC bridge can only bridge two EMQ X broker together and cannot bridge EMQ X broker to other MQTT brokers.
 
 EMQ X MQTT Bridge Configuration
@@ -102,66 +102,66 @@ EMQ X MQTT bridging principle: Create an MQTT client on the EMQ X broker, and co
 .. code-block::
 
    ## Bridge Address: Use node name for rpc bridging, use host:port for mqtt connection
-   bridge.mqtt.emqx2.address = 192.168.1.2:1883
+   bridge.mqtt.emqx2.address = "192.168.1.2:1883"
 
    ## Bridged Protocol Version
    ## Enumeration value: mqttv3 | mqttv4 | mqttv5
-   bridge.mqtt.emqx2.proto_ver = mqttv4
+   bridge.mqtt.emqx2.proto_ver = "mqttv4"
 
    ## mqtt client's clientid
-   bridge.mqtt.emqx2.clientid = bridge_emq
+   bridge.mqtt.emqx2.clientid = "bridge_emq"
 
    ## mqtt client's clean_start field
    ## Note: Some MQTT Brokers need to set the clean_start value as `true`
    bridge.mqtt.emqx2.clean_start = true
 
    ##  mqtt client's username field
-   bridge.mqtt.emqx2.username = user
+   bridge.mqtt.emqx2.username = "user"
 
    ## mqtt client's password field
-   bridge.mqtt.emqx2.password = passwd
+   bridge.mqtt.emqx2.password = "passwd"
 
    ## Whether the mqtt client uses ssl to connect to a remote serve or not
    bridge.mqtt.emqx2.ssl = off
 
    ## CA Certificate of Client SSL Connection (PEM format)
-   bridge.mqtt.emqx2.cacertfile = etc/certs/cacert.pem
+   bridge.mqtt.emqx2.cacertfile = "etc/certs/cacert.pem"
 
    ## SSL certificate of Client SSL connection 
-   bridge.mqtt.emqx2.certfile = etc/certs/client-cert.pem
+   bridge.mqtt.emqx2.certfile = "etc/certs/client-cert.pem"
 
    ## Key file of Client SSL connection 
-   bridge.mqtt.emqx2.keyfile = etc/certs/client-key.pem
+   bridge.mqtt.emqx2.keyfile = "etc/certs/client-key.pem"
 
    ## TTLS PSK password
    ## Note 'listener.ssl.external.ciphers' and 'listener.ssl.external.psk_ciphers' cannot be configured at the same time
    ##
    ## See 'https://tools.ietf.org/html/rfc4279#section-2'.
-   ## bridge.mqtt.emqx2.psk_ciphers = PSK-AES128-CBC-SHA,PSK-AES256-CBC-SHA,PSK-3DES-EDE-CBC-SHA,PSK-RC4-SHA
+   ## bridge.mqtt.emqx2.psk_ciphers = "PSK-AES128-CBC-SHA,PSK-AES256-CBC-SHA,PSK-3DES-EDE-CBC-SHA,PSK-RC4-SHA"
 
    ## Client's heartbeat interval
    bridge.mqtt.emqx2.keepalive = 60s
 
    ## Supported TLS version
-   bridge.mqtt.emqx2.tls_versions = tlsv1.2
+   bridge.mqtt.emqx2.tls_versions = "tlsv1.2"
 
    ## SSL encryption
-   bridge.mqtt.emqx2.ciphers = ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384
+   bridge.mqtt.emqx2.ciphers = "ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384"
 
    ## Forwarding topics of the message
-   bridge.mqtt.emqx2.forwards = sensor1/#,sensor2/#
+   bridge.mqtt.emqx2.forwards = "sensor1/#,sensor2/#"
 
    ## Bridged mountpoint
-   bridge.mqtt.emqx2.mountpoint = bridge/emqx2/${node}/
+   bridge.mqtt.emqx2.mountpoint = "bridge/emqx2/${node}/"
 
    ## Subscription topic for bridging
-   bridge.mqtt.emqx2.subscription.1.topic = cmd/topic1
+   bridge.mqtt.emqx2.subscription.1.topic = "cmd/topic1"
 
    ## Subscription qos for bridging
    bridge.mqtt.emqx2.subscription.1.qos = 1
 
    ## Subscription topic for bridging
-   bridge.mqtt.emqx2.subscription.2.topic = cmd/topic2
+   bridge.mqtt.emqx2.subscription.2.topic = "cmd/topic2"
 
    ## Subscription qos for bridging
    bridge.mqtt.emqx2.subscription.2.qos = 1
@@ -190,7 +190,7 @@ The bridge of EMQ X has a message caching mechanism. The caching mechanism is ap
    bridge.mqtt.emqx2.queue.batch_bytes_limit = 1000MB
 
    ## The path for placing replayq queue. If it is not specified, then replayq will run in `mem-only` mode and messages will not be cached on disk.
-   bridge.mqtt.emqx2.queue.replayq_dir = data/emqx_emqx2_bridge/
+   bridge.mqtt.emqx2.queue.replayq_dir = "data/emqx_emqx2_bridge/"
 
    ## Replayq data segment size
    bridge.mqtt.emqx2.queue.replayq_seg_bytes = 10MB

apps/emqx_exhook/etc/emqx_exhook.conf

@@ -8,8 +8,8 @@
 ## The gRPC server url
 ##
 ## exhook.server.$name.url = url()
-exhook.server.default.url = http://127.0.0.1:9000
+exhook.server.default.url = "http://127.0.0.1:9000"
 
-#exhook.server.default.ssl.cacertfile = {{ platform_etc_dir }}/certs/cacert.pem
-#exhook.server.default.ssl.certfile = {{ platform_etc_dir }}/certs/cert.pem
-#exhook.server.default.ssl.keyfile = {{ platform_etc_dir }}/certs/key.pem
+#exhook.server.default.ssl.cacertfile = "{{ platform_etc_dir }}/certs/cacert.pem"
+#exhook.server.default.ssl.certfile = "{{ platform_etc_dir }}/certs/cert.pem"
+#exhook.server.default.ssl.keyfile = "{{ platform_etc_dir }}/certs/key.pem"

apps/emqx_exhook/rebar.config

@@ -43,7 +43,6 @@
 {profiles,
  [{test,
    [{deps,
-      [{emqx_ct_helper, {git, "https://github.com/emqx/emqx-ct-helpers", {tag, "v1.3.1"}}}
-      ]}
+      []}
     ]}
 ]}.

apps/emqx_exproto/priv/emqx_exproto.schema

@@ -44,7 +44,7 @@ end}.
 %%--------------------------------------------------------------------
 %% Listeners
 
-{mapping, "exproto.listener.$proto", "emqx_exproto.listeners", [
+{mapping, "exproto.listener.$proto.endpoint", "emqx_exproto.listeners", [
   {datatype, string}
 ]}.
 
@@ -340,7 +340,7 @@ end}.
     Listeners = fun(Proto) ->
                     Prefix = string:join(["exproto","listener", Proto], "."),
                     Opts = HandlerOpts(Prefix) ++ ConnOpts(Prefix) ++ LisOpts(Prefix),
-                    case cuttlefish:conf_get(Prefix, Conf, undefined) of
+                    case cuttlefish:conf_get(Prefix ++ ".endpoint", Conf, undefined) of
                         undefined -> [];
                         ListenOn0 ->
                             case ParseListenOn(ListenOn0) of
@@ -359,6 +359,6 @@ end}.
                             end
                     end
                 end,
-    lists:flatten([Listeners(Proto) || {[_, "listener", Proto], ListenOn}
+    lists:flatten([Listeners(Proto) || {[_, "listener", Proto, "endpoint"], ListenOn}
                                    <- cuttlefish_variable:filter_by_prefix("exproto.listener", Conf)])
 end}.

apps/emqx_exproto/rebar.config

@@ -46,7 +46,6 @@
 {profiles,
  [{test,
    [{deps,
-     [{emqx_ct_helper, {git, "https://github.com/emqx/emqx-ct-helpers", {tag, "v1.3.0"}}}
-     ]}
+     []}
    ]}
 ]}.

apps/emqx_lwm2m/rebar.config

@@ -5,7 +5,6 @@
 {profiles,
  [{test,
    [{deps, [{er_coap_client, {git, "https://github.com/emqx/er_coap_client", {tag, "v1.0"}}},
-            {emqx_ct_helpers, {git, "https://github.com/emqx/emqx-ct-helpers", {tag, "1.2.2"}}},
             {emqtt, {git, "https://github.com/emqx/emqtt", {tag, "1.2.0"}}}
            ]}
    ]}

apps/emqx_management/priv/emqx_management.schema

@@ -21,7 +21,7 @@
   {datatype, string}
 ]}.
 
-{mapping, "management.listener.http", "emqx_management.listeners", [
+{mapping, "management.listener.http.port", "emqx_management.listeners", [
   {datatype, [integer, ip]}
 ]}.
 
@@ -85,7 +85,7 @@
   {datatype, {enum, [true, false]}}
 ]}.
 
-{mapping, "management.listener.https", "emqx_management.listeners", [
+{mapping, "management.listener.https.port", "emqx_management.listeners", [
   {datatype, [integer, ip]}
 ]}.
 
@@ -225,7 +225,7 @@ end}.
     lists:foldl(
       fun(Proto, Acc) ->
           Prefix = "management.listener." ++ atom_to_list(Proto),
-          case cuttlefish:conf_get(Prefix, Conf, undefined) of
+          case cuttlefish:conf_get(Prefix ++ ".port", Conf, undefined) of
               undefined -> Acc;
               Port ->
                   [{Proto, Port, TcpOpts(Prefix) ++ Opts(Prefix)

apps/emqx_management/test/emqx_bridge_mqtt_data_export_import_SUITE.erl

@@ -181,4 +181,4 @@ remove_resources() ->
     lists:foreach(fun(#resource{id = Id}) ->
         emqx_rule_engine:delete_resource(Id)
     end, emqx_rule_registry:get_resources()),
-    timer:sleep(500).
+    timer:sleep(500).

apps/emqx_prometheus/etc/emqx_prometheus.conf

@@ -5,7 +5,7 @@
 ## The Prometheus Push Gateway URL address
 ##
 ## Note: You can comment out this line to disable it
-prometheus.push.gateway.server = http://127.0.0.1:9091
+prometheus.push.gateway.server = "http://127.0.0.1:9091"
 
 ## The metrics data push interval (millisecond)
 ##

apps/emqx_psk_file/etc/emqx_psk_file.conf

@@ -1,2 +1,2 @@
-psk.file.path = {{ platform_etc_dir }}/psk.txt
-psk.file.delimiter = :
+psk.file.path = "{{ platform_etc_dir }}/psk.txt"
+psk.file.delimiter = ":"

apps/emqx_resource/Makefile

@@ -0,0 +1,43 @@
+REBAR := rebar3
+
+.PHONY: all
+all: es
+
+.PHONY: compile
+compile:
+	$(REBAR) compile
+
+.PHONY: clean
+clean: distclean
+
+.PHONY: distclean
+distclean:
+	@rm -rf _build erl_crash.dump rebar3.crashdump
+
+.PHONY: xref
+xref:
+	$(REBAR) xref
+
+.PHONY: eunit
+eunit: compile
+	$(REBAR) eunit -v -c
+	$(REBAR) cover
+
+.PHONY: ct
+ct: compile
+	$(REBAR) as test ct -v
+
+cover:
+	$(REBAR) cover
+
+.PHONY: dialyzer
+dialyzer:
+	$(REBAR) dialyzer
+
+.PHONY: es
+es: compile
+	$(REBAR) escriptize
+
+.PHONY: elvis
+elvis:
+	./scripts/elvis-check.sh

apps/emqx_resource/README.md

@@ -0,0 +1,53 @@
+# emqx_resource
+
+The `emqx_resource` is an application that manages configuration specs and runtime states
+for components that need to be configured and manipulated from the emqx-dashboard.
+
+It is intended to be used by resources, actions, acl, auth, backend_logics and more.
+
+It reads the configuration spec from *.spec (in HOCON format) and provide APIs for
+creating, updating and destroying resource instances among all nodes in the cluster.
+
+It handles the problem like storing the configs and runtime states for both resource
+and resource instances, and how porting them between different emqx_resource versions.
+
+It may maintain the config and data in JSON or HOCON files in data/ dir.
+
+After restarting the emqx_resource, it re-creates all the resource instances.
+
+There can be foreign references between resource instances via resource-id.
+So they may find each other via this Id.
+
+## Try it out
+
+    $ ./demo.sh
+    Eshell V11.1.8  (abort with ^G)
+    1> == the demo log tracer <<"log_tracer_clientid_shawn">> started.
+    config: #{<<"config">> =>
+                #{<<"bulk">> => <<"10KB">>,<<"cache_log_dir">> => <<"/tmp">>,
+                    <<"condition">> => #{<<"clientid">> => <<"abc">>},
+                    <<"level">> => <<"debug">>},
+            <<"id">> => <<"log_tracer_clientid_shawn">>,
+            <<"resource_type">> => <<"log_tracer">>}
+    1> emqx_resource_instance:health_check(<<"log_tracer_clientid_shawn">>).
+    == the demo log tracer <<"log_tracer_clientid_shawn">> is working well
+    state: #{health_checked => 1,logger_handler_id => abc}
+    ok
+
+    2> emqx_resource_instance:health_check(<<"log_tracer_clientid_shawn">>).
+    == the demo log tracer <<"log_tracer_clientid_shawn">> is working well
+    state: #{health_checked => 2,logger_handler_id => abc}
+    ok
+
+    3> emqx_resource_instance:query(<<"log_tracer_clientid_shawn">>, get_log).
+    == the demo log tracer <<"log_tracer_clientid_shawn">> received request: get_log
+    state: #{health_checked => 2,logger_handler_id => abc}
+    "this is a demo log messages..."
+
+    4> emqx_resource_instance:remove(<<"log_tracer_clientid_shawn">>).
+    == the demo log tracer <<"log_tracer_clientid_shawn">> stopped.
+    state: #{health_checked => 0,logger_handler_id => abc}
+    ok
+
+    5> emqx_resource_instance:query(<<"log_tracer_clientid_shawn">>, get_log).
+    ** exception error: {get_instance,{<<"log_tracer_clientid_shawn">>,not_found}}

apps/emqx_resource/demo.sh

@@ -0,0 +1,6 @@
+#!/bin/sh
+set -e
+
+rebar3 compile
+
+erl -sname abc -pa _build/default/lib/*/ebin _build/default/lib/emqx_resource/examples -s demo

apps/emqx_resource/elvis.config

@@ -0,0 +1,14 @@
+[{elvis, [{config, [
+
+#{dirs => ["src"],
+  filter => "*.erl",
+  %ignore => [],
+  ruleset => erl_files,
+  rules => [{elvis_style, operator_spaces, #{
+              rules => [{right, ","},
+                        {right, "|"},
+                        {left, "|"},
+                        {right, "||"},
+                        {left, "||"}]}},
+            {elvis_style, god_modules, #{limit => 100}}]}
+]}]}].

apps/emqx_resource/etc/emqx_resource.conf

@@ -0,0 +1,3 @@
+##--------------------------------------------------------------------
+## EMQ X Resource Plugin
+##--------------------------------------------------------------------

apps/emqx_resource/examples/demo.erl

@@ -0,0 +1,13 @@
+-module(demo).
+
+-export([start/0]).
+
+start() ->
+    code:load_file(log_tracer),
+    code:load_file(log_tracer_schema),
+    {ok, _} = application:ensure_all_started(minirest),
+    {ok, _} = application:ensure_all_started(emqx_resource),
+    emqx_resource:load_instances("./_build/default/lib/emqx_resource/examples"),
+    Handlers = [{"/", minirest:handler(#{modules => [log_tracer]})}],
+    Dispatch = [{"/[...]", minirest, Handlers}],
+    minirest:start_http(?MODULE, #{socket_opts => [inet, {port, 9900}]}, Dispatch).

apps/emqx_resource/examples/demo.md

@@ -0,0 +1,147 @@
+---
+theme: gaia
+color: #000
+colorSecondary: #333
+backgroundColor: #fff
+backgroundImage: url('https://marp.app/assets/hero-background.jpg')
+paginate: true
+marp: true
+---
+
+<!-- _class: lead -->
+
+# EMQX Resource
+
+---
+
+## What is it for
+
+The [emqx_resource](https://github.com/terry-xiaoyu/emqx_resource) for managing configurations and runtime states for dashboard components .
+
+![bg right](https://docs.emqx.cn/assets/img/rule_action_1@2x.73766093.png)
+
+---
+
+<!-- _class: lead -->
+
+# The Demo
+
+The little log tracer
+
+---
+
+- The hocon schema file (log_tracer_schema.erl):
+
+https://github.com/terry-xiaoyu/emqx_resource/blob/main/examples/log_tracer_schema.erl
+
+- The callback file (log_tracer.erl):
+
+https://github.com/terry-xiaoyu/emqx_resource/blob/main/examples/log_tracer.erl
+
+---
+
+Start the demo log tracer
+
+```
+./demo.sh
+```
+
+Load instance from config files (auto loaded)
+
+```
+## This will load all of the "*.conf" file under that directory:
+
+emqx_resource:load_instances("./_build/default/lib/emqx_resource/examples").
+```
+
+The config file is validated against the schema (`*_schema.erl`) before loaded.
+
+---
+
+# List Types and Instances
+
+- To list all the available resource types:
+
+```
+emqx_resource:list_types().
+emqx_resource:list_instances().
+```
+
+- And there's `*_verbose` versions for these `list_*` APIs:
+
+```
+emqx_resource:list_types_verbose().
+emqx_resource:list_instances_verbose().
+```
+
+---
+# Instance management
+
+- To get a resource types and instances:
+
+```
+emqx_resource:get_type(log_tracer).
+emqx_resource:get_instance("log_tracer_clientid_shawn").
+```
+
+- To create a resource instances:
+
+```
+emqx_resource:create("log_tracer2", log_tracer,
+#{bulk => <<"1KB">>,cache_log_dir => <<"/tmp">>,
+  cache_logs_in => <<"memory">>,chars_limit => 1024,
+  condition => #{<<"app">> => <<"emqx">>},
+  enable_cache => true,level => debug}).
+```
+
+---
+
+- To update a resource:
+
+```
+emqx_resource:update("log_tracer2", log_tracer, #{bulk => <<"100KB">>}, []).
+```
+
+- To delete a resource:
+
+```
+emqx_resource:remove("log_tracer2").
+```
+
+---
+
+<!-- _class: lead -->
+
+# HTTP APIs Demo
+
+---
+
+# Get a log tracer
+
+To list current log tracers:
+
+```
+curl -s -XGET 'http://localhost:9900/log_tracer' | jq .
+```
+
+---
+
+## Update or Create
+
+To update an existing log tracer or create a new one:
+
+```
+INST='{
+  "resource_type": "log_tracer",
+  "config": {
+    "condition": {
+      "app": "emqx"
+    },
+    "level": "debug",
+    "cache_log_dir": "/tmp",
+    "bulk": "10KB",
+    "chars_limit": 1024
+  }
+}'
+curl -sv -XPUT 'http://localhost:9900/log_tracer/log_tracer2' -d $INST | jq .
+```

apps/emqx_resource/examples/log_tracer.conf

@@ -0,0 +1,11 @@
+{
+    "id": "log_tracer_clientid_shawn"
+    "resource_type": "log_tracer"
+    "config": {
+        "condition": {"app": "emqx"}
+        "level": "debug"
+        "cache_log_dir": "/tmp"
+        "bulk": "10KB"
+        "chars_limit": 1024
+    }
+}

apps/emqx_resource/examples/log_tracer.erl

@@ -0,0 +1,45 @@
+-module(log_tracer).
+
+-include_lib("emqx_resource/include/emqx_resource_behaviour.hrl").
+
+-emqx_resource_api_path("/log_tracer").
+
+%% callbacks of behaviour emqx_resource
+-export([ on_start/2
+        , on_stop/2
+        , on_query/4
+        , on_health_check/2
+        , on_api_reply_format/1
+        , on_config_merge/3
+        ]).
+
+%% callbacks for emqx_resource config schema
+-export([fields/1]).
+
+fields(ConfPath) ->
+    log_tracer_schema:fields(ConfPath).
+
+on_start(InstId, Config) ->
+    io:format("== the demo log tracer ~p started.~nconfig: ~p~n", [InstId, Config]),
+    {ok, #{logger_handler_id => abc, health_checked => 0}}.
+
+on_stop(InstId, State) ->
+    io:format("== the demo log tracer ~p stopped.~nstate: ~p~n", [InstId, State]),
+    ok.
+
+on_query(InstId, Request, AfterQuery, State) ->
+    io:format("== the demo log tracer ~p received request: ~p~nstate: ~p~n",
+        [InstId, Request, State]),
+    emqx_resource:query_success(AfterQuery),
+    "this is a demo log messages...".
+
+on_health_check(InstId, State = #{health_checked := Checked}) ->
+    NState = State#{health_checked => Checked + 1},
+    io:format("== the demo log tracer ~p is working well~nstate: ~p~n", [InstId, NState]),
+    {ok, NState}.
+
+on_api_reply_format(#{id := Id, status := Status, state := #{health_checked := NChecked}}) ->
+    #{id => Id, status => Status, checked_count => NChecked}.
+
+on_config_merge(OldConfig, NewConfig, _Params) ->
+    maps:merge(OldConfig, NewConfig).

apps/emqx_resource/examples/log_tracer_schema.erl

@@ -0,0 +1,45 @@
+-module(log_tracer_schema).
+
+-include_lib("typerefl/include/types.hrl").
+
+-export([fields/1]).
+
+-reflect_type([t_level/0, t_cache_logs_in/0]).
+
+-type t_level() :: debug | info | notice | warning | error | critical | alert | emergency.
+
+-type t_cache_logs_in() :: memory | file.
+
+fields("config") ->
+    [ {condition, fun condition/1}
+    , {level, fun level/1}
+    , {enable_cache, fun enable_cache/1}
+    , {cache_logs_in, fun cache_logs_in/1}
+    , {cache_log_dir, fun cache_log_dir/1}
+    , {bulk, fun bulk/1}
+    ];
+fields(_) -> [].
+
+condition(mapping) -> "config.condition";
+condition(type) -> map();
+condition(_) -> undefined.
+
+level(mapping) -> "config.level";
+level(type) -> t_level();
+level(_) -> undefined.
+
+enable_cache(mapping) -> "config.enable_cache";
+enable_cache(type) -> boolean();
+enable_cache(_) -> undefined.
+
+cache_logs_in(mapping) -> "config.cache_logs_in";
+cache_logs_in(type) -> t_cache_logs_in();
+cache_logs_in(_) -> undefined.
+
+cache_log_dir(mapping) -> "config.cache_log_dir";
+cache_log_dir(type) -> typerefl:regexp_string("^(.*)$");
+cache_log_dir(_) -> undefined.
+
+bulk(mapping) -> "config.bulk";
+bulk(type) -> typerefl:regexp_string("^[. 0-9]+(B|KB|MB|GB)$");
+bulk(_) -> undefined.

apps/emqx_resource/include/emqx_resource.hrl

@@ -0,0 +1,34 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+-type resource_type() :: module().
+-type instance_id() :: binary().
+-type resource_config() :: jsx:json_term().
+-type resource_spec() :: map().
+-type resource_state() :: term().
+-type resource_data() :: #{
+    id => instance_id(),
+    mod => module(),
+    config => resource_config(),
+    state => resource_state(),
+    status => started | stopped
+}.
+
+-type after_query() :: {OnSuccess :: after_query_fun(), OnFailed :: after_query_fun()} |
+    undefined.
+
+%% the `after_query_fun()` is mainly for callbacks that increment counters or do some fallback
+%% actions upon query failure
+-type after_query_fun() :: {fun((...) -> ok), Args :: [term()]}.

apps/emqx_resource/include/emqx_resource_behaviour.hrl

@@ -0,0 +1,18 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+-include_lib("emqx_resource/include/emqx_resource.hrl").
+-behaviour(emqx_resource).
+-compile({parse_transform, emqx_resource_transform}).

apps/emqx_resource/include/emqx_resource_utils.hrl

@@ -0,0 +1,53 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+-define(CLUSTER_CALL(Func, Args), ?CLUSTER_CALL(Func, Args, ok)).
+
+-define(CLUSTER_CALL(Func, Args, ResParttern),
+%% ekka_mnesia:running_nodes()
+    fun() ->
+        case LocalResult = erlang:apply(?MODULE, Func, Args) of
+            ResParttern ->
+                case rpc:multicall(nodes(), ?MODULE, Func, Args, 5000) of
+                {ResL, []} ->
+                    Filter = fun
+                        (ResParttern) -> false;
+                        ({badrpc, {'EXIT', {undef, [{?MODULE, Func0, _, []}]}}})
+                            when Func0 =:= Func -> false;
+                        (_) -> true
+                    end,
+                    case lists:filter(Filter, ResL) of
+                        [] -> LocalResult;
+                        ErrL -> {error, ErrL}
+                    end;
+                {ResL, BadNodes} ->
+                    {error, {failed_on_nodes, BadNodes, ResL}}
+                end;
+            ErrorResult ->
+                {error, ErrorResult}
+        end
+    end()).
+
+-define(SAFE_CALL(_EXP_),
+        ?SAFE_CALL(_EXP_, _ = do_nothing)).
+
+-define(SAFE_CALL(_EXP_, _EXP_ON_FAIL_),
+        fun() ->
+            try (_EXP_)
+            catch _EXCLASS_:_EXCPTION_:_ST_ ->
+                _EXP_ON_FAIL_,
+                {error, {_EXCLASS_, _EXCPTION_, _ST_}}
+            end
+        end()).

apps/emqx_resource/priv/emqx_resource.schema

@@ -0,0 +1,2 @@
+%%-*- mode: erlang -*-
+%% emqx-resource config mapping

apps/emqx_resource/rebar.config

@@ -0,0 +1,18 @@
+{erl_opts, [ debug_info
+           , nowarn_unused_import
+           %, {d, 'RESOURCE_DEBUG'}
+           ]}.
+
+{erl_first_files, ["src/emqx_resource_transform.erl"]}.
+
+{extra_src_dirs, ["examples"]}.
+
+%% try to override the dialyzer 'race_conditions' defined in the top-level dir,
+%% but it doesn't work
+{dialyzer, [{warnings, [unmatched_returns, error_handling]}
+           ]}.
+
+{deps, [ {hocon, {git, "https://github.com/emqx/hocon", {branch, "master"}}}
+       , {jsx, {git, "https://github.com/talentdeficit/jsx", {tag, "v3.1.0"}}}
+       ]}.
+

apps/emqx_resource/scripts/elvis-check.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+
+set -euo pipefail
+
+ELVIS_VERSION='1.0.0-emqx-2'
+
+elvis_version="${2:-$ELVIS_VERSION}"
+
+echo "elvis -v: $elvis_version"
+
+if [ ! -f ./elvis ] || [ "$(./elvis -v | grep -oE '[1-9]+\.[0-9]+\.[0-9]+\-emqx-[0-9]+')" != "$elvis_version" ]; then
+    curl  -fLO "https://github.com/emqx/elvis/releases/download/$elvis_version/elvis"
+    chmod +x ./elvis
+fi
+
+./elvis rock --config elvis.config
+

apps/emqx_resource/src/emqx_resource.app.src

@@ -0,0 +1,17 @@
+{application, emqx_resource,
+ [{description, "An OTP application"},
+  {vsn, "0.1.0"},
+  {registered, []},
+  {mod, {emqx_resource_app, []}},
+  {applications,
+   [kernel,
+    stdlib,
+    gproc,
+    hocon
+   ]},
+  {env,[]},
+  {modules, []},
+
+  {licenses, ["Apache 2.0"]},
+  {links, []}
+ ]}.

apps/emqx_resource/src/emqx_resource.erl

@@ -0,0 +1,274 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(emqx_resource).
+
+-include("emqx_resource.hrl").
+-include("emqx_resource_utils.hrl").
+
+%% APIs for resource types
+
+-export([ get_type/1
+        , list_types/0
+        , list_types_verbose/0
+        ]).
+
+-export([ discover_resource_mods/0
+        , is_resource_mod/1
+        , call_instance/2
+        ]).
+
+-export([ query_success/1
+        , query_failed/1
+        ]).
+
+%% APIs for instances
+
+-export([ parse_config/2
+        , resource_type_from_str/1
+        ]).
+
+%% Sync resource instances and files
+%% provisional solution: rpc:multical to all the nodes for creating/updating/removing
+%% todo: replicate operations
+-export([ create/3 %% store the config and start the instance
+        , create_dry_run/3 %% run start/2, health_check/2 and stop/1 sequentially
+        , update/4 %% update the config, stop the old instance and start the new one
+                   %% it will create a new resource when the id does not exist
+        , remove/1 %% remove the config and stop the instance
+        ]).
+
+%% Calls to the callback module with current resource state
+%% They also save the state after the call finished (except query/2,3).
+-export([ restart/1  %% restart the instance.
+        , health_check/1 %% verify if the resource is working normally
+        , stop/1   %% stop the instance
+        , query/2  %% query the instance
+        , query/3  %% query the instance with after_query()
+        ]).
+
+%% Direct calls to the callback module
+-export([ call_start/3  %% start the instance
+        , call_health_check/3 %% verify if the resource is working normally
+        , call_stop/3   %% stop the instance
+        , call_config_merge/4 %% merge the config when updating
+        ]).
+
+-export([ list_instances/0 %% list all the instances, id only.
+        , list_instances_verbose/0 %% list all the instances
+        , get_instance/1 %% return the data of the instance
+        , get_instance_by_type/1 %% return all the instances of the same resource type
+        , load_instances/1 %% load instances from config files
+        % , dependents/1
+        % , inc_counter/2 %% increment the counter of the instance
+        % , inc_counter/3 %% increment the counter by a given integer
+        ]).
+
+-define(EXT, "*.spec").
+
+-optional_callbacks([ on_query/4
+                    , on_health_check/2
+                    , on_api_reply_format/1
+                    , on_config_merge/3
+                    ]).
+
+-callback on_api_reply_format(resource_data()) -> map().
+
+-callback on_config_merge(resource_config(), resource_config(), term()) -> resource_config().
+
+%% when calling emqx_resource:start/1
+-callback on_start(instance_id(), resource_config()) ->
+    {ok, resource_state()} | {error, Reason :: term()}.
+
+%% when calling emqx_resource:stop/1
+-callback on_stop(instance_id(), resource_state()) -> term().
+
+%% when calling emqx_resource:query/3
+-callback on_query(instance_id(), Request :: term(), after_query(), resource_state()) -> term().
+
+%% when calling emqx_resource:health_check/2
+-callback on_health_check(instance_id(), resource_state()) ->
+    {ok, resource_state()} | {error, Reason:: term(), resource_state()}.
+
+%% load specs and return the loaded resources this time.
+-spec list_types_verbose() -> [resource_spec()].
+list_types_verbose() ->
+    [get_spec(Mod) || Mod <- list_types()].
+
+-spec list_types() -> [module()].
+list_types() ->
+    discover_resource_mods().
+
+-spec get_type(module()) -> {ok, resource_spec()} | {error, not_found}.
+get_type(Mod) ->
+    case is_resource_mod(Mod) of
+        true -> {ok, get_spec(Mod)};
+        false -> {error, not_found}
+    end.
+
+-spec get_spec(module()) -> resource_spec().
+get_spec(Mod) ->
+    maps:put(<<"resource_type">>, Mod, Mod:emqx_resource_schema()).
+
+-spec discover_resource_mods() -> [module()].
+discover_resource_mods() ->
+    [Mod || {Mod, _} <- code:all_loaded(), is_resource_mod(Mod)].
+
+-spec is_resource_mod(module()) -> boolean().
+is_resource_mod(Mod) ->
+    erlang:function_exported(Mod, emqx_resource_schema, 0).
+
+-spec query_success(after_query()) -> ok.
+query_success(undefined) -> ok;
+query_success({{OnSucc, Args}, _}) ->
+    safe_apply(OnSucc, Args).
+
+-spec query_failed(after_query()) -> ok.
+query_failed(undefined) -> ok;
+query_failed({_, {OnFailed, Args}}) ->
+    safe_apply(OnFailed, Args).
+
+%% =================================================================================
+%% APIs for resource instances
+%% =================================================================================
+-spec create(instance_id(), resource_type(), resource_config()) ->
+    {ok, resource_data()} | {error, Reason :: term()}.
+create(InstId, ResourceType, Config) ->
+    ?CLUSTER_CALL(call_instance, [InstId, {create, InstId, ResourceType, Config}], {ok, _}).
+
+-spec create_dry_run(instance_id(), resource_type(), resource_config()) ->
+    ok | {error, Reason :: term()}.
+create_dry_run(InstId, ResourceType, Config) ->
+    ?CLUSTER_CALL(call_instance, [InstId, {create_dry_run, InstId, ResourceType, Config}]).
+
+-spec update(instance_id(), resource_type(), resource_config(), term()) ->
+    {ok, resource_data()} | {error, Reason :: term()}.
+update(InstId, ResourceType, Config, Params) ->
+    ?CLUSTER_CALL(call_instance, [InstId, {update, InstId, ResourceType, Config, Params}], {ok, _}).
+
+-spec remove(instance_id()) -> ok | {error, Reason :: term()}.
+remove(InstId) ->
+    ?CLUSTER_CALL(call_instance, [InstId, {remove, InstId}]).
+
+-spec query(instance_id(), Request :: term()) -> Result :: term().
+query(InstId, Request) ->
+    query(InstId, Request, undefined).
+
+%% same to above, also defines what to do when the Module:on_query success or failed
+%% it is the duty of the Moudle to apply the `after_query()` functions.
+-spec query(instance_id(), Request :: term(), after_query()) -> Result :: term().
+query(InstId, Request, AfterQuery) ->
+    case get_instance(InstId) of
+        {ok, #{mod := Mod, state := ResourceState}} ->
+            %% the resource state is readonly to Moudle:on_query/4
+            %% and the `after_query()` functions should be thread safe
+            Mod:on_query(InstId, Request, AfterQuery, ResourceState);
+        {error, Reason} ->
+            error({get_instance, {InstId, Reason}})
+    end.
+
+-spec restart(instance_id()) -> ok | {error, Reason :: term()}.
+restart(InstId) ->
+    call_instance(InstId, {restart, InstId}).
+
+-spec stop(instance_id()) -> ok | {error, Reason :: term()}.
+stop(InstId) ->
+    call_instance(InstId, {stop, InstId}).
+
+-spec health_check(instance_id()) -> ok | {error, Reason :: term()}.
+health_check(InstId) ->
+    call_instance(InstId, {health_check, InstId}).
+
+-spec get_instance(instance_id()) -> {ok, resource_data()} | {error, Reason :: term()}.
+get_instance(InstId) ->
+    emqx_resource_instance:lookup(InstId).
+
+-spec list_instances() -> [instance_id()].
+list_instances() ->
+    [Id || #{id := Id} <- list_instances_verbose()].
+
+-spec list_instances_verbose() -> [resource_data()].
+list_instances_verbose() ->
+    emqx_resource_instance:list_all().
+
+-spec get_instance_by_type(module()) -> [resource_data()].
+get_instance_by_type(ResourceType) ->
+    emqx_resource_instance:lookup_by_type(ResourceType).
+
+-spec load_instances(Dir :: string()) -> ok.
+load_instances(Dir) ->
+    emqx_resource_instance:load(Dir).
+
+-spec call_start(instance_id(), module(), resource_config()) ->
+    {ok, resource_state()} | {error, Reason :: term()}.
+call_start(InstId, Mod, Config) ->
+    ?SAFE_CALL(Mod:on_start(InstId, Config)).
+
+-spec call_health_check(instance_id(), module(), resource_state()) ->
+    {ok, resource_state()} | {error, Reason:: term(), resource_state()}.
+call_health_check(InstId, Mod, ResourceState) ->
+    ?SAFE_CALL(Mod:on_health_check(InstId, ResourceState)).
+
+-spec call_stop(instance_id(), module(), resource_state()) -> term().
+call_stop(InstId, Mod, ResourceState) ->
+    ?SAFE_CALL(Mod:on_stop(InstId, ResourceState)).
+
+-spec call_config_merge(module(), resource_config(), resource_config(), term()) ->
+    resource_config().
+call_config_merge(Mod, OldConfig, NewConfig, Params) ->
+    ?SAFE_CALL(Mod:on_config_merge(OldConfig, NewConfig, Params)).
+
+-spec parse_config(resource_type(), binary() | term()) ->
+    {ok, resource_config()} | {error, term()}.
+parse_config(ResourceType, RawConfig) when is_binary(RawConfig) ->
+    case hocon:binary(RawConfig, #{format => richmap}) of
+        {ok, MapConfig} ->
+            do_parse_config(ResourceType, MapConfig);
+        Error -> Error
+    end;
+parse_config(ResourceType, RawConfigTerm) ->
+    parse_config(ResourceType, jsx:encode(#{<<"config">> => RawConfigTerm})).
+
+-spec do_parse_config(resource_type(), map()) -> {ok, resource_config()} | {error, term()}.
+do_parse_config(ResourceType, MapConfig) ->
+    case ?SAFE_CALL(hocon_schema:generate(ResourceType, MapConfig)) of
+        {error, Reason} -> {error, Reason};
+        Config ->
+            InstConf = maps:from_list(proplists:get_value(config, Config)),
+            {ok, InstConf}
+    end.
+
+%% =================================================================================
+
+-spec resource_type_from_str(string()) -> {ok, resource_type()} | {error, term()}.
+resource_type_from_str(ResourceType) ->
+    try Mod = list_to_existing_atom(str(ResourceType)),
+        case emqx_resource:is_resource_mod(Mod) of
+            true -> {ok, Mod};
+            false -> {error, {invalid_resource, Mod}}
+        end
+    catch error:badarg ->
+        {error, {not_found, ResourceType}}
+    end.
+
+call_instance(InstId, Query) ->
+    emqx_resource_instance:hash_call(InstId, Query).
+
+safe_apply(Func, Args) ->
+    ?SAFE_CALL(erlang:apply(Func, Args)).
+
+str(S) when is_binary(S) -> binary_to_list(S);
+str(S) when is_list(S) -> S.

apps/emqx_resource/src/emqx_resource_api.erl

@@ -0,0 +1,79 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+-module(emqx_resource_api).
+
+-export([ get_all/3
+        , get/3
+        , put/3
+        , delete/3
+        ]).
+get_all(Mod, _Binding, _Params) ->
+    {200, #{code => 0, data =>
+        [format_data(Mod, Data) || Data <- emqx_resource:list_instances_verbose()]}}.
+
+get(Mod, #{id := Id}, _Params) ->
+    case emqx_resource:get_instance(stringnify(Id)) of
+        {ok, Data} ->
+            {200, #{code => 0, data => format_data(Mod, Data)}};
+        {error, not_found} ->
+            {404, #{code => 102, message => {resource_instance_not_found, stringnify(Id)}}}
+    end.
+
+put(Mod, #{id := Id}, Params) ->
+    ConfigParams = proplists:get_value(<<"config">>, Params),
+    ResourceTypeStr = proplists:get_value(<<"resource_type">>, Params),
+    case emqx_resource:resource_type_from_str(ResourceTypeStr) of
+        {ok, ResourceType} ->
+            do_put(Mod, stringnify(Id), ConfigParams, ResourceType, Params);
+        {error, Reason} ->
+            {404, #{code => 102, message => stringnify(Reason)}}
+    end.
+
+do_put(Mod, Id, ConfigParams, ResourceType, Params) ->
+    case emqx_resource:parse_config(ResourceType, ConfigParams) of
+        {ok, Config} ->
+            case emqx_resource:update(Id, ResourceType, Config, Params) of
+                {ok, Data} ->
+                    {200, #{code => 0, data => format_data(Mod, Data)}};
+                {error, Reason} ->
+                    {500, #{code => 102, message => stringnify(Reason)}}
+            end;
+        {error, Reason} ->
+            {400, #{code => 108, message => stringnify(Reason)}}
+    end.
+
+delete(_Mod, #{id := Id}, _Params) ->
+    case emqx_resource:remove(stringnify(Id)) of
+        ok -> {200, #{code => 0, data => #{}}};
+        {error, Reason} ->
+            {500, #{code => 102, message => stringnify(Reason)}}
+    end.
+
+format_data(Mod, Data) ->
+    case erlang:function_exported(Mod, on_api_reply_format, 1) of
+        false ->
+            default_api_reply_format(Data);
+        true ->
+            Mod:on_api_reply_format(Data)
+    end.
+
+default_api_reply_format(#{id := Id, status := Status, config := Config}) ->
+    #{node => node(), id => Id, status => Status, config => Config}.
+
+stringnify(Bin) when is_binary(Bin) -> Bin;
+stringnify(Str) when is_list(Str) -> list_to_binary(Str);
+stringnify(Reason) ->
+    iolist_to_binary(io_lib:format("~p", [Reason])).

apps/emqx_resource/src/emqx_resource_app.erl

@@ -0,0 +1,33 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(emqx_resource_app).
+
+-behaviour(application).
+
+-include("emqx_resource.hrl").
+
+-emqx_plugin(?MODULE).
+
+-export([start/2, stop/1]).
+
+start(_StartType, _StartArgs) ->
+    emqx_resource_sup:start_link().
+
+stop(_State) ->
+    ok.
+
+%% internal functions

apps/emqx_resource/src/emqx_resource_instance.erl

@@ -0,0 +1,295 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+-module(emqx_resource_instance).
+
+-behaviour(gen_server).
+
+-include("emqx_resource.hrl").
+-include("emqx_resource_utils.hrl").
+
+-export([start_link/2]).
+
+%% load resource instances from *.conf files
+-export([ load/1
+        , lookup/1
+        , list_all/0
+        , lookup_by_type/1
+        ]).
+
+-export([ hash_call/2
+        , hash_call/3
+        ]).
+
+%% gen_server Callbacks
+-export([ init/1
+        , handle_call/3
+        , handle_cast/2
+        , handle_info/2
+        , terminate/2
+        , code_change/3
+        ]).
+
+-record(state, {worker_pool, worker_id}).
+
+-type state() :: #state{}.
+
+%%------------------------------------------------------------------------------
+%% Start the registry
+%%------------------------------------------------------------------------------
+
+start_link(Pool, Id) ->
+    gen_server:start_link({local, proc_name(?MODULE, Id)},
+                          ?MODULE, {Pool, Id}, []).
+
+%% call the worker by the hash of resource-instance-id, to make sure we always handle
+%% operations on the same instance in the same worker.
+hash_call(InstId, Request) ->
+    hash_call(InstId, Request, infinity).
+
+hash_call(InstId, Request, Timeout) ->
+    gen_server:call(pick(InstId), Request, Timeout).
+
+-spec lookup(instance_id()) -> {ok, resource_data()} | {error, Reason :: term()}.
+lookup(InstId) ->
+    case ets:lookup(emqx_resource_instance, InstId) of
+        [] -> {error, not_found};
+        [{_, Data}] -> {ok, Data#{id => InstId}}
+    end.
+
+force_lookup(InstId) ->
+    {ok, Data} = lookup(InstId),
+    Data.
+
+-spec list_all() -> [resource_data()].
+list_all() ->
+    [Data#{id => Id} || {Id, Data} <- ets:tab2list(emqx_resource_instance)].
+
+-spec lookup_by_type(module()) -> [resource_data()].
+lookup_by_type(ResourceType) ->
+    [Data || #{mod := Mod} = Data <- list_all()
+             , Mod =:= ResourceType].
+
+-spec load(Dir :: string()) -> ok.
+load(Dir) ->
+    lists:foreach(fun load_file/1, filelib:wildcard(filename:join([Dir, "*.conf"]))).
+
+load_file(File) ->
+    case ?SAFE_CALL(hocon_token:read(File)) of
+        {error, Reason} ->
+            logger:error("load resource from ~p failed: ~p", [File, Reason]);
+        RawConfig ->
+            case hocon:binary(RawConfig, #{format => map}) of
+                {ok, #{<<"id">> := Id, <<"resource_type">> := ResourceTypeStr,
+                       <<"config">> := MapConfig}} ->
+                    case emqx_resource:resource_type_from_str(ResourceTypeStr) of
+                        {ok, ResourceType} ->
+                            parse_and_load_config(Id, ResourceType, MapConfig);
+                        {error, Reason} ->
+                            logger:error("no such resource type: ~s, ~p",
+                                [ResourceTypeStr, Reason])
+                    end;
+                {error, Reason} ->
+                    logger:error("load resource from ~p failed: ~p", [File, Reason])
+            end
+    end.
+
+parse_and_load_config(InstId, ResourceType, MapConfig) ->
+    case emqx_resource:parse_config(ResourceType, MapConfig) of
+        {error, Reason} ->
+            logger:error("parse config for resource ~p of type ~p failed: ~p",
+                [InstId, ResourceType, Reason]);
+        {ok, InstConf} ->
+            create_instance_local(InstId, ResourceType, InstConf)
+    end.
+
+create_instance_local(InstId, ResourceType, InstConf) ->
+    case do_create(InstId, ResourceType, InstConf) of
+        {ok, Data} ->
+            logger:debug("created ~p resource instance: ~p from config: ~p, Data: ~p",
+                [ResourceType, InstId, InstConf, Data]);
+        {error, Reason} ->
+            logger:error("create ~p resource instance: ~p failed: ~p, config: ~p",
+                [ResourceType, InstId, Reason, InstConf])
+    end.
+
+%%------------------------------------------------------------------------------
+%% gen_server callbacks
+%%------------------------------------------------------------------------------
+
+-spec init({atom(), integer()}) ->
+    {ok, State :: state()} | {ok, State :: state(), timeout() | hibernate | {continue, term()}} |
+    {stop, Reason :: term()} | ignore.
+init({Pool, Id}) ->
+    true = gproc_pool:connect_worker(Pool, {Pool, Id}),
+    {ok, #state{worker_pool = Pool, worker_id = Id}}.
+
+handle_call({create, InstId, ResourceType, Config}, _From, State) ->
+    {reply, do_create(InstId, ResourceType, Config), State};
+
+handle_call({create_dry_run, InstId, ResourceType, Config}, _From, State) ->
+    {reply, do_create_dry_run(InstId, ResourceType, Config), State};
+
+handle_call({update, InstId, ResourceType, Config, Params}, _From, State) ->
+    {reply, do_update(InstId, ResourceType, Config, Params), State};
+
+handle_call({remove, InstId}, _From, State) ->
+    {reply, do_remove(InstId), State};
+
+handle_call({restart, InstId}, _From, State) ->
+    {reply, do_restart(InstId), State};
+
+handle_call({stop, InstId}, _From, State) ->
+    {reply, do_stop(InstId), State};
+
+handle_call({health_check, InstId}, _From, State) ->
+    {reply, do_health_check(InstId), State};
+
+handle_call(Req, _From, State) ->
+    logger:error("Received unexpected call: ~p", [Req]),
+    {reply, ignored, State}.
+
+handle_cast(_Msg, State) ->
+    {noreply, State}.
+
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+terminate(_Reason, #state{worker_pool = Pool, worker_id = Id}) ->
+    gproc_pool:disconnect_worker(Pool, {Pool, Id}).
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%%------------------------------------------------------------------------------
+
+%% suppress the race condition check, as these functions are protected in gproc workers
+-dialyzer({nowarn_function, [do_update/4, do_create/3, do_restart/1, do_stop/1, do_health_check/1]}).
+do_update(InstId, ResourceType, NewConfig, Params) ->
+    case lookup(InstId) of
+        {ok, #{mod := ResourceType, state := ResourceState, config := OldConfig}} ->
+            Config = emqx_resource:call_config_merge(ResourceType, OldConfig,
+                        NewConfig, Params),
+            case do_create_dry_run(InstId, ResourceType, Config) of
+                ok ->
+                    do_remove(ResourceType, InstId, ResourceState),
+                    do_create(InstId, ResourceType, Config);
+                Error ->
+                    Error
+            end;
+        {ok, #{mod := Mod}} when Mod =/= ResourceType ->
+            {error, updating_to_incorrect_resource_type};
+        {error, not_found} ->
+            do_create(InstId, ResourceType, NewConfig)
+    end.
+
+do_create(InstId, ResourceType, Config) ->
+    case lookup(InstId) of
+        {ok, _} -> {error, already_created};
+        _ ->
+            case emqx_resource:call_start(InstId, ResourceType, Config) of
+                {ok, ResourceState} ->
+                    ets:insert(emqx_resource_instance, {InstId,
+                        #{mod => ResourceType, config => Config,
+                          state => ResourceState, status => stopped}}),
+                    _ = do_health_check(InstId),
+                    {ok, force_lookup(InstId)};
+                {error, Reason} ->
+                    logger:error("start ~s resource ~s failed: ~p", [ResourceType, InstId, Reason]),
+                    {error, Reason}
+            end
+    end.
+
+do_create_dry_run(InstId, ResourceType, Config) ->
+    case emqx_resource:call_start(InstId, ResourceType, Config) of
+        {ok, ResourceState0} ->
+            Return = case emqx_resource:call_health_check(InstId, ResourceType, ResourceState0) of
+                {ok, ResourceState1} -> ok;
+                {error, Reason, ResourceState1} ->
+                    {error, Reason}
+            end,
+            _ = emqx_resource:call_stop(InstId, ResourceType, ResourceState1),
+            Return;
+        {error, Reason} ->
+            {error, Reason}
+    end.
+
+do_remove(InstId) ->
+    case lookup(InstId) of
+        {ok, #{mod := Mod, state := ResourceState}} ->
+            do_remove(Mod, InstId, ResourceState);
+        Error ->
+            Error
+    end.
+
+do_remove(Mod, InstId, ResourceState) ->
+    _ = emqx_resource:call_stop(InstId, Mod, ResourceState),
+    ets:delete(emqx_resource_instance, InstId),
+    ok.
+
+do_restart(InstId) ->
+    case lookup(InstId) of
+        {ok, #{mod := Mod, state := ResourceState, config := Config} = Data} ->
+            _ = emqx_resource:call_stop(InstId, Mod, ResourceState),
+            case emqx_resource:call_start(InstId, Mod, Config) of
+                {ok, ResourceState} ->
+                    ets:insert(emqx_resource_instance,
+                        {InstId, Data#{state => ResourceState, status => started}}),
+                    ok;
+                {error, Reason} ->
+                    ets:insert(emqx_resource_instance, {InstId, Data#{status => stopped}}),
+                    {error, Reason}
+            end;
+        Error ->
+            Error
+    end.
+
+do_stop(InstId) ->
+    case lookup(InstId) of
+        {ok, #{mod := Mod, state := ResourceState} = Data} ->
+            _ = emqx_resource:call_stop(InstId, Mod, ResourceState),
+            ets:insert(emqx_resource_instance, {InstId, Data#{status => stopped}}),
+            ok;
+        Error ->
+            Error
+    end.
+
+do_health_check(InstId) ->
+    case lookup(InstId) of
+        {ok, #{mod := Mod, state := ResourceState0} = Data} ->
+            case emqx_resource:call_health_check(InstId, Mod, ResourceState0) of
+                {ok, ResourceState1} ->
+                    ets:insert(emqx_resource_instance,
+                        {InstId, Data#{status => started, state => ResourceState1}}),
+                    ok;
+                {error, Reason, ResourceState1} ->
+                    logger:error("health check for ~p failed: ~p", [InstId, Reason]),
+                    ets:insert(emqx_resource_instance,
+                        {InstId, Data#{status => stopped, state => ResourceState1}}),
+                    {error, Reason}
+            end;
+        Error ->
+            Error
+    end.
+
+%%------------------------------------------------------------------------------
+%% internal functions
+%%------------------------------------------------------------------------------
+
+proc_name(Mod, Id) ->
+    list_to_atom(lists:concat([Mod, "_", Id])).
+
+pick(InstId) ->
+    gproc_pool:pick_worker(emqx_resource_instance, InstId).

apps/emqx_resource/src/emqx_resource_sup.erl

@@ -0,0 +1,58 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+-module(emqx_resource_sup).
+
+-behaviour(supervisor).
+
+-export([start_link/0]).
+
+-export([init/1]).
+
+-define(RESOURCE_INST_MOD, emqx_resource_instance).
+-define(POOL_SIZE, 64). %% set a very large pool size in case all the workers busy
+
+start_link() ->
+    supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+init([]) ->
+    TabOpts = [named_table, set, public, {read_concurrency, true}],
+    _ = ets:new(emqx_resource_instance, TabOpts),
+
+    SupFlags = #{strategy => one_for_one, intensity => 10, period => 10},
+    Pool = ?RESOURCE_INST_MOD,
+    Mod = ?RESOURCE_INST_MOD,
+    ensure_pool(Pool, hash, [{size, ?POOL_SIZE}]),
+    {ok, {SupFlags, [
+        begin
+            ensure_pool_worker(Pool, {Pool, Idx}, Idx),
+            #{id => {Mod, Idx},
+              start => {Mod, start_link, [Pool, Idx]},
+              restart => transient,
+              shutdown => 5000, type => worker, modules => [Mod]}
+        end || Idx <- lists:seq(1, ?POOL_SIZE)]}}.
+
+%% internal functions
+ensure_pool(Pool, Type, Opts) ->
+    try gproc_pool:new(Pool, Type, Opts)
+    catch
+        error:exists -> ok
+    end.
+
+ensure_pool_worker(Pool, Name, Slot) ->
+    try gproc_pool:add_worker(Pool, Name, Slot)
+    catch
+        error:exists -> ok
+    end.

apps/emqx_resource/src/emqx_resource_transform.erl

@@ -0,0 +1,114 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+-module(emqx_resource_transform).
+
+-include_lib("syntax_tools/include/merl.hrl").
+
+-export([parse_transform/2]).
+
+parse_transform(Forms, _Opts) ->
+    Mod = hd([M || {attribute, _, module, M} <- Forms]),
+    AST = trans(Mod, proplists:delete(eof, Forms)),
+    debug_print(Mod, AST),
+    AST.
+
+-ifdef(RESOURCE_DEBUG).
+
+debug_print(Mod, Ts) ->
+    {ok, Io} = file:open("./" ++ atom_to_list(Mod) ++ ".trans.erl", [write]),
+    do_debug_print(Io, Ts),
+    file:close(Io).
+
+do_debug_print(Io, Ts) when is_list(Ts) ->
+    lists:foreach(fun(T) -> do_debug_print(Io, T) end, Ts);
+do_debug_print(Io, T) ->
+    io:put_chars(Io, erl_prettypr:format(merl:tree(T))),
+    io:nl(Io).
+-else.
+debug_print(_Mod, _AST) ->
+    ok.
+-endif.
+
+trans(Mod, Forms) ->
+    forms(Mod, Forms) ++ [erl_syntax:revert(erl_syntax:eof_marker())].
+
+forms(Mod, [F0 | Fs0]) ->
+    case form(Mod, F0) of
+        {CurrForm, AppendedForms} ->
+            CurrForm ++ forms(Mod, Fs0) ++ AppendedForms;
+        {AHeadForms, CurrForm, AppendedForms} ->
+            AHeadForms ++ CurrForm ++ forms(Mod, Fs0) ++ AppendedForms
+    end;
+forms(_, []) -> [].
+
+form(Mod, Form) ->
+    case Form of
+        ?Q("-emqx_resource_api_path('@Path').") ->
+            {fix_spec_attrs() ++ fix_api_attrs(erl_syntax:concrete(Path)) ++ fix_api_exports(),
+             [],
+             fix_spec_funcs(Mod) ++ fix_api_funcs(Mod)};
+        _ ->
+            %io:format("---other form: ~p~n", [Form]),
+            {[], [Form], []}
+    end.
+
+fix_spec_attrs() ->
+    [ ?Q("-export([emqx_resource_schema/0]).")
+    , ?Q("-export([structs/0]).")
+    , ?Q("-behaviour(hocon_schema).")
+    ].
+fix_spec_funcs(_Mod) ->
+    [ (?Q("emqx_resource_schema() -> <<\"demo_swagger_schema\">>."))
+    , ?Q("structs() -> [\"config\"].")
+    ].
+
+fix_api_attrs(Path0) ->
+    BaseName = filename:basename(Path0),
+    Path = "/" ++ BaseName,
+    [erl_syntax:revert(
+        erl_syntax:attribute(?Q("rest_api"), [
+            erl_syntax:abstract(#{
+                name => list_to_atom(Name ++ "_log_tracers"),
+                method => Method,
+                path => mk_path(Path, WithId),
+                func => Func,
+                descr => Name ++ " the " ++ BaseName})]))
+       || {Name, Method, WithId, Func} <- [
+            {"list", 'GET', noid, api_get_all},
+            {"get", 'GET', id, api_get},
+            {"update", 'PUT', id, api_put},
+            {"delete", 'DELETE', id, api_delete}]].
+
+fix_api_exports() ->
+    [?Q("-export([api_get_all/2, api_get/2, api_put/2, api_delete/2]).")].
+
+fix_api_funcs(Mod) ->
+    [erl_syntax:revert(?Q(
+        "api_get_all(Binding, Params) ->
+            emqx_resource_api:get_all('@Mod@', Binding, Params).")),
+     erl_syntax:revert(?Q(
+        "api_get(Binding, Params) ->
+            emqx_resource_api:get('@Mod@', Binding, Params).")),
+     erl_syntax:revert(?Q(
+        "api_put(Binding, Params) ->
+            emqx_resource_api:put('@Mod@', Binding, Params).")),
+     erl_syntax:revert(?Q(
+        "api_delete(Binding, Params) ->
+            emqx_resource_api:delete('@Mod@', Binding, Params)."))
+    ].
+
+mk_path(Path, id) -> Path ++ "/:bin:id";
+mk_path(Path, noid) -> Path.

apps/emqx_resource/src/emqx_resource_uitils.erl

@@ -0,0 +1,16 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+-module(emqx_resource_uitils).

apps/emqx_resource/src/emqx_resource_validator.erl

@@ -0,0 +1,63 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(emqx_resource_validator).
+
+-export([ min/2
+        , max/2
+        , equals/2
+        , enum/1
+        , required/1
+        ]).
+
+max(Type, Max) ->
+    limit(Type, '=<', Max).
+
+min(Type, Min) ->
+    limit(Type, '>=', Min).
+
+equals(Type, Expected) ->
+    limit(Type, '==', Expected).
+
+enum(Items) ->
+    fun(Value) ->
+        return(lists:member(Value, Items),
+            err_limit({enum, {is_member_of, Items}, {got, Value}}))
+    end.
+
+required(ErrMsg) ->
+    fun(undefined) -> {error, ErrMsg};
+       (_) -> ok
+    end.
+
+limit(Type, Op, Expected) ->
+    L = len(Type),
+    fun(Value) ->
+        Got = L(Value),
+        return(erlang:Op(Got, Expected),
+            err_limit({Type, {Op, Expected}, {got, Got}}))
+    end.
+
+len(array) -> fun erlang:length/1;
+len(string) -> fun string:length/1;
+len(_Type) -> fun(Val) -> Val end.
+
+err_limit({Type, {Op, Expected}, {got, Got}}) ->
+    io_lib:format("Expect the ~s value ~s ~p but got: ~p", [Type, Op, Expected, Got]).
+
+return(true, _) -> ok;
+return(false, Error) ->
+    {error, Error}.

apps/emqx_retainer/etc/emqx_retainer.conf

@@ -37,5 +37,5 @@ retainer.max_payload_size = 1MB
 ##  - 30m: 30 minutes
 ##  - 20s: 20 seconds
 ##
-## Defaut: 0
+## Default: 0
 retainer.expiry_interval = 0

apps/emqx_retainer/rebar.config

@@ -18,7 +18,7 @@
 {profiles,
  [{test,
    [{deps,
-     [{emqx_ct_helpers, {git, "https://github.com/emqx/emqx-ct-helpers", {tag, "1.2.2"}}},
+     [
       {emqtt, {git, "https://github.com/emqx/emqtt", {tag, "1.2.3"}}}]}
    ]}
  ]}.

apps/emqx_rule_engine/etc/emqx_rule_engine.conf

@@ -32,7 +32,7 @@ rule_engine.ignore_sys_message = on
 ##
 ## QoS-Level: qos0/qos1/qos2
 
-#rule_engine.events.client_connected = on, qos1
+#rule_engine.events.client_connected = "on, qos1"
 rule_engine.events.client_connected = off
 rule_engine.events.client_disconnected = off
 rule_engine.events.session_subscribed = off

apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl

@@ -2554,7 +2554,7 @@ start_apps() ->
     [start_apps(App, SchemaFile, ConfigFile) ||
         {App, SchemaFile, ConfigFile}
             <- [{emqx, deps_path(emqx, "priv/emqx.schema"),
-                       deps_path(emqx, "etc/emqx.conf")},
+                       deps_path(emqx, "etc/emqx.conf.rendered")},
                 {emqx_rule_engine, local_path("priv/emqx_rule_engine.schema"),
                                    local_path("etc/emqx_rule_engine.conf")}]].
 
@@ -2566,7 +2566,7 @@ start_apps(App, SchemaFile, ConfigFile) ->
 read_schema_configs(App, SchemaFile, ConfigFile) ->
     ct:pal("Read configs - SchemaFile: ~p, ConfigFile: ~p", [SchemaFile, ConfigFile]),
     Schema = cuttlefish_schema:files([SchemaFile]),
-    Conf = conf_parse:file(ConfigFile),
+    {ok, Conf} = hocon:load(ConfigFile, #{format => proplists}),
     NewConfig = cuttlefish_generator:map(Schema, Conf),
     Vals = proplists:get_value(App, NewConfig, []),
     [application:set_env(App, Par, Value) || {Par, Value} <- Vals].

apps/emqx_sn/etc/emqx_sn.conf

@@ -6,7 +6,7 @@
 ##
 ## Value: IP:Port | Port
 ##
-## Examples: 1884, 127.0.0.1:1884, ::1:1884
+## Examples: 1884, "127.0.0.1:1884", "::1:1884"
 mqtt.sn.port = 1884
 
 ## The duration that emqx-sn broadcast ADVERTISE message through.
@@ -37,8 +37,8 @@ mqtt.sn.idle_timeout = 30s
 ## The pre-defined topic name corresponding to the pre-defined topic id of N.
 ## Note that the pre-defined topic id of 0 is reserved.
 mqtt.sn.predefined.topic.0 = reserved
-mqtt.sn.predefined.topic.1 = /predefined/topic/name/hello
-mqtt.sn.predefined.topic.2 = /predefined/topic/name/nice
+mqtt.sn.predefined.topic.1 = "/predefined/topic/name/hello"
+mqtt.sn.predefined.topic.2 = "/predefined/topic/name/nice"
 
 ## Default username for MQTT-SN. This parameter is optional. If specified,
 ## emq-sn will connect EMQ core with this username. It is useful if any auth

apps/emqx_sn/priv/emqx_sn.schema

@@ -1,23 +1,19 @@
 %%-*- mode: erlang -*-
 %% emqx_sn config mapping
 {mapping, "mqtt.sn.port", "emqx_sn.port", [
-  {default, "1884"},
-  {datatype, string}
+  {default, 1884},
+  {datatype, [integer, ip]}
 ]}.
 
 {translation, "emqx_sn.port", fun(Conf) ->
-  case re:split(cuttlefish:conf_get("mqtt.sn.port", Conf, ""), ":", [{return, list}]) of
-      [Port] ->
-          {{0,0,0,0}, list_to_integer(Port)};
-      Tokens ->
-          Port = lists:last(Tokens),
-          IP = case inet:parse_address(lists:flatten(lists:join(":", Tokens -- [Port]))) of
-                   {error, Reason} ->
-                       throw({invalid_ip_address, Reason});
-                   {ok, X} -> X
-               end,
-          Port1 = list_to_integer(Port),
-          {IP, Port1}
+  case cuttlefish:conf_get("mqtt.sn.port", Conf, undefined) of
+      Port when is_integer(Port) ->
+          {{0,0,0,0}, Port};
+      {Ip, Port} ->
+          case inet:parse_address(Ip) of
+              {ok ,R} -> {R, Port};
+              _ -> {Ip, Port}
+          end
   end
 end}.
 

apps/emqx_sn/rebar.config

@@ -2,8 +2,7 @@
 {plugins, [rebar3_proper]}.
 
 {deps,
- [{esockd, {git, "https://github.com/emqx/esockd", {tag, "5.7.4"}}},
-  {cuttlefish, {git, "https://github.com/emqx/cuttlefish", {tag, "v3.0.0"}}}
+ [{esockd, {git, "https://github.com/emqx/esockd", {tag, "5.7.4"}}}
  ]}.
 
 {edoc_opts, [{preprocess, true}]}.

apps/emqx_stomp/priv/emqx_stomp.schema

@@ -1,7 +1,7 @@
 %%-*- mode: erlang -*-
 %% emqx_stomp config mapping
 
-{mapping, "stomp.listener", "emqx_stomp.listener", [
+{mapping, "stomp.listener.port", "emqx_stomp.listener", [
   {default, 61613},
   {datatype, [integer, ip]}
 ]}.
@@ -72,7 +72,7 @@
 ]}.
 
 {translation, "emqx_stomp.listener", fun(Conf) ->
-  Port = cuttlefish:conf_get("stomp.listener", Conf),
+  Port = cuttlefish:conf_get("stomp.listener.port", Conf),
   Acceptors = cuttlefish:conf_get("stomp.listener.acceptors", Conf),
   MaxConnections = cuttlefish:conf_get("stomp.listener.max_connections", Conf),
   Filter = fun(Opts) -> [{K, V} || {K, V} <- Opts, V =/= undefined] end,

apps/emqx_web_hook/etc/emqx_web_hook.conf

@@ -5,16 +5,16 @@
 ## Webhook URL
 ##
 ## Value: String
-web.hook.url = http://127.0.0.1:80
+web.hook.url = "http://127.0.0.1:80"
 
 ## HTTP Headers
 ##
 ## Example:
-## 1. web.hook.headers.content-type = application/json
-## 2. web.hook.headers.accept = *
+## 1. web.hook.headers.content-type = "application/json"
+## 2. web.hook.headers.accept = "*"
 ##
 ## Value: String
-web.hook.headers.content-type = application/json
+web.hook.headers.content-type = "application/json"
 
 ## The encoding format of the payload field in the HTTP body
 ## The payload field only appears in the on_message_publish and on_message_delivered actions
@@ -63,15 +63,15 @@ web.hook.pool_size = 32
 ##
 ## Format:
 ##   web.hook.rule.<HookName>.<No> = <Spec>
-#web.hook.rule.client.connect.1       = {"action": "on_client_connect"}
-#web.hook.rule.client.connack.1       = {"action": "on_client_connack"}
-#web.hook.rule.client.connected.1     = {"action": "on_client_connected"}
-#web.hook.rule.client.disconnected.1  = {"action": "on_client_disconnected"}
-#web.hook.rule.client.subscribe.1     = {"action": "on_client_subscribe"}
-#web.hook.rule.client.unsubscribe.1   = {"action": "on_client_unsubscribe"}
-#web.hook.rule.session.subscribed.1   = {"action": "on_session_subscribed"}
-#web.hook.rule.session.unsubscribed.1 = {"action": "on_session_unsubscribed"}
-#web.hook.rule.session.terminated.1   = {"action": "on_session_terminated"}
-#web.hook.rule.message.publish.1      = {"action": "on_message_publish"}
-#web.hook.rule.message.delivered.1    = {"action": "on_message_delivered"}
-#web.hook.rule.message.acked.1        = {"action": "on_message_acked"}
+#web.hook.rule.client.connect.1       = "{"action": "on_client_connect"}"
+#web.hook.rule.client.connack.1       = "{"action": "on_client_connack"}"
+#web.hook.rule.client.connected.1     = "{"action": "on_client_connected"}"
+#web.hook.rule.client.disconnected.1  = "{"action": "on_client_disconnected"}"
+#web.hook.rule.client.subscribe.1     = "{"action": "on_client_subscribe"}"
+#web.hook.rule.client.unsubscribe.1   = "{"action": "on_client_unsubscribe"}"
+#web.hook.rule.session.subscribed.1   = "{"action": "on_session_subscribed"}"
+#web.hook.rule.session.unsubscribed.1 = "{"action": "on_session_unsubscribed"}"
+#web.hook.rule.session.terminated.1   = "{"action": "on_session_terminated"}"
+#web.hook.rule.message.publish.1      = "{"action": "on_message_publish"}"
+#web.hook.rule.message.delivered.1    = "{"action": "on_message_delivered"}"
+#web.hook.rule.message.acked.1        = ""{"action": "on_message_acked"}"

apps/emqx_web_hook/rebar.config

@@ -15,4 +15,4 @@
                warnings_as_errors, deprecated_functions]}.
 {cover_enabled, true}.
 {cover_opts, [verbose]}.
-{cover_export_enabled, true}.
+{cover_export_enabled, true}.

data/loaded_plugins.tmpl

@@ -5,4 +5,5 @@
 {emqx_retainer, {{enable_plugin_emqx_retainer}}}.
 {emqx_telemetry, {{enable_plugin_emqx_telemetry}}}.
 {emqx_rule_engine, {{enable_plugin_emqx_rule_engine}}}.
+{emqx_resource, {{enable_plugin_emqx_resource}}}.
 {emqx_bridge_mqtt, {{enable_plugin_emqx_bridge_mqtt}}}.

include/emqx_release.hrl

@@ -29,7 +29,7 @@
 
 -ifndef(EMQX_ENTERPRISE).
 
--define(EMQX_RELEASE, {opensource, "4.3.2"}).
+-define(EMQX_RELEASE, {opensource, "5.0-pre"}).
 
 -else.
 

lib-ce/emqx_dashboard/priv/emqx_dashboard.schema

@@ -10,7 +10,7 @@
   {override_env, "ADMIN_PASSWORD"}
 ]}.
 
-{mapping, "dashboard.listener.http", "emqx_dashboard.listeners", [
+{mapping, "dashboard.listener.http.port", "emqx_dashboard.listeners", [
   {datatype, integer}
 ]}.
 
@@ -38,7 +38,7 @@
   {datatype, {enum, [true, false]}}
 ]}.
 
-{mapping, "dashboard.listener.https", "emqx_dashboard.listeners", [
+{mapping, "dashboard.listener.https.port", "emqx_dashboard.listeners", [
   {datatype, integer}
 ]}.
 
@@ -139,7 +139,7 @@
     lists:map(
       fun(Proto) ->
         Prefix = "dashboard.listener." ++ atom_to_list(Proto),
-        case cuttlefish:conf_get(Prefix, Conf, undefined) of
+        case cuttlefish:conf_get(Prefix ++ ".port", Conf, undefined) of
             undefined -> [];
             Port      ->
                 [{Proto, Port, case Proto of

lib-ce/emqx_telemetry/etc/emqx_telemetry.conf

@@ -13,8 +13,8 @@ telemetry.enabled = true
 ##
 ## Value: String
 ##
-## Default: https://telemetry.emqx.io/api/telemetry
-telemetry.url = https://telemetry.emqx.io/api/telemetry
+## Default: "https://telemetry.emqx.io/api/telemetry"
+telemetry.url = "https://telemetry.emqx.io/api/telemetry"
 
 ## Interval for reporting telemetry data
 ##
@@ -25,4 +25,4 @@ telemetry.url = https://telemetry.emqx.io/api/telemetry
 ## -s: second
 ##
 ## Default: 7d
-telemetry.report_interval = 7d
+telemetry.report_interval = 7d

priv/emqx.schema

@@ -103,6 +103,10 @@
   {datatype, string}
 ]}.
 
+{mapping, "cluster.etcd.version", "ekka.cluster_discovery", [
+  {datatype, {enum, [v2, v3]}}
+]}.
+
 {mapping, "cluster.etcd.prefix", "ekka.cluster_discovery", [
   {datatype, string}
 ]}.
@@ -180,6 +184,7 @@
                                         end, Options)
                             end,
                  [{server, string:tokens(cuttlefish:conf_get("cluster.etcd.server", Conf), ",")},
+                  {version, cuttlefish:conf_get("cluster.etcd.version", Conf, v3)},
                   {prefix, cuttlefish:conf_get("cluster.etcd.prefix", Conf, "emqcl")},
                   {node_ttl, cuttlefish:conf_get("cluster.etcd.node_ttl", Conf, 60)},
                   {ssl_options, SslOpts(Conf)}];
@@ -467,6 +472,15 @@ end}.
   {datatype, {enum, [debug, info, notice, warning, error, critical, alert, emergency, all]}}
 ]}.
 
+%% @doc Timezone offset to display in logs,
+%% "system" use system time zone
+%% "utc" for Universal Coordinated Time (UTC)
+%% "+hh:mm" or "-hh:mm" for a specified offset
+{mapping, "log.time_offset", "kernel.logger", [
+  {default, "system"},
+  {datatype, string}
+]}.
+
 {mapping, "log.primary_log_level", "kernel.logger_level", [
    {default, warning},
    {datatype, {enum, [debug, info, notice, warning, error, critical, alert, emergency, all]}}
@@ -512,7 +526,7 @@ end}.
   {datatype, {enum, [true, false]}}
 ]}.
 
-{mapping, "log.rotation", "kernel.logger", [
+{mapping, "log.rotation.enable", "kernel.logger", [
   {default, on},
   {datatype, flag}
 ]}.
@@ -600,7 +614,27 @@ end}.
 {translation, "kernel.logger", fun(Conf) ->
     LogTo = cuttlefish:conf_get("log.to", Conf),
     LogLevel = cuttlefish:conf_get("log.level", Conf),
-    LogType = case cuttlefish:conf_get("log.rotation", Conf) of
+    LogTimeoffset =
+        case cuttlefish:conf_get("log.time_offset", Conf) of
+            "system" -> "";
+            "utc" -> "0";
+            [S, H1, H2, $:, M1, M2] = HHMM ->
+                (S =:= $+ orelse S =:= $-) andalso
+                try
+                    begin
+                        H = list_to_integer([H1, H2]),
+                        M = list_to_integer([M1, M2]),
+                        H >=0 andalso H =< 14 andalso
+                        M >= 0 andalso M =< 59
+                    end
+                catch
+                    _ : _ ->
+                        error({"invalid_log_time_offset", HHMM})
+                end andalso HHMM;
+            Other ->
+                error({"invalid_log_time_offset", Other})
+        end,
+    LogType = case cuttlefish:conf_get("log.rotation.enable", Conf) of
                   true -> wrap;
                   false -> halt
               end,
@@ -1229,7 +1263,7 @@ end}.
 %%--------------------------------------------------------------------
 %% TCP Listeners
 
-{mapping, "listener.tcp.$name", "emqx.listeners", [
+{mapping, "listener.tcp.$name.endpoint", "emqx.listeners", [
   {datatype, [integer, ip]}
 ]}.
 
@@ -1336,7 +1370,7 @@ end}.
 %%--------------------------------------------------------------------
 %% SSL Listeners
 
-{mapping, "listener.ssl.$name", "emqx.listeners", [
+{mapping, "listener.ssl.$name.endpoint", "emqx.listeners", [
   {datatype, [integer, ip]}
 ]}.
 
@@ -1504,7 +1538,7 @@ end}.
 %%--------------------------------------------------------------------
 %% MQTT/WebSocket Listeners
 
-{mapping, "listener.ws.$name", "emqx.listeners", [
+{mapping, "listener.ws.$name.endpoint", "emqx.listeners", [
   {datatype, [integer, ip]}
 ]}.
 
@@ -1698,7 +1732,7 @@ end}.
 %%--------------------------------------------------------------------
 %% MQTT/WebSocket/SSL Listeners
 
-{mapping, "listener.wss.$name", "emqx.listeners", [
+{mapping, "listener.wss.$name.endpoint", "emqx.listeners", [
   {datatype, [integer, ip]}
 ]}.
 
@@ -1954,7 +1988,6 @@ end}.
 ]}.
 
 {translation, "emqx.listeners", fun(Conf) ->
-
     Filter  = fun(Opts) -> [{K, V} || {K, V} <- Opts, V =/= undefined] end,
 
     Atom = fun(undefined) -> undefined; (S) -> list_to_atom(S) end,
@@ -2103,7 +2136,7 @@ end}.
 
     TcpListeners = fun(Type, Name) ->
                       Prefix = string:join(["listener", Type, Name], "."),
-                      ListenOnN = case cuttlefish:conf_get(Prefix, Conf, undefined) of
+                      ListenOnN = case cuttlefish:conf_get(Prefix ++ ".endpoint", Conf, undefined) of
                           undefined -> [];
                           ListenOn  -> Listen_fix(ListenOn)
                       end,
@@ -2119,7 +2152,7 @@ end}.
                    end,
     SslListeners = fun(Type, Name) ->
                        Prefix = string:join(["listener", Type, Name], "."),
-                       case cuttlefish:conf_get(Prefix, Conf, undefined) of
+                       case cuttlefish:conf_get(Prefix ++ ".endpoint", Conf, undefined) of
                            undefined ->
                                [];
                            ListenOn ->
@@ -2135,12 +2168,11 @@ end}.
                                ]
                        end
                    end,
-
-    lists:flatten([TcpListeners(Type, Name) || {["listener", Type, Name], ListenOn}
+    lists:flatten([TcpListeners(Type, Name) || {["listener", Type, Name, "endpoint"], ListenOn}
                                                <- cuttlefish_variable:filter_by_prefix("listener.tcp", Conf)
                                                ++ cuttlefish_variable:filter_by_prefix("listener.ws", Conf)]
                   ++
-                  [SslListeners(Type, Name) || {["listener", Type, Name], ListenOn}
+                  [SslListeners(Type, Name) || {["listener", Type, Name, "endpoint"], ListenOn}
                                                <- cuttlefish_variable:filter_by_prefix("listener.ssl", Conf)
                                                ++ cuttlefish_variable:filter_by_prefix("listener.wss", Conf)])
 end}.

rebar.config

@@ -43,9 +43,9 @@
     , {jiffy, {git, "https://github.com/emqx/jiffy", {tag, "1.0.5"}}}
     , {cowboy, {git, "https://github.com/emqx/cowboy", {tag, "2.8.2"}}}
     , {esockd, {git, "https://github.com/emqx/esockd", {tag, "5.8.0"}}}
-    , {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.8.1"}}}
+    , {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.9.0"}}}
     , {gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "2.5.1"}}}
-    , {cuttlefish, {git, "https://github.com/emqx/cuttlefish", {tag, "v3.3.5"}}}
+    , {cuttlefish, {git, "https://github.com/emqx/cuttlefish", {tag, "v4.0.0"}}}
     , {minirest, {git, "https://github.com/emqx/minirest", {tag, "0.3.5"}}}
     , {ecpool, {git, "https://github.com/emqx/ecpool", {tag, "0.5.1"}}}
     , {replayq, {git, "https://github.com/emqx/replayq", {tag, "0.3.2"}}}

rebar.config.erl

@@ -106,7 +106,7 @@ test_plugins() ->
 
 test_deps() ->
     [ {bbmustache, "1.10.0"}
-    , {emqx_ct_helpers, {git, "https://github.com/emqx/emqx-ct-helpers", {tag, "1.3.9"}}}
+    , {emqx_ct_helpers, {git, "https://github.com/emqx/emqx-ct-helpers", {branch, "hocon"}}}
     , meck
     ].
 
@@ -188,6 +188,7 @@ overlay_vars_rel(RelType) ->
              end,
     [ {enable_plugin_emqx_rule_engine, RelType =:= cloud}
     , {enable_plugin_emqx_bridge_mqtt, RelType =:= edge}
+    , {enable_plugin_emqx_resource, true}
     , {enable_plugin_emqx_modules, false} %% modules is not a plugin in ce
     , {enable_plugin_emqx_recon, true}
     , {enable_plugin_emqx_retainer, true}
@@ -281,6 +282,7 @@ relx_plugin_apps(ReleaseType) ->
     , emqx_auth_mnesia
     , emqx_web_hook
     , emqx_recon
+    , emqx_resource
     , emqx_rule_engine
     , emqx_sasl
     ]

test/emqx_listeners_SUITE.erl

@@ -67,14 +67,16 @@ mustache_vars() ->
 generate_config() ->
     Schema = cuttlefish_schema:files([local_path(["priv", "emqx.schema"])]),
     ConfFile = render_config_file(),
-    Conf = conf_parse:file(ConfFile),
+    {ok, Conf} = hocon:load(ConfFile, #{format => proplists}),
     cuttlefish_generator:map(Schema, Conf).
 
 set_app_env({App, Lists}) ->
     lists:foreach(fun({acl_file, _Var}) ->
                       application:set_env(App, acl_file, local_path(["etc", "acl.conf"]));
                      ({plugins_loaded_file, _Var}) ->
-                      application:set_env(App, plugins_loaded_file, local_path(["test", "emqx_SUITE_data","loaded_plugins"]));
+                      application:set_env(App,
+                                          plugins_loaded_file,
+                                          local_path(["test", "emqx_SUITE_data","loaded_plugins"]));
                      ({Par, Var}) ->
                       application:set_env(App, Par, Var)
                   end, Lists).
@@ -91,4 +93,4 @@ get_base_dir(Module) ->
 
 get_base_dir() ->
     get_base_dir(?MODULE).
-    
+