Merge pull request #12893 from zmstone/0416-add-is-template-flag-to-dashboard-schema

0416 add `is_template` flag to dashboard schema

apps/emqx/src/emqx_schema.erl

@@ -61,6 +61,8 @@
 }.
 -type url() :: binary().
 -type json_binary() :: binary().
+-type template() :: binary().
+-type template_str() :: string().
 
 -typerefl_from_string({duration/0, emqx_schema, to_duration}).
 -typerefl_from_string({duration_s/0, emqx_schema, to_duration_s}).
@@ -78,6 +80,8 @@
 -typerefl_from_string({comma_separated_atoms/0, emqx_schema, to_comma_separated_atoms}).
 -typerefl_from_string({url/0, emqx_schema, to_url}).
 -typerefl_from_string({json_binary/0, emqx_schema, to_json_binary}).
+-typerefl_from_string({template/0, emqx_schema, to_template}).
+-typerefl_from_string({template_str/0, emqx_schema, to_template_str}).
 
 -type parsed_server() :: #{
     hostname := string(),
@@ -120,7 +124,9 @@
     to_erl_cipher_suite/1,
     to_comma_separated_atoms/1,
     to_url/1,
-    to_json_binary/1
+    to_json_binary/1,
+    to_template/1,
+    to_template_str/1
 ]).
 
 -export([
@@ -160,7 +166,9 @@
     comma_separated_atoms/0,
     url/0,
     json_binary/0,
-    port_number/0
+    port_number/0,
+    template/0,
+    template_str/0
 ]).
 
 -export([namespace/0, roots/0, roots/1, fields/1, desc/1, tags/0]).
@@ -2594,6 +2602,12 @@ to_json_binary(Str) ->
             Error
     end.
 
+to_template(Str) ->
+    {ok, iolist_to_binary(Str)}.
+
+to_template_str(Str) ->
+    {ok, unicode:characters_to_list(Str, utf8)}.
+
 %% @doc support the following format:
 %%  - 127.0.0.1:1883
 %%  - ::1:1883

apps/emqx_auth/src/emqx_authn/emqx_authn_chains.erl

@@ -353,13 +353,13 @@ init(_Opts) ->
     ok = emqx_config_handler:add_handler([listeners, '?', '?', ?CONF_ROOT], Module),
     ok = hook_deny(),
     {ok, #{hooked => false, providers => #{}, init_done => false},
-        {continue, initialize_authentication}}.
+        {continue, {initialize_authentication, init}}}.
 
 handle_call(get_providers, _From, #{providers := Providers} = State) ->
     reply(Providers, State);
 handle_call(
     {register_providers, Providers},
-    _From,
+    From,
     #{providers := Reg0} = State
 ) ->
     case lists:filter(fun({T, _}) -> maps:is_key(T, Reg0) end, Providers) of
@@ -371,7 +371,7 @@ handle_call(
                 Reg0,
                 Providers
             ),
-            reply(ok, State#{providers := Reg}, initialize_authentication);
+            reply(ok, State#{providers := Reg}, {initialize_authentication, From});
         Clashes ->
             reply({error, {authentication_type_clash, Clashes}}, State)
     end;
@@ -447,10 +447,10 @@ handle_call(Req, _From, State) ->
     ?SLOG(error, #{msg => "unexpected_call", call => Req}),
     {reply, ignored, State}.
 
-handle_continue(initialize_authentication, #{init_done := true} = State) ->
+handle_continue({initialize_authentication, _From}, #{init_done := true} = State) ->
     {noreply, State};
-handle_continue(initialize_authentication, #{providers := Providers} = State) ->
+handle_continue({initialize_authentication, From}, #{providers := Providers} = State) ->
-    InitDone = initialize_authentication(Providers),
+    InitDone = initialize_authentication(Providers, From),
     {noreply, maybe_hook(State#{init_done := InitDone})}.
 
 handle_cast(Req, State) ->
@@ -484,11 +484,13 @@ code_change(_OldVsn, State, _Extra) ->
 %% Private functions
 %%------------------------------------------------------------------------------
 
-initialize_authentication(Providers) ->
+initialize_authentication(Providers, From) ->
     ProviderTypes = maps:keys(Providers),
     Chains = chain_configs(),
     HasProviders = has_providers_for_configs(Chains, ProviderTypes),
-    do_initialize_authentication(Providers, Chains, HasProviders).
+    Result = do_initialize_authentication(Providers, Chains, HasProviders),
+    ?tp(info, authn_chains_initialization_done, #{from => From, result => Result}),
+    Result.
 
 do_initialize_authentication(_Providers, _Chains, _HasProviders = false) ->
     false;
@@ -500,7 +502,6 @@ do_initialize_authentication(Providers, Chains, _HasProviders = true) ->
         Chains
     ),
     ok = unhook_deny(),
-    ?tp(info, authn_chains_initialization_done, #{}),
     true.
 
 initialize_chain_authentication(_Providers, _ChainName, []) ->

apps/emqx_auth/test/emqx_authn/emqx_authn_init_SUITE.erl

@@ -69,9 +69,10 @@ t_initialize(_Config) ->
         emqx_access_control:authenticate(?CLIENTINFO)
     ),
 
+    Self = self(),
     ?assertWaitEvent(
         ok = emqx_authn_test_lib:register_fake_providers([{password_based, built_in_database}]),
-        #{?snk_kind := authn_chains_initialization_done},
+        #{?snk_kind := authn_chains_initialization_done, from := {Self, _}},
         100
     ),
 

apps/emqx_bridge/src/emqx_bridge_v2_api.erl

@@ -1007,7 +1007,13 @@ call_operation(NodeOrAll, OperFunc, Args = [_Nodes, _ConfRootKey, BridgeType, Br
         {error, not_implemented} ->
             ?NOT_IMPLEMENTED;
         {error, timeout} ->
-            ?BAD_REQUEST(<<"Request timeout">>);
+            BridgeId = emqx_bridge_resource:bridge_id(BridgeType, BridgeName),
+            ?SLOG(warning, #{
+                msg => "bridge_bpapi_call_timeout",
+                bridge => BridgeId,
+                call => OperFunc
+            }),
+            ?SERVICE_UNAVAILABLE(<<"Request timeout">>);
         {error, {start_pool_failed, Name, Reason}} ->
             Msg = bin(
                 io_lib:format("Failed to start ~p pool for reason ~p", [Name, redact(Reason)])
@@ -1018,9 +1024,8 @@ call_operation(NodeOrAll, OperFunc, Args = [_Nodes, _ConfRootKey, BridgeType, Br
             ?SLOG(warning, #{
                 msg => "bridge_inconsistent_in_cluster_for_call_operation",
                 reason => not_found,
-                type => BridgeType,
+                bridge => BridgeId,
-                name => BridgeName,
+                call => OperFunc
-                bridge => BridgeId
             }),
             ?SERVICE_UNAVAILABLE(<<"Bridge not found on remote node: ", BridgeId/binary>>);
         {error, {node_not_found, Node}} ->

apps/emqx_bridge_cassandra/src/emqx_bridge_cassandra.erl

@@ -181,7 +181,7 @@ fields("post", Type) ->
 cql_field() ->
     {cql,
         mk(
-            binary(),
+            emqx_schema:template(),
             #{desc => ?DESC("cql_template"), default => ?DEFAULT_CQL, format => <<"sql">>}
         )}.
 

apps/emqx_bridge_clickhouse/src/emqx_bridge_clickhouse.erl

@@ -184,8 +184,12 @@ fields("post", Type) ->
 sql_field() ->
     {sql,
         mk(
-            binary(),
+            emqx_schema:template(),
-            #{desc => ?DESC("sql_template"), default => ?DEFAULT_SQL, format => <<"sql">>}
+            #{
+                desc => ?DESC("sql_template"),
+                default => ?DEFAULT_SQL,
+                format => <<"sql">>
+            }
         )}.
 
 batch_value_separator_field() ->

apps/emqx_bridge_dynamo/src/emqx_bridge_dynamo.erl

@@ -160,13 +160,7 @@ fields(dynamo_action) ->
     );
 fields(action_parameters) ->
     Parameters =
-        [
+        [{template, template_field_schema()}] ++ emqx_bridge_dynamo_connector:fields(config),
-            {template,
-                mk(
-                    binary(),
-                    #{desc => ?DESC("template"), default => ?DEFAULT_TEMPLATE}
-                )}
-        ] ++ emqx_bridge_dynamo_connector:fields(config),
     lists:foldl(
         fun(Key, Acc) ->
             proplists:delete(Key, Acc)
@@ -199,11 +193,7 @@ fields(connector_resource_opts) ->
 fields("config") ->
     [
         {enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
-        {template,
+        {template, template_field_schema()},
-            mk(
-                binary(),
-                #{desc => ?DESC("template"), default => ?DEFAULT_TEMPLATE}
-            )},
         {local_topic,
             mk(
                 binary(),
@@ -230,6 +220,15 @@ fields("put") ->
 fields("get") ->
     emqx_bridge_schema:status_fields() ++ fields("post").
 
+template_field_schema() ->
+    mk(
+        emqx_schema:template(),
+        #{
+            desc => ?DESC("template"),
+            default => ?DEFAULT_TEMPLATE
+        }
+    ).
+
 desc("config") ->
     ?DESC("desc_config");
 desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->

apps/emqx_bridge_es/src/emqx_bridge_es.erl

@@ -135,7 +135,7 @@ overwrite() ->
 index() ->
     {index,
         ?HOCON(
-            binary(),
+            emqx_schema:template(),
             #{
                 required => true,
                 example => <<"${payload.index}">>,
@@ -146,7 +146,7 @@ index() ->
 id(Required) ->
     {id,
         ?HOCON(
-            binary(),
+            emqx_schema:template(),
             #{
                 required => Required,
                 example => <<"${payload.id}">>,
@@ -157,7 +157,7 @@ id(Required) ->
 doc() ->
     {doc,
         ?HOCON(
-            binary(),
+            emqx_schema:template(),
             #{
                 required => false,
                 example => <<"${payload.doc}">>,
@@ -187,7 +187,7 @@ doc_as_upsert() ->
 routing() ->
     {routing,
         ?HOCON(
-            binary(),
+            emqx_schema:template(),
             #{
                 required => false,
                 example => <<"${payload.routing}">>,

apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub.erl

@@ -122,7 +122,7 @@ fields(producer) ->
             )},
         {ordering_key_template,
             sc(
-                binary(),
+                emqx_schema:template(),
                 #{
                     default => <<>>,
                     desc => ?DESC("ordering_key_template")
@@ -130,7 +130,7 @@ fields(producer) ->
             )},
         {payload_template,
             sc(
-                binary(),
+                emqx_schema:template(),
                 #{
                     default => <<>>,
                     desc => ?DESC("payload_template")
@@ -201,8 +201,11 @@ fields(consumer_topic_mapping) ->
         {qos, mk(emqx_schema:qos(), #{default => 0, desc => ?DESC(consumer_mqtt_qos)})},
         {payload_template,
             mk(
-                string(),
+                emqx_schema:template(),
-                #{default => <<"${.}">>, desc => ?DESC(consumer_mqtt_payload)}
+                #{
+                    default => <<"${.}">>,
+                    desc => ?DESC(consumer_mqtt_payload)
+                }
             )}
     ];
 fields("consumer_resource_opts") ->
@@ -221,14 +224,18 @@ fields("consumer_resource_opts") ->
 fields(key_value_pair) ->
     [
         {key,
-            mk(binary(), #{
+            mk(emqx_schema:template(), #{
                 required => true,
                 validator => [
                     emqx_resource_validator:not_empty("Key templates must not be empty")
                 ],
                 desc => ?DESC(kv_pair_key)
             })},
-        {value, mk(binary(), #{required => true, desc => ?DESC(kv_pair_value)})}
+        {value,
+            mk(emqx_schema:template(), #{
+                required => true,
+                desc => ?DESC(kv_pair_value)
+            })}
     ];
 fields("get_producer") ->
     emqx_bridge_schema:status_fields() ++ fields("post_producer");

apps/emqx_bridge_hstreamdb/src/emqx_bridge_hstreamdb.erl

@@ -167,13 +167,13 @@ fields(action_parameters) ->
             })},
 
         {partition_key,
-            mk(binary(), #{
+            mk(emqx_schema:template(), #{
-                required => false, desc => ?DESC(emqx_bridge_hstreamdb_connector, "partition_key")
+                required => false,
+                desc => ?DESC(emqx_bridge_hstreamdb_connector, "partition_key")
             })},
 
         {grpc_flush_timeout, fun grpc_flush_timeout/1},
-        {record_template,
+        {record_template, record_template_schema()},
-            mk(binary(), #{default => <<"${payload}">>, desc => ?DESC("record_template")})},
         {aggregation_pool_size,
             mk(pos_integer(), #{
                 default => ?DEFAULT_AGG_POOL_SIZE, desc => ?DESC("aggregation_pool_size")
@@ -222,6 +222,12 @@ fields("put") ->
     hstream_bridge_common_fields() ++
         connector_fields().
 
+record_template_schema() ->
+    mk(emqx_schema:template(), #{
+        default => <<"${payload}">>,
+        desc => ?DESC("record_template")
+    }).
+
 grpc_timeout(type) -> emqx_schema:timeout_duration_ms();
 grpc_timeout(desc) -> ?DESC(emqx_bridge_hstreamdb_connector, "grpc_timeout");
 grpc_timeout(default) -> ?DEFAULT_GRPC_TIMEOUT_RAW;
@@ -239,8 +245,7 @@ hstream_bridge_common_fields() ->
         [
             {direction, mk(egress, #{desc => ?DESC("config_direction"), default => egress})},
             {local_topic, mk(binary(), #{desc => ?DESC("local_topic")})},
-            {record_template,
+            {record_template, record_template_schema()}
-                mk(binary(), #{default => <<"${payload}">>, desc => ?DESC("record_template")})}
         ] ++
         emqx_resource_schema:fields("resource_opts").
 

apps/emqx_bridge_http/src/emqx_bridge_http_connector.erl

@@ -128,9 +128,10 @@ fields("request") ->
                 desc => ?DESC("method"),
                 validator => fun ?MODULE:validate_method/1
             })},
-        {path, hoconsc:mk(binary(), #{required => false, desc => ?DESC("path")})},
+        {path, hoconsc:mk(emqx_schema:template(), #{required => false, desc => ?DESC("path")})},
-        {body, hoconsc:mk(binary(), #{required => false, desc => ?DESC("body")})},
+        {body, hoconsc:mk(emqx_schema:template(), #{required => false, desc => ?DESC("body")})},
-        {headers, hoconsc:mk(map(), #{required => false, desc => ?DESC("headers")})},
+        {headers,
+            hoconsc:mk(map(), #{required => false, desc => ?DESC("headers"), is_template => true})},
         {max_retries,
             sc(
                 non_neg_integer(),

apps/emqx_bridge_http/src/emqx_bridge_http_schema.erl

@@ -114,7 +114,7 @@ fields("parameters_opts") ->
     [
         {path,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{
                     desc => ?DESC("config_path"),
                     required => false
@@ -270,7 +270,8 @@ headers_field() ->
                     <<"content-type">> => <<"application/json">>,
                     <<"keep-alive">> => <<"timeout=5">>
                 },
-                desc => ?DESC("config_headers")
+                desc => ?DESC("config_headers"),
+                is_template => true
             }
         )}.
 
@@ -287,7 +288,7 @@ method_field() ->
 body_field() ->
     {body,
         mk(
-            binary(),
+            emqx_schema:template(),
             #{
                 default => undefined,
                 desc => ?DESC("config_body")

apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb.erl

@@ -42,7 +42,7 @@
 %% api
 
 write_syntax_type() ->
-    typerefl:alias("string", write_syntax()).
+    typerefl:alias("template", write_syntax()).
 
 %% Examples
 conn_bridge_examples(Method) ->

apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.erl

@@ -84,7 +84,7 @@ fields(action_parameters) ->
             )},
         {device_id,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{
                     desc => ?DESC("config_device_id")
                 }
@@ -114,7 +114,7 @@ fields(action_parameters_data) ->
             )},
         {measurement,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{
                     required => true,
                     desc => ?DESC("config_parameters_measurement")
@@ -122,7 +122,9 @@ fields(action_parameters_data) ->
             )},
         {data_type,
             mk(
-                hoconsc:union([enum([text, boolean, int32, int64, float, double]), binary()]),
+                hoconsc:union([
+                    enum([text, boolean, int32, int64, float, double]), emqx_schema:template()
+                ]),
                 #{
                     required => true,
                     desc => ?DESC("config_parameters_data_type")
@@ -130,7 +132,7 @@ fields(action_parameters_data) ->
             )},
         {value,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{
                     required => true,
                     desc => ?DESC("config_parameters_value")

apps/emqx_bridge_kafka/src/emqx_bridge_kafka.erl

@@ -389,7 +389,7 @@ fields(producer_kafka_opts) ->
             )},
         {kafka_headers,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{
                     required => false,
                     validator => fun kafka_header_validator/1,
@@ -462,12 +462,12 @@ fields(producer_kafka_ext_headers) ->
     [
         {kafka_ext_header_key,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{required => true, desc => ?DESC(producer_kafka_ext_header_key)}
             )},
         {kafka_ext_header_value,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{
                     required => true,
                     validator => fun kafka_ext_header_value_validator/1,
@@ -477,11 +477,20 @@ fields(producer_kafka_ext_headers) ->
     ];
 fields(kafka_message) ->
     [
-        {key, mk(string(), #{default => <<"${.clientid}">>, desc => ?DESC(kafka_message_key)})},
+        {key,
-        {value, mk(string(), #{default => <<"${.}">>, desc => ?DESC(kafka_message_value)})},
+            mk(emqx_schema:template(), #{
+                default => <<"${.clientid}">>,
+                desc => ?DESC(kafka_message_key)
+            })},
+        {value,
+            mk(emqx_schema:template(), #{
+                default => <<"${.}">>,
+                desc => ?DESC(kafka_message_value)
+            })},
         {timestamp,
-            mk(string(), #{
+            mk(emqx_schema:template(), #{
-                default => <<"${.timestamp}">>, desc => ?DESC(kafka_message_timestamp)
+                default => <<"${.timestamp}">>,
+                desc => ?DESC(kafka_message_timestamp)
             })}
     ];
 fields(producer_buffer) ->
@@ -536,8 +545,11 @@ fields(consumer_topic_mapping) ->
         {qos, mk(emqx_schema:qos(), #{default => 0, desc => ?DESC(consumer_mqtt_qos)})},
         {payload_template,
             mk(
-                string(),
+                emqx_schema:template(),
-                #{default => <<"${.}">>, desc => ?DESC(consumer_mqtt_payload)}
+                #{
+                    default => <<"${.}">>,
+                    desc => ?DESC(consumer_mqtt_payload)
+                }
             )}
     ];
 fields(consumer_kafka_opts) ->
@@ -744,8 +756,8 @@ producer_strategy_key_validator(
     producer_strategy_key_validator(emqx_utils_maps:binary_key_map(Conf));
 producer_strategy_key_validator(#{
     <<"partition_strategy">> := key_dispatch,
-    <<"message">> := #{<<"key">> := ""}
+    <<"message">> := #{<<"key">> := Key}
-}) ->
+}) when Key =:= "" orelse Key =:= <<>> ->
     {error, "Message key cannot be empty when `key_dispatch` strategy is used"};
 producer_strategy_key_validator(_) ->
     ok.

apps/emqx_bridge_kinesis/src/emqx_bridge_kinesis.erl

@@ -150,7 +150,7 @@ fields(producer) ->
     [
         {payload_template,
             sc(
-                binary(),
+                emqx_schema:template(),
                 #{
                     default => <<"${.}">>,
                     desc => ?DESC("payload_template")

apps/emqx_bridge_mongodb/src/emqx_bridge_mongodb.erl

@@ -44,8 +44,10 @@ roots() -> [].
 fields("config") ->
     [
         {enable, mk(boolean(), #{desc => ?DESC("enable"), default => true})},
-        {collection, mk(binary(), #{desc => ?DESC("collection"), default => <<"mqtt">>})},
+        {collection,
-        {payload_template, mk(binary(), #{required => false, desc => ?DESC("payload_template")})},
+            mk(emqx_schema:template(), #{desc => ?DESC("collection"), default => <<"mqtt">>})},
+        {payload_template,
+            mk(emqx_schema:template(), #{required => false, desc => ?DESC("payload_template")})},
         {resource_opts,
             mk(
                 ref(?MODULE, "creation_opts"),

apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt_connector_schema.erl

@@ -200,7 +200,7 @@ fields("ingress_local") ->
     [
         {topic,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{
                     validator => fun emqx_schema:non_empty_string/1,
                     desc => ?DESC("ingress_local_topic"),
@@ -217,7 +217,7 @@ fields("ingress_local") ->
             )},
         {retain,
             mk(
-                hoconsc:union([boolean(), binary()]),
+                hoconsc:union([boolean(), emqx_schema:template()]),
                 #{
                     default => <<"${retain}">>,
                     desc => ?DESC("retain")
@@ -225,7 +225,7 @@ fields("ingress_local") ->
             )},
         {payload,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{
                     default => undefined,
                     desc => ?DESC("payload")
@@ -268,7 +268,7 @@ fields("egress_remote") ->
     [
         {topic,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{
                     required => true,
                     validator => fun emqx_schema:non_empty_string/1,
@@ -286,7 +286,7 @@ fields("egress_remote") ->
             )},
         {retain,
             mk(
-                hoconsc:union([boolean(), binary()]),
+                hoconsc:union([boolean(), emqx_schema:template()]),
                 #{
                     required => false,
                     default => false,
@@ -295,7 +295,7 @@ fields("egress_remote") ->
             )},
         {payload,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{
                     default => undefined,
                     desc => ?DESC("payload")
@@ -344,7 +344,7 @@ desc(_) ->
     undefined.
 
 qos() ->
-    hoconsc:union([emqx_schema:qos(), binary()]).
+    hoconsc:union([emqx_schema:qos(), emqx_schema:template()]).
 
 parse_server(Str) ->
     #{hostname := Host, port := Port} = emqx_schema:parse_server(Str, ?MQTT_HOST_OPTS),

apps/emqx_bridge_mysql/src/emqx_bridge_mysql.erl

@@ -117,7 +117,7 @@ fields("config") ->
         {enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
         {sql,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{desc => ?DESC("sql_template"), default => ?DEFAULT_SQL, format => <<"sql">>}
             )},
         {local_topic,

apps/emqx_bridge_opents/src/emqx_bridge_opents.erl

@@ -146,7 +146,7 @@ fields(action_parameters_data) ->
     [
         {timestamp,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{
                     desc => ?DESC("config_parameters_timestamp"),
                     required => false
@@ -154,7 +154,7 @@ fields(action_parameters_data) ->
             )},
         {metric,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{
                     required => true,
                     desc => ?DESC("config_parameters_metric")
@@ -162,7 +162,7 @@ fields(action_parameters_data) ->
             )},
         {tags,
             mk(
-                hoconsc:union([map(), binary()]),
+                hoconsc:union([map(), emqx_schema:template()]),
                 #{
                     required => true,
                     desc => ?DESC("config_parameters_tags"),
@@ -188,7 +188,7 @@ fields(action_parameters_data) ->
             )},
         {value,
             mk(
-                hoconsc:union([integer(), float(), binary()]),
+                hoconsc:union([integer(), float(), emqx_schema:template()]),
                 #{
                     required => true,
                     desc => ?DESC("config_parameters_value")

apps/emqx_bridge_oracle/src/emqx_bridge_oracle.erl

@@ -158,7 +158,7 @@ fields(action_parameters) ->
     [
         {sql,
             hoconsc:mk(
-                binary(),
+                emqx_schema:template(),
                 #{desc => ?DESC("sql_template"), default => ?DEFAULT_SQL, format => <<"sql">>}
             )}
     ];
@@ -177,7 +177,7 @@ fields("config") ->
             )},
         {sql,
             hoconsc:mk(
-                binary(),
+                emqx_schema:template(),
                 #{desc => ?DESC("sql_template"), default => ?DEFAULT_SQL, format => <<"sql">>}
             )},
         {local_topic,

apps/emqx_bridge_pgsql/src/emqx_bridge_pgsql.erl

@@ -61,7 +61,7 @@ fields(action_parameters) ->
     [
         {sql,
             hoconsc:mk(
-                binary(),
+                emqx_schema:template(),
                 #{desc => ?DESC("sql_template"), default => default_sql(), format => <<"sql">>}
             )}
     ];

apps/emqx_bridge_pulsar/src/emqx_bridge_pulsar_pubsub_schema.erl

@@ -51,12 +51,12 @@ fields(action_parameters) ->
 fields(producer_pulsar_message) ->
     [
         {key,
-            ?HOCON(string(), #{
+            ?HOCON(emqx_schema:template(), #{
                 default => <<"${.clientid}">>,
                 desc => ?DESC("producer_key_template")
             })},
         {value,
-            ?HOCON(string(), #{
+            ?HOCON(emqx_schema:template(), #{
                 default => <<"${.}">>,
                 desc => ?DESC("producer_value_template")
             })}

apps/emqx_bridge_pulsar/test/emqx_bridge_pulsar_connector_SUITE.erl

@@ -1235,7 +1235,7 @@ t_resilience(Config) ->
                 after 1_000 -> ct:fail("producer didn't stop!")
                 end,
             Consumed = lists:flatmap(
-                fun(_) -> receive_consumed(5_000) end, lists:seq(1, NumProduced)
+                fun(_) -> receive_consumed(10_000) end, lists:seq(1, NumProduced)
             ),
             ?assertEqual(NumProduced, length(Consumed)),
             ExpectedPayloads = lists:map(fun integer_to_binary/1, lists:seq(1, NumProduced)),

apps/emqx_bridge_rabbitmq/src/emqx_bridge_rabbitmq_pubsub_schema.erl

@@ -99,7 +99,7 @@ fields(action_parameters) ->
             )},
         {payload_template,
             hoconsc:mk(
-                binary(),
+                emqx_schema:template(),
                 #{
                     default => <<"">>,
                     desc => ?DESC(?CONNECTOR_SCHEMA, "payload_template")

apps/emqx_bridge_redis/src/emqx_bridge_redis.erl

@@ -211,7 +211,7 @@ desc(_) ->
     undefined.
 
 command_template(type) ->
-    list(binary());
+    hoconsc:array(emqx_schema:template());
 command_template(required) ->
     true;
 command_template(validator) ->

apps/emqx_bridge_rocketmq/src/emqx_bridge_rocketmq.erl

@@ -162,8 +162,11 @@ fields(action_parameters) ->
         [
             {template,
                 mk(
-                    binary(),
+                    emqx_schema:template(),
-                    #{desc => ?DESC("template"), default => ?DEFAULT_TEMPLATE}
+                    #{
+                        desc => ?DESC("template"),
+                        default => ?DEFAULT_TEMPLATE
+                    }
                 )}
         ] ++ emqx_bridge_rocketmq_connector:fields(config),
     lists:foldl(
@@ -205,7 +208,7 @@ fields("config") ->
         {enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
         {template,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{desc => ?DESC("template"), default => ?DEFAULT_TEMPLATE}
             )},
         {local_topic,
@@ -214,8 +217,7 @@ fields("config") ->
                 #{desc => ?DESC("local_topic"), required => false}
             )}
     ] ++ emqx_resource_schema:fields("resource_opts") ++
-        (emqx_bridge_rocketmq_connector:fields(config) --
+        emqx_bridge_rocketmq_connector:fields(config);
-            emqx_connector_schema_lib:prepare_statement_fields());
 fields("post") ->
     [type_field(), name_field() | fields("config")];
 fields("put") ->

apps/emqx_bridge_rocketmq/src/emqx_bridge_rocketmq_connector.erl

@@ -47,7 +47,7 @@ fields(config) ->
         {servers, servers()},
         {topic,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{default => <<"TopicTest">>, desc => ?DESC(topic)}
             )},
         {access_key,

apps/emqx_bridge_s3/src/emqx_bridge_s3.erl

@@ -77,7 +77,7 @@ fields(s3_upload_parameters) ->
         [
             {content,
                 hoconsc:mk(
-                    string(),
+                    emqx_schema:template(),
                     #{
                         required => false,
                         default => <<"${.}">>,

apps/emqx_bridge_sqlserver/src/emqx_bridge_sqlserver.erl

@@ -192,7 +192,7 @@ fields(action_parameters) ->
     [
         {sql,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{desc => ?DESC("sql_template"), default => ?DEFAULT_SQL, format => <<"sql">>}
             )}
     ];

apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper.erl

@@ -112,7 +112,7 @@ fields("parameters") ->
     [
         {target_topic,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{desc => ?DESC("target_topic"), default => <<"${topic}">>}
             )},
         {target_qos,
@@ -122,7 +122,7 @@ fields("parameters") ->
             )},
         {template,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{desc => ?DESC("template"), default => <<"${payload}">>}
             )}
     ];

apps/emqx_bridge_tdengine/src/emqx_bridge_tdengine.erl

@@ -83,7 +83,7 @@ fields("config") ->
         {enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
         {sql,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{
                     desc => ?DESC("sql_template"),
                     default => ?DEFAULT_SQL,
@@ -125,7 +125,7 @@ fields(action_parameters) ->
         {database, fun emqx_connector_schema_lib:database/1},
         {sql,
             mk(
-                binary(),
+                emqx_schema:template(),
                 #{
                     desc => ?DESC("sql_template"),
                     default => ?DEFAULT_SQL,

apps/emqx_conf/src/emqx_conf.erl

@@ -304,12 +304,22 @@ gen_flat_doc(RootNames, #{full_name := FullName, fields := Fields} = S, DescReso
         false ->
             ok
     end,
-    #{
+    try
-        text => short_name(FullName),
+        #{
-        hash => format_hash(FullName),
+            text => short_name(FullName),
-        doc => maps:get(desc, S, <<"">>),
+            hash => format_hash(FullName),
-        fields => format_fields(Fields, DescResolver)
+            doc => maps:get(desc, S, <<"">>),
-    }.
+            fields => format_fields(Fields, DescResolver)
+        }
+    catch
+        throw:Reason ->
+            io:format(
+                standard_error,
+                "failed_to_build_doc for ~s:~n~p~n",
+                [FullName, Reason]
+            ),
+            error(failed_to_build_doc)
+    end.
 
 format_fields(Fields, DescResolver) ->
     [format_field(F, DescResolver) || F <- Fields].

apps/emqx_conf/src/emqx_conf_schema_types.erl

@@ -33,8 +33,19 @@ readable(Module, TypeStr) when is_list(TypeStr) ->
         %% Module is ignored so far as all types are distinguished by their names
         readable(TypeStr)
     catch
-        throw:unknown_type ->
+        throw:Reason ->
-            fail(#{reason => unknown_type, type => TypeStr, module => Module})
+            throw(#{
+                reason => Reason,
+                type => TypeStr,
+                module => Module
+            });
+        error:Reason:Stacktrace ->
+            throw(#{
+                reason => Reason,
+                stacktrace => Stacktrace,
+                type => TypeStr,
+                module => Module
+            })
     end.
 
 readable_swagger(Module, TypeStr) ->
@@ -49,22 +60,28 @@ readable_docgen(Module, TypeStr) ->
 get_readable(Module, TypeStr, Flavor) ->
     Map = readable(Module, TypeStr),
     case maps:get(Flavor, Map, undefined) of
-        undefined -> fail(#{reason => unknown_type, module => Module, type => TypeStr});
+        undefined -> throw(#{reason => unknown_type, module => Module, type => TypeStr});
         Value -> Value
     end.
 
-%% Fail the build or test. Production code should never get here.
--spec fail(_) -> no_return().
-fail(Reason) ->
-    io:format(standard_error, "ERROR: ~p~n", [Reason]),
-    error(Reason).
-
 readable("boolean()") ->
     #{
         swagger => #{type => boolean},
         dashboard => #{type => boolean},
         docgen => #{type => "Boolean"}
     };
+readable("template()") ->
+    #{
+        swagger => #{type => string},
+        dashboard => #{type => string, is_template => true},
+        docgen => #{type => "String", desc => ?DESC(template)}
+    };
+readable("template_str()") ->
+    #{
+        swagger => #{type => string},
+        dashboard => #{type => string, is_template => true},
+        docgen => #{type => "String", desc => ?DESC(template)}
+    };
 readable("binary()") ->
     #{
         swagger => #{type => string},

apps/emqx_dashboard/src/emqx_dashboard_schema_api.erl

@@ -44,6 +44,8 @@
     ])
 ).
 
+-define(SCHEMA_VERSION, <<"0.2.0">>).
+
 %%--------------------------------------------------------------------
 %% minirest API and schema
 %%--------------------------------------------------------------------
@@ -97,20 +99,31 @@ gen_schema(connectors) ->
     connectors_schema_json().
 
 hotconf_schema_json() ->
-    SchemaInfo = #{title => <<"EMQX Hot Conf API Schema">>, version => <<"0.1.0">>},
+    SchemaInfo = #{
+        title => <<"Hot Conf Schema">>,
+        version => ?SCHEMA_VERSION
+    },
     gen_api_schema_json_iodata(emqx_mgmt_api_configs, SchemaInfo).
 
 bridge_schema_json() ->
-    SchemaInfo = #{title => <<"EMQX Data Bridge API Schema">>, version => <<"0.1.0">>},
+    SchemaInfo = #{
+        title => <<"Data Bridge Schema">>,
+        version => ?SCHEMA_VERSION
+    },
     gen_api_schema_json_iodata(emqx_bridge_api, SchemaInfo).
 
 actions_schema_json() ->
-    SchemaInfo = #{title => <<"EMQX Data Actions API Schema">>, version => <<"0.1.0">>},
+    SchemaInfo = #{
-    %% Note: this will be moved to `emqx_actions' application in the future.
+        title => <<"Actions and Sources Schema">>,
+        version => ?SCHEMA_VERSION
+    },
     gen_api_schema_json_iodata(emqx_bridge_v2_api, SchemaInfo).
 
 connectors_schema_json() ->
-    SchemaInfo = #{title => <<"EMQX Connectors Schema">>, version => <<"0.1.0">>},
+    SchemaInfo = #{
+        title => <<"Connectors Schema">>,
+        version => ?SCHEMA_VERSION
+    },
     gen_api_schema_json_iodata(emqx_connector_api, SchemaInfo).
 
 gen_api_schema_json_iodata(SchemaMod, SchemaInfo) ->

apps/emqx_dashboard/src/emqx_dashboard_swagger.erl

@@ -57,7 +57,11 @@
     allowEmptyValue,
     deprecated,
     minimum,
-    maximum
+    maximum,
+    %% is_template is a type property,
+    %% but some exceptions are made for them to be field property
+    %% for example, HTTP headers (which is a map type)
+    is_template
 ]).
 
 -define(INIT_SCHEMA, #{
@@ -652,19 +656,6 @@ trans_required(Spec, true, _) -> Spec#{required => true};
 trans_required(Spec, _, path) -> Spec#{required => true};
 trans_required(Spec, _, _) -> Spec.
 
-trans_desc(Init, Hocon, Func, Name, Options) ->
-    Spec0 = trans_description(Init, Hocon, Options),
-    case Func =:= fun hocon_schema_to_spec/2 of
-        true ->
-            Spec0;
-        false ->
-            Spec1 = trans_label(Spec0, Hocon, Name, Options),
-            case Spec1 of
-                #{description := _} -> Spec1;
-                _ -> Spec1
-            end
-    end.
-
 trans_description(Spec, Hocon, Options) ->
     Desc =
         case desc_struct(Hocon) of
@@ -702,19 +693,6 @@ get_i18n_text(Lang, Namespace, Id, Tag, Default) ->
 get_lang(#{i18n_lang := Lang}) -> Lang;
 get_lang(_) -> emqx:get_config([dashboard, i18n_lang]).
 
-trans_label(Spec, Hocon, Default, Options) ->
-    Label =
-        case desc_struct(Hocon) of
-            ?DESC(_, _) = Struct -> get_i18n(<<"label">>, Struct, Default, Options);
-            _ -> Default
-        end,
-    case Label =:= undefined of
-        true ->
-            Spec;
-        false ->
-            Spec#{label => Label}
-    end.
-
 desc_struct(Hocon) ->
     R =
         case hocon_schema:field_schema(Hocon, desc) of
@@ -772,7 +750,7 @@ response(Status, #{content := _} = Content, {Acc, RefsAcc, Module, Options}) ->
 response(Status, ?REF(StructName), {Acc, RefsAcc, Module, Options}) ->
     response(Status, ?R_REF(Module, StructName), {Acc, RefsAcc, Module, Options});
 response(Status, ?R_REF(_Mod, _Name) = RRef, {Acc, RefsAcc, Module, Options}) ->
-    SchemaToSpec = schema_converter(Options),
+    SchemaToSpec = get_schema_converter(Options),
     {Spec, Refs} = SchemaToSpec(RRef, Module),
     Content = content(Spec),
     {
@@ -910,7 +888,7 @@ parse_object(PropList = [_ | _], Module, Options) when is_list(PropList) ->
 parse_object(Other, Module, Options) ->
     erlang:throw(
         {error, #{
-            msg => <<"Object only supports not empty proplists">>,
+            msg => <<"Object only supports non-empty fields list">>,
             args => Other,
             module => Module,
             options => Options
@@ -950,10 +928,10 @@ parse_object_loop([{Name, Hocon} | Rest], Module, Options, Props, Required, Refs
         true ->
             HoconType = hocon_schema:field_schema(Hocon, type),
             Init0 = init_prop([default | ?DEFAULT_FIELDS], #{}, Hocon),
-            SchemaToSpec = schema_converter(Options),
+            SchemaToSpec = get_schema_converter(Options),
             Init = maps:remove(
                 summary,
-                trans_desc(Init0, Hocon, SchemaToSpec, NameBin, Options)
+                trans_description(Init0, Hocon, Options)
             ),
             {Prop, Refs1} = SchemaToSpec(HoconType, Module),
             NewRequiredAcc =
@@ -1002,7 +980,7 @@ to_ref(Mod, StructName, Acc, RefsAcc) ->
     Ref = #{<<"$ref">> => ?TO_COMPONENTS_PARAM(Mod, StructName)},
     {[Ref | Acc], [{Mod, StructName, parameter} | RefsAcc]}.
 
-schema_converter(Options) ->
+get_schema_converter(Options) ->
     maps:get(schema_converter, Options, fun hocon_schema_to_spec/2).
 
 hocon_error_msg(Reason) ->

apps/emqx_dashboard/test/emqx_swagger_requestBody_SUITE.erl

@@ -359,7 +359,7 @@ t_bad_ref(_Config) ->
     Refs = [{?MODULE, bad_ref}],
     Fields = fields(bad_ref),
     ?assertThrow(
-        {error, #{msg := <<"Object only supports not empty proplists">>, args := Fields}},
+        {error, #{msg := <<"Object only supports non-empty fields list">>, args := Fields}},
         validate(Path, Spec, Refs)
     ),
     ok.

apps/emqx_dashboard/test/emqx_swagger_response_SUITE.erl

@@ -189,7 +189,7 @@ t_nest_object(_Config) ->
 t_empty(_Config) ->
     ?assertThrow(
         {error, #{
-            msg := <<"Object only supports not empty proplists">>,
+            msg := <<"Object only supports non-empty fields list">>,
             args := [],
             module := ?MODULE
         }},
@@ -273,7 +273,7 @@ t_bad_ref(_Config) ->
     ?assertThrow(
         {error, #{
             module := ?MODULE,
-            msg := <<"Object only supports not empty proplists">>
+            msg := <<"Object only supports non-empty fields list">>
         }},
         validate(Path, Object, ExpectRefs)
     ),

apps/emqx_resource/src/emqx_resource_manager.erl

@@ -1058,7 +1058,8 @@ channels_health_check(?status_connected = _ConnectorStatus, Data0) ->
         get_config_for_channels(Data0, ChannelsNotAdded),
     Data1 = add_channels_in_list(ChannelsNotAddedWithConfigs, Data0),
     %% Now that we have done the adding, we can get the status of all channels
-    trigger_health_check_for_added_channels(Data1);
+    Data2 = trigger_health_check_for_added_channels(Data1),
+    update_state(Data2, Data0);
 channels_health_check(?status_connecting = _ConnectorStatus, Data0) ->
     %% Whenever the resource is connecting:
     %% 1. Change the status of all added channels to connecting

apps/emqx_s3/src/emqx_s3.app.src

@@ -1,6 +1,6 @@
 {application, emqx_s3, [
     {description, "EMQX S3"},
-    {vsn, "5.0.14"},
+    {vsn, "5.1.0"},
     {modules, []},
     {registered, [emqx_s3_sup]},
     {applications, [

apps/emqx_s3/src/emqx_s3_schema.erl

@@ -74,7 +74,7 @@ fields(s3_upload) ->
     [
         {bucket,
             mk(
-                string(),
+                emqx_schema:template_str(),
                 #{
                     desc => ?DESC("bucket"),
                     required => true
@@ -82,7 +82,7 @@ fields(s3_upload) ->
             )},
         {key,
             mk(
-                string(),
+                emqx_schema:template_str(),
                 #{
                     desc => ?DESC("key"),
                     required => true

rel/i18n/emqx_conf_schema_types.hocon

@@ -9,4 +9,7 @@ emqx_conf_schema_types {
     secret.desc:
     """A string holding some sensitive information, such as a password. When secret starts with <code>file://</code>, the rest of the string is interpreted as a path to a file containing the secret itself: whole content of the file except any trailing whitespace characters is considered a secret value. Note: when clustered, all EMQX nodes should have the same file present before using <code>file://</code> secrets."""
 
+    template.desc: """~
+        A string for `${.path.to.var}` style value interpolation,
+        where the leading dot is optional, and `${.}` represents all values as an object."""
 }

scripts/test/emqx-smoke-test.sh

@@ -82,8 +82,10 @@ main() {
     ## The json status feature was added after hotconf and bridges schema API
     if [ "$JSON_STATUS" != 'NOT_JSON' ]; then
         check_swagger_json
-        check_schema_json hotconf "EMQX Hot Conf API Schema"
+        check_schema_json hotconf "Hot Conf Schema"
-        check_schema_json bridges "EMQX Data Bridge API Schema"
+        check_schema_json bridges "Data Bridge Schema"
+        check_schema_json actions "Actions and Sources Schema"
+        check_schema_json connectors "Connectors Schema"
     fi
 }