feat: add ee influxdb connector & bridge

apps/emqx_connector/src/emqx_connector_schema.erl

@@ -45,7 +45,7 @@ post_request() ->
 
 http_schema(Method) ->
     Broker = [?R_REF(schema_mod(Type), Method) || Type <- ?CONN_TYPES],
-    EE = [?R_REF(Module, Method) || Module <- schema_modules()],
+    EE = ee_schemas(Method),
     Schemas = Broker ++ EE,
     ?UNION(Schemas).
 
@@ -70,8 +70,8 @@ fields("connectors") ->
     Broker ++ EE.
 
 -if(?EMQX_RELEASE_EDITION == ee).
-schema_modules() ->
+ee_schemas(Method) ->
-    emqx_ee_connector:schema_modules().
+    emqx_ee_connector:api_schemas(Method).
 
 ee_fields_connectors() ->
     emqx_ee_connector:fields(connectors).
@@ -79,7 +79,7 @@ ee_fields_connectors() ->
 ee_fields_connectors() ->
     [].
 
-schema_modules() ->
+ee_schemas(_) ->
     [].
 -endif.
 

lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_hstream.conf

@@ -13,10 +13,10 @@ will be forwarded.
 注意：如果这个 Bridge 被用作规则（EMQX 规则引擎）的输出，同时也配置了 'local_topic' ，那么这两部分的消息都会被转发到 HStreamDB。
 """
         }
-    label {
+        label {
-            en: "Local Topic"
+                en: "Local Topic"
-            zh: "本地 Topic"
+                zh: "本地 Topic"
-        }
+            }
     }
     payload {
         desc {

lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_influxdb.conf

@@ -0,0 +1,94 @@
+emqx_ee_bridge_influxdb {
+    local_topic {
+        desc {
+            en: """
+The MQTT topic filter to be forwarded to the InfluxDB. All MQTT 'PUBLISH' messages with the topic
+matching the local_topic will be forwarded.</br>
+NOTE: if this bridge is used as the action of a rule (EMQX rule engine), and also local_topic is
+configured, then both the data got from the rule and the MQTT messages that match local_topic
+will be forwarded.
+"""
+            zh: """
+发送到 'local_topic' 的消息都会转发到 InfluxDB。 </br>
+注意：如果这个 Bridge 被用作规则（EMQX 规则引擎）的输出，同时也配置了 'local_topic' ，那么这两部分的消息都会被转发到 InfluxDB。
+"""
+        }
+        label {
+                en: "Local Topic"
+                zh: "本地 Topic"
+            }
+    }
+    payload {
+        desc {
+            en: """The payload to be forwarded to the InfluxDB. Placeholders supported."""
+            zh: """要转发到 InfluxDB 的数据内容，支持占位符"""
+            }
+        label {
+            en: "Payload"
+            zh: "消息内容"
+        }
+    }
+    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 InfluxDB bridge."""
+            zh: """InfluxDB 桥接配置"""
+        }
+        label: {
+            en: "InfluxDB Bridge Configuration"
+            zh: "InfluxDB 桥接配置"
+        }
+    }
+
+    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

@@ -13,7 +13,7 @@
 ]).
 
 schema_modules() ->
-    [emqx_ee_bridge_hstream].
+    [emqx_ee_bridge_hstream, emqx_ee_bridge_influxdb].
 
 conn_bridge_examples(Method) ->
     Fun =
@@ -23,8 +23,9 @@ conn_bridge_examples(Method) ->
         end,
     lists:foldl(Fun, #{}, schema_modules()).
 
+resource_type(Type) when is_binary(Type) -> resource_type(binary_to_atom(Type, utf8));
 resource_type(hstreamdb) -> emqx_ee_connector_hstream;
-resource_type(<<"hstreamdb">>) -> emqx_ee_connector_hstream.
+resource_type(influxdb) -> emqx_ee_connector_influxdb.
 
 fields(bridges) ->
     [
@@ -32,5 +33,10 @@ fields(bridges) ->
             mk(
                 hoconsc:map(name, ref(emqx_ee_bridge_hstream, "config")),
                 #{desc => <<"EMQX Enterprise Config">>}
+            )},
+        {influxdb,
+            mk(
+                hoconsc:map(name, ref(emqx_ee_bridge_influxdb, "config")),
+                #{desc => <<"EMQX Enterprise Config">>}
             )}
     ].

lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_influxdb.erl

@@ -0,0 +1,99 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+-module(emqx_ee_bridge_influxdb).
+
+-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_example/1
+]).
+
+-export([
+    namespace/0,
+    roots/0,
+    fields/1,
+    desc/1
+]).
+
+%% -------------------------------------------------------------------------------------------------
+%% api
+
+conn_bridge_example(Method) ->
+    #{
+        <<"influxdb">> => #{
+            summary => <<"InfluxDB Bridge">>,
+            value => values(Method)
+        }
+    }.
+
+values(get) ->
+    maps:merge(values(post), ?METRICS_EXAMPLE);
+values(post) ->
+    #{
+        type => influxdb,
+        name => <<"demo">>,
+        connector => <<"influxdb:api_v2_connector">>,
+        enable => true,
+        direction => egress,
+        local_topic => <<"local/topic/#">>,
+        payload => <<"${payload}">>
+    };
+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})},
+        {local_topic, mk(binary(), #{desc => ?DESC("local_topic")})},
+        {payload, mk(binary(), #{default => <<"${payload}">>, desc => ?DESC("payload")})},
+        {connector, field(connector)}
+    ];
+fields("post") ->
+    [type_field(), name_field() | fields("config")];
+fields("put") ->
+    fields("config");
+fields("get") ->
+    emqx_bridge_schema:metrics_status_fields() ++ fields("post").
+
+field(connector) ->
+    ConnectorConfigRef =
+        [
+            ref(emqx_ee_connector_influxdb, udp),
+            ref(emqx_ee_connector_influxdb, api_v1),
+            ref(emqx_ee_connector_influxdb, api_v2)
+        ],
+    mk(
+        hoconsc:union([binary() | ConnectorConfigRef]),
+        #{
+            required => true,
+            example => <<"influxdb:demo">>,
+            desc => ?DESC("desc_connector")
+        }
+    ).
+
+desc("config") ->
+    ?DESC("desc_config");
+desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->
+    ["Configuration for HStream using `", string:to_upper(Method), "` method."];
+desc(_) ->
+    undefined.
+
+%% -------------------------------------------------------------------------------------------------
+%% internal
+type_field() ->
+    {type, mk(enum([influxdb]), #{required => true, desc => ?DESC("desc_type")})}.
+
+name_field() ->
+    {name, mk(binary(), #{required => true, desc => ?DESC("desc_name")})}.

lib-ee/emqx_ee_connector/i18n/emqx_ee_connector_influxdb.conf

@@ -0,0 +1,165 @@
+
+emqx_ee_connector_influxdb {
+    type {
+        desc {
+                en: """The Connector Type."""
+                zh: """连接器类型。"""
+        }
+        label: {
+                en: """Connector Type"""
+                zh: """连接器类型"""
+            }
+    }
+
+    name {
+        desc {
+                en: """Connector name, used as a human-readable description of the connector."""
+                zh: """连接器名称，人类可读的连接器描述。"""
+        }
+        label: {
+                en: """Connector Name"""
+                zh: """连接器名称"""
+            }
+    }
+    host {
+        desc {
+                en: """InfluxDB host."""
+                zh: """InfluxDB 主机地址。"""
+        }
+        label: {
+                en: """Host"""
+                zh: """主机"""
+            }
+    }
+    port {
+        desc {
+                en: """InfluxDB port."""
+                zh: """InfluxDB 端口。"""
+        }
+        label: {
+                en: """Port"""
+                zh: """端口"""
+            }
+    }
+    protocol {
+        desc {
+                en: """InfluxDB protocol. UDP or HTTP API or HTTP API V2"""
+                zh: """InfluxDB 协议。UDP 或 HTTP API 或 HTTP API V2"""
+        }
+        label: {
+                en: """Protocol"""
+                zh: """协议"""
+            }
+    }
+    protocol_udp {
+        desc {
+                en: """InfluxDB protocol."""
+                zh: """InfluxDB UDP 协议"""
+        }
+        label: {
+                en: """UDP Protocol"""
+                zh: """UDP 协议"""
+            }
+    }
+    protocol_api_v1 {
+        desc {
+                en: """InfluxDB protocol. Support InfluxDB v1.8 and before."""
+                zh: """InfluxDB HTTP API 协议。支持 Influxdb v1.8 以及之前的版本"""
+        }
+        label: {
+                en: """HTTP API Protocol"""
+                zh: """HTTP API 协议"""
+            }
+    }
+    protocol_api_v2 {
+        desc {
+                en: """InfluxDB protocol. Support InfluxDB v2.0 and after."""
+                zh: """InfluxDB HTTP API V2 协议。支持 Influxdb v2.0 以及之后的版本"""
+        }
+        label: {
+                en: """HTTP API V2 Protocol"""
+                zh: """HTTP API V2 协议"""
+            }
+    }
+    database {
+        desc {
+                en: """InfluxDB database."""
+                zh: """InfluxDB 数据库。"""
+        }
+        label: {
+                en: "Database"
+                zh: "数据库"
+            }
+    }
+    username {
+        desc {
+                en: "InfluxDB username."
+                zh: "InfluxDB 用户名。"
+        }
+        label: {
+                en: "Username"
+                zh: "用户名"
+            }
+    }
+    password {
+        desc {
+                en: "InfluxDB password."
+                zh: "InfluxDB 密码。"
+        }
+        label: {
+                en: "Password"
+                zh: "密码"
+            }
+    }
+    bucket {
+        desc {
+                en: "InfluxDB bucket name."
+                zh: "InfluxDB bucket 名称"
+        }
+        label: {
+                en: "Bucket"
+                zh: "Bucket"
+            }
+    }
+    org {
+        desc {
+                en: """InfluxDB organization name."""
+                zh: """InfluxDB 组织名称。"""
+        }
+        label: {
+                en: """Organization"""
+                zh: """组织"""
+            }
+    }
+    token {
+        desc {
+                en: """InfluxDB token."""
+                zh: """InfluxDB token。"""
+        }
+        label: {
+                en: """Token"""
+                zh: """Token"""
+            }
+    }
+    precision {
+        desc {
+                en: """InfluxDB time precision."""
+                zh: """InfluxDB 时间精度。"""
+        }
+        label: {
+                en: """Time Precision"""
+                zh: """时间精度"""
+            }
+    }
+    pool_size {
+        desc {
+                en: """InfluxDB Pool Size"""
+                zh: """InfluxDB 连接池大小"""
+            }
+        label {
+                en: """InfluxDB Pool Size"""
+                zh: """InfluxDB 连接池大小"""
+            }
+    }
+
+}

lib-ee/emqx_ee_connector/src/emqx_ee_connector.erl

@@ -6,13 +6,18 @@
 -import(hoconsc, [mk/2, enum/1, ref/2]).
 
 -export([
-    schema_modules/0,
     fields/1,
-    connector_examples/1
+    connector_examples/1,
+    api_schemas/1
 ]).
 
-schema_modules() ->
+api_schemas(Method) ->
-    [emqx_ee_connector_hstream].
+    [
+        ref(emqx_ee_connector_hstream, Method),
+        ref(emqx_ee_connector_influxdb, Method ++ "_udp"),
+        ref(emqx_ee_connector_influxdb, Method ++ "_api_v1"),
+        ref(emqx_ee_connector_influxdb, Method ++ "_api_v2")
+    ].
 
 fields(connectors) ->
     [
@@ -21,12 +26,27 @@ fields(connectors) ->
                 hoconsc:map(name, ref(emqx_ee_connector_hstream, config)),
                 #{desc => <<"EMQX Enterprise Config">>}
             )}
+    ] ++ fields(influxdb);
+fields(influxdb) ->
+    [
+        {Protocol,
+            mk(hoconsc:map(name, ref(emqx_ee_connector_influxdb, Protocol)), #{
+                desc => <<"EMQX Enterprise Config">>
+            })}
+     || Protocol <- [udp, api_v1, api_v2]
     ].
 
 connector_examples(Method) ->
-    Fun =
+    MergeFun =
-        fun(Module, Examples) ->
+        fun(Example, Examples) ->
-            Example = erlang:apply(Module, connector_example, [Method]),
             maps:merge(Examples, Example)
         end,
+    Fun =
+        fun(Module, Examples) ->
+            ConnectorExamples = erlang:apply(Module, connector_examples, [Method]),
+            lists:foldl(MergeFun, Examples, ConnectorExamples)
+        end,
     lists:foldl(Fun, #{}, schema_modules()).
+
+schema_modules() ->
+    [emqx_ee_connector_hstream, emqx_ee_connector_influxdb].

lib-ee/emqx_ee_connector/src/emqx_ee_connector_hstream.erl

@@ -26,7 +26,7 @@
 -export([
     roots/0,
     fields/1,
-    connector_example/1
+    connector_examples/1
 ]).
 
 %% -------------------------------------------------------------------------------------------------
@@ -96,13 +96,15 @@ fields("post") ->
         {name, mk(binary(), #{required => true, desc => ?DESC("name")})}
     ] ++ fields("put").
 
-connector_example(Method) ->
+connector_examples(Method) ->
-    #{
+    [
-        <<"hstreamdb">> => #{
+        #{
-            summary => <<"HStreamDB Connector">>,
+            <<"hstreamdb">> => #{
-            value => values(Method)
+                summary => <<"HStreamDB Connector">>,
+                value => values(Method)
+            }
         }
-    }.
+    ].
 
 values(post) ->
     maps:merge(values(put), #{name => <<"connector">>});

lib-ee/emqx_ee_connector/src/emqx_ee_connector_influxdb.erl

@@ -0,0 +1,240 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+-module(emqx_ee_connector_influxdb).
+
+-include_lib("hocon/include/hoconsc.hrl").
+-include_lib("typerefl/include/types.hrl").
+-include_lib("emqx/include/logger.hrl").
+
+-import(hoconsc, [mk/2, enum/1, ref/2]).
+
+-behaviour(emqx_resource).
+
+-define(PUT_FIELDS_FILTER, fun({Name, _}) -> not lists:member(Name, [type, name]) end).
+
+%% callbacks of behaviour emqx_resource
+-export([
+    on_start/2,
+    on_stop/2,
+    on_query/4,
+    on_get_status/2
+]).
+
+-export([
+    fields/1,
+    connector_examples/1
+]).
+
+%% -------------------------------------------------------------------------------------------------
+%% resource callback
+
+on_start(InstId, Config) ->
+    start_client(InstId, Config).
+
+on_stop(_InstId, _State) ->
+    ok.
+
+on_query(_InstId, {send_message, _Data}, _AfterQuery, _State) ->
+    ok.
+
+on_get_status(_InstId, _State) ->
+    % connected;
+    disconnected.
+
+%% -------------------------------------------------------------------------------------------------
+%% schema
+
+fields("put_udp") ->
+    lists:filter(?PUT_FIELDS_FILTER, fields(udp));
+fields("put_api_v1") ->
+    lists:filter(?PUT_FIELDS_FILTER, fields(api_v1));
+fields("put_api_v2") ->
+    lists:filter(?PUT_FIELDS_FILTER, fields(api_v2));
+fields("get_udp") ->
+    fields(udp);
+fields("get_api_v1") ->
+    fields(api_v1);
+fields("get_api_v2") ->
+    fields(api_v2);
+fields("post_udp") ->
+    fields(udp);
+fields("post_api_v1") ->
+    fields(api_v1);
+fields("post_api_v2") ->
+    fields(api_v2);
+fields(basic) ->
+    [
+        {host,
+            mk(binary(), #{required => true, default => <<"120.0.0.1">>, desc => ?DESC("host")})},
+        {port, mk(pos_integer(), #{required => true, default => 8086, desc => ?DESC("port")})},
+        {precision,
+            mk(enum([ns, us, ms, s, m, h]), #{
+                required => false, default => ms, desc => ?DESC("precision")
+            })},
+        {pool_size, mk(pos_integer(), #{required => true, desc => ?DESC("pool_size")})},
+        {type, mk(enum([influxdb]), #{required => true, desc => ?DESC("type")})},
+        {name, mk(binary(), #{required => true, desc => ?DESC("name")})}
+    ];
+fields(udp) ->
+    [
+        {protocol, mk(enum([udp]), #{required => true, desc => ?DESC("protocol_udp")})}
+    ] ++ fields(basic);
+fields(api_v1) ->
+    [
+        {protocol, mk(enum([api_v1]), #{required => true, desc => ?DESC("protocol_api_v1")})},
+        {database, mk(binary(), #{required => true, desc => ?DESC("database")})},
+        {username, mk(binary(), #{required => true, desc => ?DESC("username")})},
+        {password, mk(binary(), #{required => true, desc => ?DESC("password")})}
+    ] ++ emqx_connector_schema_lib:ssl_fields() ++ fields(basic);
+fields(api_v2) ->
+    [
+        {protocol, mk(enum([api_v2]), #{required => true, desc => ?DESC("protocol_api_v2")})},
+        {bucket, mk(binary(), #{required => true, desc => ?DESC("bucket")})},
+        {org, mk(binary(), #{required => true, desc => ?DESC("org")})},
+        {token, mk(binary(), #{required => true, desc => ?DESC("token")})}
+    ] ++ emqx_connector_schema_lib:ssl_fields() ++ fields(basic).
+
+connector_examples(Method) ->
+    [
+        #{
+            <<"influxdb_udp">> => #{
+                summary => <<"InfluxDB UDP Connector">>,
+                value => values(udp, Method)
+            }
+        },
+        #{
+            <<"influxdb_api_v1">> => #{
+                summary => <<"InfluxDB HTTP API V1 Connector">>,
+                value => values(api_v1, Method)
+            }
+        },
+        #{
+            <<"influxdb_api_v2">> => #{
+                summary => <<"InfluxDB HTTP API V2 Connector">>,
+                value => values(api_v2, Method)
+            }
+        }
+    ].
+
+values(Protocol, get) ->
+    values(Protocol, post);
+values(Protocol, post) ->
+    ConnectorName = list_to_binary(io_lib:format("~p_connector", [Protocol])),
+    maps:merge(values(Protocol, put), #{type => influxdb, name => ConnectorName});
+values(udp, put) ->
+    #{
+        protocol => udp,
+        host => <<"127.0.0.1">>,
+        port => 8089,
+        precision => ms,
+        pool_size => 8
+    };
+values(api_v1, put) ->
+    #{
+        protocol => api_v1,
+        host => <<"127.0.0.1">>,
+        port => 8086,
+        precision => ms,
+        pool_size => 8,
+        database => <<"my_db">>,
+        username => <<"my_user">>,
+        password => <<"my_password">>,
+        ssl => #{enable => false}
+    };
+values(api_v2, put) ->
+    #{
+        protocol => api_v2,
+        host => <<"127.0.0.1">>,
+        port => 8086,
+        precision => ms,
+        pool_size => 8,
+        bucket => <<"my_bucket">>,
+        org => <<"my_org">>,
+        token => <<"my_token">>,
+        ssl => #{enable => false}
+    }.
+%% -------------------------------------------------------------------------------------------------
+%% internal functions
+
+start_client(InstId, Config) ->
+    io:format("InstId ~p~n", [InstId]),
+    client_config(InstId, Config).
+
+% ClientConfig = client_config(InstId, Config),
+% case influxdb:start_client(ClientConfig) of
+%     {ok, Client} ->
+%         true = influxdb:is_alive(Client),
+%         maybe_pool_size(Client, Params);
+%     {error, {already_started, Client0}} ->
+%         _ = influxdb:stop_client(Client0),
+%         {ok, Client} = influxdb:start_client(Options),
+%         true = influxdb:is_alive(Client),
+%         maybe_pool_size(Client, Params);
+%     {error, Reason} ->
+%         logger:log(error, "Initiate influxdb failed ~0p", [Reason]),
+%         error({start_pool_failed, ResId})
+% end.
+
+client_config(
+    _InstId,
+    Config = #{
+        host := Host,
+        port := Port,
+        pool_size := PoolSize
+    }
+) ->
+    [
+        {host, Host},
+        {port, Port},
+        {pool_size, PoolSize},
+        {pool, atom_pname_todo},
+        {precision, maps:get(precision, Config, ms)}
+    ] ++ protocol_config(Config).
+
+protocol_config(#{
+    protocol := udp
+}) ->
+    [
+        {protocol, udp}
+    ];
+protocol_config(#{
+    protocol := api_v1,
+    username := Username,
+    password := Password,
+    database := DB,
+    ssl := SSL
+}) ->
+    [
+        {protocol, http},
+        {version, v1},
+        {username, Username},
+        {password, Password},
+        {database, DB},
+        {ssl, SSL}
+    ] ++ ssl_config(SSL);
+protocol_config(#{
+    protocol := api_v2,
+    bucket := Bucket,
+    org := Org,
+    token := Token,
+    ssl := SSL
+}) ->
+    [
+        {protocol, http},
+        {version, v2},
+        {bucket, Bucket},
+        {org, Org},
+        {token, Token},
+        {ssl, SSL}
+    ] ++ ssl_config(SSL).
+
+ssl_config(#{enable := false}) ->
+    [
+        {https_enabled, false}
+    ];
+ssl_config(SSL = #{enable := true}) ->
+    [
+        {https_enabled, true},
+        {transport, ssl}
+    ] ++ maps:to_list(maps:remove(enable, SSL)).