From 8ceeafb0de8f49e02afdf9d63ec402ad6a8a2b42 Mon Sep 17 00:00:00 2001 From: JimMoen Date: Wed, 22 Mar 2023 10:28:37 +0800 Subject: [PATCH 01/22] chore: fix file license in lib-ee --- .../emqx_connector/src/emqx_connector_mysql.erl | 2 +- .../src/emqx_ee_connector_cassa.erl | 13 +------------ .../src/emqx_ee_connector_clickhouse.erl | 14 +------------- .../test/ee_connector_clickhouse_SUITE.erl | 17 +++-------------- .../test/emqx_ee_connector_cassa_SUITE.erl | 13 +------------ 5 files changed, 7 insertions(+), 52 deletions(-) diff --git a/apps/emqx_connector/src/emqx_connector_mysql.erl b/apps/emqx_connector/src/emqx_connector_mysql.erl index fe495252a..c69cf5ca0 100644 --- a/apps/emqx_connector/src/emqx_connector_mysql.erl +++ b/apps/emqx_connector/src/emqx_connector_mysql.erl @@ -437,7 +437,7 @@ do_sql_query(SQLFunc, Conn, SQLOrKey, Params, Timeout, LogMeta) -> error, LogMeta#{msg => "mysql_connector_do_sql_query_failed", reason => disconnected} ), - %% kill the poll worker to trigger reconnection + %% kill the pool worker to trigger reconnection _ = exit(Conn, restart), {error, {recoverable_error, disconnected}}; {error, not_prepared} = Error -> diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_cassa.erl b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_cassa.erl index 1e1882a1f..ce964dd17 100644 --- a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_cassa.erl +++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_cassa.erl @@ -1,18 +1,7 @@ %%-------------------------------------------------------------------- %% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved. -%% -%% 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. %%-------------------------------------------------------------------- + -module(emqx_ee_connector_cassa). -behaviour(emqx_resource). diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_clickhouse.erl b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_clickhouse.erl index 5d68687dc..3749ecc6e 100644 --- a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_clickhouse.erl +++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_clickhouse.erl @@ -1,17 +1,5 @@ %%-------------------------------------------------------------------- -%% Copyright (c) 2020-2023 EMQ Technologies Co., Ltd. All Rights Reserved. -%% -%% 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. +%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved. %%-------------------------------------------------------------------- -module(emqx_ee_connector_clickhouse). diff --git a/lib-ee/emqx_ee_connector/test/ee_connector_clickhouse_SUITE.erl b/lib-ee/emqx_ee_connector/test/ee_connector_clickhouse_SUITE.erl index 111deba06..c608588a1 100644 --- a/lib-ee/emqx_ee_connector/test/ee_connector_clickhouse_SUITE.erl +++ b/lib-ee/emqx_ee_connector/test/ee_connector_clickhouse_SUITE.erl @@ -1,17 +1,6 @@ -% %%-------------------------------------------------------------------- -% %% Copyright (c) 2020-2023 EMQ Technologies Co., Ltd. All Rights Reserved. -% %% -% %% 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. -% %%-------------------------------------------------------------------- +%%-------------------------------------------------------------------- +%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved. +%%-------------------------------------------------------------------- -module(ee_connector_clickhouse_SUITE). diff --git a/lib-ee/emqx_ee_connector/test/emqx_ee_connector_cassa_SUITE.erl b/lib-ee/emqx_ee_connector/test/emqx_ee_connector_cassa_SUITE.erl index 95b4407cf..52ed03a62 100644 --- a/lib-ee/emqx_ee_connector/test/emqx_ee_connector_cassa_SUITE.erl +++ b/lib-ee/emqx_ee_connector/test/emqx_ee_connector_cassa_SUITE.erl @@ -1,16 +1,5 @@ %%-------------------------------------------------------------------- -%% Copyright (c) 2020-2023 EMQ Technologies Co., Ltd. All Rights Reserved. -%% -%% 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. +%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved. %%-------------------------------------------------------------------- -module(emqx_ee_connector_cassa_SUITE). From 8e62afecdd31cb905426bae1c90e55c767db9dfe Mon Sep 17 00:00:00 2001 From: JimMoen Date: Mon, 27 Mar 2023 13:55:16 +0800 Subject: [PATCH 02/22] chore: suite module rename --- ...ckhouse_SUITE.erl => emqx_ee_connector_clickhouse_SUITE.erl} | 2 +- ...streamdb_SUITE.erl => emqx_ee_connector_hstreamdb_SUITE.erl} | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) rename lib-ee/emqx_ee_connector/test/{ee_connector_clickhouse_SUITE.erl => emqx_ee_connector_clickhouse_SUITE.erl} (99%) rename lib-ee/emqx_ee_connector/test/{ee_connector_hstreamdb_SUITE.erl => emqx_ee_connector_hstreamdb_SUITE.erl} (90%) diff --git a/lib-ee/emqx_ee_connector/test/ee_connector_clickhouse_SUITE.erl b/lib-ee/emqx_ee_connector/test/emqx_ee_connector_clickhouse_SUITE.erl similarity index 99% rename from lib-ee/emqx_ee_connector/test/ee_connector_clickhouse_SUITE.erl rename to lib-ee/emqx_ee_connector/test/emqx_ee_connector_clickhouse_SUITE.erl index c608588a1..73018e14f 100644 --- a/lib-ee/emqx_ee_connector/test/ee_connector_clickhouse_SUITE.erl +++ b/lib-ee/emqx_ee_connector/test/emqx_ee_connector_clickhouse_SUITE.erl @@ -2,7 +2,7 @@ %% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved. %%-------------------------------------------------------------------- --module(ee_connector_clickhouse_SUITE). +-module(emqx_ee_connector_clickhouse_SUITE). -compile(nowarn_export_all). -compile(export_all). diff --git a/lib-ee/emqx_ee_connector/test/ee_connector_hstreamdb_SUITE.erl b/lib-ee/emqx_ee_connector/test/emqx_ee_connector_hstreamdb_SUITE.erl similarity index 90% rename from lib-ee/emqx_ee_connector/test/ee_connector_hstreamdb_SUITE.erl rename to lib-ee/emqx_ee_connector/test/emqx_ee_connector_hstreamdb_SUITE.erl index 8acabbef4..ad49d9f62 100644 --- a/lib-ee/emqx_ee_connector/test/ee_connector_hstreamdb_SUITE.erl +++ b/lib-ee/emqx_ee_connector/test/emqx_ee_connector_hstreamdb_SUITE.erl @@ -2,7 +2,7 @@ %% Copyright (c) 2022-2023 EMQ Technologies Co., Ltd. All Rights Reserved. %%-------------------------------------------------------------------- --module(ee_connector_hstreamdb_SUITE). +-module(emqx_ee_connector_hstreamdb_SUITE). -compile(nowarn_export_all). -compile(export_all). From 58419698775275da71fa16229ebb94aa21b4e3ec Mon Sep 17 00:00:00 2001 From: JimMoen Date: Mon, 20 Mar 2023 10:28:34 +0800 Subject: [PATCH 03/22] feat: implement Microsoft SQL Server bridge (e5.0) --- apps/emqx_bridge/src/emqx_bridge.erl | 3 +- changes/ee/feat-10363.en.md | 1 + lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl | 23 +- .../src/emqx_ee_bridge_sqlserver.erl | 128 +++++ .../src/emqx_ee_connector.app.src | 4 +- .../src/emqx_ee_connector_sqlserver.erl | 531 ++++++++++++++++++ rel/i18n/emqx_ee_bridge_sqlserver.hocon | 87 +++ rel/i18n/emqx_ee_connector_sqlserver.hocon | 22 + 8 files changed, 793 insertions(+), 6 deletions(-) create mode 100644 changes/ee/feat-10363.en.md create mode 100644 lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_sqlserver.erl create mode 100644 lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl create mode 100644 rel/i18n/emqx_ee_bridge_sqlserver.hocon create mode 100644 rel/i18n/emqx_ee_connector_sqlserver.hocon diff --git a/apps/emqx_bridge/src/emqx_bridge.erl b/apps/emqx_bridge/src/emqx_bridge.erl index bf91d20f7..5d1850ebd 100644 --- a/apps/emqx_bridge/src/emqx_bridge.erl +++ b/apps/emqx_bridge/src/emqx_bridge.erl @@ -69,7 +69,8 @@ T == tdengine; T == dynamo; T == rocketmq; - T == cassandra + T == cassandra; + T == sqlserver ). load() -> diff --git a/changes/ee/feat-10363.en.md b/changes/ee/feat-10363.en.md new file mode 100644 index 000000000..c3b53a538 --- /dev/null +++ b/changes/ee/feat-10363.en.md @@ -0,0 +1 @@ +Add MicroSoft SQL Server data Bridge support. 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 84b0b98b0..3ad5cbbb4 100644 --- a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl +++ b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl @@ -34,7 +34,8 @@ api_schemas(Method) -> ref(emqx_ee_bridge_clickhouse, Method), ref(emqx_ee_bridge_dynamo, Method), ref(emqx_ee_bridge_rocketmq, Method), - ref(emqx_ee_bridge_cassa, Method) + ref(emqx_ee_bridge_cassa, Method), + ref(emqx_ee_bridge_sqlserver, Method) ]. schema_modules() -> @@ -53,7 +54,8 @@ schema_modules() -> emqx_ee_bridge_clickhouse, emqx_ee_bridge_dynamo, emqx_ee_bridge_rocketmq, - emqx_ee_bridge_cassa + emqx_ee_bridge_cassa, + emqx_ee_bridge_sqlserver ]. examples(Method) -> @@ -91,7 +93,8 @@ resource_type(tdengine) -> emqx_ee_connector_tdengine; resource_type(clickhouse) -> emqx_ee_connector_clickhouse; resource_type(dynamo) -> emqx_ee_connector_dynamo; resource_type(rocketmq) -> emqx_ee_connector_rocketmq; -resource_type(cassandra) -> emqx_ee_connector_cassa. +resource_type(cassandra) -> emqx_ee_connector_cassa; +resource_type(sqlserver) -> emqx_ee_connector_sqlserver. fields(bridges) -> [ @@ -152,7 +155,7 @@ fields(bridges) -> } )} ] ++ kafka_structs() ++ mongodb_structs() ++ influxdb_structs() ++ redis_structs() ++ - pgsql_structs() ++ clickhouse_structs(). + pgsql_structs() ++ clickhouse_structs() ++ sqlserver_structs(). mongodb_structs() -> [ @@ -249,3 +252,15 @@ clickhouse_structs() -> } )} ]. + +sqlserver_structs() -> + [ + {sqlserver, + mk( + hoconsc:map(name, ref(emqx_ee_bridge_sqlserver, "config")), + #{ + desc => <<"Microsoft SQL Server Bridge Config">>, + required => false + } + )} + ]. diff --git a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_sqlserver.erl b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_sqlserver.erl new file mode 100644 index 000000000..49a5ed0ce --- /dev/null +++ b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_sqlserver.erl @@ -0,0 +1,128 @@ +%%-------------------------------------------------------------------- +%% Copyright (c) 2022-2023 EMQ Technologies Co., Ltd. All Rights Reserved. +%%-------------------------------------------------------------------- +-module(emqx_ee_bridge_sqlserver). + +-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 +]). + +-export([ + namespace/0, + roots/0, + fields/1, + desc/1 +]). + +-define(DEFAULT_SQL, << + "insert into t_mqtt_msg(msgid, topic, qos, payload)" + "values (${id}, ${topic}, ${qos}, ${payload})" +>>). + +-define(DEFAULT_DRIVER, <<"ms-sqlserver-18">>). + +conn_bridge_examples(Method) -> + [ + #{ + <<"sqlserver">> => #{ + summary => <<"Microsoft SQL Server Bridge">>, + value => values(Method) + } + } + ]. + +values(get) -> + values(post); +values(post) -> + #{ + enable => true, + type => sqlserver, + name => <<"bar">>, + server => <<"127.0.0.1:1433">>, + database => <<"test">>, + pool_size => 8, + username => <<"sa">>, + password => <<"******">>, + sql => ?DEFAULT_SQL, + driver => ?DEFAULT_DRIVER, + 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 => async, + max_queue_bytes => ?DEFAULT_QUEUE_SIZE + } + }; +values(put) -> + values(post). + +%% ------------------------------------------------------------------------------------------------- +%% Hocon Schema Definitions +namespace() -> "bridge_sqlserver". + +roots() -> []. + +fields("config") -> + [ + {enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})}, + {sql, + mk( + binary(), + #{desc => ?DESC("sql_template"), default => ?DEFAULT_SQL, format => <<"sql">>} + )}, + {driver, mk(binary(), #{desc => ?DESC("driver"), default => ?DEFAULT_DRIVER})}, + {local_topic, + mk( + binary(), + #{desc => ?DESC("local_topic"), default => undefined} + )}, + {resource_opts, + mk( + ref(?MODULE, "creation_opts"), + #{ + required => false, + default => #{}, + desc => ?DESC(emqx_resource_schema, <<"resource_opts">>) + } + )} + ] ++ + (emqx_ee_connector_sqlserver:fields(config) -- + emqx_connector_schema_lib:prepare_statement_fields()); +fields("creation_opts") -> + emqx_resource_schema:fields("creation_opts"); +fields("post") -> + fields("post", sqlserver); +fields("put") -> + fields("config"); +fields("get") -> + emqx_bridge_schema:status_fields() ++ fields("post"). + +fields("post", Type) -> + [type_field(Type), name_field() | fields("config")]. + +desc("config") -> + ?DESC("desc_config"); +desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" -> + ["Configuration for Microsoft SQL Server using `", string:to_upper(Method), "` method."]; +desc("creation_opts" = Name) -> + emqx_resource_schema:desc(Name); +desc(_) -> + undefined. + +%% ------------------------------------------------------------------------------------------------- + +type_field(Type) -> + {type, mk(enum([Type]), #{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/src/emqx_ee_connector.app.src b/lib-ee/emqx_ee_connector/src/emqx_ee_connector.app.src index 2e6406d70..2e2ff2a62 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 @@ -5,13 +5,15 @@ {applications, [ kernel, stdlib, + ecpool, hstreamdb_erl, influxdb, tdengine, clickhouse, erlcloud, rocketmq, - ecql + ecql, + odbc ]}, {env, []}, {modules, []}, diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl new file mode 100644 index 000000000..4cc92e80a --- /dev/null +++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl @@ -0,0 +1,531 @@ +%%-------------------------------------------------------------------- +%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved. +%%-------------------------------------------------------------------- + +-module(emqx_ee_connector_sqlserver). + +-behaviour(emqx_resource). + +-include_lib("kernel/include/file.hrl"). +-include_lib("emqx/include/logger.hrl"). +-include_lib("emqx_resource/include/emqx_resource.hrl"). +-include_lib("emqx_ee_connector/include/emqx_ee_connector.hrl"). + +-include_lib("typerefl/include/types.hrl"). +-include_lib("hocon/include/hoconsc.hrl"). + +-include_lib("snabbkaffe/include/snabbkaffe.hrl"). + +%%==================================================================== +%% Exports +%%==================================================================== + +%% Hocon config schema exports +-export([ + roots/0, + fields/1 +]). + +%% callbacks for behaviour emqx_resource +-export([ + callback_mode/0, + is_buffer_supported/0, + on_start/2, + on_stop/2, + on_query/3, + on_batch_query/3, + on_query_async/4, + on_batch_query_async/4, + on_get_status/2 +]). + +%% callbacks for ecpool +-export([connect/1]). + +%% Internal exports used to execute code with ecpool worker +-export([do_get_status/2, worker_do_insert/3, do_async_reply/2]). + +-import(emqx_plugin_libs_rule, [str/1]). +-import(hoconsc, [mk/2, enum/1, ref/2]). + +-define(ACTION_SEND_MESSAGE, send_message). + +-define(SINGLE_INSERT, single_insert). + +-define(SYNC_QUERY_MODE, handover). +-define(ASYNC_QUERY_MODE(REPLY), {handover_async, {?MODULE, do_async_reply, [REPLY]}}). + +-define(SQLSERVER_HOST_OPTIONS, #{ + default_port => 1433 +}). + +-define(REQUEST_TIMEOUT(RESOURCE_OPTS), + maps:get(request_timeout, RESOURCE_OPTS, ?DEFAULT_REQUEST_TIMEOUT) +). + +-define(SINGLE_INSERT_TEMP, single_insert_temp). +-define(BATCH_INSERT_TEMP, batch_insert_temp). + +-define(BATCH_INSERT_PART, batch_insert_part). +-define(BATCH_PARAMS_TOKENS, batch_insert_tks). + +%% Copied from odbc reference page +%% https://www.erlang.org/doc/man/odbc.html + +-type bridge_id() :: binary(). +%% as returned by connect/2 +-type connection_reference() :: pid(). +-type time_out() :: milliseconds() | infinity. +-type sql() :: string() | binary(). +-type milliseconds() :: pos_integer(). +%% Tuple of column values e.g. one row of the result set. +%% it's a variable size tuple of column values. +-type row() :: tuple(). +%% Some kind of explanation of what went wrong +-type common_reason() :: connection_closed | extended_error() | term(). +%% extended error type with ODBC +%% and native database error codes, as well as the base reason that would have been +%% returned had extended_errors not been enabled. +-type extended_error() :: {string(), integer(), _Reason :: term()}. +%% Name of column in the result set +-type col_name() :: string(). +%% e.g. a list of the names of the selected columns in the result set. +-type col_names() :: [col_name()]. +%% A list of rows from the result set. +-type rows() :: list(row()). + +%% -type result_tuple() :: {updated, n_rows()} | {selected, col_names(), rows()}. +-type updated_tuple() :: {updated, n_rows()}. +-type selected_tuple() :: {selected, col_names(), rows()}. +%% The number of affected rows for UPDATE, +%% INSERT, or DELETE queries. For other query types the value +%% is driver defined, and hence should be ignored. +-type n_rows() :: integer(). + +%% These type was not used in this module, but we may use it later +%% -type odbc_data_type() :: +%% sql_integer +%% | sql_smallint +%% | sql_tinyint +%% | {sql_decimal, precision(), scale()} +%% | {sql_numeric, precision(), scale()} +%% | {sql_char, size()} +%% | {sql_wchar, size()} +%% | {sql_varchar, size()} +%% | {sql_wvarchar, size()} +%% | {sql_float, precision()} +%% | {sql_wlongvarchar, size()} +%% | {sql_float, precision()} +%% | sql_real +%% | sql_double +%% | sql_bit +%% | atom(). +%% -type precision() :: integer(). +%% -type scale() :: integer(). +%% -type size() :: integer(). + +-type state() :: #{ + poolname := binary(), + resource_opts := map(), + sql_templates := map() +}. + +%%==================================================================== +%% Configuration and default values +%%==================================================================== + +roots() -> + [{config, #{type => hoconsc:ref(?MODULE, config)}}]. + +fields(config) -> + [ + {server, server()} + | add_default_username(emqx_connector_schema_lib:relational_db_fields()) + ]. + +add_default_username(Fields) -> + lists:map( + fun + ({username, OrigUsernameFn}) -> + {username, add_default_fn(OrigUsernameFn, <<"sa">>)}; + (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, ?SQLSERVER_HOST_OPTIONS). + +%%==================================================================== +%% Callbacks defined in emqx_resource +%%==================================================================== + +callback_mode() -> async_if_possible. + +is_buffer_supported() -> false. + +on_start( + BridgeId = PoolName, + #{ + server := Server, + username := Username, + password := Password, + driver := Driver, + database := Database, + pool_size := PoolSize, + resource_opts := ResourceOpts + } = Config +) -> + ?SLOG(info, #{ + msg => "starting_sqlserver_connector", + connector => BridgeId, + config => emqx_misc:redact(Config) + }), + + ODBCDir = code:priv_dir(odbc), + OdbcserverDir = filename:join(ODBCDir, "bin/odbcserver"), + {ok, Info = #file_info{mode = Mode}} = file:read_file_info(OdbcserverDir), + case 33261 =:= Mode of + true -> + ok; + false -> + _ = file:write_file_info(OdbcserverDir, Info#file_info{mode = 33261}), + ok + end, + + Options = [ + {server, to_bin(Server)}, + {username, Username}, + {password, Password}, + {driver, Driver}, + {database, Database}, + {pool_size, PoolSize}, + {poolname, PoolName} + ], + + State = #{ + %% also BridgeId + poolname => PoolName, + sql_templates => parse_sql_template(Config), + resource_opts => ResourceOpts + }, + case emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Options) of + ok -> + {ok, State}; + {error, Reason} -> + ?tp( + sqlserver_connector_start_failed, + #{error => Reason} + ), + {error, Reason} + end. + +on_stop(InstanceId, #{poolname := PoolName} = _State) -> + ?SLOG(info, #{ + msg => "stopping_sqlserver_connector", + connector => InstanceId + }), + emqx_plugin_libs_pool:stop_pool(PoolName). + +-spec on_query( + bridge_id(), + {?ACTION_SEND_MESSAGE, map()}, + state() +) -> + ok + | {ok, list()} + | {error, {recoverable_error, term()}} + | {error, term()}. +on_query(BridgeId, {?ACTION_SEND_MESSAGE, _Msg} = Query, State) -> + ?TRACE( + "SINGLE_QUERY_SYNC", + "bridge_sqlserver_received", + #{requests => Query, connector => BridgeId, state => State} + ), + do_query(BridgeId, Query, ?SYNC_QUERY_MODE, State). + +-spec on_query_async( + bridge_id(), + {?ACTION_SEND_MESSAGE, map()}, + {ReplyFun :: function(), Args :: list()}, + state() +) -> + {ok, any()} + | {error, term()}. +on_query_async( + BridgeId, + {?ACTION_SEND_MESSAGE, _Msg} = Query, + ReplyFunAndArgs, + %% #{poolname := PoolName, sql_templates := Templates} = State + State +) -> + ?TRACE( + "SINGLE_QUERY_ASYNC", + "bridge_sqlserver_received", + #{requests => Query, connector => BridgeId, state => State} + ), + do_query(BridgeId, Query, ?ASYNC_QUERY_MODE(ReplyFunAndArgs), State). + +-spec on_batch_query( + bridge_id(), + [{?ACTION_SEND_MESSAGE, map()}], + state() +) -> + ok + | {ok, list()} + | {error, {recoverable_error, term()}} + | {error, term()}. +on_batch_query(BridgeId, BatchRequests, State) -> + ?TRACE( + "BATCH_QUERY_SYNC", + "bridge_sqlserver_received", + #{requests => BatchRequests, connector => BridgeId, state => State} + ), + do_query(BridgeId, BatchRequests, ?SYNC_QUERY_MODE, State). + +-spec on_batch_query_async( + bridge_id(), + [{?ACTION_SEND_MESSAGE, map()}], + {ReplyFun :: function(), Args :: list()}, + state() +) -> {ok, any()}. +on_batch_query_async(BridgeId, Requests, ReplyFunAndArgs, State) -> + ?TRACE( + "BATCH_QUERY_ASYNC", + "bridge_sqlserver_received", + #{requests => Requests, connector => BridgeId, state => State} + ), + do_query(BridgeId, Requests, ?ASYNC_QUERY_MODE(ReplyFunAndArgs), State). + +on_get_status(_InstanceId, #{poolname := Pool, resource_opts := ResourceOpts} = _State) -> + RequestTimeout = ?REQUEST_TIMEOUT(ResourceOpts), + Health = emqx_plugin_libs_pool:health_check_ecpool_workers( + Pool, {?MODULE, do_get_status, [RequestTimeout]}, RequestTimeout + ), + status_result(Health). + +status_result(_Status = true) -> connected; +status_result(_Status = false) -> connecting. +%% TODO: +%% case for disconnected + +%%==================================================================== +%% ecpool callback fns +%%==================================================================== + +-spec connect(Options :: list()) -> {ok, connection_reference()} | {error, term()}. +connect(Options) -> + ConnectStr = lists:concat(conn_str(Options, [])), + Opts = proplists:get_value(options, Options, []), + odbc:connect(ConnectStr, Opts). + +-spec do_get_status(connection_reference(), time_out()) -> Result :: boolean(). +do_get_status(Conn, RequestTimeout) -> + case execute(Conn, <<"SELECT 1">>, RequestTimeout) of + {selected, [[]], [{1}]} -> true; + _ -> false + end. + +%%==================================================================== +%% Internal Helper fns +%%==================================================================== + +%% TODO && FIXME: +%% About the connection string attribute `Encrypt`: +%% The default value is `yes` in odbc version 18.0+ and `no` in previous versions. +%% And encrypted connections always verify the server's certificate. +%% So `Encrypt=YES;TrustServerCertificate=YES` must be set in the connection string when connecting to a server that has a self-signed certificate. +%% See also: +%% https://learn.microsoft.com/en-us/sql/connect/odbc/dsn-connection-string-attribute?source=recommendations&view=sql-server-ver16#encrypt +conn_str([], Acc) -> + %% we should use this for msodbcsql 18+ + %% lists:join(";", ["Encrypt=YES", "TrustServerCertificate=YES" | Acc]); + lists:join(";", Acc); +conn_str([{driver, Driver} | Opts], Acc) -> + conn_str(Opts, ["Driver=" ++ str(Driver) | Acc]); +conn_str([{server, Server} | Opts], Acc) -> + {Host, Port} = emqx_schema:parse_server(Server, ?SQLSERVER_HOST_OPTIONS), + conn_str(Opts, ["Server=" ++ str(Host) ++ "," ++ str(Port) | Acc]); +conn_str([{database, Database} | Opts], Acc) -> + conn_str(Opts, ["Database=" ++ str(Database) | Acc]); +conn_str([{username, Username} | Opts], Acc) -> + conn_str(Opts, ["UID=" ++ str(Username) | Acc]); +conn_str([{password, Password} | Opts], Acc) -> + conn_str(Opts, ["PWD=" ++ str(Password) | Acc]); +conn_str([{_, _} | Opts], Acc) -> + conn_str(Opts, Acc). + +%% Sync & Async query with singe & batch sql statement +-spec do_query( + bridge_id(), + Query :: {?ACTION_SEND_MESSAGE, map()} | [{?ACTION_SEND_MESSAGE, map()}], + ApplyMode :: + handover + | {handover_async, {?MODULE, do_async_reply, [{ReplyFun :: function(), Args :: list()}]}}, + state() +) -> + {ok, list()} + | {error, {recoverable_error, term()}} + | {error, term()}. +do_query( + BridgeId, + Query, + ApplyMode, + #{poolname := PoolName, sql_templates := Templates} = State +) -> + ?TRACE( + "SINGLE_QUERY_SYNC", + "sqlserver_connector_received", + #{query => Query, connector => BridgeId, state => State} + ), + + %% only insert sql statement for single query and batch query + case apply_template(Query, Templates) of + {?ACTION_SEND_MESSAGE, SQL} -> + Result = ecpool:pick_and_do( + PoolName, + {?MODULE, worker_do_insert, [SQL, State]}, + ApplyMode + ); + Query -> + Result = {error, {unrecoverable_error, invalid_query}}; + _ -> + Result = {error, {unrecoverable_error, failed_to_apply_sql_template}} + end, + case Result of + {error, Reason} -> + ?tp( + sqlserver_connector_query_return, + #{error => Reason} + ), + ?SLOG(error, #{ + msg => "sqlserver_connector_do_query_failed", + connector => BridgeId, + query => Query, + reason => Reason + }), + Result; + _ -> + ?tp( + sqlserver_connector_query_return, + #{result => Result} + ), + Result + end. + +worker_do_insert( + Conn, SQL, #{resource_opts := ResourceOpts, poolname := BridgeId} = State +) -> + LogMeta = #{connector => BridgeId, state => State}, + try + case execute(Conn, SQL, ?REQUEST_TIMEOUT(ResourceOpts)) of + {selected, Rows, _} -> + {ok, Rows}; + {updated, _} -> + ok; + {error, ErrStr} -> + ?SLOG(error, LogMeta#{msg => "invalid_request", reason => ErrStr}), + {error, {unrecoverable_error, {invalid_request, ErrStr}}} + end + catch + _Type:Reason -> + ?SLOG(error, LogMeta#{msg => "invalid_request", reason => Reason}), + {error, {unrecoverable_error, {invalid_request, Reason}}} + end. + +-spec execute(pid(), sql(), time_out()) -> + updated_tuple() + | selected_tuple() + | [updated_tuple()] + | [selected_tuple()] + | {error, common_reason()}. +execute(Conn, SQL, Timeout) -> + odbc:sql_query(Conn, str(SQL), Timeout). + +to_bin(List) when is_list(List) -> + unicode:characters_to_binary(List, utf8). + +%% for bridge data to sql server +parse_sql_template(Config) -> + RawSQLTemplates = + case maps:get(sql, Config, undefined) of + undefined -> #{}; + <<>> -> #{}; + SQLTemplate -> #{?ACTION_SEND_MESSAGE => SQLTemplate} + end, + + SingleInsertTks = #{}, + BatchInsertTks = #{}, + parse_sql_template(maps:to_list(RawSQLTemplates), SingleInsertTks, BatchInsertTks). + +parse_sql_template([{Key, H} | T], SingleInsertTks, BatchInsertTks) -> + case emqx_plugin_libs_rule:detect_sql_type(H) of + {ok, select} -> + parse_sql_template(T, SingleInsertTks, BatchInsertTks); + {ok, insert} -> + case emqx_plugin_libs_rule:split_insert_sql(H) of + {ok, {InsertSQL, Params}} -> + parse_sql_template( + T, + SingleInsertTks#{Key => emqx_plugin_libs_rule:preproc_tmpl(H)}, + BatchInsertTks#{ + Key => + #{ + ?BATCH_INSERT_PART => InsertSQL, + ?BATCH_PARAMS_TOKENS => emqx_plugin_libs_rule:preproc_tmpl( + Params + ) + } + } + ); + {error, Reason} -> + ?SLOG(error, #{msg => "split sql failed", sql => H, reason => Reason}), + parse_sql_template(T, SingleInsertTks, BatchInsertTks) + end; + {error, Reason} -> + ?SLOG(error, #{msg => "detect sql type failed", sql => H, reason => Reason}), + parse_sql_template(T, SingleInsertTks, BatchInsertTks) + end; +parse_sql_template([], SingleInsertTks, BatchInsertTks) -> + #{ + ?SINGLE_INSERT_TEMP => SingleInsertTks, + ?BATCH_INSERT_TEMP => BatchInsertTks + }. + +%% single insert +apply_template( + {?ACTION_SEND_MESSAGE = Key, Msg} = Query, + #{?SINGLE_INSERT_TEMP := SingleInsertTks} = _Templates +) -> + case maps:get(Key, SingleInsertTks, undefined) of + undefined -> + Query; + Template -> + {Key, emqx_plugin_libs_rule:proc_tmpl(Template, Msg)} + end; +%% batch inserts +apply_template( + [{?ACTION_SEND_MESSAGE = Key, _Msg} | _T] = BatchReqs, + #{?BATCH_INSERT_TEMP := BatchInsertsTks} = _Templates +) -> + case maps:get(Key, BatchInsertsTks, undefined) of + undefined -> + BatchReqs; + #{?BATCH_INSERT_PART := BatchInserts, ?BATCH_PARAMS_TOKENS := BatchParamsTks} -> + SQL = emqx_plugin_libs_rule:proc_batch_sql(BatchReqs, BatchInserts, BatchParamsTks), + {Key, SQL} + end; +apply_template(Query, Templates) -> + %% TODO: more detail infomatoin + ?SLOG(error, #{msg => "apply sql template failed", query => Query, templates => Templates}). + +do_async_reply(Result, {ReplyFun, Args}) -> + erlang:apply(ReplyFun, Args ++ Result). diff --git a/rel/i18n/emqx_ee_bridge_sqlserver.hocon b/rel/i18n/emqx_ee_bridge_sqlserver.hocon new file mode 100644 index 000000000..bb82c82f8 --- /dev/null +++ b/rel/i18n/emqx_ee_bridge_sqlserver.hocon @@ -0,0 +1,87 @@ +emqx_ee_bridge_sqlserver { + + local_topic { + desc { + en: """The MQTT topic filter to be forwarded to MicroSoft SQL Server. All MQTT 'PUBLISH' messages with the topic +matching the local_topic will be forwarded.
+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' 的消息都会转发到 MicroSoft SQL Server。
+注意：如果这个 Bridge 被用作规则（EMQX 规则引擎）的输出，同时也配置了 'local_topic' ，那么这两部分的消息都会被转发。 +""" + } + label { + en: """Local Topic""" + zh: """本地 Topic""" + } + } + + sql_template { + desc { + en: """SQL Template""" + zh: """SQL 模板""" + } + label { + en: """SQL Template""" + zh: """SQL 模板""" + } + } + + driver { + desc { + en: """SQL Server Driver Name""" + zh: """SQL Server Driver 名称""" + } + desc { + en: """SQL Server Driver Name""" + zh: """SQL Server Driver 名称""" + } + } + + config_enable { + desc { + en: """Enable or disable this bridge""" + zh: """启用/禁用桥接""" + } + label { + en: """Enable Or Disable Bridge""" + zh: """启用/禁用桥接""" + } + } + + desc_config { + desc { + en: """Configuration for an MicroSoft SQL Server bridge.""" + zh: """MicroSoft SQL Server 桥接配置""" + } + label: { + en: """MicroSoft SQL Server Bridge Configuration""" + zh: """MicroSoft SQL Server 桥接配置""" + } + } + + 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/rel/i18n/emqx_ee_connector_sqlserver.hocon b/rel/i18n/emqx_ee_connector_sqlserver.hocon new file mode 100644 index 000000000..9ff24c1c1 --- /dev/null +++ b/rel/i18n/emqx_ee_connector_sqlserver.hocon @@ -0,0 +1,22 @@ +emqx_ee_connector_sqlserver { + + server { + desc { + en: """ +The IPv4 or IPv6 address or the hostname to connect to.
+A host entry has the following form: `Host[:Port]`.
+The SQL Server default port 1433 is used if `[:Port]` is not specified. +""" + zh: """ +将要连接的 IPv4 或 IPv6 地址，或者主机名。
+主机名具有以下形式：`Host[:Port]`。
+如果未指定 `[:Port]`，则使用 SQL Server 默认端口 1433。 +""" + } + label: { + en: "Server Host" + zh: "服务器地址" + } + } + +} From 611edf422719069297219370fe120968737fc0cf Mon Sep 17 00:00:00 2001 From: JimMoen Date: Mon, 10 Apr 2023 14:18:25 +0800 Subject: [PATCH 04/22] build: docker file add unixodbc and msodbcsql17 --- .gitignore | 1 + deploy/docker/Dockerfile | 16 +++++++++++++--- 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/.gitignore b/.gitignore index 62e8ddc81..3f7e268a2 100644 --- a/.gitignore +++ b/.gitignore @@ -71,3 +71,4 @@ apps/emqx/test/emqx_static_checks_data/master.bpapi lux_logs/ /.prepare bom.json +.dockerignore diff --git a/deploy/docker/Dockerfile b/deploy/docker/Dockerfile index 8f04c433c..dcebe9b1a 100644 --- a/deploy/docker/Dockerfile +++ b/deploy/docker/Dockerfile @@ -29,9 +29,19 @@ COPY --from=builder /emqx-rel/emqx /opt/emqx RUN ln -s /opt/emqx/bin/* /usr/local/bin/ -RUN apt-get update; \ - apt-get install -y --no-install-recommends ca-certificates procps; \ - rm -rf /var/lib/apt/lists/* +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates procps + +RUN apt-get update \ + && apt-get install -y gnupg2 curl apt-utils \ + && curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add - \ + && curl https://packages.microsoft.com/config/debian/11/prod.list > /etc/apt/sources.list.d/mssql-release.list \ + && apt-get update \ + && ACCEPT_EULA=Y apt-get install -y msodbcsql17 unixodbc-dev \ + && sed -i 's/ODBC Driver 17 for SQL Server/ms-sql/g' /etc/odbcinst.ini + +RUN apt-get clean && \ + rm -rf /var/lib/apt/lists/*; WORKDIR /opt/emqx From c366267b0faf7290ce5479690acfe2fefb0a2a9b Mon Sep 17 00:00:00 2001 From: JimMoen Date: Fri, 7 Apr 2023 09:59:07 +0800 Subject: [PATCH 05/22] test: MS SQL Server data bridge --- .ci/docker-compose-file/.env | 3 + .../docker-compose-sqlserver.yaml | 19 + .../docker-compose-toxiproxy.yaml | 1 + .ci/docker-compose-file/docker-compose.yaml | 1 + .ci/docker-compose-file/odbc/odbcinst.ini | 10 + .ci/docker-compose-file/toxiproxy.json | 6 + lib-ee/emqx_ee_bridge/docker-ct | 1 + .../test/emqx_ee_bridge_sqlserver_SUITE.erl | 654 ++++++++++++++++++ lib-ee/emqx_ee_connector/docker-ct | 1 + scripts/ct/run.sh | 3 + 10 files changed, 699 insertions(+) create mode 100644 .ci/docker-compose-file/docker-compose-sqlserver.yaml create mode 100644 .ci/docker-compose-file/odbc/odbcinst.ini create mode 100644 lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl diff --git a/.ci/docker-compose-file/.env b/.ci/docker-compose-file/.env index 956750e00..b394d6bb1 100644 --- a/.ci/docker-compose-file/.env +++ b/.ci/docker-compose-file/.env @@ -8,4 +8,7 @@ TDENGINE_TAG=3.0.2.4 DYNAMO_TAG=1.21.0 CASSANDRA_TAG=3.11.6 +MS_IMAGE_ADDR=mcr.microsoft.com/mssql/server +SQLSERVER_TAG=2019-CU11-ubuntu-20.04 + TARGET=emqx/emqx diff --git a/.ci/docker-compose-file/docker-compose-sqlserver.yaml b/.ci/docker-compose-file/docker-compose-sqlserver.yaml new file mode 100644 index 000000000..37799e20b --- /dev/null +++ b/.ci/docker-compose-file/docker-compose-sqlserver.yaml @@ -0,0 +1,19 @@ +version: '3.9' + +services: + sql_server: + container_name: sqlserver + # See also: + # https://mcr.microsoft.com/en-us/product/mssql/server/about + # https://hub.docker.com/_/microsoft-mssql-server + image: ${MS_IMAGE_ADDR}:${SQLSERVER_TAG} + environment: + # See also: + # https://learn.microsoft.com/en-us/sql/linux/sql-server-linux-configure-environment-variables + ACCEPT_EULA: "Y" + SA_PASSWORD: "mqtt_public" + restart: always + # ports: + # - "1433:1433" + networks: + - emqx_bridge diff --git a/.ci/docker-compose-file/docker-compose-toxiproxy.yaml b/.ci/docker-compose-file/docker-compose-toxiproxy.yaml index 9a1d08ba6..ba5e831a5 100644 --- a/.ci/docker-compose-file/docker-compose-toxiproxy.yaml +++ b/.ci/docker-compose-file/docker-compose-toxiproxy.yaml @@ -16,6 +16,7 @@ services: - 8474:8474 - 8086:8086 - 8087:8087 + - 11433:1433 - 13306:3306 - 13307:3307 - 15432:5432 diff --git a/.ci/docker-compose-file/docker-compose.yaml b/.ci/docker-compose-file/docker-compose.yaml index 48d900400..8f97fa9d6 100644 --- a/.ci/docker-compose-file/docker-compose.yaml +++ b/.ci/docker-compose-file/docker-compose.yaml @@ -24,6 +24,7 @@ services: - /tmp/emqx-ci/emqx-shared-secret:/var/lib/secret - ./kerberos/krb5.conf:/etc/kdc/krb5.conf - ./kerberos/krb5.conf:/etc/krb5.conf + - ./odbc/odbcinst.ini:/etc/odbcinst.ini working_dir: /emqx tty: true user: "${DOCKER_USER:-root}" diff --git a/.ci/docker-compose-file/odbc/odbcinst.ini b/.ci/docker-compose-file/odbc/odbcinst.ini new file mode 100644 index 000000000..16701cfd2 --- /dev/null +++ b/.ci/docker-compose-file/odbc/odbcinst.ini @@ -0,0 +1,10 @@ +[ms-sql] +Description=Microsoft ODBC Driver 17 for SQL Server +Driver=/opt/microsoft/msodbcsql17/lib64/libmsodbcsql-17.10.so.2.1 +UsageCount=1 + +[ODBC Driver 17 for SQL Server] +Description=Microsoft ODBC Driver 17 for SQL Server +Driver=/opt/microsoft/msodbcsql17/lib64/libmsodbcsql-17.10.so.2.1 +UsageCount=1 + diff --git a/.ci/docker-compose-file/toxiproxy.json b/.ci/docker-compose-file/toxiproxy.json index 708cbf1ef..da2dff763 100644 --- a/.ci/docker-compose-file/toxiproxy.json +++ b/.ci/docker-compose-file/toxiproxy.json @@ -95,5 +95,11 @@ "listen": "0.0.0.0:9142", "upstream": "cassandra:9142", "enabled": true + }, + { + "name": "sqlserver", + "listen": "0.0.0.0:1433", + "upstream": "sqlserver:1433", + "enabled": true } ] diff --git a/lib-ee/emqx_ee_bridge/docker-ct b/lib-ee/emqx_ee_bridge/docker-ct index 0e947d89e..35d6b9d5b 100644 --- a/lib-ee/emqx_ee_bridge/docker-ct +++ b/lib-ee/emqx_ee_bridge/docker-ct @@ -11,3 +11,4 @@ clickhouse dynamo rocketmq cassandra +sqlserver diff --git a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl new file mode 100644 index 000000000..8b0cc53ce --- /dev/null +++ b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl @@ -0,0 +1,654 @@ +%%-------------------------------------------------------------------- +% Copyright (c) 2022-2023 EMQ Technologies Co., Ltd. All Rights Reserved. +%%-------------------------------------------------------------------- + +-module(emqx_ee_bridge_sqlserver_SUITE). + +-compile(nowarn_export_all). +-compile(export_all). + +-include_lib("eunit/include/eunit.hrl"). +-include_lib("common_test/include/ct.hrl"). +-include_lib("snabbkaffe/include/snabbkaffe.hrl"). + +% SQL definitions +-define(SQL_BRIDGE, + "insert into t_mqtt_msg(msgid, topic, qos, payload) values ( ${id}, ${topic}, ${qos}, ${payload})" +). +-define(SQL_SERVER_DRIVER, "ms-sql"). + +-define(SQL_CREATE_DATABASE_IF_NOT_EXISTS, + " IF NOT EXISTS(SELECT name FROM sys.databases WHERE name = 'mqtt')" + " BEGIN" + " CREATE DATABASE mqtt;" + " END" +). + +-define(SQL_CREATE_TABLE_IN_DB_MQTT, + " CREATE TABLE mqtt.dbo.t_mqtt_msg" + " (id int PRIMARY KEY IDENTITY(1000000001,1) NOT NULL," + " msgid VARCHAR(64) NULL," + " topic VARCHAR(100) NULL," + " qos tinyint NOT NULL DEFAULT 0," + " payload NVARCHAR(100) NULL," + " arrived DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP)" +). + +-define(SQL_DROP_DB_MQTT, "DROP DATABASE mqtt"). +-define(SQL_DROP_TABLE, "DROP TABLE mqtt.dbo.t_mqtt_msg"). +-define(SQL_DELETE, "DELETE from mqtt.dbo.t_mqtt_msg"). +-define(SQL_SELECT, "SELECT payload FROM mqtt.dbo.t_mqtt_msg"). +-define(SQL_SELECT_COUNT, "SELECT COUNT(*) FROM mqtt.dbo.t_mqtt_msg"). +% DB defaults +-define(SQL_SERVER_DATABASE, "mqtt"). +-define(SQL_SERVER_USERNAME, "sa"). +-define(SQL_SERVER_PASSWORD, "mqtt_public"). +-define(BATCH_SIZE, 10). +-define(REQUEST_TIMEOUT_MS, 500). + +-define(WORKER_POOL_SIZE, 4). + +-define(WITH_CON(Process), + Con = connect_direct_sqlserver(Config), + Process, + ok = disconnect(Con) +). + +%% How to run it locally: +%% A: run ct on host +%% 1. Start all deps services +%% sudo docker compose -f .ci/docker-compose-file/docker-compose.yaml \ +%% -f .ci/docker-compose-file/docker-compose-sqlserver.yaml \ +%% -f .ci/docker-compose-file/docker-compose-toxiproxy.yaml \ +%% up --build +%% +%% 2. Run use cases with special environment variables +%% 11433 is toxiproxy exported port +%% Local: +%% ``` +%% SQLSERVER_HOST=toxiproxy SQLSERVER_PORT=11433 \ +%% PROXY_HOST=toxiproxy PROXY_PORT=1433 \ +%% ./rebar3 as test ct -c -v --readable true --name ct@127.0.0.1 --suite lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl +%% ``` +%% +%% B: run ct in docker container +%% run script: +%% ./scripts/ct/run.sh --ci --app lib-ee/emqx_ee_bridge/ \ +%% -- --name 'test@127.0.0.1' -c -v --readable true --suite lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl + +%%------------------------------------------------------------------------------ +%% CT boilerplate +%%------------------------------------------------------------------------------ + +all() -> + [ + {group, async}, + {group, sync} + ]. + +groups() -> + TCs = emqx_common_test_helpers:all(?MODULE), + NonBatchCases = [t_write_timeout], + BatchingGroups = [{group, with_batch}, {group, without_batch}], + [ + {async, BatchingGroups}, + {sync, BatchingGroups}, + {with_batch, TCs -- NonBatchCases}, + {without_batch, TCs} + ]. + +init_per_group(async, Config) -> + [{query_mode, async} | Config]; +init_per_group(sync, Config) -> + [{query_mode, sync} | Config]; +init_per_group(with_batch, Config0) -> + Config = [{enable_batch, true} | Config0], + common_init(Config); +init_per_group(without_batch, Config0) -> + Config = [{enable_batch, false} | Config0], + common_init(Config); +init_per_group(_Group, Config) -> + Config. + +end_per_group(Group, Config) when Group =:= with_batch; Group =:= without_batch -> + connect_and_drop_table(Config), + connect_and_drop_db(Config), + ProxyHost = ?config(proxy_host, Config), + ProxyPort = ?config(proxy_port, Config), + emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort), + ok; +end_per_group(_Group, _Config) -> + ok. + +init_per_suite(Config) -> + Config. + +end_per_suite(_Config) -> + emqx_mgmt_api_test_util:end_suite(), + ok = emqx_common_test_helpers:stop_apps([emqx_bridge, emqx_conf]), + ok. + +init_per_testcase(_Testcase, Config) -> + %% drop database and table + %% connect_and_clear_table(Config), + %% create a new one + %% TODO: create a new database for each test case + delete_bridge(Config), + snabbkaffe:start_trace(), + Config. + +end_per_testcase(_Testcase, Config) -> + ProxyHost = ?config(proxy_host, Config), + ProxyPort = ?config(proxy_port, Config), + emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort), + connect_and_clear_table(Config), + ok = snabbkaffe:stop(), + delete_bridge(Config), + ok. + +%%------------------------------------------------------------------------------ +%% Testcases +%%------------------------------------------------------------------------------ + +t_setup_via_config_and_publish(Config) -> + ?assertMatch( + {ok, _}, + create_bridge(Config) + ), + Val = str(erlang:unique_integer()), + SentData = sent_data(Val), + ?check_trace( + begin + ?wait_async_action( + ?assertEqual(ok, send_message(Config, SentData)), + #{?snk_kind := sqlserver_connector_query_return}, + 10_000 + ), + ?assertMatch( + [{Val}], + connect_and_get_payload(Config) + ), + ok + end, + fun(Trace0) -> + Trace = ?of_kind(sqlserver_connector_query_return, Trace0), + ?assertMatch([#{result := ok}], Trace), + ok + end + ), + ok. + +t_setup_via_http_api_and_publish(Config) -> + BridgeType = ?config(sqlserver_bridge_type, Config), + Name = ?config(sqlserver_name, Config), + SQLServerConfig0 = ?config(sqlserver_config, Config), + SQLServerConfig = SQLServerConfig0#{ + <<"name">> => Name, + <<"type">> => BridgeType + }, + ?assertMatch( + {ok, _}, + create_bridge_http(SQLServerConfig) + ), + Val = str(erlang:unique_integer()), + SentData = sent_data(Val), + ?check_trace( + begin + ?wait_async_action( + ?assertEqual(ok, send_message(Config, SentData)), + #{?snk_kind := sqlserver_connector_query_return}, + 10_000 + ), + ?assertMatch( + [{Val}], + connect_and_get_payload(Config) + ), + ok + end, + fun(Trace0) -> + Trace = ?of_kind(sqlserver_connector_query_return, Trace0), + ?assertMatch([#{result := ok}], Trace), + ok + end + ), + ok. + +t_get_status(Config) -> + ?assertMatch( + {ok, _}, + create_bridge(Config) + ), + ProxyPort = ?config(proxy_port, Config), + ProxyHost = ?config(proxy_host, Config), + ProxyName = ?config(proxy_name, Config), + + Name = ?config(sqlserver_name, Config), + BridgeType = ?config(sqlserver_bridge_type, Config), + ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name), + + ?assertEqual({ok, connected}, emqx_resource_manager:health_check(ResourceID)), + emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() -> + case emqx_resource_manager:health_check(ResourceID) of + {ok, Status} when Status =:= disconnected orelse Status =:= connecting -> + ok; + {error, timeout} -> + ok; + Other -> + ?assert( + false, lists:flatten(io_lib:format("invalid health check result:~p~n", [Other])) + ) + end + end), + ok. + +t_create_disconnected(Config) -> + ProxyPort = ?config(proxy_port, Config), + ProxyHost = ?config(proxy_host, Config), + ProxyName = ?config(proxy_name, Config), + ?check_trace( + emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() -> + ?assertMatch({ok, _}, create_bridge(Config)) + end), + fun(Trace) -> + ?assertMatch( + [#{error := {start_pool_failed, _, _}}], + ?of_kind(sqlserver_connector_start_failed, Trace) + ), + ok + end + ), + ok. + +t_write_failure(Config) -> + ProxyName = ?config(proxy_name, Config), + ProxyPort = ?config(proxy_port, Config), + ProxyHost = ?config(proxy_host, Config), + QueryMode = ?config(query_mode, Config), + Val = str(erlang:unique_integer()), + SentData = sent_data(Val), + {{ok, _}, {ok, _}} = + ?wait_async_action( + create_bridge(Config), + #{?snk_kind := resource_connected_enter}, + 20_000 + ), + emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() -> + case QueryMode of + sync -> + ?assertMatch( + {error, {resource_error, #{reason := timeout}}}, + send_message(Config, SentData) + ); + async -> + ?assertMatch( + ok, send_message(Config, SentData) + ) + end + end), + ok. + +t_write_timeout(_Config) -> + %% msodbc driver handled all connection exceptions + %% the case is same as t_write_failure/1 + ok. + +t_simple_query(Config) -> + BatchSize = batch_size(Config), + ?assertMatch( + {ok, _}, + create_bridge(Config) + ), + {Requests, Vals} = gen_batch_req(BatchSize), + ?check_trace( + begin + ?wait_async_action( + begin + [?assertEqual(ok, query_resource(Config, Request)) || Request <- Requests] + end, + #{?snk_kind := sqlserver_connector_query_return}, + 10_000 + ), + %% just assert the data count is correct + ?assertMatch( + BatchSize, + connect_and_get_count(Config) + ), + %% assert the data order is correct + ?assertMatch( + Vals, + connect_and_get_payload(Config) + ) + end, + fun(Trace0) -> + Trace = ?of_kind(sqlserver_connector_query_return, Trace0), + case BatchSize of + 1 -> + ?assertMatch([#{result := ok}], Trace); + _ -> + [?assertMatch(#{result := ok}, Trace1) || Trace1 <- Trace] + end, + ok + end + ), + ok. + +-define(MISSING_TINYINT_ERROR, + "[Microsoft][ODBC Driver 17 for SQL Server][SQL Server]" + "Conversion failed when converting the varchar value 'undefined' to data type tinyint. SQLSTATE IS: 22018" +). + +t_missing_data(Config) -> + QueryMode = ?config(query_mode, Config), + ?assertMatch( + {ok, _}, + create_bridge(Config) + ), + Result = send_message(Config, #{}), + case QueryMode of + sync -> + ?assertMatch( + {error, {unrecoverable_error, {invalid_request, ?MISSING_TINYINT_ERROR}}}, + Result + ); + async -> + ?assertMatch( + ok, send_message(Config, #{}) + ) + end, + ok. + +t_bad_parameter(Config) -> + QueryMode = ?config(query_mode, Config), + ?assertMatch( + {ok, _}, + create_bridge(Config) + ), + Result = send_message(Config, #{}), + case QueryMode of + sync -> + ?assertMatch( + {error, {unrecoverable_error, {invalid_request, ?MISSING_TINYINT_ERROR}}}, + Result + ); + async -> + ?assertMatch( + ok, send_message(Config, #{}) + ) + end, + ok. + +%%------------------------------------------------------------------------------ +%% Helper fns +%%------------------------------------------------------------------------------ + +common_init(ConfigT) -> + Host = os:getenv("SQLSERVER_HOST", "toxiproxy"), + Port = list_to_integer(os:getenv("SQLSERVER_PORT", "1433")), + + Config0 = [ + {sqlserver_host, Host}, + {sqlserver_port, Port}, + %% see also for `proxy_name` : $PROJ_ROOT/.ci/docker-compose-file/toxiproxy.json + {proxy_name, "sqlserver"}, + {batch_size, ?BATCH_SIZE} + | ConfigT + ], + + BridgeType = proplists:get_value(bridge_type, Config0, <<"sqlserver">>), + case emqx_common_test_helpers:is_tcp_server_available(Host, Port) of + true -> + % Setup toxiproxy + ProxyHost = os:getenv("PROXY_HOST", "toxiproxy"), + ProxyPort = list_to_integer(os:getenv("PROXY_PORT", "8474")), + emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort), + % Ensure EE bridge module is loaded + _ = application:load(emqx_ee_bridge), + _ = emqx_ee_bridge:module_info(), + ok = emqx_common_test_helpers:start_apps([emqx_conf, emqx_bridge]), + emqx_mgmt_api_test_util:init_suite(), + % Connect to sqlserver directly + % drop old db and table, and then create new ones + connect_and_create_db_and_table(Config0), + {Name, SQLServerConf} = sqlserver_config(BridgeType, Config0), + Config = + [ + {sqlserver_config, SQLServerConf}, + {sqlserver_bridge_type, BridgeType}, + {sqlserver_name, Name}, + {proxy_host, ProxyHost}, + {proxy_port, ProxyPort} + | Config0 + ], + Config; + false -> + case os:getenv("IS_CI") of + "yes" -> + throw(no_sqlserver); + _ -> + {skip, no_sqlserver} + end + end. + +sqlserver_config(BridgeType, Config) -> + Port = integer_to_list(?config(sqlserver_port, Config)), + Server = ?config(sqlserver_host, Config) ++ ":" ++ Port, + Name = atom_to_binary(?MODULE), + BatchSize = batch_size(Config), + QueryMode = ?config(query_mode, Config), + ConfigString = + io_lib:format( + "bridges.~s.~s {\n" + " enable = true\n" + " server = ~p\n" + " database = ~p\n" + " username = ~p\n" + " password = ~p\n" + " sql = ~p\n" + " driver = ~p\n" + " resource_opts = {\n" + " request_timeout = 500ms\n" + " batch_size = ~b\n" + " query_mode = ~s\n" + " worker_pool_size = ~b\n" + " }\n" + "}", + [ + BridgeType, + Name, + Server, + ?SQL_SERVER_DATABASE, + ?SQL_SERVER_USERNAME, + ?SQL_SERVER_PASSWORD, + ?SQL_BRIDGE, + ?SQL_SERVER_DRIVER, + BatchSize, + QueryMode, + ?WORKER_POOL_SIZE + ] + ), + {Name, parse_and_check(ConfigString, BridgeType, Name)}. + +parse_and_check(ConfigString, BridgeType, Name) -> + {ok, RawConf} = hocon:binary(ConfigString, #{format => map}), + hocon_tconf:check_plain(emqx_bridge_schema, RawConf, #{required => false, atom_key => false}), + #{<<"bridges">> := #{BridgeType := #{Name := Config}}} = RawConf, + Config. + +create_bridge(Config) -> + create_bridge(Config, _Overrides = #{}). + +create_bridge(Config, Overrides) -> + BridgeType = ?config(sqlserver_bridge_type, Config), + Name = ?config(sqlserver_name, Config), + SSConfig0 = ?config(sqlserver_config, Config), + SSConfig = emqx_map_lib:deep_merge(SSConfig0, Overrides), + emqx_bridge:create(BridgeType, Name, SSConfig). + +delete_bridge(Config) -> + BridgeType = ?config(sqlserver_bridge_type, Config), + Name = ?config(sqlserver_name, Config), + emqx_bridge:remove(BridgeType, Name). + +create_bridge_http(Params) -> + Path = emqx_mgmt_api_test_util:api_path(["bridges"]), + AuthHeader = emqx_mgmt_api_test_util:auth_header_(), + case emqx_mgmt_api_test_util:request_api(post, Path, "", AuthHeader, Params) of + {ok, Res} -> {ok, emqx_json:decode(Res, [return_maps])}; + Error -> Error + end. + +send_message(Config, Payload) -> + Name = ?config(sqlserver_name, Config), + BridgeType = ?config(sqlserver_bridge_type, Config), + BridgeID = emqx_bridge_resource:bridge_id(BridgeType, Name), + emqx_bridge:send_message(BridgeID, Payload). + +query_resource(Config, Request) -> + Name = ?config(sqlserver_name, Config), + BridgeType = ?config(sqlserver_bridge_type, Config), + ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name), + emqx_resource:query(ResourceID, Request, #{timeout => 1_000}). + +query_resource_async(Config, Request) -> + Name = ?config(sqlserver_name, Config), + BridgeType = ?config(sqlserver_bridge_type, Config), + Ref = alias([reply]), + AsyncReplyFun = fun(Result) -> Ref ! {result, Ref, Result} end, + ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name), + Return = emqx_resource:query(ResourceID, Request, #{ + timeout => 500, async_reply_fun => {AsyncReplyFun, []} + }), + {Return, Ref}. + +receive_result(Ref, Timeout) -> + receive + {result, Ref, Result} -> + {ok, Result}; + {Ref, Result} -> + {ok, Result} + after Timeout -> + timeout + end. + +connect_direct_sqlserver(Config) -> + Opts = [ + {host, ?config(sqlserver_host, Config)}, + {port, ?config(sqlserver_port, Config)}, + {username, ?SQL_SERVER_USERNAME}, + {password, ?SQL_SERVER_PASSWORD}, + {driver, ?SQL_SERVER_DRIVER}, + {pool_size, 8} + ], + {ok, Con} = connect(Opts), + Con. + +connect(Options) -> + ConnectStr = lists:concat(conn_str(Options, [])), + Opts = proplists:get_value(options, Options, []), + odbc:connect(ConnectStr, Opts). + +disconnect(Ref) -> + odbc:disconnect(Ref). + +% These funs connect and then stop the sqlserver connection +connect_and_create_db_and_table(Config) -> + ?WITH_CON(begin + {updated, undefined} = directly_query(Con, ?SQL_CREATE_DATABASE_IF_NOT_EXISTS), + {updated, undefined} = directly_query(Con, ?SQL_CREATE_TABLE_IN_DB_MQTT) + end). + +connect_and_drop_db(Config) -> + ?WITH_CON({updated, undefined} = directly_query(Con, ?SQL_DROP_DB_MQTT)). + +connect_and_drop_table(Config) -> + ?WITH_CON({updated, undefined} = directly_query(Con, ?SQL_DROP_TABLE)). + +connect_and_clear_table(Config) -> + ?WITH_CON({updated, _} = directly_query(Con, ?SQL_DELETE)). + +connect_and_get_payload(Config) -> + ?WITH_CON( + {selected, ["payload"], Rows} = directly_query(Con, ?SQL_SELECT) + ), + Rows. + +connect_and_get_count(Config) -> + ?WITH_CON( + {selected, [[]], [{Count}]} = directly_query(Con, ?SQL_SELECT_COUNT) + ), + Count. + +directly_query(Con, Query) -> + directly_query(Con, Query, ?REQUEST_TIMEOUT_MS). + +directly_query(Con, Query, Timeout) -> + odbc:sql_query(Con, Query, Timeout). + +%%-------------------------------------------------------------------- +%% help functions +%%-------------------------------------------------------------------- + +batch_size(Config) -> + case ?config(enable_batch, Config) of + true -> ?BATCH_SIZE; + false -> 1 + end. + +conn_str([], Acc) -> + %% TODO: for msodbc 18+, we need to add "Encrypt=YES;TrustServerCertificate=YES" + %% but havn't tested now + %% we should use this for msodbcsql 18+ + %% lists:join(";", ["Encrypt=YES", "TrustServerCertificate=YES" | Acc]); + lists:join(";", Acc); +conn_str([{driver, Driver} | Opts], Acc) -> + conn_str(Opts, ["Driver=" ++ str(Driver) | Acc]); +conn_str([{host, Host} | Opts], Acc) -> + Port = proplists:get_value(port, Opts, "1433"), + NOpts = proplists:delete(port, Opts), + conn_str(NOpts, ["Server=" ++ str(Host) ++ "," ++ str(Port) | Acc]); +conn_str([{port, Port} | Opts], Acc) -> + Host = proplists:get_value(host, Opts, "localhost"), + NOpts = proplists:delete(host, Opts), + conn_str(NOpts, ["Server=" ++ str(Host) ++ "," ++ str(Port) | Acc]); +conn_str([{database, Database} | Opts], Acc) -> + conn_str(Opts, ["Database=" ++ str(Database) | Acc]); +conn_str([{username, Username} | Opts], Acc) -> + conn_str(Opts, ["UID=" ++ str(Username) | Acc]); +conn_str([{password, Password} | Opts], Acc) -> + conn_str(Opts, ["PWD=" ++ str(Password) | Acc]); +conn_str([{_, _} | Opts], Acc) -> + conn_str(Opts, Acc). + +sent_data(Payload) -> + #{ + payload => to_bin(Payload), + id => <<"0005F8F84FFFAFB9F44200000D810002">>, + topic => <<"test/topic">>, + qos => 0 + }. + +gen_batch_req(Count) when + is_integer(Count) andalso Count > 0 +-> + Vals = [{str(erlang:unique_integer())} || _Seq <- lists:seq(1, Count)], + Requests = [{send_message, sent_data(Payload)} || {Payload} <- Vals], + {Requests, Vals}; +gen_batch_req(Count) -> + ct:pal("Gen batch requests failed with unexpected Count: ~p", [Count]). + +str(List) when is_list(List) -> + unicode:characters_to_list(List, utf8); +str(Bin) when is_binary(Bin) -> + unicode:characters_to_list(Bin, utf8); +str(Num) when is_number(Num) -> + number_to_list(Num). + +number_to_list(Int) when is_integer(Int) -> + integer_to_list(Int); +number_to_list(Float) when is_float(Float) -> + float_to_list(Float, [{decimals, 10}, compact]). + +to_bin(List) when is_list(List) -> + unicode:characters_to_binary(List, utf8); +to_bin(Bin) when is_binary(Bin) -> + Bin. diff --git a/lib-ee/emqx_ee_connector/docker-ct b/lib-ee/emqx_ee_connector/docker-ct index 446c8ee4e..fc8e75e68 100644 --- a/lib-ee/emqx_ee_connector/docker-ct +++ b/lib-ee/emqx_ee_connector/docker-ct @@ -2,3 +2,4 @@ toxiproxy influxdb clickhouse cassandra +sqlserver diff --git a/scripts/ct/run.sh b/scripts/ct/run.sh index 83e69d1ad..17a30e08b 100755 --- a/scripts/ct/run.sh +++ b/scripts/ct/run.sh @@ -180,6 +180,9 @@ for dep in ${CT_DEPS}; do cassandra) FILES+=( '.ci/docker-compose-file/docker-compose-cassandra.yaml' ) ;; + sqlserver) + FILES+=( '.ci/docker-compose-file/docker-compose-sqlserver.yaml' ) + ;; *) echo "unknown_ct_dependency $dep" exit 1 From a379d909bfb6101ccc55f32e04ffa9dea3ac3b89 Mon Sep 17 00:00:00 2001 From: JimMoen Date: Mon, 10 Apr 2023 16:44:21 +0800 Subject: [PATCH 06/22] test: use type VARCHAR to use utf8 encoding in sqlserver --- lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl index 8b0cc53ce..27a85d10f 100644 --- a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl +++ b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl @@ -30,7 +30,9 @@ " msgid VARCHAR(64) NULL," " topic VARCHAR(100) NULL," " qos tinyint NOT NULL DEFAULT 0," - " payload NVARCHAR(100) NULL," + %% use VARCHAR to use utf8 encoding + %% for default, sqlserver use utf16 encoding NVARCHAR() + " payload VARCHAR(100) NULL," " arrived DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP)" ). From 27fdf644aa5fa75906b5791c6d8cba08f96ae9fd Mon Sep 17 00:00:00 2001 From: JimMoen Date: Mon, 10 Apr 2023 16:18:29 +0800 Subject: [PATCH 07/22] fix(sqlserver): let a single query be a length 1 batch Need to be fixed: `emqx_plugin_libs_rule:proc_tmpl/2` won't add single quotes for string --- .../src/emqx_ee_connector_sqlserver.erl | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl index 4cc92e80a..214663b22 100644 --- a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl +++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl @@ -502,15 +502,18 @@ parse_sql_template([], SingleInsertTks, BatchInsertTks) -> %% single insert apply_template( - {?ACTION_SEND_MESSAGE = Key, Msg} = Query, - #{?SINGLE_INSERT_TEMP := SingleInsertTks} = _Templates + {?ACTION_SEND_MESSAGE = _Key, _Msg} = Query, + #{?SINGLE_INSERT_TEMP := _SingleInsertTks} = Templates ) -> - case maps:get(Key, SingleInsertTks, undefined) of - undefined -> - Query; - Template -> - {Key, emqx_plugin_libs_rule:proc_tmpl(Template, Msg)} - end; + %% TODO: fix emqx_plugin_libs_rule:proc_tmpl/2 + %% it won't add single quotes for string + apply_template([Query], Templates); +%% case maps:get(Key, SingleInsertTks, undefined) of +%% undefined -> +%% Query; +%% Template -> +%% {Key, emqx_plugin_libs_rule:proc_tmpl(Template, Msg)} +%% end; %% batch inserts apply_template( [{?ACTION_SEND_MESSAGE = Key, _Msg} | _T] = BatchReqs, From 59433b8b33288bf8b9abcb0a2b708350efaa9eaa Mon Sep 17 00:00:00 2001 From: JimMoen Date: Tue, 11 Apr 2023 03:41:40 +0800 Subject: [PATCH 08/22] ci: install msodbcsql17 drier data bridge --- scripts/install-odbc-driver.sh | 12 ++++++++++++ 1 file changed, 12 insertions(+) create mode 100755 scripts/install-odbc-driver.sh diff --git a/scripts/install-odbc-driver.sh b/scripts/install-odbc-driver.sh new file mode 100755 index 000000000..c7b94cf13 --- /dev/null +++ b/scripts/install-odbc-driver.sh @@ -0,0 +1,12 @@ +#!/usr/bin/env bash + +## this script install msodbcsql17 and unixodbc-dev on ci environment +set -euo pipefail + +# install msodbcsql17 +curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add - && \ +curl https://packages.microsoft.com/config/debian/11/prod.list > /etc/apt/sources.list.d/mssql-release.list && \ +ACCEPT_EULA=Y apt-get install -y msodbcsql17 unixodbc-dev && \ +## and not needed to modify /etc/odbcinst.ini +## docker-compose will mount one in .ci/docker-compose-file/odbc +sed -i 's/ODBC Driver 17 for SQL Server/ms-sql/g' /etc/odbcinst.ini From d7ad07f1dc27dca9f20fc980163d036d203bc754 Mon Sep 17 00:00:00 2001 From: JimMoen Date: Tue, 11 Apr 2023 04:17:04 +0800 Subject: [PATCH 09/22] ci: install odbc driver for ms sqlserver bridge --- scripts/ct/run.sh | 9 +++++++++ scripts/install-odbc-driver.sh | 2 ++ 2 files changed, 11 insertions(+) diff --git a/scripts/ct/run.sh b/scripts/ct/run.sh index 17a30e08b..4e79476e0 100755 --- a/scripts/ct/run.sh +++ b/scripts/ct/run.sh @@ -39,6 +39,7 @@ ONLY_UP='no' ATTACH='no' STOP='no' IS_CI='no' +ODBC_REQUEST='no' while [ "$#" -gt 0 ]; do case $1 in -h|--help) @@ -181,6 +182,7 @@ for dep in ${CT_DEPS}; do FILES+=( '.ci/docker-compose-file/docker-compose-cassandra.yaml' ) ;; sqlserver) + ODBC_REQUEST='yes' FILES+=( '.ci/docker-compose-file/docker-compose-sqlserver.yaml' ) ;; *) @@ -190,6 +192,12 @@ for dep in ${CT_DEPS}; do esac done +if [ "$ODBC_REQUEST" = 'yes' ]; then + INSTALL_ODBC="./scripts/install-odbc-driver.sh" +else + INSTALL_ODBC="echo 'Driver msodbcsql driver not requested'" +fi + F_OPTIONS="" for file in "${FILES[@]}"; do @@ -232,6 +240,7 @@ docker exec -i $TTY -u root:root "$ERLANG_CONTAINER" bash -c "openssl rand -base # the user must exist inside the container for `whoami` to work docker exec -i $TTY -u root:root "$ERLANG_CONTAINER" bash -c "useradd --uid $DOCKER_USER -M -d / emqx" || true docker exec -i $TTY -u root:root "$ERLANG_CONTAINER" bash -c "chown -R $DOCKER_USER /var/lib/secret" || true +docker exec -i $TTY -u root:root "$ERLANG_CONTAINER" bash -c "$INSTALL_ODBC" || true if [ "$ONLY_UP" = 'yes' ]; then exit 0 diff --git a/scripts/install-odbc-driver.sh b/scripts/install-odbc-driver.sh index c7b94cf13..b1fd894a0 100755 --- a/scripts/install-odbc-driver.sh +++ b/scripts/install-odbc-driver.sh @@ -6,6 +6,8 @@ set -euo pipefail # install msodbcsql17 curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add - && \ curl https://packages.microsoft.com/config/debian/11/prod.list > /etc/apt/sources.list.d/mssql-release.list && \ +## TODO: upgrade builder image +apt-get update && \ ACCEPT_EULA=Y apt-get install -y msodbcsql17 unixodbc-dev && \ ## and not needed to modify /etc/odbcinst.ini ## docker-compose will mount one in .ci/docker-compose-file/odbc From 437096985b7c43d02ae2cccebc19166c284a689b Mon Sep 17 00:00:00 2001 From: JimMoen Date: Tue, 11 Apr 2023 04:39:15 +0800 Subject: [PATCH 10/22] test: fix SUITE case `t_create_disconnected/1` --- .../test/emqx_ee_bridge_sqlserver_SUITE.erl | 54 +++++++++---------- 1 file changed, 27 insertions(+), 27 deletions(-) diff --git a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl index 27a85d10f..7a9c1baf9 100644 --- a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl +++ b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl @@ -224,22 +224,9 @@ t_get_status(Config) -> ProxyHost = ?config(proxy_host, Config), ProxyName = ?config(proxy_name, Config), - Name = ?config(sqlserver_name, Config), - BridgeType = ?config(sqlserver_bridge_type, Config), - ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name), - - ?assertEqual({ok, connected}, emqx_resource_manager:health_check(ResourceID)), + health_check_resource_ok(Config), emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() -> - case emqx_resource_manager:health_check(ResourceID) of - {ok, Status} when Status =:= disconnected orelse Status =:= connecting -> - ok; - {error, timeout} -> - ok; - Other -> - ?assert( - false, lists:flatten(io_lib:format("invalid health check result:~p~n", [Other])) - ) - end + health_check_resource_down(Config) end), ok. @@ -247,18 +234,11 @@ t_create_disconnected(Config) -> ProxyPort = ?config(proxy_port, Config), ProxyHost = ?config(proxy_host, Config), ProxyName = ?config(proxy_name, Config), - ?check_trace( - emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() -> - ?assertMatch({ok, _}, create_bridge(Config)) - end), - fun(Trace) -> - ?assertMatch( - [#{error := {start_pool_failed, _, _}}], - ?of_kind(sqlserver_connector_start_failed, Trace) - ), - ok - end - ), + emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() -> + ?assertMatch({ok, _}, create_bridge(Config)), + health_check_resource_down(Config) + end), + health_check_resource_ok(Config), ok. t_write_failure(Config) -> @@ -522,6 +502,26 @@ query_resource_async(Config, Request) -> }), {Return, Ref}. +resource_id(Config) -> + Name = ?config(sqlserver_name, Config), + BridgeType = ?config(sqlserver_bridge_type, Config), + _ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name). + +health_check_resource_ok(Config) -> + ?assertEqual({ok, connected}, emqx_resource_manager:health_check(resource_id(Config))). + +health_check_resource_down(Config) -> + case emqx_resource_manager:health_check(resource_id(Config)) of + {ok, Status} when Status =:= disconnected orelse Status =:= connecting -> + ok; + {error, timeout} -> + ok; + Other -> + ?assert( + false, lists:flatten(io_lib:format("invalid health check result:~p~n", [Other])) + ) + end. + receive_result(Ref, Timeout) -> receive {result, Ref, Result} -> From 57505cdfb3fc84c559e47912e3cff9e016eed214 Mon Sep 17 00:00:00 2001 From: JimMoen Date: Tue, 11 Apr 2023 05:15:24 +0800 Subject: [PATCH 11/22] ci: unmount odbcinst.ini file --- .ci/docker-compose-file/docker-compose.yaml | 2 +- .ci/docker-compose-file/odbc/odbcinst.ini | 1 - scripts/install-odbc-driver.sh | 11 +++++++++-- 3 files changed, 10 insertions(+), 4 deletions(-) diff --git a/.ci/docker-compose-file/docker-compose.yaml b/.ci/docker-compose-file/docker-compose.yaml index 8f97fa9d6..6f4b7c04b 100644 --- a/.ci/docker-compose-file/docker-compose.yaml +++ b/.ci/docker-compose-file/docker-compose.yaml @@ -24,7 +24,7 @@ services: - /tmp/emqx-ci/emqx-shared-secret:/var/lib/secret - ./kerberos/krb5.conf:/etc/kdc/krb5.conf - ./kerberos/krb5.conf:/etc/krb5.conf - - ./odbc/odbcinst.ini:/etc/odbcinst.ini + # - ./odbc/odbcinst.ini:/etc/odbcinst.ini working_dir: /emqx tty: true user: "${DOCKER_USER:-root}" diff --git a/.ci/docker-compose-file/odbc/odbcinst.ini b/.ci/docker-compose-file/odbc/odbcinst.ini index 16701cfd2..dd0241543 100644 --- a/.ci/docker-compose-file/odbc/odbcinst.ini +++ b/.ci/docker-compose-file/odbc/odbcinst.ini @@ -7,4 +7,3 @@ UsageCount=1 Description=Microsoft ODBC Driver 17 for SQL Server Driver=/opt/microsoft/msodbcsql17/lib64/libmsodbcsql-17.10.so.2.1 UsageCount=1 - diff --git a/scripts/install-odbc-driver.sh b/scripts/install-odbc-driver.sh index b1fd894a0..4ca71d88d 100755 --- a/scripts/install-odbc-driver.sh +++ b/scripts/install-odbc-driver.sh @@ -1,14 +1,21 @@ #!/usr/bin/env bash ## this script install msodbcsql17 and unixodbc-dev on ci environment +## specific to ubuntu 16.04, 18.04, 20.04, 22.04 set -euo pipefail # install msodbcsql17 +if ! [[ "16.04 18.04 20.04 22.04" == *"$(lsb_release -rs)"* ]]; +then + echo "Ubuntu $(lsb_release -rs) is not currently supported."; + exit 1; +fi + curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add - && \ -curl https://packages.microsoft.com/config/debian/11/prod.list > /etc/apt/sources.list.d/mssql-release.list && \ +curl https://packages.microsoft.com/config/ubuntu/$(lsb_release -rs)/prod.list > /etc/apt/sources.list.d/mssql-release.list && \ ## TODO: upgrade builder image apt-get update && \ -ACCEPT_EULA=Y apt-get install -y msodbcsql17 unixodbc-dev && \ +ACCEPT_EULA=Y apt-get install -y msodbcsql17 unixodbc-dev mssql-tools && \ ## and not needed to modify /etc/odbcinst.ini ## docker-compose will mount one in .ci/docker-compose-file/odbc sed -i 's/ODBC Driver 17 for SQL Server/ms-sql/g' /etc/odbcinst.ini From f8c0aa49a240b65114edb99be2f1e68401cebc56 Mon Sep 17 00:00:00 2001 From: JimMoen Date: Tue, 11 Apr 2023 09:54:58 +0800 Subject: [PATCH 12/22] style: make elvis and shellcheck happy --- .../emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl | 6 ++++-- scripts/install-odbc-driver.sh | 7 ++++--- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl index 214663b22..55a5d488f 100644 --- a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl +++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl @@ -342,9 +342,11 @@ do_get_status(Conn, RequestTimeout) -> %% About the connection string attribute `Encrypt`: %% The default value is `yes` in odbc version 18.0+ and `no` in previous versions. %% And encrypted connections always verify the server's certificate. -%% So `Encrypt=YES;TrustServerCertificate=YES` must be set in the connection string when connecting to a server that has a self-signed certificate. +%% So `Encrypt=YES;TrustServerCertificate=YES` must be set in the connection string +%% when connecting to a server that has a self-signed certificate. %% See also: -%% https://learn.microsoft.com/en-us/sql/connect/odbc/dsn-connection-string-attribute?source=recommendations&view=sql-server-ver16#encrypt +%% 'https://learn.microsoft.com/en-us/sql/connect/odbc/ +%% dsn-connection-string-attribute?source=recommendations&view=sql-server-ver16#encrypt' conn_str([], Acc) -> %% we should use this for msodbcsql 18+ %% lists:join(";", ["Encrypt=YES", "TrustServerCertificate=YES" | Acc]); diff --git a/scripts/install-odbc-driver.sh b/scripts/install-odbc-driver.sh index 4ca71d88d..7ceab95a0 100755 --- a/scripts/install-odbc-driver.sh +++ b/scripts/install-odbc-driver.sh @@ -5,14 +5,15 @@ set -euo pipefail # install msodbcsql17 -if ! [[ "16.04 18.04 20.04 22.04" == *"$(lsb_release -rs)"* ]]; +VERSION=$(lsb_release -rs) +if ! [[ "16.04 18.04 20.04 22.04" == *"$VERSION"* ]]; then - echo "Ubuntu $(lsb_release -rs) is not currently supported."; + echo "Ubuntu $VERSION is not currently supported."; exit 1; fi curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add - && \ -curl https://packages.microsoft.com/config/ubuntu/$(lsb_release -rs)/prod.list > /etc/apt/sources.list.d/mssql-release.list && \ +curl https://packages.microsoft.com/config/ubuntu/"$VERSION"/prod.list > /etc/apt/sources.list.d/mssql-release.list && \ ## TODO: upgrade builder image apt-get update && \ ACCEPT_EULA=Y apt-get install -y msodbcsql17 unixodbc-dev mssql-tools && \ From 675f9894ab5335bdd867ae6803bf47852faf9bf3 Mon Sep 17 00:00:00 2001 From: JimMoen Date: Tue, 11 Apr 2023 10:43:35 +0800 Subject: [PATCH 13/22] fix: fix single quotes not applied for single query - alwasy use `emqx_plugin_libs_rule:proc_batch_sql/3` to add single quotes for string --- .../src/emqx_ee_connector_sqlserver.erl | 30 ++++++------------- 1 file changed, 9 insertions(+), 21 deletions(-) diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl index 55a5d488f..99b5c2809 100644 --- a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl +++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl @@ -50,8 +50,6 @@ -define(ACTION_SEND_MESSAGE, send_message). --define(SINGLE_INSERT, single_insert). - -define(SYNC_QUERY_MODE, handover). -define(ASYNC_QUERY_MODE(REPLY), {handover_async, {?MODULE, do_async_reply, [REPLY]}}). @@ -63,7 +61,6 @@ maps:get(request_timeout, RESOURCE_OPTS, ?DEFAULT_REQUEST_TIMEOUT) ). --define(SINGLE_INSERT_TEMP, single_insert_temp). -define(BATCH_INSERT_TEMP, batch_insert_temp). -define(BATCH_INSERT_PART, batch_insert_part). @@ -464,20 +461,18 @@ parse_sql_template(Config) -> SQLTemplate -> #{?ACTION_SEND_MESSAGE => SQLTemplate} end, - SingleInsertTks = #{}, BatchInsertTks = #{}, - parse_sql_template(maps:to_list(RawSQLTemplates), SingleInsertTks, BatchInsertTks). + parse_sql_template(maps:to_list(RawSQLTemplates), BatchInsertTks). -parse_sql_template([{Key, H} | T], SingleInsertTks, BatchInsertTks) -> +parse_sql_template([{Key, H} | T], BatchInsertTks) -> case emqx_plugin_libs_rule:detect_sql_type(H) of {ok, select} -> - parse_sql_template(T, SingleInsertTks, BatchInsertTks); + parse_sql_template(T, BatchInsertTks); {ok, insert} -> case emqx_plugin_libs_rule:split_insert_sql(H) of {ok, {InsertSQL, Params}} -> parse_sql_template( T, - SingleInsertTks#{Key => emqx_plugin_libs_rule:preproc_tmpl(H)}, BatchInsertTks#{ Key => #{ @@ -490,32 +485,24 @@ parse_sql_template([{Key, H} | T], SingleInsertTks, BatchInsertTks) -> ); {error, Reason} -> ?SLOG(error, #{msg => "split sql failed", sql => H, reason => Reason}), - parse_sql_template(T, SingleInsertTks, BatchInsertTks) + parse_sql_template(T, BatchInsertTks) end; {error, Reason} -> ?SLOG(error, #{msg => "detect sql type failed", sql => H, reason => Reason}), - parse_sql_template(T, SingleInsertTks, BatchInsertTks) + parse_sql_template(T, BatchInsertTks) end; -parse_sql_template([], SingleInsertTks, BatchInsertTks) -> +parse_sql_template([], BatchInsertTks) -> #{ - ?SINGLE_INSERT_TEMP => SingleInsertTks, ?BATCH_INSERT_TEMP => BatchInsertTks }. %% single insert apply_template( - {?ACTION_SEND_MESSAGE = _Key, _Msg} = Query, - #{?SINGLE_INSERT_TEMP := _SingleInsertTks} = Templates + {?ACTION_SEND_MESSAGE = _Key, _Msg} = Query, Templates ) -> %% TODO: fix emqx_plugin_libs_rule:proc_tmpl/2 %% it won't add single quotes for string apply_template([Query], Templates); -%% case maps:get(Key, SingleInsertTks, undefined) of -%% undefined -> -%% Query; -%% Template -> -%% {Key, emqx_plugin_libs_rule:proc_tmpl(Template, Msg)} -%% end; %% batch inserts apply_template( [{?ACTION_SEND_MESSAGE = Key, _Msg} | _T] = BatchReqs, @@ -530,7 +517,8 @@ apply_template( end; apply_template(Query, Templates) -> %% TODO: more detail infomatoin - ?SLOG(error, #{msg => "apply sql template failed", query => Query, templates => Templates}). + ?SLOG(error, #{msg => "apply sql template failed", query => Query, templates => Templates}), + {error, failed_to_apply_sql_template}. do_async_reply(Result, {ReplyFun, Args}) -> erlang:apply(ReplyFun, Args ++ Result). From 35ab1c01900ada9d1b1610c0e1d325babc5a664d Mon Sep 17 00:00:00 2001 From: JimMoen Date: Tue, 11 Apr 2023 11:11:09 +0800 Subject: [PATCH 14/22] style: make hocon style check and spellcheck happy - with more human readable i18n style check format --- changes/ee/feat-10363.en.md | 2 +- rel/i18n/emqx_ee_bridge_dynamo.hocon | 2 +- rel/i18n/emqx_ee_bridge_pgsql.hocon | 2 +- rel/i18n/emqx_ee_bridge_sqlserver.hocon | 18 ++++++++---------- rel/i18n/emqx_ee_bridge_tdengine.hocon | 2 +- rel/i18n/emqx_ee_connector_sqlserver.hocon | 12 ++++-------- scripts/check-i18n-style.escript | 6 +++--- 7 files changed, 19 insertions(+), 25 deletions(-) diff --git a/changes/ee/feat-10363.en.md b/changes/ee/feat-10363.en.md index c3b53a538..84816031d 100644 --- a/changes/ee/feat-10363.en.md +++ b/changes/ee/feat-10363.en.md @@ -1 +1 @@ -Add MicroSoft SQL Server data Bridge support. +Implement Microsoft SQL Server bridge. diff --git a/rel/i18n/emqx_ee_bridge_dynamo.hocon b/rel/i18n/emqx_ee_bridge_dynamo.hocon index 664b13174..b93b12166 100644 --- a/rel/i18n/emqx_ee_bridge_dynamo.hocon +++ b/rel/i18n/emqx_ee_bridge_dynamo.hocon @@ -39,7 +39,7 @@ will be forwarded.""" desc_config { desc { - en: """Configuration for an DynamoDB bridge.""" + en: """Configuration for a DynamoDB bridge.""" zh: """DynamoDB 桥接配置""" } label: { diff --git a/rel/i18n/emqx_ee_bridge_pgsql.hocon b/rel/i18n/emqx_ee_bridge_pgsql.hocon index 81d2330c5..1ce2818ca 100644 --- a/rel/i18n/emqx_ee_bridge_pgsql.hocon +++ b/rel/i18n/emqx_ee_bridge_pgsql.hocon @@ -39,7 +39,7 @@ will be forwarded.""" desc_config { desc { - en: """Configuration for an PostgreSQL bridge.""" + en: """Configuration for a PostgreSQL bridge.""" zh: """PostgreSQL 桥接配置""" } label: { diff --git a/rel/i18n/emqx_ee_bridge_sqlserver.hocon b/rel/i18n/emqx_ee_bridge_sqlserver.hocon index bb82c82f8..007b64ae6 100644 --- a/rel/i18n/emqx_ee_bridge_sqlserver.hocon +++ b/rel/i18n/emqx_ee_bridge_sqlserver.hocon @@ -2,15 +2,13 @@ emqx_ee_bridge_sqlserver { local_topic { desc { - en: """The MQTT topic filter to be forwarded to MicroSoft SQL Server. All MQTT 'PUBLISH' messages with the topic + en: """The MQTT topic filter to be forwarded to Microsoft SQL Server. All MQTT 'PUBLISH' messages with the topic matching the local_topic will be forwarded.
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' 的消息都会转发到 MicroSoft SQL Server。
-注意：如果这个 Bridge 被用作规则（EMQX 规则引擎）的输出，同时也配置了 'local_topic' ，那么这两部分的消息都会被转发。 -""" +will be forwarded.""" + zh: """发送到 'local_topic' 的消息都会转发到 Microsoft SQL Server。
+注意：如果这个 Bridge 被用作规则（EMQX 规则引擎）的输出，同时也配置了 'local_topic' ，那么这两部分的消息都会被转发。""" } label { en: """Local Topic""" @@ -53,12 +51,12 @@ will be forwarded. desc_config { desc { - en: """Configuration for an MicroSoft SQL Server bridge.""" - zh: """MicroSoft SQL Server 桥接配置""" + en: """Configuration for a Microsoft SQL Server bridge.""" + zh: """Microsoft SQL Server 桥接配置""" } label: { - en: """MicroSoft SQL Server Bridge Configuration""" - zh: """MicroSoft SQL Server 桥接配置""" + en: """Microsoft SQL Server Bridge Configuration""" + zh: """Microsoft SQL Server 桥接配置""" } } diff --git a/rel/i18n/emqx_ee_bridge_tdengine.hocon b/rel/i18n/emqx_ee_bridge_tdengine.hocon index 21fc013df..12eb6f062 100644 --- a/rel/i18n/emqx_ee_bridge_tdengine.hocon +++ b/rel/i18n/emqx_ee_bridge_tdengine.hocon @@ -39,7 +39,7 @@ will be forwarded.""" desc_config { desc { - en: """Configuration for an TDengine bridge.""" + en: """Configuration for a TDengine bridge.""" zh: """TDengine 桥接配置""" } label: { diff --git a/rel/i18n/emqx_ee_connector_sqlserver.hocon b/rel/i18n/emqx_ee_connector_sqlserver.hocon index 9ff24c1c1..85280d833 100644 --- a/rel/i18n/emqx_ee_connector_sqlserver.hocon +++ b/rel/i18n/emqx_ee_connector_sqlserver.hocon @@ -2,16 +2,12 @@ emqx_ee_connector_sqlserver { server { desc { - en: """ -The IPv4 or IPv6 address or the hostname to connect to.
+ en: """The IPv4 or IPv6 address or the hostname to connect to.
A host entry has the following form: `Host[:Port]`.
-The SQL Server default port 1433 is used if `[:Port]` is not specified. -""" - zh: """ -将要连接的 IPv4 或 IPv6 地址，或者主机名。
+The SQL Server default port 1433 is used if `[:Port]` is not specified.""" + zh: """将要连接的 IPv4 或 IPv6 地址，或者主机名。
主机名具有以下形式：`Host[:Port]`。
-如果未指定 `[:Port]`，则使用 SQL Server 默认端口 1433。 -""" +如果未指定 `[:Port]`，则使用 SQL Server 默认端口 1433。""" } label: { en: "Server Host" diff --git a/scripts/check-i18n-style.escript b/scripts/check-i18n-style.escript index cbe79c82e..a76fce90e 100755 --- a/scripts/check-i18n-style.escript +++ b/scripts/check-i18n-style.escript @@ -9,7 +9,7 @@ -define(RESET, "\e[39m"). main([Files0]) -> - io:format(user, "checking i18n file styles", []), + io:format(user, "checking i18n file styles~n", []), _ = put(errors, 0), Files = string:tokens(Files0, "\n"), ok = load_hocon(), @@ -20,7 +20,7 @@ main([Files0]) -> N when is_integer(N) andalso N > 1 -> logerr("~p errors found~n", [N]); _ -> - io:format(user, "OK~n", []) + io:format(user, "~nOK~n", []) end. load_hocon() -> @@ -42,7 +42,7 @@ die(Msg, Args) -> halt(1). logerr(Fmt, Args) -> - io:format(standard_error, ?RED ++ "ERROR: " ++ Fmt ++ ?RESET, Args), + io:format(standard_error, "~n" ++ ?RED ++ "ERROR: " ++ Fmt ++ ?RESET, Args), N = get(errors), _ = put(errors, N + 1), ok. From 1b2cf6d421f7e3dae05ec1be1ccaa1e057b449b5 Mon Sep 17 00:00:00 2001 From: JimMoen Date: Thu, 13 Apr 2023 17:58:06 +0800 Subject: [PATCH 15/22] ci: newest SQL Serevr 2019 image and rm deprecated env `SA_PASSWORD` --- .ci/docker-compose-file/.env | 2 +- .ci/docker-compose-file/docker-compose-sqlserver.yaml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.ci/docker-compose-file/.env b/.ci/docker-compose-file/.env index b394d6bb1..d33637ea0 100644 --- a/.ci/docker-compose-file/.env +++ b/.ci/docker-compose-file/.env @@ -9,6 +9,6 @@ DYNAMO_TAG=1.21.0 CASSANDRA_TAG=3.11.6 MS_IMAGE_ADDR=mcr.microsoft.com/mssql/server -SQLSERVER_TAG=2019-CU11-ubuntu-20.04 +SQLSERVER_TAG=2019-CU19-ubuntu-20.04 TARGET=emqx/emqx diff --git a/.ci/docker-compose-file/docker-compose-sqlserver.yaml b/.ci/docker-compose-file/docker-compose-sqlserver.yaml index 37799e20b..a578983f8 100644 --- a/.ci/docker-compose-file/docker-compose-sqlserver.yaml +++ b/.ci/docker-compose-file/docker-compose-sqlserver.yaml @@ -11,7 +11,7 @@ services: # See also: # https://learn.microsoft.com/en-us/sql/linux/sql-server-linux-configure-environment-variables ACCEPT_EULA: "Y" - SA_PASSWORD: "mqtt_public" + MSSQL_SA_PASSWORD: "mqtt_public" restart: always # ports: # - "1433:1433" From 2d1b94f7f79e841802e2e3de2e898f4a8ff90af0 Mon Sep 17 00:00:00 2001 From: JimMoen Date: Fri, 14 Apr 2023 01:34:30 +0800 Subject: [PATCH 16/22] fix: file mode 755 magic number --- .../src/emqx_ee_connector_sqlserver.erl | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl index 99b5c2809..e48570c96 100644 --- a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl +++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl @@ -66,6 +66,11 @@ -define(BATCH_INSERT_PART, batch_insert_part). -define(BATCH_PARAMS_TOKENS, batch_insert_tks). +-define(FILE_MODE_755, 33261). +%% 32768 + 8#00400 + 8#00200 + 8#00100 + 8#00040 + 8#00010 + 8#00004 + 8#00001 +%% See also +%% https://www.erlang.org/doc/man/file.html#read_file_info-2 + %% Copied from odbc reference page %% https://www.erlang.org/doc/man/odbc.html @@ -190,11 +195,11 @@ on_start( ODBCDir = code:priv_dir(odbc), OdbcserverDir = filename:join(ODBCDir, "bin/odbcserver"), {ok, Info = #file_info{mode = Mode}} = file:read_file_info(OdbcserverDir), - case 33261 =:= Mode of + case ?FILE_MODE_755 =:= Mode of true -> ok; false -> - _ = file:write_file_info(OdbcserverDir, Info#file_info{mode = 33261}), + _ = file:write_file_info(OdbcserverDir, Info#file_info{mode = ?FILE_MODE_755}), ok end, From 65c4234829a9f52758803b4987a8275e8e90a24d Mon Sep 17 00:00:00 2001 From: JimMoen Date: Fri, 14 Apr 2023 01:39:08 +0800 Subject: [PATCH 17/22] refactor: rename instance_id, which is also PoolName --- .../src/emqx_ee_connector_sqlserver.erl | 51 +++++++++---------- 1 file changed, 25 insertions(+), 26 deletions(-) diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl index e48570c96..586444947 100644 --- a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl +++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl @@ -74,7 +74,6 @@ %% Copied from odbc reference page %% https://www.erlang.org/doc/man/odbc.html --type bridge_id() :: binary(). %% as returned by connect/2 -type connection_reference() :: pid(). -type time_out() :: milliseconds() | infinity. @@ -175,7 +174,7 @@ callback_mode() -> async_if_possible. is_buffer_supported() -> false. on_start( - BridgeId = PoolName, + InstanceId = PoolName, #{ server := Server, username := Username, @@ -188,7 +187,7 @@ on_start( ) -> ?SLOG(info, #{ msg => "starting_sqlserver_connector", - connector => BridgeId, + connector => InstanceId, config => emqx_misc:redact(Config) }), @@ -214,7 +213,7 @@ on_start( ], State = #{ - %% also BridgeId + %% also InstanceId poolname => PoolName, sql_templates => parse_sql_template(Config), resource_opts => ResourceOpts @@ -238,7 +237,7 @@ on_stop(InstanceId, #{poolname := PoolName} = _State) -> emqx_plugin_libs_pool:stop_pool(PoolName). -spec on_query( - bridge_id(), + manager_id(), {?ACTION_SEND_MESSAGE, map()}, state() ) -> @@ -246,16 +245,16 @@ on_stop(InstanceId, #{poolname := PoolName} = _State) -> | {ok, list()} | {error, {recoverable_error, term()}} | {error, term()}. -on_query(BridgeId, {?ACTION_SEND_MESSAGE, _Msg} = Query, State) -> +on_query(InstanceId, {?ACTION_SEND_MESSAGE, _Msg} = Query, State) -> ?TRACE( "SINGLE_QUERY_SYNC", "bridge_sqlserver_received", - #{requests => Query, connector => BridgeId, state => State} + #{requests => Query, connector => InstanceId, state => State} ), - do_query(BridgeId, Query, ?SYNC_QUERY_MODE, State). + do_query(InstanceId, Query, ?SYNC_QUERY_MODE, State). -spec on_query_async( - bridge_id(), + manager_id(), {?ACTION_SEND_MESSAGE, map()}, {ReplyFun :: function(), Args :: list()}, state() @@ -263,7 +262,7 @@ on_query(BridgeId, {?ACTION_SEND_MESSAGE, _Msg} = Query, State) -> {ok, any()} | {error, term()}. on_query_async( - BridgeId, + InstanceId, {?ACTION_SEND_MESSAGE, _Msg} = Query, ReplyFunAndArgs, %% #{poolname := PoolName, sql_templates := Templates} = State @@ -272,12 +271,12 @@ on_query_async( ?TRACE( "SINGLE_QUERY_ASYNC", "bridge_sqlserver_received", - #{requests => Query, connector => BridgeId, state => State} + #{requests => Query, connector => InstanceId, state => State} ), - do_query(BridgeId, Query, ?ASYNC_QUERY_MODE(ReplyFunAndArgs), State). + do_query(InstanceId, Query, ?ASYNC_QUERY_MODE(ReplyFunAndArgs), State). -spec on_batch_query( - bridge_id(), + manager_id(), [{?ACTION_SEND_MESSAGE, map()}], state() ) -> @@ -285,27 +284,27 @@ on_query_async( | {ok, list()} | {error, {recoverable_error, term()}} | {error, term()}. -on_batch_query(BridgeId, BatchRequests, State) -> +on_batch_query(InstanceId, BatchRequests, State) -> ?TRACE( "BATCH_QUERY_SYNC", "bridge_sqlserver_received", - #{requests => BatchRequests, connector => BridgeId, state => State} + #{requests => BatchRequests, connector => InstanceId, state => State} ), - do_query(BridgeId, BatchRequests, ?SYNC_QUERY_MODE, State). + do_query(InstanceId, BatchRequests, ?SYNC_QUERY_MODE, State). -spec on_batch_query_async( - bridge_id(), + manager_id(), [{?ACTION_SEND_MESSAGE, map()}], {ReplyFun :: function(), Args :: list()}, state() ) -> {ok, any()}. -on_batch_query_async(BridgeId, Requests, ReplyFunAndArgs, State) -> +on_batch_query_async(InstanceId, Requests, ReplyFunAndArgs, State) -> ?TRACE( "BATCH_QUERY_ASYNC", "bridge_sqlserver_received", - #{requests => Requests, connector => BridgeId, state => State} + #{requests => Requests, connector => InstanceId, state => State} ), - do_query(BridgeId, Requests, ?ASYNC_QUERY_MODE(ReplyFunAndArgs), State). + do_query(InstanceId, Requests, ?ASYNC_QUERY_MODE(ReplyFunAndArgs), State). on_get_status(_InstanceId, #{poolname := Pool, resource_opts := ResourceOpts} = _State) -> RequestTimeout = ?REQUEST_TIMEOUT(ResourceOpts), @@ -369,7 +368,7 @@ conn_str([{_, _} | Opts], Acc) -> %% Sync & Async query with singe & batch sql statement -spec do_query( - bridge_id(), + manager_id(), Query :: {?ACTION_SEND_MESSAGE, map()} | [{?ACTION_SEND_MESSAGE, map()}], ApplyMode :: handover @@ -380,7 +379,7 @@ conn_str([{_, _} | Opts], Acc) -> | {error, {recoverable_error, term()}} | {error, term()}. do_query( - BridgeId, + InstanceId, Query, ApplyMode, #{poolname := PoolName, sql_templates := Templates} = State @@ -388,7 +387,7 @@ do_query( ?TRACE( "SINGLE_QUERY_SYNC", "sqlserver_connector_received", - #{query => Query, connector => BridgeId, state => State} + #{query => Query, connector => InstanceId, state => State} ), %% only insert sql statement for single query and batch query @@ -412,7 +411,7 @@ do_query( ), ?SLOG(error, #{ msg => "sqlserver_connector_do_query_failed", - connector => BridgeId, + connector => InstanceId, query => Query, reason => Reason }), @@ -426,9 +425,9 @@ do_query( end. worker_do_insert( - Conn, SQL, #{resource_opts := ResourceOpts, poolname := BridgeId} = State + Conn, SQL, #{resource_opts := ResourceOpts, poolname := InstanceId} = State ) -> - LogMeta = #{connector => BridgeId, state => State}, + LogMeta = #{connector => InstanceId, state => State}, try case execute(Conn, SQL, ?REQUEST_TIMEOUT(ResourceOpts)) of {selected, Rows, _} -> From 06b1ec862efda931c975521c831b316331f964cc Mon Sep 17 00:00:00 2001 From: JimMoen Date: Fri, 14 Apr 2023 10:24:40 +0800 Subject: [PATCH 18/22] fix: hocon label field --- rel/i18n/emqx_ee_bridge_sqlserver.hocon | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rel/i18n/emqx_ee_bridge_sqlserver.hocon b/rel/i18n/emqx_ee_bridge_sqlserver.hocon index 007b64ae6..a6f96f86d 100644 --- a/rel/i18n/emqx_ee_bridge_sqlserver.hocon +++ b/rel/i18n/emqx_ee_bridge_sqlserver.hocon @@ -32,7 +32,7 @@ will be forwarded.""" en: """SQL Server Driver Name""" zh: """SQL Server Driver 名称""" } - desc { + label { en: """SQL Server Driver Name""" zh: """SQL Server Driver 名称""" } From 8cec62bd877f7e59e283b4e73486d02dde5cd97b Mon Sep 17 00:00:00 2001 From: JimMoen Date: Fri, 14 Apr 2023 12:41:34 +0800 Subject: [PATCH 19/22] fix: async reply arg list --- lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl index 586444947..dc9759341 100644 --- a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl +++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl @@ -525,4 +525,4 @@ apply_template(Query, Templates) -> {error, failed_to_apply_sql_template}. do_async_reply(Result, {ReplyFun, Args}) -> - erlang:apply(ReplyFun, Args ++ Result). + erlang:apply(ReplyFun, Args ++ [Result]). From 2484a79c7aa0d712464b7c413069f3b615e0b601 Mon Sep 17 00:00:00 2001 From: JimMoen Date: Fri, 14 Apr 2023 12:42:09 +0800 Subject: [PATCH 20/22] test: create bridge with invalid password --- .../docker-compose-sqlserver.yaml | 2 +- .../test/emqx_ee_bridge_sqlserver_SUITE.erl | 32 +++++++++++++++++-- 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/.ci/docker-compose-file/docker-compose-sqlserver.yaml b/.ci/docker-compose-file/docker-compose-sqlserver.yaml index a578983f8..63fcfeecd 100644 --- a/.ci/docker-compose-file/docker-compose-sqlserver.yaml +++ b/.ci/docker-compose-file/docker-compose-sqlserver.yaml @@ -11,7 +11,7 @@ services: # See also: # https://learn.microsoft.com/en-us/sql/linux/sql-server-linux-configure-environment-variables ACCEPT_EULA: "Y" - MSSQL_SA_PASSWORD: "mqtt_public" + MSSQL_SA_PASSWORD: "mqtt_public1" restart: always # ports: # - "1433:1433" diff --git a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl index 7a9c1baf9..152bdf2b7 100644 --- a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl +++ b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl @@ -44,7 +44,7 @@ % DB defaults -define(SQL_SERVER_DATABASE, "mqtt"). -define(SQL_SERVER_USERNAME, "sa"). --define(SQL_SERVER_PASSWORD, "mqtt_public"). +-define(SQL_SERVER_PASSWORD, "mqtt_public1"). -define(BATCH_SIZE, 10). -define(REQUEST_TIMEOUT_MS, 500). @@ -56,7 +56,7 @@ ok = disconnect(Con) ). -%% How to run it locally: +%% How to run it locally (all commands are run in $PROJ_ROOT dir): %% A: run ct on host %% 1. Start all deps services %% sudo docker compose -f .ci/docker-compose-file/docker-compose.yaml \ @@ -65,7 +65,7 @@ %% up --build %% %% 2. Run use cases with special environment variables -%% 11433 is toxiproxy exported port +%% 11433 is toxiproxy exported port. %% Local: %% ``` %% SQLSERVER_HOST=toxiproxy SQLSERVER_PORT=11433 \ @@ -241,6 +241,32 @@ t_create_disconnected(Config) -> health_check_resource_ok(Config), ok. +t_create_with_invalid_password(Config) -> + BridgeType = ?config(sqlserver_bridge_type, Config), + Name = ?config(sqlserver_name, Config), + SQLServerConfig0 = ?config(sqlserver_config, Config), + SQLServerConfig = SQLServerConfig0#{ + <<"name">> => Name, + <<"type">> => BridgeType, + <<"password">> => <<"wrong_password">> + }, + ?check_trace( + begin + ?assertMatch( + {ok, _}, + create_bridge_http(SQLServerConfig) + ) + end, + fun(Trace) -> + ?assertMatch( + [#{error := {start_pool_failed, _, _}}], + ?of_kind(sqlserver_connector_start_failed, Trace) + ), + ok + end + ), + ok. + t_write_failure(Config) -> ProxyName = ?config(proxy_name, Config), ProxyPort = ?config(proxy_port, Config), From 9ce0dbae7b297f27d97bd27c7a536f9e8ae02d4e Mon Sep 17 00:00:00 2001 From: JimMoen Date: Fri, 14 Apr 2023 13:11:45 +0800 Subject: [PATCH 21/22] test: fix batch_size parameter --- lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl index 152bdf2b7..c1b33c3c9 100644 --- a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl +++ b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl @@ -398,7 +398,7 @@ common_init(ConfigT) -> {sqlserver_port, Port}, %% see also for `proxy_name` : $PROJ_ROOT/.ci/docker-compose-file/toxiproxy.json {proxy_name, "sqlserver"}, - {batch_size, ?BATCH_SIZE} + {batch_size, batch_size(ConfigT)} | ConfigT ], From 987ef6c51daece05188de4943190d2eb6142b34e Mon Sep 17 00:00:00 2001 From: JimMoen Date: Fri, 14 Apr 2023 15:42:44 +0800 Subject: [PATCH 22/22] revert: "build: docker file add unixodbc and msodbcsql17" This reverts commit 611edf422719069297219370fe120968737fc0cf. --- .gitignore | 1 - deploy/docker/Dockerfile | 16 +++------------- 2 files changed, 3 insertions(+), 14 deletions(-) diff --git a/.gitignore b/.gitignore index 3f7e268a2..62e8ddc81 100644 --- a/.gitignore +++ b/.gitignore @@ -71,4 +71,3 @@ apps/emqx/test/emqx_static_checks_data/master.bpapi lux_logs/ /.prepare bom.json -.dockerignore diff --git a/deploy/docker/Dockerfile b/deploy/docker/Dockerfile index dcebe9b1a..8f04c433c 100644 --- a/deploy/docker/Dockerfile +++ b/deploy/docker/Dockerfile @@ -29,19 +29,9 @@ COPY --from=builder /emqx-rel/emqx /opt/emqx RUN ln -s /opt/emqx/bin/* /usr/local/bin/ -RUN apt-get update \ - && apt-get install -y --no-install-recommends ca-certificates procps - -RUN apt-get update \ - && apt-get install -y gnupg2 curl apt-utils \ - && curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add - \ - && curl https://packages.microsoft.com/config/debian/11/prod.list > /etc/apt/sources.list.d/mssql-release.list \ - && apt-get update \ - && ACCEPT_EULA=Y apt-get install -y msodbcsql17 unixodbc-dev \ - && sed -i 's/ODBC Driver 17 for SQL Server/ms-sql/g' /etc/odbcinst.ini - -RUN apt-get clean && \ - rm -rf /var/lib/apt/lists/*; +RUN apt-get update; \ + apt-get install -y --no-install-recommends ca-certificates procps; \ + rm -rf /var/lib/apt/lists/* WORKDIR /opt/emqx