From 4ad3579966945a40750e9ee2b3c87fe4148a6d89 Mon Sep 17 00:00:00 2001 From: firest Date: Mon, 20 Mar 2023 11:14:25 +0800 Subject: [PATCH] test(bridges): add test suite for RocketMQ --- .../docker-compose-rocketmq.yaml | 34 +++ .../docker-compose-toxiproxy.yaml | 1 + .../rocketmq/conf/broker.conf | 22 ++ .../rocketmq/logs/.gitkeep | 0 .../rocketmq/store/.gitkeep | 0 .ci/docker-compose-file/toxiproxy.json | 6 + lib-ee/emqx_ee_bridge/docker-ct | 1 + .../test/emqx_ee_bridge_rocketmq_SUITE.erl | 267 ++++++++++++++++++ .../src/emqx_ee_connector_rocketmq.erl | 4 - scripts/ct/run.sh | 3 + 10 files changed, 334 insertions(+), 4 deletions(-) create mode 100644 .ci/docker-compose-file/docker-compose-rocketmq.yaml create mode 100644 .ci/docker-compose-file/rocketmq/conf/broker.conf create mode 100644 .ci/docker-compose-file/rocketmq/logs/.gitkeep create mode 100644 .ci/docker-compose-file/rocketmq/store/.gitkeep create mode 100644 lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_rocketmq_SUITE.erl diff --git a/.ci/docker-compose-file/docker-compose-rocketmq.yaml b/.ci/docker-compose-file/docker-compose-rocketmq.yaml new file mode 100644 index 000000000..3c872a7c2 --- /dev/null +++ b/.ci/docker-compose-file/docker-compose-rocketmq.yaml @@ -0,0 +1,34 @@ +version: '3.9' + +services: + mqnamesrv: + image: apache/rocketmq:4.9.4 + container_name: rocketmq_namesrv +# ports: +# - 9876:9876 + volumes: + - ./rocketmq/logs:/opt/logs + - ./rocketmq/store:/opt/store + command: ./mqnamesrv + networks: + - emqx_bridge + + mqbroker: + image: apache/rocketmq:4.9.4 + container_name: rocketmq_broker +# ports: +# - 10909:10909 +# - 10911:10911 + volumes: + - ./rocketmq/logs:/opt/logs + - ./rocketmq/store:/opt/store + - ./rocketmq/conf/broker.conf:/etc/rocketmq/broker.conf + environment: + NAMESRV_ADDR: "rocketmq_namesrv:9876" + JAVA_OPTS: " -Duser.home=/opt" + JAVA_OPT_EXT: "-server -Xms1024m -Xmx1024m -Xmn1024m" + command: ./mqbroker -c /etc/rocketmq/broker.conf + depends_on: + - mqnamesrv + networks: + - emqx_bridge diff --git a/.ci/docker-compose-file/docker-compose-toxiproxy.yaml b/.ci/docker-compose-file/docker-compose-toxiproxy.yaml index 16f18b6c2..24f1d90b2 100644 --- a/.ci/docker-compose-file/docker-compose-toxiproxy.yaml +++ b/.ci/docker-compose-file/docker-compose-toxiproxy.yaml @@ -22,6 +22,7 @@ services: - 15433:5433 - 16041:6041 - 18000:8000 + - 19876:9876 command: - "-host=0.0.0.0" - "-config=/config/toxiproxy.json" diff --git a/.ci/docker-compose-file/rocketmq/conf/broker.conf b/.ci/docker-compose-file/rocketmq/conf/broker.conf new file mode 100644 index 000000000..c343090e4 --- /dev/null +++ b/.ci/docker-compose-file/rocketmq/conf/broker.conf @@ -0,0 +1,22 @@ +brokerClusterName=DefaultCluster +brokerName=broker-a +brokerId=0 + +brokerIP1=rocketmq_broker + +defaultTopicQueueNums=4 +autoCreateTopicEnable=true +autoCreateSubscriptionGroup=true + +listenPort=10911 +deleteWhen=04 + +fileReservedTime=120 +mapedFileSizeCommitLog=1073741824 +mapedFileSizeConsumeQueue=300000 +diskMaxUsedSpaceRatio=100 +maxMessageSize=65536 + +brokerRole=ASYNC_MASTER + +flushDiskType=ASYNC_FLUSH diff --git a/.ci/docker-compose-file/rocketmq/logs/.gitkeep b/.ci/docker-compose-file/rocketmq/logs/.gitkeep new file mode 100644 index 000000000..e69de29bb diff --git a/.ci/docker-compose-file/rocketmq/store/.gitkeep b/.ci/docker-compose-file/rocketmq/store/.gitkeep new file mode 100644 index 000000000..e69de29bb diff --git a/.ci/docker-compose-file/toxiproxy.json b/.ci/docker-compose-file/toxiproxy.json index 2f8c4341b..e22735091 100644 --- a/.ci/docker-compose-file/toxiproxy.json +++ b/.ci/docker-compose-file/toxiproxy.json @@ -77,5 +77,11 @@ "listen": "0.0.0.0:9295", "upstream": "kafka-1.emqx.net:9295", "enabled": true + }, + { + "name": "rocketmq", + "listen": "0.0.0.0:9876", + "upstream": "rocketmq_namesrv:9876", + "enabled": true } ] diff --git a/lib-ee/emqx_ee_bridge/docker-ct b/lib-ee/emqx_ee_bridge/docker-ct index ac1728ad2..116bc44ad 100644 --- a/lib-ee/emqx_ee_bridge/docker-ct +++ b/lib-ee/emqx_ee_bridge/docker-ct @@ -10,3 +10,4 @@ pgsql tdengine clickhouse dynamo +rocketmq diff --git a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_rocketmq_SUITE.erl b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_rocketmq_SUITE.erl new file mode 100644 index 000000000..cd02b65d0 --- /dev/null +++ b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_rocketmq_SUITE.erl @@ -0,0 +1,267 @@ +%%-------------------------------------------------------------------- +% Copyright (c) 2022-2023 EMQ Technologies Co., Ltd. All Rights Reserved. +%%-------------------------------------------------------------------- + +-module(emqx_ee_bridge_rocketmq_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"). + +% Bridge defaults +-define(TOPIC, "TopicTest"). +-define(BATCH_SIZE, 10). +-define(PAYLOAD, <<"HELLO">>). + +-define(GET_CONFIG(KEY__, CFG__), proplists:get_value(KEY__, CFG__)). + +%%------------------------------------------------------------------------------ +%% CT boilerplate +%%------------------------------------------------------------------------------ + +all() -> + [ + {group, with_batch}, + {group, without_batch} + ]. + +groups() -> + TCs = emqx_common_test_helpers:all(?MODULE), + [ + {with_batch, TCs}, + {without_batch, TCs} + ]. + +init_per_group(with_batch, Config0) -> + Config = [{batch_size, ?BATCH_SIZE} | Config0], + common_init(Config); +init_per_group(without_batch, Config0) -> + Config = [{batch_size, 1} | Config0], + common_init(Config); +init_per_group(_Group, Config) -> + Config. + +end_per_group(Group, Config) when Group =:= with_batch; Group =:= without_batch -> + 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) -> + delete_bridge(Config), + Config. + +end_per_testcase(_Testcase, Config) -> + ProxyHost = ?config(proxy_host, Config), + ProxyPort = ?config(proxy_port, Config), + emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort), + ok = snabbkaffe:stop(), + delete_bridge(Config), + ok. + +%%------------------------------------------------------------------------------ +%% Helper fns +%%------------------------------------------------------------------------------ + +common_init(ConfigT) -> + BridgeType = <<"rocketmq">>, + Host = os:getenv("ROCKETMQ_HOST", "toxiproxy"), + Port = list_to_integer(os:getenv("ROCKETMQ_PORT", "9876")), + + Config0 = [ + {host, Host}, + {port, Port}, + {query_mode, sync}, + {proxy_name, "rocketmq"} + | ConfigT + ], + + 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(), + {Name, RocketMQConf} = rocketmq_config(BridgeType, Config0), + Config = + [ + {rocketmq_config, RocketMQConf}, + {rocketmq_bridge_type, BridgeType}, + {rocketmq_name, Name}, + {proxy_host, ProxyHost}, + {proxy_port, ProxyPort} + | Config0 + ], + Config; + false -> + case os:getenv("IS_CI") of + false -> + {skip, no_rocketmq}; + _ -> + throw(no_rocketmq) + end + end. + +rocketmq_config(BridgeType, Config) -> + Port = integer_to_list(?GET_CONFIG(port, Config)), + Server = ?GET_CONFIG(host, Config) ++ ":" ++ Port, + Name = atom_to_binary(?MODULE), + BatchSize = ?config(batch_size, Config), + QueryMode = ?config(query_mode, Config), + ConfigString = + io_lib:format( + "bridges.~s.~s {\n" + " enable = true\n" + " server = ~p\n" + " topic = ~p\n" + " resource_opts = {\n" + " request_timeout = 1500ms\n" + " batch_size = ~b\n" + " query_mode = ~s\n" + " }\n" + "}", + [ + BridgeType, + Name, + Server, + ?TOPIC, + BatchSize, + QueryMode + ] + ), + {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 = ?GET_CONFIG(rocketmq_bridge_type, Config), + Name = ?GET_CONFIG(rocketmq_name, Config), + RocketMQConf = ?GET_CONFIG(rocketmq_config, Config), + emqx_bridge:create(BridgeType, Name, RocketMQConf). + +delete_bridge(Config) -> + BridgeType = ?GET_CONFIG(rocketmq_bridge_type, Config), + Name = ?GET_CONFIG(rocketmq_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 = ?GET_CONFIG(rocketmq_name, Config), + BridgeType = ?GET_CONFIG(rocketmq_bridge_type, Config), + BridgeID = emqx_bridge_resource:bridge_id(BridgeType, Name), + emqx_bridge:send_message(BridgeID, Payload). + +query_resource(Config, Request) -> + Name = ?GET_CONFIG(rocketmq_name, Config), + BridgeType = ?GET_CONFIG(rocketmq_bridge_type, Config), + ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name), + emqx_resource:query(ResourceID, Request, #{timeout => 500}). + +%%------------------------------------------------------------------------------ +%% Testcases +%%------------------------------------------------------------------------------ + +t_setup_via_config_and_publish(Config) -> + ?assertMatch( + {ok, _}, + create_bridge(Config) + ), + SentData = #{payload => ?PAYLOAD}, + ?check_trace( + begin + ?wait_async_action( + ?assertEqual(ok, send_message(Config, SentData)), + #{?snk_kind := rocketmq_connector_query_return}, + 10_000 + ), + ok + end, + fun(Trace0) -> + Trace = ?of_kind(rocketmq_connector_query_return, Trace0), + ?assertMatch([#{result := ok}], Trace), + ok + end + ), + ok. + +t_setup_via_http_api_and_publish(Config) -> + BridgeType = ?GET_CONFIG(rocketmq_bridge_type, Config), + Name = ?GET_CONFIG(rocketmq_name, Config), + RocketMQConf = ?GET_CONFIG(rocketmq_config, Config), + RocketMQConf2 = RocketMQConf#{ + <<"name">> => Name, + <<"type">> => BridgeType + }, + ?assertMatch( + {ok, _}, + create_bridge_http(RocketMQConf2) + ), + SentData = #{payload => ?PAYLOAD}, + ?check_trace( + begin + ?wait_async_action( + ?assertEqual(ok, send_message(Config, SentData)), + #{?snk_kind := rocketmq_connector_query_return}, + 10_000 + ), + ok + end, + fun(Trace0) -> + Trace = ?of_kind(rocketmq_connector_query_return, Trace0), + ?assertMatch([#{result := ok}], Trace), + ok + end + ), + ok. + +t_get_status(Config) -> + ?assertMatch( + {ok, _}, + create_bridge(Config) + ), + + Name = ?GET_CONFIG(rocketmq_name, Config), + BridgeType = ?GET_CONFIG(rocketmq_bridge_type, Config), + ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name), + + ?assertEqual({ok, connected}, emqx_resource_manager:health_check(ResourceID)), + ok. + +t_simple_query(Config) -> + ?assertMatch( + {ok, _}, + create_bridge(Config) + ), + Request = {send_message, #{message => <<"Hello">>}}, + Result = query_resource(Config, Request), + ?assertEqual(ok, Result), + ok. diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_rocketmq.erl b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_rocketmq.erl index d41f83e1d..84f2e2a89 100644 --- a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_rocketmq.erl +++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_rocketmq.erl @@ -31,10 +31,6 @@ default_port => 9876 }). --ifdef(TEST). --export([execute/2]). --endif. - %%===================================================================== %% Hocon schema roots() -> diff --git a/scripts/ct/run.sh b/scripts/ct/run.sh index bf7b2073d..a4eeb366d 100755 --- a/scripts/ct/run.sh +++ b/scripts/ct/run.sh @@ -170,6 +170,9 @@ for dep in ${CT_DEPS}; do dynamo) FILES+=( '.ci/docker-compose-file/docker-compose-dynamo.yaml' ) ;; + rocketmq) + FILES+=( '.ci/docker-compose-file/docker-compose-rocketmq.yaml' ) + ;; *) echo "unknown_ct_dependency $dep" exit 1