From 9853d00cadb9ba446c769353d565ceb2c5d54230 Mon Sep 17 00:00:00 2001 From: firest Date: Mon, 13 Mar 2023 11:17:53 +0800 Subject: [PATCH] feat(bridges): integrate RocketMQ into data bridges --- apps/emqx_bridge/src/emqx_bridge.erl | 3 +- .../i18n/emqx_ee_bridge_rocketmq.conf | 70 ++++ lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl | 17 +- .../src/emqx_ee_bridge_rocketmq.erl | 120 ++++++ .../i18n/emqx_ee_connector_rocketmq.conf | 66 ++++ lib-ee/emqx_ee_connector/rebar.config | 1 + .../src/emqx_ee_connector.app.src | 3 +- .../src/emqx_ee_connector_rocketmq.erl | 342 ++++++++++++++++++ 8 files changed, 617 insertions(+), 5 deletions(-) create mode 100644 lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_rocketmq.conf create mode 100644 lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_rocketmq.erl create mode 100644 lib-ee/emqx_ee_connector/i18n/emqx_ee_connector_rocketmq.conf create mode 100644 lib-ee/emqx_ee_connector/src/emqx_ee_connector_rocketmq.erl diff --git a/apps/emqx_bridge/src/emqx_bridge.erl b/apps/emqx_bridge/src/emqx_bridge.erl index 98ce6a8b0..087c72dc3 100644 --- a/apps/emqx_bridge/src/emqx_bridge.erl +++ b/apps/emqx_bridge/src/emqx_bridge.erl @@ -67,7 +67,8 @@ T == timescale; T == matrix; T == tdengine; - T == dynamo + T == dynamo; + T == rocketmq ). load() -> diff --git a/lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_rocketmq.conf b/lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_rocketmq.conf new file mode 100644 index 000000000..2e33e6c07 --- /dev/null +++ b/lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_rocketmq.conf @@ -0,0 +1,70 @@ +emqx_ee_bridge_rocketmq { + + local_topic { + desc { + en: """The MQTT topic filter to be forwarded to RocketMQ. All MQTT `PUBLISH` messages with the topic +matching the `local_topic` will be forwarded.
+NOTE: if the bridge is used as a rule action, `local_topic` should be left empty otherwise the messages will be duplicated.""" + zh: """发送到 'local_topic' 的消息都会转发到 RocketMQ。
+注意：如果这个 Bridge 被用作规则（EMQX 规则引擎）的输出，同时也配置了 'local_topic' ，那么这两部分的消息都会被转发。""" + } + label { + en: "Local Topic" + zh: "本地 Topic" + } + } + + template { + desc { + en: """Template, the default value is empty. When this value is empty the whole message will be stored in the RocketMQ""" + zh: """模板, 默认为空，为空时将会将整个消息转发给 RocketMQ""" + } + label { + en: "Template" + zh: "模板" + } + } + config_enable { + desc { + en: """Enable or disable this bridge""" + zh: """启用/禁用桥接""" + } + label { + en: "Enable Or Disable Bridge" + zh: "启用/禁用桥接" + } + } + + desc_config { + desc { + en: """Configuration for a RocketMQ bridge.""" + zh: """RocketMQ 桥接配置""" + } + label: { + en: "RocketMQ Bridge Configuration" + zh: "RocketMQ 桥接配置" + } + } + + desc_type { + desc { + en: """The Bridge Type""" + zh: """Bridge 类型""" + } + label { + en: "Bridge Type" + zh: "桥接类型" + } + } + + desc_name { + desc { + en: """Bridge name.""" + zh: """桥接名字""" + } + label { + en: "Bridge Name" + zh: "桥接名字" + } + } +} diff --git a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl index ec81b7935..3989c3ab2 100644 --- a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl +++ b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl @@ -32,7 +32,8 @@ api_schemas(Method) -> ref(emqx_ee_bridge_matrix, Method), ref(emqx_ee_bridge_tdengine, Method), ref(emqx_ee_bridge_clickhouse, Method), - ref(emqx_ee_bridge_dynamo, Method) + ref(emqx_ee_bridge_dynamo, Method), + ref(emqx_ee_bridge_rocketmq, Method) ]. schema_modules() -> @@ -49,7 +50,8 @@ schema_modules() -> emqx_ee_bridge_matrix, emqx_ee_bridge_tdengine, emqx_ee_bridge_clickhouse, - emqx_ee_bridge_dynamo + emqx_ee_bridge_dynamo, + emqx_ee_bridge_rocketmq ]. examples(Method) -> @@ -85,7 +87,8 @@ resource_type(timescale) -> emqx_connector_pgsql; resource_type(matrix) -> emqx_connector_pgsql; resource_type(tdengine) -> emqx_ee_connector_tdengine; resource_type(clickhouse) -> emqx_ee_connector_clickhouse; -resource_type(dynamo) -> emqx_ee_connector_dynamo. +resource_type(dynamo) -> emqx_ee_connector_dynamo; +resource_type(rocketmq) -> emqx_ee_connector_rocketmq. fields(bridges) -> [ @@ -128,6 +131,14 @@ fields(bridges) -> desc => <<"Dynamo Bridge Config">>, required => false } + )}, + {rocketmq, + mk( + hoconsc:map(name, ref(emqx_ee_bridge_rocketmq, "config")), + #{ + desc => <<"RocketMQ Bridge Config">>, + required => false + } )} ] ++ kafka_structs() ++ mongodb_structs() ++ influxdb_structs() ++ redis_structs() ++ pgsql_structs() ++ clickhouse_structs(). diff --git a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_rocketmq.erl b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_rocketmq.erl new file mode 100644 index 000000000..d81ffc54c --- /dev/null +++ b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_rocketmq.erl @@ -0,0 +1,120 @@ +%%-------------------------------------------------------------------- +%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved. +%%-------------------------------------------------------------------- +-module(emqx_ee_bridge_rocketmq). + +-include_lib("typerefl/include/types.hrl"). +-include_lib("hocon/include/hoconsc.hrl"). +-include_lib("emqx_bridge/include/emqx_bridge.hrl"). +-include_lib("emqx_resource/include/emqx_resource.hrl"). + +-import(hoconsc, [mk/2, enum/1, ref/2]). + +-export([ + conn_bridge_examples/1, + values/1 +]). + +-export([ + namespace/0, + roots/0, + fields/1, + desc/1 +]). + +-define(DEFAULT_TEMPLATE, <<>>). +-define(DEFFAULT_REQ_TIMEOUT, <<"15s">>). + +%% ------------------------------------------------------------------------------------------------- +%% api + +conn_bridge_examples(Method) -> + [ + #{ + <<"rocketmq">> => #{ + summary => <<"RocketMQ Bridge">>, + value => values(Method) + } + } + ]. + +values(get) -> + maps:merge(values(post), ?METRICS_EXAMPLE); +values(post) -> + #{ + enable => true, + type => rocketmq, + name => <<"foo">>, + server => <<"127.0.0.1:9876">>, + topic => <<"TopicTest">>, + template => ?DEFAULT_TEMPLATE, + local_topic => <<"local/topic/#">>, + resource_opts => #{ + worker_pool_size => 1, + health_check_interval => ?HEALTHCHECK_INTERVAL_RAW, + auto_restart_interval => ?AUTO_RESTART_INTERVAL_RAW, + batch_size => ?DEFAULT_BATCH_SIZE, + batch_time => ?DEFAULT_BATCH_TIME, + query_mode => sync, + max_queue_bytes => ?DEFAULT_QUEUE_SIZE + } + }; +values(put) -> + values(post). + +%% ------------------------------------------------------------------------------------------------- +%% Hocon Schema Definitions +namespace() -> "bridge_rocketmq". + +roots() -> []. + +fields("config") -> + [ + {enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})}, + {template, + mk( + binary(), + #{desc => ?DESC("template"), default => ?DEFAULT_TEMPLATE} + )}, + {local_topic, + mk( + binary(), + #{desc => ?DESC("local_topic"), required => false} + )}, + {resource_opts, + mk( + ref(?MODULE, "creation_opts"), + #{ + required => false, + default => #{<<"request_timeout">> => ?DEFFAULT_REQ_TIMEOUT}, + desc => ?DESC(emqx_resource_schema, <<"resource_opts">>) + } + )} + ] ++ + (emqx_ee_connector_rocketmq:fields(config) -- + emqx_connector_schema_lib:prepare_statement_fields()); +fields("creation_opts") -> + emqx_resource_schema:fields("creation_opts_sync_only"); +fields("post") -> + [type_field(), name_field() | fields("config")]; +fields("put") -> + fields("config"); +fields("get") -> + emqx_bridge_schema:status_fields() ++ fields("post"). + +desc("config") -> + ?DESC("desc_config"); +desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" -> + ["Configuration for RocketMQ using `", string:to_upper(Method), "` method."]; +desc("creation_opts" = Name) -> + emqx_resource_schema:desc(Name); +desc(_) -> + undefined. + +%% ------------------------------------------------------------------------------------------------- + +type_field() -> + {type, mk(enum([rocketmq]), #{required => true, desc => ?DESC("desc_type")})}. + +name_field() -> + {name, mk(binary(), #{required => true, desc => ?DESC("desc_name")})}. diff --git a/lib-ee/emqx_ee_connector/i18n/emqx_ee_connector_rocketmq.conf b/lib-ee/emqx_ee_connector/i18n/emqx_ee_connector_rocketmq.conf new file mode 100644 index 000000000..d4a610212 --- /dev/null +++ b/lib-ee/emqx_ee_connector/i18n/emqx_ee_connector_rocketmq.conf @@ -0,0 +1,66 @@ +emqx_ee_connector_rocketmq { + + server { + desc { + en: """ +The IPv4 or IPv6 address or the hostname to connect to.
+A host entry has the following form: `Host[:Port]`.
+The RocketMQ default port 9876 is used if `[:Port]` is not specified. +""" + zh: """ +将要连接的 IPv4 或 IPv6 地址，或者主机名。
+主机名具有以下形式：`Host[:Port]`。
+如果未指定 `[:Port]`，则使用 RocketMQ 默认端口 9876。 +""" + } + label: { + en: "Server Host" + zh: "服务器地址" + } + } + + topic { + desc { + en: """RocketMQ Topic""" + zh: """RocketMQ 主题""" + } + label: { + en: "RocketMQ Topic" + zh: "RocketMQ 主题" + } + } + + refresh_interval { + desc { + en: """RocketMQ Topic Route Refresh Interval.""" + zh: """RocketMQ 主题路由更新间隔。""" + } + label: { + en: "Topic Route Refresh Interval" + zh: "主题路由更新间隔" + } + } + + send_buffer { + desc { + en: """The socket send buffer size of the RocketMQ driver client.""" + zh: """RocketMQ 驱动的套字节发送消息的缓冲区大小""" + } + label: { + en: "Send Buffer Size" + zh: "发送消息的缓冲区大小" + } + } + + security_token { + desc { + en: """RocketMQ Server Security Token""" + zh: """RocketMQ 服务器安全令牌""" + } + label: { + en: "Security Token" + zh: "安全令牌" + } + } + +} diff --git a/lib-ee/emqx_ee_connector/rebar.config b/lib-ee/emqx_ee_connector/rebar.config index 76f6ccfba..96b3df6a3 100644 --- a/lib-ee/emqx_ee_connector/rebar.config +++ b/lib-ee/emqx_ee_connector/rebar.config @@ -5,6 +5,7 @@ {tdengine, {git, "https://github.com/emqx/tdengine-client-erl", {tag, "0.1.5"}}}, {clickhouse, {git, "https://github.com/emqx/clickhouse-client-erl", {tag, "0.2"}}}, {erlcloud, {git, "https://github.com/emqx/erlcloud.git", {tag,"3.5.16-emqx-1"}}}, + {rocketmq, {git, "https://github.com/emqx/rocketmq-client-erl.git", {tag, "v0.5.1"}}}, {emqx, {path, "../../apps/emqx"}} ]}. diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector.app.src b/lib-ee/emqx_ee_connector/src/emqx_ee_connector.app.src index 6f40f7158..d8921198c 100644 --- a/lib-ee/emqx_ee_connector/src/emqx_ee_connector.app.src +++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector.app.src @@ -11,7 +11,8 @@ wolff, brod, clickhouse, - erlcloud + erlcloud, + rocketmq ]}, {env, []}, {modules, []}, diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_rocketmq.erl b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_rocketmq.erl new file mode 100644 index 000000000..d41f83e1d --- /dev/null +++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_rocketmq.erl @@ -0,0 +1,342 @@ +%-------------------------------------------------------------------- +%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved. +%%-------------------------------------------------------------------- + +-module(emqx_ee_connector_rocketmq). + +-behaviour(emqx_resource). + +-include_lib("emqx_resource/include/emqx_resource.hrl"). +-include_lib("typerefl/include/types.hrl"). +-include_lib("emqx/include/logger.hrl"). +-include_lib("snabbkaffe/include/snabbkaffe.hrl"). +-include_lib("hocon/include/hoconsc.hrl"). + +-export([roots/0, fields/1]). + +%% `emqx_resource' API +-export([ + callback_mode/0, + is_buffer_supported/0, + on_start/2, + on_stop/2, + on_query/3, + on_batch_query/3, + on_get_status/2 +]). + +-import(hoconsc, [mk/2, enum/1, ref/2]). + +-define(ROCKETMQ_HOST_OPTIONS, #{ + default_port => 9876 +}). + +-ifdef(TEST). +-export([execute/2]). +-endif. + +%%===================================================================== +%% Hocon schema +roots() -> + [{config, #{type => hoconsc:ref(?MODULE, config)}}]. + +fields(config) -> + [ + {server, server()}, + {topic, + mk( + binary(), + #{default => <<"TopicTest">>, desc => ?DESC(topic)} + )}, + {refresh_interval, + mk( + emqx_schema:duration(), + #{default => <<"3s">>, desc => ?DESC(refresh_interval)} + )}, + {send_buffer, + mk( + emqx_schema:bytesize(), + #{default => <<"1024KB">>, desc => ?DESC(send_buffer)} + )}, + {security_token, mk(binary(), #{default => <<>>, desc => ?DESC(security_token)})} + | relational_fields() + ]. + +add_default_username(Fields) -> + lists:map( + fun + ({username, OrigUsernameFn}) -> + {username, add_default_fn(OrigUsernameFn, <<"">>)}; + (Field) -> + Field + end, + Fields + ). + +add_default_fn(OrigFn, Default) -> + fun + (default) -> Default; + (Field) -> OrigFn(Field) + end. + +server() -> + Meta = #{desc => ?DESC("server")}, + emqx_schema:servers_sc(Meta, ?ROCKETMQ_HOST_OPTIONS). + +relational_fields() -> + Fields = [username, password, auto_reconnect], + Values = lists:filter( + fun({E, _}) -> lists:member(E, Fields) end, + emqx_connector_schema_lib:relational_db_fields() + ), + add_default_username(Values). + +%%======================================================================================== +%% `emqx_resource' API +%%======================================================================================== + +callback_mode() -> always_sync. + +is_buffer_supported() -> false. + +on_start( + InstanceId, + #{server := Server, topic := Topic} = Config1 +) -> + ?SLOG(info, #{ + msg => "starting_rocketmq_connector", + connector => InstanceId, + config => redact(Config1) + }), + Config = maps:merge(default_security_info(), Config1), + {Host, Port} = emqx_schema:parse_server(Server, ?ROCKETMQ_HOST_OPTIONS), + + Server1 = [{Host, Port}], + ClientId = client_id(InstanceId), + ClientCfg = #{acl_info => #{}}, + + TopicTks = emqx_plugin_libs_rule:preproc_tmpl(Topic), + ProducerOpts = make_producer_opts(Config), + Templates = parse_template(Config), + ProducersMapPID = create_producers_map(ClientId), + State = #{ + client_id => ClientId, + topic_tokens => TopicTks, + config => Config, + templates => Templates, + producers_map_pid => ProducersMapPID, + producers_opts => ProducerOpts + }, + + case rocketmq:ensure_supervised_client(ClientId, Server1, ClientCfg) of + {ok, _Pid} -> + {ok, State}; + {error, _Reason} = Error -> + ?tp( + rocketmq_connector_start_failed, + #{error => _Reason} + ), + Error + end. + +on_stop(InstanceId, #{client_id := ClientId, producers_map_pid := Pid} = _State) -> + ?SLOG(info, #{ + msg => "stopping_rocketmq_connector", + connector => InstanceId + }), + Pid ! ok, + ok = rocketmq:stop_and_delete_supervised_client(ClientId). + +on_query(InstanceId, Query, State) -> + do_query(InstanceId, Query, send_sync, State). + +%% We only support batch inserts and all messages must have the same topic +on_batch_query(InstanceId, [{send_message, _Msg} | _] = Query, State) -> + do_query(InstanceId, Query, batch_send_sync, State); +on_batch_query(_InstanceId, Query, _State) -> + {error, {unrecoverable_error, {invalid_request, Query}}}. + +on_get_status(_InstanceId, #{client_id := ClientId}) -> + case rocketmq_client_sup:find_client(ClientId) of + {ok, _Pid} -> + connected; + _ -> + connecting + end. + +%%======================================================================================== +%% Helper fns +%%======================================================================================== + +do_query( + InstanceId, + Query, + QueryFunc, + #{ + templates := Templates, + client_id := ClientId, + topic_tokens := TopicTks, + producers_opts := ProducerOpts, + config := #{topic := RawTopic, resource_opts := #{request_timeout := RequestTimeout}} + } = State +) -> + ?TRACE( + "QUERY", + "rocketmq_connector_received", + #{connector => InstanceId, query => Query, state => State} + ), + + TopicKey = get_topic_key(Query, RawTopic, TopicTks), + Data = apply_template(Query, Templates), + + Result = safe_do_produce( + InstanceId, QueryFunc, ClientId, TopicKey, Data, ProducerOpts, RequestTimeout + ), + case Result of + {error, Reason} -> + ?tp( + rocketmq_connector_query_return, + #{error => Reason} + ), + ?SLOG(error, #{ + msg => "rocketmq_connector_do_query_failed", + connector => InstanceId, + query => Query, + reason => Reason + }), + Result; + _ -> + ?tp( + rocketmq_connector_query_return, + #{result => Result} + ), + Result + end. + +safe_do_produce(InstanceId, QueryFunc, ClientId, TopicKey, Data, ProducerOpts, RequestTimeout) -> + try + Producers = get_producers(ClientId, TopicKey, ProducerOpts), + produce(InstanceId, QueryFunc, Producers, Data, RequestTimeout) + catch + _Type:Reason -> + {error, {unrecoverable_error, Reason}} + end. + +produce(_InstanceId, QueryFunc, Producers, Data, RequestTimeout) -> + rocketmq:QueryFunc(Producers, Data, RequestTimeout). + +parse_template(Config) -> + Templates = + case maps:get(template, Config, undefined) of + undefined -> #{}; + <<>> -> #{}; + Template -> #{send_message => Template} + end, + + parse_template(maps:to_list(Templates), #{}). + +parse_template([{Key, H} | T], Templates) -> + ParamsTks = emqx_plugin_libs_rule:preproc_tmpl(H), + parse_template( + T, + Templates#{Key => ParamsTks} + ); +parse_template([], Templates) -> + Templates. + +get_topic_key({_, Msg}, RawTopic, TopicTks) -> + {RawTopic, emqx_plugin_libs_rule:proc_tmpl(TopicTks, Msg)}; +get_topic_key([Query | _], RawTopic, TopicTks) -> + get_topic_key(Query, RawTopic, TopicTks). + +apply_template({Key, Msg} = _Req, Templates) -> + case maps:get(Key, Templates, undefined) of + undefined -> + emqx_json:encode(Msg); + Template -> + emqx_plugin_libs_rule:proc_tmpl(Template, Msg) + end; +apply_template([{Key, _} | _] = Reqs, Templates) -> + case maps:get(Key, Templates, undefined) of + undefined -> + [emqx_json:encode(Msg) || {_, Msg} <- Reqs]; + Template -> + [emqx_plugin_libs_rule:proc_tmpl(Template, Msg) || {_, Msg} <- Reqs] + end. + +client_id(InstanceId) -> + Name = emqx_resource_manager:manager_id_to_resource_id(InstanceId), + erlang:binary_to_atom(Name, utf8). + +redact(Msg) -> + emqx_misc:redact(Msg, fun is_sensitive_key/1). + +is_sensitive_key(security_token) -> + true; +is_sensitive_key(_) -> + false. + +make_producer_opts( + #{ + username := Username, + password := Password, + security_token := SecurityToken, + send_buffer := SendBuff, + refresh_interval := RefreshInterval + } +) -> + ACLInfo = acl_info(Username, Password, SecurityToken), + #{ + tcp_opts => [{sndbuf, SendBuff}], + ref_topic_route_interval => RefreshInterval, + acl_info => ACLInfo + }. + +acl_info(<<>>, <<>>, <<>>) -> + #{}; +acl_info(Username, Password, <<>>) when is_binary(Username), is_binary(Password) -> + #{ + access_key => Username, + secret_key => Password + }; +acl_info(Username, Password, SecurityToken) when + is_binary(Username), is_binary(Password), is_binary(SecurityToken) +-> + #{ + access_key => Username, + secret_key => Password, + security_token => SecurityToken + }; +acl_info(_, _, _) -> + #{}. + +create_producers_map(ClientId) -> + erlang:spawn(fun() -> + case ets:whereis(ClientId) of + undefined -> + _ = ets:new(ClientId, [public, named_table]), + ok; + _ -> + ok + end, + receive + _Msg -> + ok + end + end). + +get_producers(ClientId, {_, Topic1} = TopicKey, ProducerOpts) -> + case ets:lookup(ClientId, TopicKey) of + [{_, Producers0}] -> + Producers0; + _ -> + ProducerGroup = iolist_to_binary([atom_to_list(ClientId), "_", Topic1]), + {ok, Producers0} = rocketmq:ensure_supervised_producers( + ClientId, ProducerGroup, Topic1, ProducerOpts + ), + ets:insert(ClientId, {TopicKey, Producers0}), + Producers0 + end. + +default_security_info() -> + #{username => <<>>, password => <<>>, security_token => <<>>}.