feat: support tablestore data integration

apps/emqx_bridge_tablestore/.gitignore

@@ -0,0 +1,19 @@
+.rebar3
+_*
+.eunit
+*.o
+*.beam
+*.plt
+*.swp
+*.swo
+.erlang.cookie
+ebin
+log
+erl_crash.dump
+.rebar
+logs
+_build
+.idea
+*.iml
+rebar3.crashdump
+*~

apps/emqx_bridge_tablestore/BSL.txt

@@ -0,0 +1,94 @@
+Business Source License 1.1
+
+Licensor:             Hangzhou EMQ Technologies Co., Ltd.
+Licensed Work:        EMQX Enterprise Edition
+                      The Licensed Work is (c) 2023
+                      Hangzhou EMQ Technologies Co., Ltd.
+Additional Use Grant: Students and educators are granted right to copy,
+                      modify, and create derivative work for research
+                      or education.
+Change Date:          2028-01-26
+Change License:       Apache License, Version 2.0
+
+For information about alternative licensing arrangements for the Software,
+please contact Licensor: https://www.emqx.com/en/contact
+
+Notice
+
+The Business Source License (this document, or the “License”) is not an Open
+Source license. However, the Licensed Work will eventually be made available
+under an Open Source License, as stated in this License.
+
+License text copyright (c) 2017 MariaDB Corporation Ab, All Rights Reserved.
+“Business Source License” is a trademark of MariaDB Corporation Ab.
+
+-----------------------------------------------------------------------------
+
+Business Source License 1.1
+
+Terms
+
+The Licensor hereby grants you the right to copy, modify, create derivative
+works, redistribute, and make non-production use of the Licensed Work. The
+Licensor may make an Additional Use Grant, above, permitting limited
+production use.
+
+Effective on the Change Date, or the fourth anniversary of the first publicly
+available distribution of a specific version of the Licensed Work under this
+License, whichever comes first, the Licensor hereby grants you rights under
+the terms of the Change License, and the rights granted in the paragraph
+above terminate.
+
+If your use of the Licensed Work does not comply with the requirements
+currently in effect as described in this License, you must purchase a
+commercial license from the Licensor, its affiliated entities, or authorized
+resellers, or you must refrain from using the Licensed Work.
+
+All copies of the original and modified Licensed Work, and derivative works
+of the Licensed Work, are subject to this License. This License applies
+separately for each version of the Licensed Work and the Change Date may vary
+for each version of the Licensed Work released by Licensor.
+
+You must conspicuously display this License on each original or modified copy
+of the Licensed Work. If you receive the Licensed Work in original or
+modified form from a third party, the terms and conditions set forth in this
+License apply to your use of that work.
+
+Any use of the Licensed Work in violation of this License will automatically
+terminate your rights under this License for the current and all other
+versions of the Licensed Work.
+
+This License does not grant you any right in any trademark or logo of
+Licensor or its affiliates (provided that you may use a trademark or logo of
+Licensor as expressly required by this License).
+
+TO THE EXTENT PERMITTED BY APPLICABLE LAW, THE LICENSED WORK IS PROVIDED ON
+AN “AS IS” BASIS. LICENSOR HEREBY DISCLAIMS ALL WARRANTIES AND CONDITIONS,
+EXPRESS OR IMPLIED, INCLUDING (WITHOUT LIMITATION) WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, NON-INFRINGEMENT, AND
+TITLE.
+
+MariaDB hereby grants you permission to use this License’s text to license
+your works, and to refer to it using the trademark “Business Source License”,
+as long as you comply with the Covenants of Licensor below.
+
+Covenants of Licensor
+
+In consideration of the right to use this License’s text and the “Business
+Source License” name and trademark, Licensor covenants to MariaDB, and to all
+other recipients of the licensed work to be provided by Licensor:
+
+1. To specify as the Change License the GPL Version 2.0 or any later version,
+   or a license that is compatible with GPL Version 2.0 or a later version,
+   where “compatible” means that software provided under the Change License can
+   be included in a program with software provided under GPL Version 2.0 or a
+   later version. Licensor may specify additional Change Licenses without
+   limitation.
+
+2. To either: (a) specify an additional grant of rights to use that does not
+   impose any additional restriction on the right granted in this License, as
+   the Additional Use Grant; or (b) insert the text “None”.
+
+3. To specify a Change Date.
+
+4. Not to modify this License in any other way.

apps/emqx_bridge_tablestore/LICENSE

@@ -0,0 +1,191 @@
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   Copyright 2024, Shawn <506895667@qq.com>.
+
+   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.
+

apps/emqx_bridge_tablestore/README.md

@@ -0,0 +1,9 @@
+emqx_bridge_tablestore
+=====
+
+An OTP library
+
+Build
+-----
+
+    $ rebar3 compile

apps/emqx_bridge_tablestore/docker-ct

@@ -0,0 +1,2 @@
+toxiproxy
+datalayers

apps/emqx_bridge_tablestore/include/emqx_bridge_tablestore.hrl

@@ -0,0 +1 @@
+-define(CLIENT_REF_FOR_TEST, dummy_client_ref).

apps/emqx_bridge_tablestore/mix.exs

@@ -0,0 +1,32 @@
+defmodule EMQXBridgeTablestore.MixProject do
+  use Mix.Project
+  alias EMQXUmbrella.MixProject, as: UMP
+
+  def project do
+    [
+      app: :emqx_bridge_tablestore,
+      version: "0.1.0",
+      build_path: "../../_build",
+      erlc_options: UMP.erlc_options(),
+      erlc_paths: UMP.erlc_paths(),
+      deps_path: "../../deps",
+      lockfile: "../../mix.lock",
+      elixir: "~> 1.14",
+      start_permanent: Mix.env() == :prod,
+      deps: deps()
+    ]
+  end
+
+  def application do
+    [extra_applications: UMP.extra_applications()]
+  end
+
+  def deps() do
+    [
+      UMP.common_dep(:tablestore),
+      {:emqx_connector, in_umbrella: true, runtime: false},
+      {:emqx_resource, in_umbrella: true},
+      {:emqx_bridge, in_umbrella: true, runtime: false}
+    ]
+  end
+end

apps/emqx_bridge_tablestore/rebar.config

@@ -0,0 +1,7 @@
+{erl_opts, [debug_info]}.
+{deps, [
+    {ots_erl, {git, "https://github.com/emqx/ots_erl.git", {tag, "0.2.2"}}},
+    {emqx_connector, {path, "../../apps/emqx_connector"}},
+    {emqx_resource, {path, "../../apps/emqx_resource"}},
+    {emqx_bridge, {path, "../../apps/emqx_bridge"}}
+]}.

apps/emqx_bridge_tablestore/src/emqx_bridge_tablestore.app.src

@@ -0,0 +1,16 @@
+{application, emqx_bridge_tablestore, [
+    {description, "EMQX Enterprise Tablestore Bridge"},
+    {vsn, "0.1.0"},
+    {registered, []},
+    {applications, [
+        kernel,
+        stdlib,
+        ots_erl
+    ]},
+    {env, [
+        {emqx_action_info_modules, [emqx_bridge_tablestore_action_info]},
+        {emqx_connector_info_modules, [emqx_bridge_tablestore_connector_info]}
+    ]},
+    {modules, []},
+    {links, []}
+]}.

apps/emqx_bridge_tablestore/src/emqx_bridge_tablestore.erl

@@ -0,0 +1,274 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+-module(emqx_bridge_tablestore).
+
+-behaviour(emqx_connector_examples).
+
+-include_lib("emqx/include/logger.hrl").
+-include_lib("emqx_connector/include/emqx_connector.hrl").
+-include_lib("typerefl/include/types.hrl").
+-include_lib("hocon/include/hoconsc.hrl").
+
+-import(hoconsc, [mk/2, enum/1, ref/2]).
+
+-export([
+    namespace/0,
+    roots/0,
+    fields/1,
+    desc/1
+]).
+
+%% Examples
+-export([
+    bridge_v2_examples/1,
+    conn_bridge_examples/1,
+    connector_examples/1
+]).
+
+-define(CONNECTOR_TYPE, tablestore).
+-define(ACTION_TYPE, tablestore).
+
+%% Examples
+conn_bridge_examples(Method) ->
+    [
+        #{
+            <<"tablestore_timeseries">> => #{
+                summary => <<"Tablestore Timeseries Bridge">>,
+                value => values(timeseries, Method)
+            }
+        }
+    ].
+
+bridge_v2_examples(Method) ->
+    [
+        #{
+            <<"tablestore_timeseries">> => #{
+                summary => <<"Tablestore Timeseries Action">>,
+                value => emqx_bridge_v2_schema:action_values(
+                    Method, tablestore, tablestore, action_parameters_example(timeseries)
+                )
+            }
+        }
+    ].
+
+connector_examples(Method) ->
+    [
+        #{
+            <<"tablestore_timeseries">> => #{
+                summary => <<"Tablestore Timeseries Connector">>,
+                value => emqx_connector_schema:connector_values(
+                    Method, tablestore, connector_values(timeseries)
+                )
+            }
+        }
+    ].
+
+connector_values(timeseries) ->
+    #{
+        name => <<"tablestore_connector">>,
+        enable => true,
+        endpoint => <<"https://myinstance.cn-hangzhou.ots.aliyuncs.com">>,
+        storage_model_type => timeseries,
+        instance_name => <<"myinstance">>,
+        access_key_id => <<"******">>,
+        access_key_secret => <<"******">>
+    }.
+
+action_parameters_example(timeseries) ->
+    #{
+        parameters => #{
+            storage_model_type => timeseries,
+            table_name => <<"table_name">>,
+            data_source => <<"${data_source}">>,
+            measurement => <<"${measurement}">>,
+            tags => #{
+                tag1 => <<"${tag1}">>,
+                tag2 => <<"${tag2}">>
+            },
+            fields => [
+                #{
+                    column => <<"${column}">>,
+                    value => <<"${value}">>,
+                    isint => true
+                }
+            ],
+            meta_update_model => <<"MUM_IGNORE">>
+        }
+    }.
+
+values(StorageType, get) ->
+    values(StorageType, post);
+values(StorageType, put) ->
+    values(StorageType, post);
+values(timeseries, post) ->
+    #{
+        name => <<"tablestore_connector">>,
+        enable => true,
+        local_topic => <<"local/topic/#">>,
+        endpoint => <<"https://myinstance.cn-hangzhou.ots.aliyuncs.com">>,
+        storage_model_type => timeseries,
+        instance_name => <<"myinstance">>,
+        access_key_id => <<"******">>,
+        access_key_secret => <<"******">>,
+        table_name => <<"table_name">>,
+        measurement => <<"measurement">>,
+        tags => #{
+            tag1 => <<"${tag1}">>,
+            tag2 => <<"${tag2}">>
+        },
+        fields => [
+            #{
+                column => <<"${column}">>,
+                value => <<"${value}">>,
+                isint => true
+            }
+        ],
+        data_source => <<"${data_source}">>,
+        meta_update_model => <<"MUM_IGNORE">>,
+        resource_opts => #{
+            batch_size => 1,
+            batch_time => <<"20ms">>
+        },
+        pool_size => 8,
+        ssl => #{enable => false}
+    }.
+
+%% -------------------------------------------------------------------------------------------------
+%% Hocon Schema Definitions
+namespace() -> "bridge_tablestore".
+
+roots() -> [].
+
+fields("connector") ->
+    [
+        storage_model_type_field(),
+        {endpoint, mk(binary(), #{required => true, desc => ?DESC("desc_endpoint")})},
+        {instance_name, mk(binary(), #{required => true, desc => ?DESC("desc_instance_name")})},
+        {access_key_id, mk(binary(), #{required => true, desc => ?DESC("desc_access_key_id")})},
+        {access_key_secret,
+            mk(binary(), #{required => true, desc => ?DESC("desc_access_key_secret")})},
+        {pool_size,
+            mk(
+                integer(),
+                #{
+                    required => false,
+                    default => 8,
+                    desc => ?DESC("pool_size")
+                }
+            )}
+    ] ++ emqx_connector_schema_lib:ssl_fields();
+fields("config_connector") ->
+    emqx_connector_schema:common_fields() ++
+        fields("connector") ++
+        emqx_connector_schema:resource_opts_ref(?MODULE, connector_resource_opts);
+fields(action) ->
+    {tablestore,
+        mk(
+            hoconsc:map(name, ref(?MODULE, tablestore_action)),
+            #{desc => <<"Tablestore Action Config">>, required => false}
+        )};
+fields(tablestore_action) ->
+    emqx_bridge_v2_schema:make_producer_action_schema(
+        mk(ref(?MODULE, action_parameters), #{
+            required => true, desc => ?DESC(action_parameters)
+        })
+    );
+fields(action_parameters) ->
+    [
+        storage_model_type_field(),
+        {table_name, mk(binary(), #{required => true, desc => ?DESC("desc_table_name")})},
+        {measurement, mk(binary(), #{required => true, desc => ?DESC("desc_measurement")})},
+        tags_field(),
+        fields_field(),
+        {data_source, mk(binary(), #{required => true, desc => ?DESC("desc_data_source")})},
+        {timestamp, mk(binary(), #{required => false, desc => ?DESC("desc_timestamp")})},
+        {meta_update_model,
+            mk(
+                hoconsc:enum(['MUM_IGNORE', 'MUM_NORMAL']), #{
+                    required => false,
+                    default => 'MUM_IGNORE',
+                    desc => ?DESC("desc_meta_update_model")
+                }
+            )}
+    ];
+fields(connector_resource_opts) ->
+    emqx_connector_schema:resource_opts_fields();
+fields(Field) when
+    Field == "get_connector";
+    Field == "put_connector";
+    Field == "post_connector"
+->
+    Fields =
+        fields("connector") ++
+            emqx_connector_schema:resource_opts_ref(?MODULE, connector_resource_opts),
+    emqx_connector_schema:api_fields(Field, ?CONNECTOR_TYPE, Fields);
+fields(Field) when
+    Field == "get_bridge_v2";
+    Field == "post_bridge_v2";
+    Field == "put_bridge_v2"
+->
+    emqx_bridge_v2_schema:api_fields(Field, ?ACTION_TYPE, fields(tablestore_action));
+fields("tablestore_fields") ->
+    [
+        {column,
+            mk(binary(), #{
+                required => true,
+                desc => ?DESC("tablestore_fields_column")
+            })},
+        {value,
+            mk(hoconsc:union([binary(), boolean(), number()]), #{
+                required => true,
+                desc => ?DESC("tablestore_fields_value")
+            })},
+        {isint,
+            mk(boolean(), #{
+                required => false,
+                desc => ?DESC("tablestore_fields_isint")
+            })},
+        {isbinary,
+            mk(boolean(), #{
+                required => false,
+                desc => ?DESC("tablestore_fields_isbinary")
+            })}
+    ].
+
+storage_model_type_field() ->
+    {storage_model_type,
+        mk(
+            hoconsc:enum([timeseries]), #{
+                required => false,
+                default => timeseries,
+                desc => ?DESC("storage_model_type")
+            }
+        )}.
+
+tags_field() ->
+    {tags,
+        mk(
+            map(), #{
+                default => #{},
+                desc => ?DESC("desc_tags"),
+                is_template => true
+            }
+        )}.
+
+fields_field() ->
+    {fields,
+        mk(
+            hoconsc:array(ref(?MODULE, "tablestore_fields")), #{
+                default => [],
+                desc => ?DESC("desc_fields")
+            }
+        )}.
+
+desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->
+    ["Configuration for Tablestore using `", string:to_upper(Method), "` method."];
+desc("connector") ->
+    ?DESC("connector");
+desc("config_connector") ->
+    ?DESC("desc_config");
+desc(connector_resource_opts) ->
+    ?DESC(emqx_resource_schema, "resource_opts");
+desc(_) ->
+    undefined.

apps/emqx_bridge_tablestore/src/emqx_bridge_tablestore_action_info.erl

@@ -0,0 +1,20 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+-module(emqx_bridge_tablestore_action_info).
+
+-behaviour(emqx_action_info).
+
+-export([
+    action_type_name/0,
+    connector_type_name/0,
+    schema_module/0
+]).
+
+-define(SCHEMA_MODULE, emqx_bridge_tablestore).
+
+action_type_name() -> tablestore.
+
+connector_type_name() -> tablestore.
+
+schema_module() -> ?SCHEMA_MODULE.

apps/emqx_bridge_tablestore/src/emqx_bridge_tablestore_connector.erl

@@ -0,0 +1,306 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+-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").
+-include_lib("emqx_resource/include/emqx_resource.hrl").
+
+-behaviour(emqx_resource).
+
+%% callbacks of behaviour emqx_resource
+-export([
+    resource_type/0,
+    callback_mode/0,
+    on_start/2,
+    on_stop/2,
+    on_add_channel/4,
+    on_remove_channel/3,
+    on_get_channel_status/3,
+    on_query/3,
+    on_batch_query/3,
+    on_get_status/2
+]).
+
+-export([
+    mk_tablestore_data/3,
+    mk_tablestore_batch_data/2
+]).
+
+-define(OTS_CLIENT_NAME(ID), list_to_binary("tablestore:" ++ str(ID))).
+-define(TKS(STR), {tmpl_tokens, STR}).
+
+%%--------------------------------------------------------------------
+%% resource callback
+
+resource_type() -> tablestore.
+
+callback_mode() -> always_sync.
+
+on_add_channel(_InstId, #{channels := Channels} = OldState, ChannelId, Conf) ->
+    #{parameters := #{storage_model_type := timeseries} = Params} = Conf,
+    Channels1 = maps:get(ChannelId, Channels, #{}),
+    NewState = OldState#{
+        channels => Channels1#{
+            ChannelId => #{
+                timestamp => maybe_preproc(maps:get(timestamp, Params, <<"now">>)),
+                table_name => maybe_preproc(maps:get(table_name, Params)),
+                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)),
+                meta_update_model => maps:get(meta_update_model, Params)
+            }
+        }
+    },
+    {ok, NewState}.
+
+on_remove_channel(_InstId, #{channels := Channels} = State, ChannelId) ->
+    {ok, State#{
+        channels => maps:remove(ChannelId, Channels)
+    }}.
+
+on_get_channel_status(InstId, _ChannelId, State) ->
+    on_get_status(InstId, State).
+
+on_start(?CLIENT_REF_FOR_TEST, Config) ->
+    {ok, #{client_ref => ?CLIENT_REF_FOR_TEST, channels => #{}, config => Config}};
+on_start(InstId, Config) ->
+    OtsOpts = [
+        {instance, maps:get(instance_name, Config)},
+        {pool, ?OTS_CLIENT_NAME(InstId)},
+        {endpoint, maps:get(endpoint, Config)},
+        {access_key, maps:get(access_key_id, Config)},
+        {access_secret, maps:get(access_key_secret, Config)},
+        {pool_size, maps:get(pool_size, Config)}
+    ],
+    {ok, ClientRef} = ots_ts_client:start(OtsOpts),
+    case list_ots_tables(ClientRef) of
+        {ok, _} ->
+            {ok, #{client_ref => ClientRef, channels => #{}, config => Config}};
+        {error, Reason} ->
+            _ = ots_ts_client:stop(ClientRef),
+            {error, Reason}
+    end.
+
+on_stop(_InstId, #{client_ref := ?CLIENT_REF_FOR_TEST} = State) ->
+    {ok, State};
+on_stop(_InstId, #{client_ref := ClientRef} = State) ->
+    ots_ts_client:stop(ClientRef),
+    State.
+
+on_get_status(_InstId, #{client_ref := ?CLIENT_REF_FOR_TEST}) ->
+    ?status_connected;
+on_get_status(_InstId, #{client_ref := ClientRef}) ->
+    case list_ots_tables(ClientRef) of
+        {ok, _} -> ?status_connected;
+        _ -> ?status_connecting
+    end.
+
+on_query(_InstId, {ChannelId, Message}, #{client_ref := ClientRef, channels := Channels}) ->
+    case maps:find(ChannelId, Channels) of
+        {ok, #{table_name := TableName0, meta_update_model := MetaUpdateMode} = ChannelState} ->
+            try
+                Row = mk_tablestore_data_row(Message, ChannelState),
+                TableName = render_table_name(TableName0, Message),
+                mk_tablestore_data(TableName, MetaUpdateMode, [Row])
+            of
+                Data ->
+                    LogMsg = #{msg => ots_query, channel => ChannelId, data => Data},
+                    ?SLOG(debug, LogMsg, #{tag => "TABLE_STORE"}),
+                    case ots_ts_client:put(ClientRef, Data) of
+                        {ok, _Res} -> ok;
+                        {error, Reason} -> {error, {unrecoverable_error, Reason}}
+                    end
+            catch
+                throw:{bad_ots_data, _} = Reason ->
+                    {error, {unrecoverable_error, Reason}}
+            end;
+        error ->
+            {error, {unrecoverable_error, channel_not_found}}
+    end.
+
+on_batch_query(_, [{ChannelId, _} | _] = MsgList, #{client_ref := ClientRef, channels := Channels}) ->
+    case maps:find(ChannelId, Channels) of
+        {ok, ChannelState} ->
+            try mk_tablestore_batch_data(MsgList, ChannelState) of
+                BatchDataList ->
+                    send_batch_data(BatchDataList, ClientRef, ChannelId)
+            catch
+                throw:{bad_ots_data, _} = Reason ->
+                    {error, {unrecoverable_error, Reason}}
+            end;
+        error ->
+            {error, {unrecoverable_error, channel_not_found}}
+    end.
+
+send_batch_data(BatchDataList, ClientRef, ChannelId) ->
+    LogMsg = #{msg => ots_batch_query, channel => ChannelId, batch_data_list => BatchDataList},
+    ?SLOG(debug, LogMsg, #{tag => "TABLE_STORE"}),
+    Res = [ots_ts_client:put(ClientRef, BatchData) || BatchData <- BatchDataList],
+    Filter = fun
+        ({error, _}) -> true;
+        (_) -> false
+    end,
+    case lists:filter(Filter, Res) of
+        [] -> ok;
+        Errors -> {error, {unrecoverable_error, Errors}}
+    end.
+
+%%--------------------------------------------------------------------
+%% Internal Functions
+%%--------------------------------------------------------------------
+list_ots_tables(?CLIENT_REF_FOR_TEST) ->
+    {ok, []};
+list_ots_tables(ClientRef) ->
+    ots_ts_client:list_tables(ClientRef).
+
+preproc_tags(Tags) ->
+    maps:fold(
+        fun(K, V, AccIn) ->
+            AccIn#{maybe_preproc(bin(K)) => maybe_preproc(V)}
+        end,
+        #{},
+        Tags
+    ).
+
+preproc_fields(Fields) ->
+    lists:map(
+        fun(#{column := C, value := V} = Row0) ->
+            #{
+                column => maybe_preproc(C),
+                value => maybe_preproc(V),
+                isint => maybe_preproc(maps:get(isint, Row0, undefined)),
+                isbinary => maybe_preproc(maps:get(isbinary, Row0, undefined))
+            }
+        end,
+        Fields
+    ).
+
+render_table_name(TableName, Message) ->
+    case render_tmpl(TableName, Message) of
+        undefined -> throw({bad_ots_data, no_table_name});
+        TName -> TName
+    end.
+
+render_tags(Tags, Message) ->
+    maps:fold(
+        fun(K, V, AccIn) ->
+            case {render_tmpl(K, Message), render_tmpl(V, Message)} of
+                {Key, Value} when Key =/= undefined, Value =/= undefined ->
+                    AccIn#{Key => Value};
+                _ ->
+                    AccIn
+            end
+        end,
+        #{},
+        Tags
+    ).
+
+render_fields(Fields, Message) ->
+    lists:filtermap(
+        fun(#{column := Column, value := Value} = Row) ->
+            case {render_tmpl(Column, Message), render_tmpl(Value, Message)} of
+                {Col, Val} when Col =/= undefined, Val =/= undefined ->
+                    {true, {Col, Val, field_opts([isint, isbinary], Row, Message, #{})}};
+                {_Col, _Val} ->
+                    false
+            end
+        end,
+        Fields
+    ).
+
+field_opts([Key | Keys], Row, Message, Opts) ->
+    case render_tmpl(maps:get(Key, Row, undefined), Message) of
+        undefined ->
+            field_opts(Keys, Row, Message, Opts);
+        Val ->
+            field_opts(Keys, Row, Message, Opts#{Key => Val})
+    end;
+field_opts([], _, _, Opts) ->
+    Opts.
+
+maybe_preproc(Str) when is_binary(Str) ->
+    case string:find(Str, "${") of
+        nomatch -> Str;
+        _ -> ?TKS(emqx_placeholder:preproc_tmpl(Str))
+    end;
+maybe_preproc(Any) ->
+    Any.
+
+mk_tablestore_batch_data(MsgList, #{table_name := TableName0} = ChannelState) ->
+    #{meta_update_model := MetaUpdateMode} = ChannelState,
+    GrpRows = lists:foldr(
+        fun({_, Message}, Res) ->
+            TableName = render_table_name(TableName0, Message),
+            Row = mk_tablestore_data_row(Message, ChannelState),
+            Res#{TableName => [Row | maps:get(TableName, Res, [])]}
+        end,
+        #{},
+        MsgList
+    ),
+    [
+        mk_tablestore_data(TableName, MetaUpdateMode, Rows)
+     || {TableName, Rows} <- maps:to_list(GrpRows)
+    ].
+
+mk_tablestore_data(TableName, MetaUpdateMode, Rows) ->
+    #{
+        table_name => TableName,
+        rows_data => Rows,
+        meta_update_mode => MetaUpdateMode
+    }.
+
+mk_tablestore_data_row(Message, #{measurement := Measurement0} = ChannelState) ->
+    Measurement = render_tmpl(Measurement0, Message),
+    do_mk_tablestore_data_row(Message, ChannelState, Measurement).
+
+do_mk_tablestore_data_row(_, _, undefined) ->
+    throw({bad_ots_data, no_measurement});
+do_mk_tablestore_data_row(Message, ChannelState, Measurement) ->
+    #{tags := Tags0, fields := Fields0, data_source := DataSource0, timestamp := Ts0} =
+        ChannelState,
+    DataSource = trans_data_source(render_tmpl(DataSource0, Message)),
+    Tags = render_tags(Tags0, Message),
+    Fields = render_fields(Fields0, Message),
+    Timestamp =
+        case render_tmpl(Ts0, Message) of
+            <<"now">> -> os:system_time(microsecond);
+            undefined -> os:system_time(microsecond);
+            Ts1 -> Ts1
+        end,
+    #{
+        measurement => Measurement,
+        data_source => DataSource,
+        tags => Tags,
+        fields => Fields,
+        time => Timestamp
+    }.
+
+trans_data_source(undefined) -> <<>>;
+trans_data_source(DataSource) -> DataSource.
+
+render_tmpl(?TKS(Tokens), Message) ->
+    do_render_tmpl(Tokens, Message);
+render_tmpl(Val, _) ->
+    Val.
+
+do_render_tmpl(Tokens, Message) ->
+    RawResult = emqx_placeholder:proc_tmpl(Tokens, Message, #{return => rawlist}),
+    filter_vars(RawResult).
+
+filter_vars([undefined]) ->
+    undefined;
+filter_vars([RawResult]) ->
+    RawResult;
+filter_vars(RawResult) when is_list(RawResult) ->
+    erlang:iolist_to_binary([str(R) || R <- RawResult]).
+
+str(Data) ->
+    emqx_utils_conv:str(Data).
+
+bin(Data) ->
+    emqx_utils_conv:bin(Data).

apps/emqx_bridge_tablestore/src/emqx_bridge_tablestore_connector_info.erl

@@ -0,0 +1,43 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_bridge_tablestore_connector_info).
+
+-behaviour(emqx_connector_info).
+
+-export([
+    type_name/0,
+    bridge_types/0,
+    resource_callback_module/0,
+    config_schema/0,
+    schema_module/0,
+    api_schema/1
+]).
+
+type_name() ->
+    tablestore.
+
+bridge_types() ->
+    [tablestore, tablestore_timeseries].
+
+resource_callback_module() ->
+    emqx_bridge_tablestore_connector.
+
+config_schema() ->
+    {tablestore,
+        hoconsc:mk(
+            hoconsc:map(name, hoconsc:ref(emqx_bridge_tablestore, "config_connector")),
+            #{
+                desc => <<"Tablestore Connector Config">>,
+                required => false
+            }
+        )}.
+
+schema_module() ->
+    emqx_bridge_tablestore.
+
+api_schema(Method) ->
+    emqx_connector_schema:api_ref(
+        emqx_bridge_tablestore, <<"tablestore">>, Method ++ "_connector"
+    ).

apps/emqx_machine/priv/reboot_lists.eterm

@@ -118,6 +118,7 @@
             emqx_bridge_rabbitmq,
             emqx_bridge_azure_event_hub,
             emqx_bridge_datalayers,
+            emqx_bridge_tablestore,
             emqx_s3,
             emqx_bridge_s3,
             emqx_bridge_azure_blob_storage,

apps/emqx_machine/src/emqx_machine.app.src

@@ -3,7 +3,7 @@
     {id, "emqx_machine"},
     {description, "The EMQX Machine"},
     % strict semver, bump manually!
-    {vsn, "0.3.7"},
+    {vsn, "0.3.8"},
     {modules, []},
     {registered, []},
     {applications, [kernel, stdlib, emqx_ctl, redbug]},

changes/ee/feat-14410.en.md

@@ -0,0 +1 @@
+Data integration adds support for the [Aliyun Tablestore](https://cn.aliyun.com/product/ots).

mix.exs

@@ -269,6 +269,9 @@ defmodule EMQXUmbrella.MixProject do
       system_env: emqx_app_system_env()
     }
 
+  def common_dep(:tablestore),
+    do: {:tablestore, github: "emqx/ots_erl", tag: "0.2.2", override: true}
+
   def common_dep(:influxdb),
     do: {:influxdb, github: "emqx/influxdb-client-erl", tag: "1.1.13", override: true}
 
@@ -396,6 +399,7 @@ defmodule EMQXUmbrella.MixProject do
       :emqx_ds_builtin_raft,
       :emqx_auth_kerberos,
       :emqx_bridge_datalayers,
+      :emqx_bridge_tablestore,
       :emqx_auth_cinfo
     ])
   end

rebar.config.erl

@@ -101,6 +101,7 @@ is_community_umbrella_app("apps/emqx_bridge_timescale") -> false;
 is_community_umbrella_app("apps/emqx_bridge_oracle") -> false;
 is_community_umbrella_app("apps/emqx_bridge_sqlserver") -> false;
 is_community_umbrella_app("apps/emqx_bridge_datalayers") -> false;
+is_community_umbrella_app("apps/emqx_bridge_tablestore") -> false;
 is_community_umbrella_app("apps/emqx_oracle") -> false;
 is_community_umbrella_app("apps/emqx_bridge_rabbitmq") -> false;
 is_community_umbrella_app("apps/emqx_ft") -> false;

rel/i18n/emqx_bridge_tablestore.hocon

@@ -0,0 +1,108 @@
+emqx_bridge_tablestore {
+
+desc_config.label:
+"""Tablestore Bridge Configuration"""
+desc_config.desc:
+"""Configuration for a Tablestore bridge."""
+
+desc_endpoint.label:
+"""Endpoint"""
+desc_endpoint.desc:
+"""Endpoint for the Tablestore. e.g. https://myinstance.cn-hangzhou.ots.aliyuncs.com"""
+
+desc_instance_name.label:
+"""Instance Name"""
+desc_instance_name.desc:
+"""instance name."""
+
+desc_access_key_id.label:
+"""Key ID"""
+desc_access_key_id.desc:
+"""Key ID. e.g NTS**********************"""
+
+desc_access_key_secret.label:
+"""Key Secret"""
+desc_access_key_secret.desc:
+"""Key secret. e.g 7NR2****************************************"""
+
+pool_size.label:
+"""Pool Size"""
+pool_size.desc:
+"""The pool size."""
+
+desc_table_name.label:
+"""Table Name"""
+desc_table_name.desc:
+"""Table name."""
+
+desc_measurement.label:
+"""Measurement"""
+desc_measurement.desc:
+"""The measurement."""
+
+desc_data_source.label:
+"""Data Source"""
+desc_data_source.desc:
+"""The data source."""
+
+desc_timestamp.label:
+"""Timestamp"""
+desc_timestamp.desc:
+"""The timestamp."""
+
+desc_meta_update_model.label:
+"""Meta Update Model"""
+desc_meta_update_model.desc:
+"""The meta-update-module."""
+
+tablestore_fields_column.label:
+"""Column"""
+tablestore_fields_column.desc:
+"""Column name of the field"""
+
+tablestore_fields_value.label:
+"""Value"""
+tablestore_fields_value.desc:
+"""Value of the field"""
+
+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."""
+
+tablestore_fields_isbinary.label:
+"""Is Binary"""
+tablestore_fields_isbinary.desc:
+"""Whether try write binary values as binary type. Defaults to false, means that write binary values as strings."""
+
+storage_model_type.label:
+"""Storage Model Type"""
+storage_model_type.desc:
+"""Storage model type. Can be one of "timeseries" or "order"."""
+
+desc_tags.label:
+"""Tags"""
+desc_tags.desc:
+"""Tags"""
+
+desc_fields.label:
+"""Fields"""
+desc_fields.desc:
+"""Fields"""
+
+connector.label:
+"""Tablestore Connector Configuration"""
+connector.desc:
+"""Configuration for a Tablestore connector."""
+
+action_parameters.label:
+"""Action Parameters"""
+action_parameters.desc:
+"""Additional parameters specific to this action type"""
+
+tablestore_action.label:
+"""Tablestore Action"""
+tablestore_action.desc:
+"""Action to interact with a Tablestore connector"""
+
+}