From 3e679ceb572586576a6fba79f8049ad068ce73d0 Mon Sep 17 00:00:00 2001 From: Erik Timan Date: Fri, 18 Nov 2022 15:10:33 +0100 Subject: [PATCH 1/5] test: add basic tests for EE mysql bridge --- lib-ee/emqx_ee_bridge/docker-ct | 1 + .../test/emqx_ee_bridge_mysql_SUITE.erl | 260 ++++++++++++++++++ 2 files changed, 261 insertions(+) create mode 100644 lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl diff --git a/lib-ee/emqx_ee_bridge/docker-ct b/lib-ee/emqx_ee_bridge/docker-ct index 3be129d94..1548a3203 100644 --- a/lib-ee/emqx_ee_bridge/docker-ct +++ b/lib-ee/emqx_ee_bridge/docker-ct @@ -2,3 +2,4 @@ influxdb kafka mongo mongo_rs_sharded +mysql diff --git a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl new file mode 100644 index 000000000..b6f876437 --- /dev/null +++ b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl @@ -0,0 +1,260 @@ +%%-------------------------------------------------------------------- +%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved. +%%-------------------------------------------------------------------- + +-module(emqx_ee_bridge_mysql_SUITE). + +-compile(nowarn_export_all). +-compile(export_all). + +-include_lib("eunit/include/eunit.hrl"). +-include_lib("common_test/include/ct.hrl"). + +% SQL definitions +-define(SQL_BRIDGE, + "INSERT INTO mqtt_test(payload, arrived) " + "VALUES (${payload}, FROM_UNIXTIME(${timestamp}/1000))" +). +-define(SQL_CREATE_TABLE, + "CREATE TABLE IF NOT EXISTS mqtt_test (payload blob, arrived datetime NOT NULL) " + "DEFAULT CHARSET=utf8MB4;" +). +-define(SQL_DROP_TABLE, "DROP TABLE mqtt_test"). +-define(SQL_DELETE, "DELETE from mqtt_test"). +-define(SQL_SELECT, "SELECT payload FROM mqtt_test"). + +% DB defaults +-define(MYSQL_DATABASE, "mqtt"). +-define(MYSQL_USERNAME, "root"). +-define(MYSQL_PASSWORD, "public"). + +%%------------------------------------------------------------------------------ +%% CT boilerplate +%%------------------------------------------------------------------------------ + +all() -> + [ + {group, tcp}, + {group, tls} + | (emqx_common_test_helpers:all(?MODULE) -- group_tests()) + ]. + +group_tests() -> + [ + t_setup_via_config_and_publish, + t_setup_via_http_api_and_publish + ]. + +groups() -> + [ + {tcp, group_tests()}, + {tls, group_tests()} + ]. + +init_per_group(GroupType = tcp, Config) -> + MysqlHost = os:getenv("MYSQL_TCP_HOST", "mysql"), + MysqlPort = list_to_integer(os:getenv("MYSQL_TCP_PORT", "3306")), + common_init(GroupType, Config, MysqlHost, MysqlPort); +init_per_group(GroupType = tls, Config) -> + MysqlHost = os:getenv("MYSQL_TLS_HOST", "mysql-tls"), + MysqlPort = list_to_integer(os:getenv("MYSQL_TLS_PORT", "3306")), + common_init(GroupType, Config, MysqlHost, MysqlPort). + +end_per_group(GroupType, Config) -> + drop_table_raw(GroupType, 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) -> + catch clear_table(Config), + delete_bridge(Config), + Config. + +end_per_testcase(_Testcase, Config) -> + catch clear_table(Config), + delete_bridge(Config), + ok. + +%%------------------------------------------------------------------------------ +%% Helper fns +%%------------------------------------------------------------------------------ + +common_init(GroupType, Config0, MysqlHost, MysqlPort) -> + BridgeType = <<"mysql">>, + case emqx_common_test_helpers:is_tcp_server_available(MysqlHost, MysqlPort) of + true -> + % 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(), + {Name, MysqlConfig} = mysql_config(MysqlHost, MysqlPort, GroupType, BridgeType), + Config = + [ + {mysql_host, MysqlHost}, + {mysql_port, MysqlPort}, + {mysql_config, MysqlConfig}, + {mysql_bridge_type, BridgeType}, + {mysql_name, Name} + | Config0 + ], + create_table_raw(GroupType, Config), + Config; + false -> + {skip, no_mysql} + end. + +mysql_config(MysqlHost, MysqlPort0, GroupType, BridgeType) -> + MysqlPort = integer_to_list(MysqlPort0), + Server = MysqlHost ++ ":" ++ MysqlPort, + Name = iolist_to_binary(io_lib:format("~s-~s", [?MODULE, GroupType])), + SslEnabled = + case GroupType of + tcp -> "false"; + tls -> "true" + end, + 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" + " ssl = {\n" + " enable = ~w\n" + " }\n" + "}", + [ + BridgeType, + Name, + Server, + ?MYSQL_DATABASE, + ?MYSQL_USERNAME, + ?MYSQL_PASSWORD, + ?SQL_BRIDGE, + SslEnabled + ] + ), + {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) -> + BridgeType = ?config(mysql_bridge_type, Config), + Name = ?config(mysql_name, Config), + MysqlConfig = ?config(mysql_config, Config), + emqx_bridge:create(BridgeType, Name, MysqlConfig). + +delete_bridge(Config) -> + BridgeType = ?config(mysql_bridge_type, Config), + Name = ?config(mysql_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. + +query(Config, SqlQuery) -> + BridgeType = ?config(mysql_bridge_type, Config), + Name = ?config(mysql_name, Config), + ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name), + ?assertMatch({ok, connected}, emqx_resource:health_check(ResourceID)), + emqx_resource:simple_sync_query(ResourceID, {sql, SqlQuery}). + +send_message(Config, Payload) -> + Name = ?config(mysql_name, Config), + BridgeType = ?config(mysql_bridge_type, Config), + ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name), + ?assertMatch({ok, connected}, emqx_resource:health_check(ResourceID)), + % TODO: Check why we can't use send_message directly! + % BridgeID = emqx_bridge_resource:bridge_id(Type, Name), + % emqx_bridge:send_message(BridgeID, Payload). + emqx_resource:simple_sync_query(ResourceID, {send_message, Payload}). + +clear_table(Config) -> + query(Config, ?SQL_DELETE). + +get_payload(Config) -> + query(Config, ?SQL_SELECT). + +% We need to create and drop the test table outside of using bridges +% since a bridge expects the table to exist when enabling it. We +% therefore call the mysql module directly. +connect_raw_and_run_sql(GroupType, Config, Sql) -> + Opts = [ + {host, ?config(mysql_host, Config)}, + {port, ?config(mysql_port, Config)}, + {user, ?MYSQL_USERNAME}, + {password, ?MYSQL_PASSWORD}, + {database, ?MYSQL_DATABASE} + ], + SslOpts = + case GroupType of + tls -> + [{ssl, emqx_tls_lib:to_client_opts(#{enable => true})}]; + tcp -> + [] + end, + {ok, Pid} = mysql:start_link(Opts ++ SslOpts), + ok = mysql:query(Pid, Sql), + mysql:stop(Pid). + +create_table_raw(GroupType, Config) -> + connect_raw_and_run_sql(GroupType, Config, ?SQL_CREATE_TABLE). + +drop_table_raw(GroupType, Config) -> + connect_raw_and_run_sql(GroupType, Config, ?SQL_DROP_TABLE). + +%%------------------------------------------------------------------------------ +%% Testcases +%%------------------------------------------------------------------------------ + +t_setup_via_config_and_publish(Config) -> + ?assertMatch( + {ok, _}, + create_bridge(Config) + ), + Val = integer_to_binary(erlang:unique_integer()), + ?assertMatch(ok, send_message(Config, #{payload => Val, timestamp => 1668602148000})), + ?assertMatch( + {ok, [<<"payload">>], [[Val]]}, + get_payload(Config) + ), + ok. + +t_setup_via_http_api_and_publish(Config) -> + BridgeType = ?config(mysql_bridge_type, Config), + Name = ?config(mysql_name, Config), + MysqlConfig0 = ?config(mysql_config, Config), + MysqlConfig = MysqlConfig0#{ + <<"name">> => Name, + <<"type">> => BridgeType + }, + ?assertMatch( + {ok, _}, + create_bridge_http(MysqlConfig) + ), + Val = integer_to_binary(erlang:unique_integer()), + ?assertMatch(ok, send_message(Config, #{payload => Val, timestamp => 1668602148000})), + ?assertMatch( + {ok, [<<"payload">>], [[Val]]}, + get_payload(Config) + ), + ok. From 96bff1d32e53c3ab1585b4f9b6e77666f8ecbd7a Mon Sep 17 00:00:00 2001 From: Erik Timan Date: Fri, 25 Nov 2022 10:54:28 +0100 Subject: [PATCH 2/5] test: improve basic tests for EE mysql bridge --- .../src/emqx_connector_mysql.erl | 9 + .../test/emqx_ee_bridge_mysql_SUITE.erl | 210 ++++++++++++------ 2 files changed, 150 insertions(+), 69 deletions(-) diff --git a/apps/emqx_connector/src/emqx_connector_mysql.erl b/apps/emqx_connector/src/emqx_connector_mysql.erl index b35f0b018..afc6fbdd2 100644 --- a/apps/emqx_connector/src/emqx_connector_mysql.erl +++ b/apps/emqx_connector/src/emqx_connector_mysql.erl @@ -19,6 +19,7 @@ -include_lib("typerefl/include/types.hrl"). -include_lib("hocon/include/hoconsc.hrl"). -include_lib("emqx/include/logger.hrl"). +-include_lib("snabbkaffe/include/snabbkaffe.hrl"). -behaviour(emqx_resource). @@ -398,6 +399,10 @@ on_sql_query( ?TRACE("QUERY", "mysql_connector_received", LogMeta), Worker = ecpool:get_client(PoolName), {ok, Conn} = ecpool_worker:client(Worker), + ?tp( + mysql_connector_send_query, + #{sql_or_key => SQLOrKey, data => Data} + ), try mysql:SQLFunc(Conn, SQLOrKey, Data, Timeout) of {error, disconnected} = Result -> ?SLOG( @@ -427,6 +432,10 @@ on_sql_query( ), Result; Result -> + ?tp( + mysql_connector_query_return, + #{result => Result} + ), Result catch error:badarg -> diff --git a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl index b6f876437..753a9f3f9 100644 --- a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl +++ b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl @@ -9,6 +9,7 @@ -include_lib("eunit/include/eunit.hrl"). -include_lib("common_test/include/ct.hrl"). +-include_lib("snabbkaffe/include/snabbkaffe.hrl"). % SQL definitions -define(SQL_BRIDGE, @@ -34,34 +35,68 @@ all() -> [ - {group, tcp}, - {group, tls} - | (emqx_common_test_helpers:all(?MODULE) -- group_tests()) - ]. - -group_tests() -> - [ - t_setup_via_config_and_publish, - t_setup_via_http_api_and_publish + {group, with_batch}, + {group, without_batch} ]. groups() -> + TCs = emqx_common_test_helpers:all(?MODULE), [ - {tcp, group_tests()}, - {tls, group_tests()} + {with_batch, [ + {group, sync_query}, + {group, async_query} + ]}, + {without_batch, [ + {group, sync_query}, + {group, async_query} + ]}, + {sync_query, [ + {group, tcp}, + {group, tls} + ]}, + {async_query, [ + {group, tcp}, + {group, tls} + ]}, + {tcp, TCs}, + {tls, TCs} ]. -init_per_group(GroupType = tcp, Config) -> +init_per_group(tcp, Config0) -> MysqlHost = os:getenv("MYSQL_TCP_HOST", "mysql"), MysqlPort = list_to_integer(os:getenv("MYSQL_TCP_PORT", "3306")), - common_init(GroupType, Config, MysqlHost, MysqlPort); -init_per_group(GroupType = tls, Config) -> + Config = [ + {mysql_host, MysqlHost}, + {mysql_port, MysqlPort}, + {enable_tls, false} + | Config0 + ], + common_init(Config); +init_per_group(tls, Config0) -> MysqlHost = os:getenv("MYSQL_TLS_HOST", "mysql-tls"), MysqlPort = list_to_integer(os:getenv("MYSQL_TLS_PORT", "3306")), - common_init(GroupType, Config, MysqlHost, MysqlPort). + Config = [ + {mysql_host, MysqlHost}, + {mysql_port, MysqlPort}, + {enable_tls, true} + | Config0 + ], + common_init(Config); +init_per_group(sync_query, Config) -> + [{query_mode, sync} | Config]; +init_per_group(async_query, Config) -> + [{query_mode, async} | Config]; +init_per_group(with_batch, Config) -> + [{enable_batch, true} | Config]; +init_per_group(without_batch, Config) -> + [{enable_batch, false} | Config]; +init_per_group(_Group, Config) -> + Config. -end_per_group(GroupType, Config) -> - drop_table_raw(GroupType, Config), +end_per_group(Group, Config) when Group =:= tcp; Group =:= tls -> + connect_and_drop_table(Config), + ok; +end_per_group(_Group, _Config) -> ok. init_per_suite(Config) -> @@ -72,13 +107,16 @@ end_per_suite(_Config) -> ok = emqx_common_test_helpers:stop_apps([emqx_bridge, emqx_conf]), ok. -init_per_testcase(_Testcase, Config) -> +init_per_testcase(_Testcase, Config0) -> + Config = [{mysql_direct_pid, connect_direct_mysql(Config0)} | Config0], catch clear_table(Config), delete_bridge(Config), Config. end_per_testcase(_Testcase, Config) -> catch clear_table(Config), + DirectPid = ?config(mysql_direct_pid, Config), + mysql:stop(DirectPid), delete_bridge(Config), ok. @@ -86,8 +124,10 @@ end_per_testcase(_Testcase, Config) -> %% Helper fns %%------------------------------------------------------------------------------ -common_init(GroupType, Config0, MysqlHost, MysqlPort) -> +common_init(Config0) -> BridgeType = <<"mysql">>, + MysqlHost = ?config(mysql_host, Config0), + MysqlPort = ?config(mysql_port, Config0), case emqx_common_test_helpers:is_tcp_server_available(MysqlHost, MysqlPort) of true -> % Ensure EE bridge module is loaded @@ -95,31 +135,28 @@ common_init(GroupType, Config0, MysqlHost, MysqlPort) -> _ = emqx_ee_bridge:module_info(), ok = emqx_common_test_helpers:start_apps([emqx_conf, emqx_bridge]), emqx_mgmt_api_test_util:init_suite(), - {Name, MysqlConfig} = mysql_config(MysqlHost, MysqlPort, GroupType, BridgeType), + % Connect to mysql directly and create the table + connect_and_create_table(Config0), + {Name, MysqlConfig} = mysql_config(BridgeType, Config0), Config = [ - {mysql_host, MysqlHost}, - {mysql_port, MysqlPort}, {mysql_config, MysqlConfig}, {mysql_bridge_type, BridgeType}, {mysql_name, Name} | Config0 ], - create_table_raw(GroupType, Config), Config; false -> {skip, no_mysql} end. -mysql_config(MysqlHost, MysqlPort0, GroupType, BridgeType) -> - MysqlPort = integer_to_list(MysqlPort0), - Server = MysqlHost ++ ":" ++ MysqlPort, - Name = iolist_to_binary(io_lib:format("~s-~s", [?MODULE, GroupType])), - SslEnabled = - case GroupType of - tcp -> "false"; - tls -> "true" - end, +mysql_config(BridgeType, Config) -> + MysqlPort = integer_to_list(?config(mysql_port, Config)), + Server = ?config(mysql_host, Config) ++ ":" ++ MysqlPort, + Name = atom_to_binary(?MODULE), + EnableBatch = ?config(enable_batch, Config), + QueryMode = ?config(query_mode, Config), + TlsEnabled = ?config(enable_tls, Config), ConfigString = io_lib:format( "bridges.~s.~s {\n" @@ -129,6 +166,10 @@ mysql_config(MysqlHost, MysqlPort0, GroupType, BridgeType) -> " username = ~p\n" " password = ~p\n" " sql = ~p\n" + " resource_opts = {\n" + " enable_batch = ~p\n" + " query_mode = ~s\n" + " }\n" " ssl = {\n" " enable = ~w\n" " }\n" @@ -141,7 +182,9 @@ mysql_config(MysqlHost, MysqlPort0, GroupType, BridgeType) -> ?MYSQL_USERNAME, ?MYSQL_PASSWORD, ?SQL_BRIDGE, - SslEnabled + EnableBatch, + QueryMode, + TlsEnabled ] ), {Name, parse_and_check(ConfigString, BridgeType, Name)}. @@ -171,33 +214,19 @@ create_bridge_http(Params) -> Error -> Error end. -query(Config, SqlQuery) -> - BridgeType = ?config(mysql_bridge_type, Config), - Name = ?config(mysql_name, Config), - ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name), - ?assertMatch({ok, connected}, emqx_resource:health_check(ResourceID)), - emqx_resource:simple_sync_query(ResourceID, {sql, SqlQuery}). - send_message(Config, Payload) -> Name = ?config(mysql_name, Config), BridgeType = ?config(mysql_bridge_type, Config), + BridgeID = emqx_bridge_resource:bridge_id(BridgeType, Name), ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name), ?assertMatch({ok, connected}, emqx_resource:health_check(ResourceID)), - % TODO: Check why we can't use send_message directly! - % BridgeID = emqx_bridge_resource:bridge_id(Type, Name), - % emqx_bridge:send_message(BridgeID, Payload). - emqx_resource:simple_sync_query(ResourceID, {send_message, Payload}). - -clear_table(Config) -> - query(Config, ?SQL_DELETE). - -get_payload(Config) -> - query(Config, ?SQL_SELECT). + emqx_bridge:send_message(BridgeID, Payload). % We need to create and drop the test table outside of using bridges % since a bridge expects the table to exist when enabling it. We -% therefore call the mysql module directly. -connect_raw_and_run_sql(GroupType, Config, Sql) -> +% therefore call the mysql module directly, in addition to using it +% for querying the DB directly. +connect_direct_mysql(Config) -> Opts = [ {host, ?config(mysql_host, Config)}, {port, ?config(mysql_port, Config)}, @@ -206,21 +235,34 @@ connect_raw_and_run_sql(GroupType, Config, Sql) -> {database, ?MYSQL_DATABASE} ], SslOpts = - case GroupType of - tls -> + case ?config(enable_tls, Config) of + true -> [{ssl, emqx_tls_lib:to_client_opts(#{enable => true})}]; - tcp -> + false -> [] end, {ok, Pid} = mysql:start_link(Opts ++ SslOpts), - ok = mysql:query(Pid, Sql), - mysql:stop(Pid). + Pid. -create_table_raw(GroupType, Config) -> - connect_raw_and_run_sql(GroupType, Config, ?SQL_CREATE_TABLE). +% These funs connect and then stop the mysql connection +connect_and_create_table(Config) -> + DirectPid = connect_direct_mysql(Config), + ok = mysql:query(DirectPid, ?SQL_CREATE_TABLE), + mysql:stop(DirectPid). -drop_table_raw(GroupType, Config) -> - connect_raw_and_run_sql(GroupType, Config, ?SQL_DROP_TABLE). +connect_and_drop_table(Config) -> + DirectPid = connect_direct_mysql(Config), + ok = mysql:query(DirectPid, ?SQL_DROP_TABLE), + mysql:stop(DirectPid). + +% These funs expects a connection to already exist +clear_table(Config) -> + DirectPid = ?config(mysql_direct_pid, Config), + ok = mysql:query(DirectPid, ?SQL_DELETE). + +get_payload(Config) -> + DirectPid = ?config(mysql_direct_pid, Config), + mysql:query(DirectPid, ?SQL_SELECT). %%------------------------------------------------------------------------------ %% Testcases @@ -232,10 +274,25 @@ t_setup_via_config_and_publish(Config) -> create_bridge(Config) ), Val = integer_to_binary(erlang:unique_integer()), - ?assertMatch(ok, send_message(Config, #{payload => Val, timestamp => 1668602148000})), - ?assertMatch( - {ok, [<<"payload">>], [[Val]]}, - get_payload(Config) + SentData = #{payload => Val, timestamp => 1668602148000}, + ?check_trace( + begin + ?wait_async_action( + ?assertEqual(ok, send_message(Config, SentData)), + #{?snk_kind := mysql_connector_query_return}, + 10_000 + ), + ?assertMatch( + {ok, [<<"payload">>], [[Val]]}, + get_payload(Config) + ), + ok + end, + fun(Trace0) -> + Trace = ?of_kind(mysql_connector_query_return, Trace0), + ?assertMatch([#{result := ok}], Trace), + ok + end ), ok. @@ -252,9 +309,24 @@ t_setup_via_http_api_and_publish(Config) -> create_bridge_http(MysqlConfig) ), Val = integer_to_binary(erlang:unique_integer()), - ?assertMatch(ok, send_message(Config, #{payload => Val, timestamp => 1668602148000})), - ?assertMatch( - {ok, [<<"payload">>], [[Val]]}, - get_payload(Config) + SentData = #{payload => Val, timestamp => 1668602148000}, + ?check_trace( + begin + ?wait_async_action( + ?assertEqual(ok, send_message(Config, SentData)), + #{?snk_kind := mysql_connector_query_return}, + 10_000 + ), + ?assertMatch( + {ok, [<<"payload">>], [[Val]]}, + get_payload(Config) + ), + ok + end, + fun(Trace0) -> + Trace = ?of_kind(mysql_connector_query_return, Trace0), + ?assertMatch([#{result := ok}], Trace), + ok + end ), ok. From eb62192838768458e85cef38c348eb23d7684ce8 Mon Sep 17 00:00:00 2001 From: Erik Timan Date: Mon, 28 Nov 2022 13:45:52 +0100 Subject: [PATCH 3/5] test: expand EE mysql bridge test with toxiproxy --- .../docker-compose-toxiproxy.yaml | 2 + .ci/docker-compose-file/toxiproxy.json | 12 +++ lib-ee/emqx_ee_bridge/docker-ct | 1 + .../test/emqx_ee_bridge_mysql_SUITE.erl | 81 ++++++++++++++----- scripts/ct/run.sh | 6 +- 5 files changed, 78 insertions(+), 24 deletions(-) diff --git a/.ci/docker-compose-file/docker-compose-toxiproxy.yaml b/.ci/docker-compose-file/docker-compose-toxiproxy.yaml index 924c9e6ae..66e7ec308 100644 --- a/.ci/docker-compose-file/docker-compose-toxiproxy.yaml +++ b/.ci/docker-compose-file/docker-compose-toxiproxy.yaml @@ -13,6 +13,8 @@ services: - 8474:8474 - 8086:8086 - 8087:8087 + - 13306:3306 + - 13307:3307 command: - "-host=0.0.0.0" - "-config=/config/toxiproxy.json" diff --git a/.ci/docker-compose-file/toxiproxy.json b/.ci/docker-compose-file/toxiproxy.json index 176b35d36..2d3a30b6b 100644 --- a/.ci/docker-compose-file/toxiproxy.json +++ b/.ci/docker-compose-file/toxiproxy.json @@ -10,5 +10,17 @@ "listen": "0.0.0.0:8087", "upstream": "influxdb_tls:8086", "enabled": true + }, + { + "name": "mysql_tcp", + "listen": "0.0.0.0:3306", + "upstream": "mysql:3306", + "enabled": true + }, + { + "name": "mysql_tls", + "listen": "0.0.0.0:3307", + "upstream": "mysql-tls:3306", + "enabled": true } ] diff --git a/lib-ee/emqx_ee_bridge/docker-ct b/lib-ee/emqx_ee_bridge/docker-ct index 1548a3203..94f9379df 100644 --- a/lib-ee/emqx_ee_bridge/docker-ct +++ b/lib-ee/emqx_ee_bridge/docker-ct @@ -1,3 +1,4 @@ +toxiproxy influxdb kafka mongo diff --git a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl index 753a9f3f9..655fee6e5 100644 --- a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl +++ b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl @@ -63,22 +63,24 @@ groups() -> ]. init_per_group(tcp, Config0) -> - MysqlHost = os:getenv("MYSQL_TCP_HOST", "mysql"), + MysqlHost = os:getenv("MYSQL_TCP_HOST", "toxiproxy"), MysqlPort = list_to_integer(os:getenv("MYSQL_TCP_PORT", "3306")), Config = [ {mysql_host, MysqlHost}, {mysql_port, MysqlPort}, - {enable_tls, false} + {enable_tls, false}, + {proxy_name, "mysql_tcp"} | Config0 ], common_init(Config); init_per_group(tls, Config0) -> - MysqlHost = os:getenv("MYSQL_TLS_HOST", "mysql-tls"), - MysqlPort = list_to_integer(os:getenv("MYSQL_TLS_PORT", "3306")), + MysqlHost = os:getenv("MYSQL_TLS_HOST", "toxiproxy"), + MysqlPort = list_to_integer(os:getenv("MYSQL_TLS_PORT", "3307")), Config = [ {mysql_host, MysqlHost}, {mysql_port, MysqlPort}, - {enable_tls, true} + {enable_tls, true}, + {proxy_name, "mysql_tls"} | Config0 ], common_init(Config); @@ -95,6 +97,9 @@ init_per_group(_Group, Config) -> end_per_group(Group, Config) when Group =:= tcp; Group =:= tls -> connect_and_drop_table(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. @@ -107,16 +112,18 @@ end_per_suite(_Config) -> ok = emqx_common_test_helpers:stop_apps([emqx_bridge, emqx_conf]), ok. -init_per_testcase(_Testcase, Config0) -> - Config = [{mysql_direct_pid, connect_direct_mysql(Config0)} | Config0], - catch clear_table(Config), +init_per_testcase(_Testcase, Config) -> + % Config = [{mysql_direct_pid, connect_direct_mysql(Config0)} | Config0], + connect_and_clear_table(Config), delete_bridge(Config), Config. end_per_testcase(_Testcase, Config) -> - catch clear_table(Config), - DirectPid = ?config(mysql_direct_pid, Config), - mysql:stop(DirectPid), + 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. @@ -130,6 +137,10 @@ common_init(Config0) -> MysqlPort = ?config(mysql_port, Config0), case emqx_common_test_helpers:is_tcp_server_available(MysqlHost, MysqlPort) 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(), @@ -142,7 +153,9 @@ common_init(Config0) -> [ {mysql_config, MysqlConfig}, {mysql_bridge_type, BridgeType}, - {mysql_name, Name} + {mysql_name, Name}, + {proxy_host, ProxyHost}, + {proxy_port, ProxyPort} | Config0 ], Config; @@ -171,7 +184,7 @@ mysql_config(BridgeType, Config) -> " query_mode = ~s\n" " }\n" " ssl = {\n" - " enable = ~w\n" + " enable = ~w\n" " }\n" "}", [ @@ -255,14 +268,16 @@ connect_and_drop_table(Config) -> ok = mysql:query(DirectPid, ?SQL_DROP_TABLE), mysql:stop(DirectPid). -% These funs expects a connection to already exist -clear_table(Config) -> - DirectPid = ?config(mysql_direct_pid, Config), - ok = mysql:query(DirectPid, ?SQL_DELETE). +connect_and_clear_table(Config) -> + DirectPid = connect_direct_mysql(Config), + ok = mysql:query(DirectPid, ?SQL_DELETE), + mysql:stop(DirectPid). -get_payload(Config) -> - DirectPid = ?config(mysql_direct_pid, Config), - mysql:query(DirectPid, ?SQL_SELECT). +connect_and_get_payload(Config) -> + DirectPid = connect_direct_mysql(Config), + Result = mysql:query(DirectPid, ?SQL_SELECT), + mysql:stop(DirectPid), + Result. %%------------------------------------------------------------------------------ %% Testcases @@ -284,7 +299,7 @@ t_setup_via_config_and_publish(Config) -> ), ?assertMatch( {ok, [<<"payload">>], [[Val]]}, - get_payload(Config) + connect_and_get_payload(Config) ), ok end, @@ -319,7 +334,7 @@ t_setup_via_http_api_and_publish(Config) -> ), ?assertMatch( {ok, [<<"payload">>], [[Val]]}, - get_payload(Config) + connect_and_get_payload(Config) ), ok end, @@ -330,3 +345,25 @@ t_setup_via_http_api_and_publish(Config) -> 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(mysql_name, Config), + BridgeType = ?config(mysql_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() -> + ?assertMatch( + {ok, Status} when Status =:= disconnected orelse Status =:= connecting, + emqx_resource_manager:health_check(ResourceID) + ) + end), + ok. diff --git a/scripts/ct/run.sh b/scripts/ct/run.sh index d1e24558b..18dfb2525 100755 --- a/scripts/ct/run.sh +++ b/scripts/ct/run.sh @@ -92,9 +92,11 @@ for dep in ${CT_DEPS}; do erlang24) FILES+=( '.ci/docker-compose-file/docker-compose.yaml' ) ;; + toxiproxy) + FILES+=( '.ci/docker-compose-file/docker-compose-toxiproxy.yaml' ) + ;; influxdb) - FILES+=( '.ci/docker-compose-file/docker-compose-toxiproxy.yaml' - '.ci/docker-compose-file/docker-compose-influxdb-tcp.yaml' + FILES+=( '.ci/docker-compose-file/docker-compose-influxdb-tcp.yaml' '.ci/docker-compose-file/docker-compose-influxdb-tls.yaml' ) ;; mongo) From 5461311edc1ea34225297ed8cd861ea54f6d1d4b Mon Sep 17 00:00:00 2001 From: Erik Timan Date: Wed, 30 Nov 2022 14:28:50 +0100 Subject: [PATCH 4/5] test: more EE mysql bridge tests --- .../src/emqx_connector_mysql.erl | 10 ++++- .../test/emqx_ee_bridge_mysql_SUITE.erl | 45 +++++++++++++++++-- 2 files changed, 50 insertions(+), 5 deletions(-) diff --git a/apps/emqx_connector/src/emqx_connector_mysql.erl b/apps/emqx_connector/src/emqx_connector_mysql.erl index afc6fbdd2..fc3068c66 100644 --- a/apps/emqx_connector/src/emqx_connector_mysql.erl +++ b/apps/emqx_connector/src/emqx_connector_mysql.erl @@ -119,8 +119,14 @@ on_start( Prepares = parse_prepare_sql(Config), State = maps:merge(#{poolname => PoolName, auto_reconnect => AutoReconn}, Prepares), case emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Options ++ SslOpts) of - ok -> {ok, init_prepare(State)}; - {error, Reason} -> {error, Reason} + ok -> + {ok, init_prepare(State)}; + {error, Reason} -> + ?tp( + mysql_connector_start_failed, + #{error => Reason} + ), + {error, Reason} end. on_stop(InstId, #{poolname := PoolName}) -> diff --git a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl index 655fee6e5..0d8f68b39 100644 --- a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl +++ b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl @@ -113,7 +113,6 @@ end_per_suite(_Config) -> ok. init_per_testcase(_Testcase, Config) -> - % Config = [{mysql_direct_pid, connect_direct_mysql(Config0)} | Config0], connect_and_clear_table(Config), delete_bridge(Config), Config. @@ -231,8 +230,6 @@ send_message(Config, Payload) -> Name = ?config(mysql_name, Config), BridgeType = ?config(mysql_bridge_type, Config), BridgeID = emqx_bridge_resource:bridge_id(BridgeType, Name), - ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name), - ?assertMatch({ok, connected}, emqx_resource:health_check(ResourceID)), emqx_bridge:send_message(BridgeID, Payload). % We need to create and drop the test table outside of using bridges @@ -367,3 +364,45 @@ t_get_status(Config) -> ) 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(mysql_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), + {ok, _} = create_bridge(Config), + Val = integer_to_binary(erlang:unique_integer()), + SentData = #{payload => Val, timestamp => 1668602148000}, + ?check_trace( + emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() -> + send_message(Config, SentData) + end), + fun(Result, _Trace) -> + case QueryMode of + sync -> + ?assertMatch({error, {resource_error, _}}, Result); + async -> + ?assertEqual(ok, Result) + end, + ok + end + ), + ok. From 499a32ce36bd34e50cccd03147e7883918af9540 Mon Sep 17 00:00:00 2001 From: Erik Timan Date: Thu, 1 Dec 2022 16:38:03 +0100 Subject: [PATCH 5/5] test: remove unnecessary async tests in EE mysql bridge The async test cases is not needed since the mysql connector uses the always_sync callback mode. --- .../test/emqx_ee_bridge_mysql_SUITE.erl | 20 +++---------------- 1 file changed, 3 insertions(+), 17 deletions(-) diff --git a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl index 0d8f68b39..292c02580 100644 --- a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl +++ b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl @@ -43,21 +43,15 @@ groups() -> TCs = emqx_common_test_helpers:all(?MODULE), [ {with_batch, [ - {group, sync_query}, - {group, async_query} + {group, sync_query} ]}, {without_batch, [ - {group, sync_query}, - {group, async_query} + {group, sync_query} ]}, {sync_query, [ {group, tcp}, {group, tls} ]}, - {async_query, [ - {group, tcp}, - {group, tls} - ]}, {tcp, TCs}, {tls, TCs} ]. @@ -86,8 +80,6 @@ init_per_group(tls, Config0) -> common_init(Config); init_per_group(sync_query, Config) -> [{query_mode, sync} | Config]; -init_per_group(async_query, Config) -> - [{query_mode, async} | Config]; init_per_group(with_batch, Config) -> [{enable_batch, true} | Config]; init_per_group(without_batch, Config) -> @@ -387,7 +379,6 @@ t_write_failure(Config) -> ProxyName = ?config(proxy_name, Config), ProxyPort = ?config(proxy_port, Config), ProxyHost = ?config(proxy_host, Config), - QueryMode = ?config(query_mode, Config), {ok, _} = create_bridge(Config), Val = integer_to_binary(erlang:unique_integer()), SentData = #{payload => Val, timestamp => 1668602148000}, @@ -396,12 +387,7 @@ t_write_failure(Config) -> send_message(Config, SentData) end), fun(Result, _Trace) -> - case QueryMode of - sync -> - ?assertMatch({error, {resource_error, _}}, Result); - async -> - ?assertEqual(ok, Result) - end, + ?assertMatch({error, {resource_error, _}}, Result), ok end ),