Merge pull request #8645 from lafirest/feat/bridge_mysql

feat(bridge): add mysql sink

apps/emqx_connector/src/emqx_connector_mysql.erl

@@ -44,6 +44,16 @@
     default_port => ?MYSQL_DEFAULT_PORT
 }).
 
+-type prepares() :: #{atom() => binary()}.
+-type params_tokens() :: #{atom() => list()}.
+-type state() ::
+    #{
+        poolname := atom(),
+        prepare_statement := prepares(),
+        auto_reconnect := boolean(),
+        params_tokens := params_tokens()
+    }.
+
 %%=====================================================================
 %% Hocon schema
 roots() ->
@@ -63,6 +73,7 @@ server(desc) -> ?DESC("server");
 server(_) -> undefined.
 
 %% ===================================================================
+-spec on_start(binary(), hoconsc:config()) -> {ok, state()} | {error, _}.
 on_start(
     InstId,
     #{
@@ -97,8 +108,8 @@ on_start(
         {pool_size, PoolSize}
     ],
     PoolName = emqx_plugin_libs_pool:pool_name(InstId),
-    Prepares = maps:get(prepare_statement, Config, #{}),
-    State = #{poolname => PoolName, prepare_statement => Prepares, auto_reconnect => AutoReconn},
+    Prepares = parse_prepare_sql(maps:get(prepare_statement, Config, #{})),
+    State = maps:merge(#{poolname => PoolName, auto_reconnect => AutoReconn}, Prepares),
     case emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Options ++ SslOpts) of
         ok -> {ok, init_prepare(State)};
         {error, Reason} -> {error, Reason}
@@ -111,13 +122,13 @@ on_stop(InstId, #{poolname := PoolName}) ->
     }),
     emqx_plugin_libs_pool:stop_pool(PoolName).
 
-on_query(InstId, {Type, SQLOrKey}, AfterQuery, State) ->
-    on_query(InstId, {Type, SQLOrKey, [], default_timeout}, AfterQuery, State);
-on_query(InstId, {Type, SQLOrKey, Params}, AfterQuery, State) ->
-    on_query(InstId, {Type, SQLOrKey, Params, default_timeout}, AfterQuery, State);
+on_query(InstId, {TypeOrKey, SQLOrKey}, AfterQuery, State) ->
+    on_query(InstId, {TypeOrKey, SQLOrKey, [], default_timeout}, AfterQuery, State);
+on_query(InstId, {TypeOrKey, SQLOrKey, Params}, AfterQuery, State) ->
+    on_query(InstId, {TypeOrKey, SQLOrKey, Params, default_timeout}, AfterQuery, State);
 on_query(
     InstId,
-    {Type, SQLOrKey, Params, Timeout},
+    {TypeOrKey, SQLOrKey, Params, Timeout},
     AfterQuery,
     #{poolname := PoolName, prepare_statement := Prepares} = State
 ) ->
@@ -125,8 +136,9 @@ on_query(
     ?TRACE("QUERY", "mysql_connector_received", LogMeta),
     Worker = ecpool:get_client(PoolName),
     {ok, Conn} = ecpool_worker:client(Worker),
-    MySqlFunction = mysql_function(Type),
-    Result = erlang:apply(mysql, MySqlFunction, [Conn, SQLOrKey, Params, Timeout]),
+    MySqlFunction = mysql_function(TypeOrKey),
+    {SQLOrKey2, Data} = proc_sql_params(TypeOrKey, SQLOrKey, Params, State),
+    Result = erlang:apply(mysql, MySqlFunction, [Conn, SQLOrKey2, Data, Timeout]),
     case Result of
         {error, disconnected} ->
             ?SLOG(
@@ -145,7 +157,7 @@ on_query(
             case prepare_sql(Prepares, PoolName) of
                 ok ->
                     %% not return result, next loop will try again
-                    on_query(InstId, {Type, SQLOrKey, Params, Timeout}, AfterQuery, State);
+                    on_query(InstId, {TypeOrKey, SQLOrKey, Params, Timeout}, AfterQuery, State);
                 {error, Reason} ->
                     ?SLOG(
                         error,
@@ -166,8 +178,13 @@ on_query(
             Result
     end.
 
-mysql_function(sql) -> query;
-mysql_function(prepared_query) -> execute.
+mysql_function(sql) ->
+    query;
+mysql_function(prepared_query) ->
+    execute;
+%% for bridge
+mysql_function(_) ->
+    mysql_function(prepared_query).
 
 on_get_status(_InstId, #{poolname := Pool, auto_reconnect := AutoReconn} = State) ->
     case emqx_plugin_libs_pool:health_check_ecpool_workers(Pool, fun ?MODULE:do_get_status/1) of
@@ -287,3 +304,24 @@ prepare_sql_to_conn(Conn, [{Key, SQL} | PrepareList]) when is_pid(Conn) ->
 
 unprepare_sql_to_conn(Conn, PrepareSqlKey) ->
     mysql:unprepare(Conn, PrepareSqlKey).
+
+parse_prepare_sql(SQL) ->
+    parse_prepare_sql(maps:to_list(SQL), #{}, #{}).
+
+parse_prepare_sql([{Key, H} | T], SQL, Tokens) ->
+    {PrepareSQL, ParamsTokens} = emqx_plugin_libs_rule:preproc_sql(H),
+    parse_prepare_sql(T, SQL#{Key => PrepareSQL}, Tokens#{Key => ParamsTokens});
+parse_prepare_sql([], SQL, Tokens) ->
+    #{prepare_statement => SQL, params_tokens => Tokens}.
+
+proc_sql_params(query, SQLOrKey, Params, _State) ->
+    {SQLOrKey, Params};
+proc_sql_params(prepared_query, SQLOrKey, Params, _State) ->
+    {SQLOrKey, Params};
+proc_sql_params(TypeOrKey, SQLOrData, Params, #{params_tokens := ParamsTokens}) ->
+    case maps:get(TypeOrKey, ParamsTokens, undefined) of
+        undefined ->
+            {SQLOrData, Params};
+        Tokens ->
+            {TypeOrKey, emqx_plugin_libs_rule:proc_sql(Tokens, SQLOrData)}
+    end.

lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_mysql.conf

@@ -0,0 +1,75 @@
+emqx_ee_bridge_mysql {
+    sql {
+        desc {
+            en: """SQL Template"""
+            zh: """SQL 模板"""
+            }
+        label {
+            en: "SQL Template"
+            zh: "SQL 模板"
+        }
+    }
+    config_enable {
+        desc {
+            en: """Enable or disable this bridge"""
+            zh: """启用/禁用桥接"""
+        }
+        label {
+            en: "Enable Or Disable Bridge"
+            zh: "启用/禁用桥接"
+        }
+        }
+    config_direction {
+        desc {
+            en: """The direction of this bridge, MUST be 'egress'"""
+            zh: """桥接的方向， 必须是 egress"""
+        }
+        label {
+            en: "Bridge Direction"
+            zh: "桥接方向"
+        }
+    }
+
+    desc_config {
+        desc {
+            en: """Configuration for an HStreamDB bridge."""
+            zh: """HStreamDB 桥接配置"""
+        }
+        label: {
+            en: "HStreamDB Bridge Configuration"
+            zh: "HStreamDB 桥接配置"
+        }
+    }
+
+    desc_type {
+        desc {
+            en: """The Bridge Type"""
+            zh: """Bridge 类型"""
+        }
+        label {
+            en: "Bridge Type"
+            zh: "桥接类型"
+        }
+    }
+
+    desc_name {
+        desc {
+            en: """Bridge name, used as a human-readable description of the bridge."""
+            zh: """桥接名字，可读描述"""
+        }
+        label {
+            en: "Bridge Name"
+            zh: "桥接名字"
+        }
+    }
+    desc_connector {
+        desc {
+                en: """Generic configuration for the connector."""
+                zh: """连接器的通用配置。"""
+        }
+        label: {
+                en: "Connector Generic Configuration"
+                zh: "连接器通用配置。"
+        }
+    }
+}

lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl

@@ -15,6 +15,7 @@
 api_schemas(Method) ->
     [
         ref(emqx_ee_bridge_hstream, Method),
+        ref(emqx_ee_bridge_mysql, Method),
         ref(emqx_ee_bridge_influxdb, Method ++ "_udp"),
         ref(emqx_ee_bridge_influxdb, Method ++ "_api_v1"),
         ref(emqx_ee_bridge_influxdb, Method ++ "_api_v2")
@@ -23,7 +24,8 @@ api_schemas(Method) ->
 schema_modules() ->
     [
         emqx_ee_bridge_hstream,
-        emqx_ee_bridge_influxdb
+        emqx_ee_bridge_influxdb,
+        emqx_ee_bridge_mysql
     ].
 
 conn_bridge_examples(Method) ->
@@ -40,6 +42,7 @@ conn_bridge_examples(Method) ->
 
 resource_type(Type) when is_binary(Type) -> resource_type(binary_to_atom(Type, utf8));
 resource_type(hstreamdb) -> emqx_ee_connector_hstream;
+resource_type(mysql) -> emqx_connector_mysql;
 resource_type(influxdb_udp) -> emqx_ee_connector_influxdb;
 resource_type(influxdb_api_v1) -> emqx_ee_connector_influxdb;
 resource_type(influxdb_api_v2) -> emqx_ee_connector_influxdb.
@@ -50,6 +53,11 @@ fields(bridges) ->
             mk(
                 hoconsc:map(name, ref(emqx_ee_bridge_hstream, "config")),
                 #{desc => <<"EMQX Enterprise Config">>}
+            )},
+        {mysql,
+            mk(
+                hoconsc:map(name, ref(emqx_ee_bridge_mysql, "config")),
+                #{desc => <<"EMQX Enterprise Config">>}
             )}
     ] ++ fields(influxdb);
 fields(influxdb) ->

lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_mysql.erl

@@ -0,0 +1,116 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+-module(emqx_ee_bridge_mysql).
+
+-include_lib("typerefl/include/types.hrl").
+-include_lib("hocon/include/hoconsc.hrl").
+-include("emqx_ee_bridge.hrl").
+
+-import(hoconsc, [mk/2, enum/1, ref/2]).
+
+-export([
+    conn_bridge_examples/1
+]).
+
+-export([
+    namespace/0,
+    roots/0,
+    fields/1,
+    desc/1
+]).
+
+-define(DEFAULT_SQL, <<
+    "insert into t_mqtt_msg(msgid, topic, qos, payload, arrived) "
+    "values (${id}, ${topic}, ${qos}, ${payload}, FROM_UNIXTIME(${timestamp}/1000))"
+>>).
+
+%% -------------------------------------------------------------------------------------------------
+%% api
+
+conn_bridge_examples(Method) ->
+    [
+        #{
+            <<"mysql">> => #{
+                summary => <<"MySQL Bridge">>,
+                value => values(Method)
+            }
+        }
+    ].
+
+values(get) ->
+    maps:merge(values(post), ?METRICS_EXAMPLE);
+values(post) ->
+    #{
+        type => mysql,
+        name => <<"mysql">>,
+        connector => #{
+            server => <<"127.0.0.1:3306">>,
+            database => <<"test">>,
+            pool_size => 8,
+            username => <<"root">>,
+            password => <<"public">>,
+            auto_reconnect => true,
+            prepare_statement => #{send_message => ?DEFAULT_SQL}
+        },
+        enable => true,
+        direction => egress
+    };
+values(put) ->
+    values(post).
+
+%% -------------------------------------------------------------------------------------------------
+%% Hocon Schema Definitions
+namespace() -> "bridge".
+
+roots() -> [].
+
+fields("config") ->
+    [
+        {enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
+        {direction, mk(egress, #{desc => ?DESC("config_direction"), default => egress})},
+        {connector,
+            mk(
+                ref(?MODULE, connector),
+                #{
+                    required => true,
+                    desc => ?DESC("desc_connector")
+                }
+            )}
+    ];
+fields("post") ->
+    [type_field(), name_field() | fields("config")];
+fields("put") ->
+    fields("config");
+fields("get") ->
+    emqx_bridge_schema:metrics_status_fields() ++ fields("post");
+fields(connector) ->
+    (emqx_connector_mysql:fields(config) --
+        emqx_connector_schema_lib:prepare_statement_fields()) ++ prepare_statement_fields().
+
+desc("config") ->
+    ?DESC("desc_config");
+desc(connector) ->
+    ?DESC("desc_connector");
+desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->
+    ["Configuration for MySQL using `", string:to_upper(Method), "` method."];
+desc(_) ->
+    undefined.
+
+%% -------------------------------------------------------------------------------------------------
+%% internal
+type_field() ->
+    {type, mk(enum([mysql]), #{required => true, desc => ?DESC("desc_type")})}.
+
+name_field() ->
+    {name, mk(binary(), #{required => true, desc => ?DESC("desc_name")})}.
+
+prepare_statement_fields() ->
+    [
+        {prepare_statement,
+            mk(map(), #{
+                desc => ?DESC(emqx_connector_schema_lib, prepare_statement),
+                default => #{<<"send_message">> => ?DEFAULT_SQL},
+                example => #{<<"send_message">> => ?DEFAULT_SQL}
+            })}
+    ].