fix: some problems in tablestore

apps/emqx_bridge_tablestore/src/emqx_bridge_tablestore.erl

@@ -149,6 +149,7 @@ fields("connector") ->
             emqx_schema_secret:mk(
                 #{
                     required => true,
+                    sensitive => true,
                     desc => ?DESC("desc_access_key_id")
                 }
             )},
@@ -156,6 +157,7 @@ fields("connector") ->
             emqx_schema_secret:mk(
                 #{
                     required => true,
+                    sensitive => true,
                     desc => ?DESC("desc_access_key_secret")
                 }
             )},
@@ -204,12 +206,12 @@ fields(action_parameters) ->
         fields_field(),
         {data_source,
             mk(binary(), #{
-                required => true,
+                required => false,
                 is_template => true,
                 desc => ?DESC("desc_data_source")
             })},
         {timestamp,
-            mk(binary(), #{
+            mk(integer(), #{
                 required => false,
                 is_template => true,
                 desc => ?DESC("desc_timestamp")
@@ -218,7 +220,7 @@ fields(action_parameters) ->
             mk(
                 hoconsc:enum(['MUM_IGNORE', 'MUM_NORMAL']), #{
                     required => false,
-                    default => 'MUM_IGNORE',
+                    default => 'MUM_NORMAL',
                     desc => ?DESC("desc_meta_update_model")
                 }
             )}

apps/emqx_bridge_tablestore/src/emqx_bridge_tablestore_connector.erl

@@ -3,7 +3,6 @@
 %%--------------------------------------------------------------------
 -module(emqx_bridge_tablestore_connector).
 
--include("emqx_bridge_tablestore.hrl").
 -include_lib("emqx/include/logger.hrl").
 -include_lib("snabbkaffe/include/snabbkaffe.hrl").
 -include_lib("emqx_connector/include/emqx_connector.hrl").
@@ -20,6 +19,7 @@
     on_add_channel/4,
     on_remove_channel/3,
     on_get_channel_status/3,
+    on_get_channels/1,
     on_query/3,
     on_batch_query/3,
     on_get_status/2
@@ -52,7 +52,7 @@ on_add_channel(_InstId, #{channels := Channels} = OldState, ChannelId, Conf) ->
                 measurement => maybe_preproc(maps:get(measurement, Params)),
                 tags => preproc_tags(maps:get(tags, Params)),
                 fields => preproc_fields(maps:get(fields, Params)),
-                data_source => maybe_preproc(maps:get(data_source, Params)),
+                data_source => maybe_preproc(maps:get(data_source, Params, <<>>)),
                 meta_update_model => maps:get(meta_update_model, Params)
             }
         }
@@ -67,20 +67,25 @@ on_remove_channel(_InstId, #{channels := Channels} = State, ChannelId) ->
 on_get_channel_status(InstId, _ChannelId, State) ->
     on_get_status(InstId, State).
 
+on_get_channels(InstId) ->
+    emqx_bridge_influxdb_connector:on_get_channels(InstId).
+
 on_start(InstId, Config) ->
-    OtsOpts = [
+    BaseOpts = [
         {instance, maps:get(instance_name, Config)},
         {pool, ?OTS_CLIENT_NAME(InstId)},
         {endpoint, maps:get(endpoint, Config)},
-        {access_key, emqx_secret:unwrap(maps:get(access_key_id, Config))},
-        {access_secret, emqx_secret:unwrap(maps:get(access_key_secret, Config))},
         {pool_size, maps:get(pool_size, Config)}
     ],
-    ?LOG_T(info, #{msg => ots_start, opts => OtsOpts}),
-    {ok, ClientRef} = start_ots_ts_client(InstId, OtsOpts),
+    SecretOpts = [
+        {access_key, maps:get(access_key_id, Config)},
+        {access_secret, maps:get(access_key_secret, Config)}
+    ],
+    ?LOG_T(info, #{msg => ots_start, ots_opts => BaseOpts}),
+    {ok, ClientRef} = start_ots_ts_client(InstId, BaseOpts ++ SecretOpts),
     case list_ots_tables(ClientRef) of
         {ok, _} ->
-            {ok, #{client_ref => ClientRef, channels => #{}, ots_opts => OtsOpts}};
+            {ok, #{client_ref => ClientRef, channels => #{}, ots_opts => BaseOpts}};
         {error, Reason} ->
             _ = ots_ts_client:stop(ClientRef),
             {error, Reason}
@@ -273,7 +278,8 @@ do_mk_tablestore_data_row(Message, ChannelState, Measurement) ->
             <<"now">> -> os:system_time(microsecond);
             <<"NOW">> -> os:system_time(microsecond);
             undefined -> os:system_time(microsecond);
-            Ts1 -> Ts1
+            Ts1 when is_integer(Ts1) -> Ts1;
+            Ts2 -> throw({bad_ots_data, {bad_timestamp, Ts2}})
         end,
     #{
         measurement => Measurement,

apps/emqx_bridge_tablestore/test/emqx_bridge_tablestore_connector_tests.erl

@@ -1,7 +1,6 @@
 -module(emqx_bridge_tablestore_connector_tests).
 
 -include_lib("eunit/include/eunit.hrl").
--include_lib("emqx_bridge_tablestore/include/emqx_bridge_tablestore.hrl").
 
 -define(CONF, #{
     instance_name => <<"instance">>,

apps/emqx_utils/src/emqx_utils_redact.erl

@@ -41,6 +41,12 @@ is_sensitive_key(<<"secret">>) -> true;
 is_sensitive_key(secret_access_key) -> true;
 is_sensitive_key("secret_access_key") -> true;
 is_sensitive_key(<<"secret_access_key">>) -> true;
+is_sensitive_key(access_key_secret) -> true;
+is_sensitive_key("access_key_secret") -> true;
+is_sensitive_key(<<"access_key_secret">>) -> true;
+is_sensitive_key(access_key_id) -> true;
+is_sensitive_key("access_key_id") -> true;
+is_sensitive_key(<<"access_key_id">>) -> true;
 is_sensitive_key(secret_key) -> true;
 is_sensitive_key("secret_key") -> true;
 is_sensitive_key(<<"secret_key">>) -> true;

rel/i18n/emqx_bridge_tablestore.hocon

@@ -33,62 +33,72 @@ pool_size.desc:
 desc_table_name.label:
 """Table Name"""
 desc_table_name.desc:
-"""Table name."""
+"""Table name. It can either be a static value or a placeholder like `${payload.table_name}`."""
 
 desc_measurement.label:
 """Measurement"""
 desc_measurement.desc:
-"""The measurement."""
+"""The measurement. It can either be a static value or a placeholder like `${payload.measurement}`."""
 
 desc_data_source.label:
 """Data Source"""
 desc_data_source.desc:
-"""The data source."""
+"""The data source. It can either be a static value or a placeholder like `${payload.data_source}`."""
 
 desc_timestamp.label:
 """Timestamp"""
 desc_timestamp.desc:
-"""The timestamp."""
+"""~
+The timestamp in microsecond of the field.
+It can either be a static value or a placeholder like `${payload.microsecond_timestamp}`.
+If not provided or set to `NOW`, the millisecond timestamp when EMQX writes to Tablestore will be used.~"""
 
 desc_meta_update_model.label:
 """Meta Update Model"""
 desc_meta_update_model.desc:
-"""The meta-update-module."""
+"""~
+The update mode for time-series metadata. Can be one of:
+- MUM_NORMAL: Normal mode. When sending messages in this mode, Tablestore will create the timeseries metadata if not exits.
+- MUM_IGNORE: Do not update metadata. When sending messages in this mode, Tablestore will not try to create the timeseries metadata.
+Defaults to MUM_NORMAL~"""
 
 tablestore_fields_column.label:
 """Column"""
 tablestore_fields_column.desc:
-"""Column name of the field"""
+"""Column name of the field. It can either be a static value or a placeholder like `${payload.column}`"""
 
 tablestore_fields_value.label:
 """Value"""
 tablestore_fields_value.desc:
-"""Value of the field"""
+"""Value of the field. It can either be a static value or a placeholder like `${payload.value}`"""
 
 tablestore_fields_isint.label:
 """Is Int"""
 tablestore_fields_isint.desc:
-"""Whether try to write numeric value as integer. Defaults to false, means that write integers as floats."""
+"""~
+Whether try to write numeric value as `integer`. Defaults to `false`, means that write integers as floats.
+It can either be a static value or a placeholder like `${payload.is_int}`.~"""
 
 tablestore_fields_isbinary.label:
 """Is Binary"""
 tablestore_fields_isbinary.desc:
-"""Whether try to write binary values as binary type. Defaults to false, means that write binary values as strings."""
+"""~Whether try to write binary values as `binary` type. Defaults to `false`, means that write binary values as strings.
+It can either be a static value or a placeholder like `${payload.is_binary}`.~"""
 
 storage_model_type.label:
 """Storage Model Type"""
 storage_model_type.desc:
-"""Storage model type. Can be one of "timeseries" or "order"."""
+"""Storage model type. Can be one of `timeseries` or `order`."""
 
 desc_tags.label:
 """Tags"""
 desc_tags.desc:
-"""Tags"""
+"""Tags. The tag key and tag value can either be static strings or a placeholder like `${payload.tag_key}` and `${payload.tag_value}`."""
 
 desc_fields.label:
 """Fields"""
 desc_fields.desc:
-"""Fields"""
+"""Fields. The field column and value can either be static values or a placeholder like `${payload.column_name}` and `${payload.column_value}`."""
 
 connector.label:
 """Tablestore Connector Configuration"""

scripts/spellcheck/dicts/emqx.txt

@@ -79,7 +79,6 @@ XMLs
 acceptors
 ack
 acked
-ACL
 addr
 AES
 aliyun