yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: Bridge V2 compatiblilty layer progress and local topic 

* Most Bridge V1 HTTP API calls are now compatible with Bridge V2
* Local topics works for Bridge V2 now
* A lot of work on trying to get the old Kafka producer test suite
  to work after the refactorings
This commit is contained in:
Kjell Winblad 2023-10-04 19:26:28 +02:00 committed by Zaiming (Stone) Shi
parent 16d7f4d3e6
commit c0df85ac09
12 changed files with 851 additions and 348 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
apps/emqx/src/emqx_config.erl
View File

@ -325,7 +325,7 @@ init_load(SchemaMod, Conf) when is_list(Conf) orelse is_binary(Conf) ->
ok = save_schema_mod_and_names(SchemaMod), ok = save_schema_mod_and_names(SchemaMod),
HasDeprecatedFile = has_deprecated_file(), HasDeprecatedFile = has_deprecated_file(),
RawConf0 = load_config_files(HasDeprecatedFile, Conf), RawConf0 = load_config_files(HasDeprecatedFile, Conf),
RawConf1 = emqx_connector_schema:transform_old_style_bridges_to_connector_and_actions(RawConf0), RawConf1 = emqx_connector_schema:transform_bridges_v1_to_connectors_and_bridges_v2(RawConf0),
warning_deprecated_root_key(RawConf1), warning_deprecated_root_key(RawConf1),
RawConf2 = RawConf2 =
case HasDeprecatedFile of case HasDeprecatedFile of

2
apps/emqx/test/emqx_common_test_http.erl
View File

@ -61,7 +61,7 @@ request_api(Method, Url, QueryParams, Auth, Body, HttpOpts) ->
do_request_api(Method, Request, HttpOpts). do_request_api(Method, Request, HttpOpts).
do_request_api(Method, Request, HttpOpts) -> do_request_api(Method, Request, HttpOpts) ->
ct:pal("Method: ~p, Request: ~p", [Method, Request]), % ct:pal("Method: ~p, Request: ~p", [Method, Request]),
case httpc:request(Method, Request, HttpOpts, [{body_format, binary}]) of case httpc:request(Method, Request, HttpOpts, [{body_format, binary}]) of
{error, socket_closed_remotely} -> {error, socket_closed_remotely} ->
{error, socket_closed_remotely}; {error, socket_closed_remotely};

92
apps/emqx_bridge/src/emqx_bridge.erl
View File

@ -279,23 +279,26 @@ post_config_update([?ROOT_KEY], _Req, NewConf, OldConf, _AppEnv) ->
Result. Result.
list() -> list() ->
% OldStyleBridges = BridgeV1Bridges =
maps:fold( maps:fold(
fun(Type, NameAndConf, Bridges) -> fun(Type, NameAndConf, Bridges) ->
maps:fold( maps:fold(
fun(Name, RawConf, Acc) -> fun(Name, RawConf, Acc) ->
case lookup(Type, Name, RawConf) of case lookup(Type, Name, RawConf) of
{error, not_found} -> Acc; {error, not_found} -> Acc;
{ok, Res} -> [Res | Acc] {ok, Res} -> [Res | Acc]
end end
end, end,
Bridges, Bridges,
NameAndConf NameAndConf
) )
end, end,
[], [],
emqx:get_raw_config([bridges], #{}) emqx:get_raw_config([bridges], #{})
). ),
BridgeV2Bridges =
emqx_bridge_v2:list_and_transform_to_bridge_v1(),
BridgeV1Bridges ++ BridgeV2Bridges.
%%BridgeV2Bridges = emqx_bridge_v2:list(). %%BridgeV2Bridges = emqx_bridge_v2:list().
lookup(Id) -> lookup(Id) ->
@ -325,7 +328,12 @@ lookup(Type, Name, RawConf) ->
end. end.
get_metrics(Type, Name) -> get_metrics(Type, Name) ->
emqx_resource:get_metrics(emqx_bridge_resource:resource_id(Type, Name)). case emqx_bridge_v2:is_bridge_v2_type(Type) of
true ->
emqx_bridge_v2:get_metrics(Type, Name);
false ->
emqx_resource:get_metrics(emqx_bridge_resource:resource_id(Type, Name))
end.
maybe_upgrade(mqtt, Config) -> maybe_upgrade(mqtt, Config) ->
emqx_bridge_compatible_config:maybe_upgrade(Config); emqx_bridge_compatible_config:maybe_upgrade(Config);
@ -337,11 +345,16 @@ maybe_upgrade(_Other, Config) ->
disable_enable(Action, BridgeType, BridgeName) when disable_enable(Action, BridgeType, BridgeName) when
Action =:= disable; Action =:= enable Action =:= disable; Action =:= enable
-> ->
emqx_conf:update( case emqx_bridge_v2:is_bridge_v2_type(BridgeType) of
config_key_path() ++ [BridgeType, BridgeName], true ->
{Action, BridgeType, BridgeName}, emqx_bridge_v2:disable_enable(Action, BridgeType, BridgeName);
#{override_to => cluster} false ->
). emqx_conf:update(
config_key_path() ++ [BridgeType, BridgeName],
{Action, BridgeType, BridgeName},
#{override_to => cluster}
)
end.
create(BridgeType, BridgeName, RawConf) -> create(BridgeType, BridgeName, RawConf) ->
?SLOG(debug, #{ ?SLOG(debug, #{
@ -350,24 +363,47 @@ create(BridgeType, BridgeName, RawConf) ->
bridge_name => BridgeName, bridge_name => BridgeName,
bridge_raw_config => emqx_utils:redact(RawConf) bridge_raw_config => emqx_utils:redact(RawConf)
}), }),
emqx_conf:update( case emqx_bridge_v2:is_bridge_v2_type(BridgeType) of
emqx_bridge:config_key_path() ++ [BridgeType, BridgeName], true ->
RawConf, emqx_bridge_v2:split_bridge_v1_config_and_create(BridgeType, BridgeName, RawConf);
#{override_to => cluster} false ->
). emqx_conf:update(
emqx_bridge:config_key_path() ++ [BridgeType, BridgeName],
RawConf,
#{override_to => cluster}
)
end.
%% NOTE: This function can cause broken references but it is only called from
%% test cases.
remove(BridgeType, BridgeName) -> remove(BridgeType, BridgeName) ->
?SLOG(debug, #{ ?SLOG(debug, #{
bridge_action => remove, bridge_action => remove,
bridge_type => BridgeType, bridge_type => BridgeType,
bridge_name => BridgeName bridge_name => BridgeName
}), }),
case emqx_bridge_v2:is_bridge_v2_type(BridgeType) of
true ->
emqx_bridge_v2:remove(BridgeType, BridgeName);
false ->
remove_v1(BridgeType, BridgeName)
end.
remove_v1(BridgeType, BridgeName) ->
emqx_conf:remove( emqx_conf:remove(
emqx_bridge:config_key_path() ++ [BridgeType, BridgeName], emqx_bridge:config_key_path() ++ [BridgeType, BridgeName],
#{override_to => cluster} #{override_to => cluster}
). ).
check_deps_and_remove(BridgeType, BridgeName, RemoveDeps) -> check_deps_and_remove(BridgeType, BridgeName, RemoveDeps) ->
case emqx_bridge_v2:is_bridge_v2_type(BridgeType) of
true ->
emqx_bridge_v2:check_deps_and_remove(BridgeType, BridgeName, RemoveDeps);
false ->
check_deps_and_remove_v1(BridgeType, BridgeName, RemoveDeps)
end.
check_deps_and_remove_v1(BridgeType, BridgeName, RemoveDeps) ->
BridgeId = emqx_bridge_resource:bridge_id(BridgeType, BridgeName), BridgeId = emqx_bridge_resource:bridge_id(BridgeType, BridgeName),
%% NOTE: This violates the design: Rule depends on data-bridge but not vice versa. %% NOTE: This violates the design: Rule depends on data-bridge but not vice versa.
case emqx_rule_engine:get_rule_ids_by_action(BridgeId) of case emqx_rule_engine:get_rule_ids_by_action(BridgeId) of

20
apps/emqx_bridge/src/emqx_bridge_api.erl
View File

@ -485,8 +485,9 @@ schema("/bridges_probe") ->
?TRY_PARSE_ID( ?TRY_PARSE_ID(
Id, Id,
case emqx_bridge:lookup(BridgeType, BridgeName) of case emqx_bridge:lookup(BridgeType, BridgeName) of
{ok, _} -> {ok, #{raw_config := RawConf}} ->
RawConf = emqx:get_raw_config([bridges, BridgeType, BridgeName], #{}), %% TODO will the maybe_upgrade step done by emqx_bridge:lookup cause any problems
%%RawConf = emqx:get_raw_config([bridges, BridgeType, BridgeName], #{}),
Conf = deobfuscate(Conf1, RawConf), Conf = deobfuscate(Conf1, RawConf),
update_bridge(BridgeType, BridgeName, Conf); update_bridge(BridgeType, BridgeName, Conf);
{error, not_found} -> {error, not_found} ->
@ -562,8 +563,9 @@ schema("/bridges_probe") ->
maybe_deobfuscate_bridge_probe(#{<<"type">> := BridgeType, <<"name">> := BridgeName} = Params) -> maybe_deobfuscate_bridge_probe(#{<<"type">> := BridgeType, <<"name">> := BridgeName} = Params) ->
case emqx_bridge:lookup(BridgeType, BridgeName) of case emqx_bridge:lookup(BridgeType, BridgeName) of
{ok, _} -> {ok, #{raw_config := RawConf}} ->
RawConf = emqx:get_raw_config([bridges, BridgeType, BridgeName], #{}), %% TODO check if RawConf optained above is compatible with the commented out code below
%% RawConf = emqx:get_raw_config([bridges, BridgeType, BridgeName], #{}),
deobfuscate(Params, RawConf); deobfuscate(Params, RawConf);
_ -> _ ->
%% A bridge may be probed before it's created, so not finding it here is fine %% A bridge may be probed before it's created, so not finding it here is fine
@ -693,12 +695,12 @@ get_metrics_from_local_node(BridgeType, BridgeName) ->
). ).
is_enabled_bridge(BridgeType, BridgeName) -> is_enabled_bridge(BridgeType, BridgeName) ->
try emqx:get_config([bridges, BridgeType, binary_to_existing_atom(BridgeName)]) of try emqx_bridge:lookup(BridgeType, binary_to_existing_atom(BridgeName)) of
ConfMap -> {ok, #{raw_config := ConfMap}} ->
maps:get(enable, ConfMap, false) maps:get(<<"enable">>, ConfMap, false);
{error, not_found} ->
throw(not_found)
catch catch
error:{config_not_found, _} ->
throw(not_found);
error:badarg -> error:badarg ->
%% catch non-existing atom, %% catch non-existing atom,
%% none-existing atom means it is not available in config PT storage. %% none-existing atom means it is not available in config PT storage.

1
apps/emqx_bridge/src/emqx_bridge_app.erl
View File

@ -18,7 +18,6 @@
-behaviour(application). -behaviour(application).
-include_lib("snabbkaffe/include/snabbkaffe.hrl"). -include_lib("snabbkaffe/include/snabbkaffe.hrl").
-export([start/2, stop/1]). -export([start/2, stop/1]).
-export([ -export([

51
apps/emqx_bridge/src/emqx_bridge_resource.erl
View File

@ -80,7 +80,17 @@ bridge_impl_module(_BridgeType) -> undefined.
-endif. -endif.
resource_id(BridgeId) when is_binary(BridgeId) -> resource_id(BridgeId) when is_binary(BridgeId) ->
<<"bridge:", BridgeId/binary>>. case binary:split(BridgeId, <<":">>) of
[Type, _Name] ->
case emqx_bridge_v2:is_bridge_v2_type(Type) of
true ->
emqx_bridge_v2:bridge_v1_id_to_connector_resource_id(BridgeId);
false ->
<<"bridge:", BridgeId/binary>>
end;
_ ->
invalid_data(<<"should be of pattern {type}:{name}, but got ", BridgeId/binary>>)
end.
resource_id(BridgeType, BridgeName) -> resource_id(BridgeType, BridgeName) ->
BridgeId = bridge_id(BridgeType, BridgeName), BridgeId = bridge_id(BridgeType, BridgeName),
@ -154,16 +164,38 @@ to_type_atom(Type) ->
end. end.
reset_metrics(ResourceId) -> reset_metrics(ResourceId) ->
emqx_resource:reset_metrics(ResourceId). %% TODO we should not create atoms here
{Type, Name} = parse_bridge_id(ResourceId),
case emqx_bridge_v2:is_bridge_v2_type(Type) of
false ->
emqx_resource:reset_metrics(ResourceId);
true ->
emqx_bridge_v2:reset_metrics(Type, Name)
end.
restart(Type, Name) -> restart(Type, Name) ->
emqx_resource:restart(resource_id(Type, Name)). case emqx_bridge_v2:is_bridge_v2_type(Type) of
false ->
emqx_resource:restart(resource_id(Type, Name));
true ->
emqx_bridge_v2:restart(Type, Name)
end.
stop(Type, Name) -> stop(Type, Name) ->
emqx_resource:stop(resource_id(Type, Name)). case emqx_bridge_v2:is_bridge_v2_type(Type) of
false ->
emqx_resource:stop(resource_id(Type, Name));
true ->
emqx_bridge_v2:stop(Type, Name)
end.
start(Type, Name) -> start(Type, Name) ->
emqx_resource:start(resource_id(Type, Name)). case emqx_bridge_v2:is_bridge_v2_type(Type) of
false ->
emqx_resource:start(resource_id(Type, Name));
true ->
emqx_bridge_v2:start(Type, Name)
end.
create(BridgeId, Conf) -> create(BridgeId, Conf) ->
{BridgeType, BridgeName} = parse_bridge_id(BridgeId), {BridgeType, BridgeName} = parse_bridge_id(BridgeId),
@ -258,6 +290,14 @@ recreate(Type, Name, Conf0, Opts) ->
). ).
create_dry_run(Type, Conf0) -> create_dry_run(Type, Conf0) ->
case emqx_bridge_v2:is_bridge_v2_type(Type) of
false ->
create_dry_run_bridge_v1(Type, Conf0);
true ->
emqx_bridge_v2:bridge_v1_create_dry_run(Type, Conf0)
end.
create_dry_run_bridge_v1(Type, Conf0) ->
TmpName = iolist_to_binary([?TEST_ID_PREFIX, emqx_utils:gen_id(8)]), TmpName = iolist_to_binary([?TEST_ID_PREFIX, emqx_utils:gen_id(8)]),
TmpPath = emqx_utils:safe_filename(TmpName), TmpPath = emqx_utils:safe_filename(TmpName),
%% Already typechecked, no need to catch errors %% Already typechecked, no need to catch errors
@ -297,6 +337,7 @@ remove(Type, Name) ->
%% just for perform_bridge_changes/1 %% just for perform_bridge_changes/1
remove(Type, Name, _Conf, _Opts) -> remove(Type, Name, _Conf, _Opts) ->
%% TODO we need to handle bridge_v2 here
?SLOG(info, #{msg => "remove_bridge", type => Type, name => Name}), ?SLOG(info, #{msg => "remove_bridge", type => Type, name => Name}),
emqx_resource:remove_local(resource_id(Type, Name)). emqx_resource:remove_local(resource_id(Type, Name)).

462
apps/emqx_bridge/src/emqx_bridge_v2.erl
View File

@ -22,6 +22,7 @@
-include_lib("emqx/include/logger.hrl"). -include_lib("emqx/include/logger.hrl").
-include_lib("emqx/include/emqx_hooks.hrl"). -include_lib("emqx/include/emqx_hooks.hrl").
-include_lib("snabbkaffe/include/snabbkaffe.hrl"). -include_lib("snabbkaffe/include/snabbkaffe.hrl").
-include_lib("emqx_resource/include/emqx_resource.hrl").
-export([ -export([
load/0, load/0,
@ -30,8 +31,9 @@
id/2, id/2,
id/3, id/3,
parse_id/1, parse_id/1,
send_message/4,
bridge_v2_type_to_connector_type/1, bridge_v2_type_to_connector_type/1,
bridge_v1_type_to_bridge_v2_type/1,
bridge_v1_id_to_connector_resource_id/1,
is_bridge_v2_id/1, is_bridge_v2_id/1,
extract_connector_id_from_bridge_v2_id/1, extract_connector_id_from_bridge_v2_id/1,
is_bridge_v2_installed_in_connector_state/2, is_bridge_v2_installed_in_connector_state/2,
@ -41,7 +43,11 @@
%% Compatibility API %% Compatibility API
-export([ -export([
lookup_and_transform_to_bridge_v1/2 lookup_and_transform_to_bridge_v1/2,
list_and_transform_to_bridge_v1/0,
check_deps_and_remove_transform_to_bridge_v1/3,
split_bridge_v1_config_and_create/3,
bridge_v1_create_dry_run/2
]). ]).
%% CRUD API %% CRUD API
@ -55,15 +61,29 @@
disable_enable/3, disable_enable/3,
create/3, create/3,
remove/2, remove/2,
health_check/2 check_deps_and_remove/3
]).
%% Operations
-export([
health_check/2,
send_message/4,
start/2,
stop/2,
restart/2
]). ]).
%% Config Update Handler API %% Config Update Handler API
-export([ -export([
post_config_update/5 post_config_update/5,
pre_config_update/3
]). ]).
%% On message publish hook
-export([on_message_publish/1]).
-define(ROOT_KEY, bridges_v2). -define(ROOT_KEY, bridges_v2).
get_channels_for_connector(ConnectorId) -> get_channels_for_connector(ConnectorId) ->
@ -88,41 +108,11 @@ get_channels_for_connector(ConnectorName, BridgeV2Type) ->
]. ].
load() -> load() ->
% Bridge_V2s = emqx:get_config([?ROOT_KEY], #{}), load_message_publish_hook(),
% lists:foreach(
% fun({Type, NamedConf}) ->
% lists:foreach(
% fun({Name, Conf}) ->
% install_bridge_v2(
% Type,
% Name,
% Conf
% )
% end,
% maps:to_list(NamedConf)
% )
% end,
% maps:to_list(Bridge_V2s)
% ),
ok. ok.
unload() -> unload() ->
% Bridge_V2s = emqx:get_config([?ROOT_KEY], #{}), unload_message_publish_hook(),
% lists:foreach(
% fun({Type, NamedConf}) ->
% lists:foreach(
% fun({Name, Conf}) ->
% uninstall_bridge_v2(
% Type,
% Name,
% Conf
% )
% end,
% maps:to_list(NamedConf)
% )
% end,
% maps:to_list(Bridge_V2s)
% ),
ok. ok.
install_bridge_v2( install_bridge_v2(
@ -216,6 +206,50 @@ health_check(BridgeType, BridgeName) ->
Error Error
end. end.
disable_enable(Action, BridgeType, BridgeName) when
Action =:= disable; Action =:= enable
->
emqx_conf:update(
config_key_path() ++ [BridgeType, BridgeName],
{Action, BridgeType, BridgeName},
#{override_to => cluster}
).
restart(Type, Name) ->
stop(Type, Name),
start(Type, Name).
%% TODO: The following functions just restart the bridge_v2 as a temporary solution.
stop(Type, Name) ->
%% Stop means that we should remove the channel from the connector and reset the metrrics
%% The emqx_resource_buffer_worker is not stopped
stop_helper(Type, Name, lookup_raw_conf(Type, Name)).
stop_helper(_Type, _Name, #{enable := false}) ->
ok;
stop_helper(BridgeV2Type, BridgeName, #{connector := ConnectorName}) ->
BridgeV2Id = id(BridgeV2Type, BridgeName, ConnectorName),
ok = emqx_resource:clear_metrics(BridgeV2Id),
ConnectorId = emqx_connector_resource:resource_id(
bridge_v2_type_to_connector_type(BridgeV2Type), ConnectorName
),
emqx_resource_manager:remove_channel(ConnectorId, BridgeV2Id).
start(Type, Name) ->
%% Start means that we should add the channel to the connector (if it is not already there)
start_helper(Type, Name, lookup_raw_conf(Type, Name)).
start_helper(_Type, _Name, #{enable := false}) ->
ok;
start_helper(BridgeV2Type, BridgeName, #{connector := ConnectorName}) ->
BridgeV2Id = id(BridgeV2Type, BridgeName, ConnectorName),
%% Deinstall from connector
ConnectorId = emqx_connector_resource:resource_id(
bridge_v2_type_to_connector_type(BridgeV2Type), ConnectorName
),
emqx_resource_manager:add_channel(ConnectorId, BridgeV2Id, #{connector => ConnectorName}).
% do_send_msg_with_enabled_config(BridgeType, BridgeName, Message, QueryOpts0, Config) -> % do_send_msg_with_enabled_config(BridgeType, BridgeName, Message, QueryOpts0, Config) ->
% BridgeV2Id = emqx_bridge_v2:id(BridgeType, BridgeName), % BridgeV2Id = emqx_bridge_v2:id(BridgeType, BridgeName),
% ConnectorResourceId = emqx_bridge_v2:extract_connector_id_from_bridge_v2_id(BridgeV2Id), % ConnectorResourceId = emqx_bridge_v2:extract_connector_id_from_bridge_v2_id(BridgeV2Id),
@ -285,13 +319,29 @@ id(BridgeType, BridgeName, ConnectorName) ->
<<"bridge_v2:", (bin(BridgeType))/binary, ":", (bin(BridgeName))/binary, ":connector:", <<"bridge_v2:", (bin(BridgeType))/binary, ":", (bin(BridgeName))/binary, ":connector:",
(bin(ConnectorType))/binary, ":", (bin(ConnectorName))/binary>>. (bin(ConnectorType))/binary, ":", (bin(ConnectorName))/binary>>.
%% Creates the external id for the bridge_v2 that is used by the rule actions
%% to refer to the bridge_v2
external_id(BridgeType, BridgeName) ->
Name = bin(BridgeName),
Type = bin(BridgeType),
<<Type/binary, ":", Name/binary>>.
bridge_v2_type_to_connector_type(Bin) when is_binary(Bin) -> bridge_v2_type_to_connector_type(Bin) when is_binary(Bin) ->
bridge_v2_type_to_connector_type(binary_to_existing_atom(Bin)); bridge_v2_type_to_connector_type(binary_to_existing_atom(Bin));
bridge_v2_type_to_connector_type(kafka) -> bridge_v2_type_to_connector_type(kafka) ->
kafka. kafka.
is_bridge_v2_type(kafka) -> true; bridge_v1_type_to_bridge_v2_type(Bin) when is_binary(Bin) ->
is_bridge_v2_type(_) -> false. bridge_v1_type_to_bridge_v2_type(binary_to_existing_atom(Bin));
bridge_v1_type_to_bridge_v2_type(kafka) ->
kafka.
is_bridge_v2_type(Atom) when is_atom(Atom) ->
is_bridge_v2_type(atom_to_binary(Atom, utf8));
is_bridge_v2_type(<<"kafka">>) ->
true;
is_bridge_v2_type(_) ->
false.
is_bridge_v2_id(<<"bridge_v2:", _/binary>>) -> true; is_bridge_v2_id(<<"bridge_v2:", _/binary>>) -> true;
is_bridge_v2_id(_) -> false. is_bridge_v2_id(_) -> false.
@ -311,6 +361,12 @@ bin(Atom) when is_atom(Atom) -> atom_to_binary(Atom, utf8).
%% Basic CRUD Operations %% Basic CRUD Operations
list() -> list() ->
list_with_lookup_fun(fun lookup/2).
list_and_transform_to_bridge_v1() ->
list_with_lookup_fun(fun lookup_and_transform_to_bridge_v1/2).
list_with_lookup_fun(LookupFun) ->
maps:fold( maps:fold(
fun(Type, NameAndConf, Bridges) -> fun(Type, NameAndConf, Bridges) ->
maps:fold( maps:fold(
@ -318,7 +374,7 @@ list() ->
[ [
begin begin
{ok, BridgeInfo} = {ok, BridgeInfo} =
lookup(Type, Name), LookupFun(Type, Name),
BridgeInfo BridgeInfo
end end
| Acc | Acc
@ -336,11 +392,17 @@ lookup(Id) ->
{Type, Name} = parse_id(Id), {Type, Name} = parse_id(Id),
lookup(Type, Name). lookup(Type, Name).
%% TODO should not call this
% to_atom(Bin) when is_binary(Bin) ->
% binary_to_atom(Bin);
% to_atom(Atom) when is_atom(Atom) ->
% Atom.
lookup(Type, Name) -> lookup(Type, Name) ->
case emqx:get_config([?ROOT_KEY, Type, Name], not_found) of case emqx:get_raw_config([?ROOT_KEY, Type, Name], not_found) of
not_found -> not_found ->
{error, bridge_not_found}; {error, bridge_not_found};
#{connector := BridgeConnector} = RawConf -> #{<<"connector">> := BridgeConnector} = RawConf ->
ConnectorId = emqx_connector_resource:resource_id( ConnectorId = emqx_connector_resource:resource_id(
bridge_v2_type_to_connector_type(Type), BridgeConnector bridge_v2_type_to_connector_type(Type), BridgeConnector
), ),
@ -362,7 +424,7 @@ lookup(Type, Name) ->
lookup_and_transform_to_bridge_v1(Type, Name) -> lookup_and_transform_to_bridge_v1(Type, Name) ->
case lookup(Type, Name) of case lookup(Type, Name) of
{ok, #{raw_config := #{connector := ConnectorName}} = BridgeV2} -> {ok, #{raw_config := #{<<"connector">> := ConnectorName}} = BridgeV2} ->
ConnectorType = bridge_v2_type_to_connector_type(Type), ConnectorType = bridge_v2_type_to_connector_type(Type),
case emqx_connector:lookup(ConnectorType, ConnectorName) of case emqx_connector:lookup(ConnectorType, ConnectorName) of
{ok, Connector} -> {ok, Connector} ->
@ -391,7 +453,7 @@ lookup_and_transform_to_bridge_v1_helper(BridgeV2Type, BridgeV2, ConnectorType,
<<"bridges_v2">>, <<"bridges_v2">>,
emqx_bridge_v2_schema emqx_bridge_v2_schema
), ),
BridgeV1Config1 = maps:remove(connector, BridgeV2RawConfig2), BridgeV1Config1 = maps:remove(<<"connector">>, BridgeV2RawConfig2),
BridgeV1Config2 = maps:merge(BridgeV1Config1, ConnectorRawConfig2), BridgeV1Config2 = maps:merge(BridgeV1Config1, ConnectorRawConfig2),
BridgeV1 = maps:put(raw_config, BridgeV1Config2, BridgeV2), BridgeV1 = maps:put(raw_config, BridgeV1Config2, BridgeV2),
{ok, BridgeV1}. {ok, BridgeV1}.
@ -410,15 +472,6 @@ get_metrics(Type, Name) ->
config_key_path() -> config_key_path() ->
[?ROOT_KEY]. [?ROOT_KEY].
disable_enable(Action, BridgeType, BridgeName) when
Action =:= disable; Action =:= enable
->
emqx_conf:update(
config_key_path() ++ [BridgeType, BridgeName],
{Action, BridgeType, BridgeName},
#{override_to => cluster}
).
create(BridgeType, BridgeName, RawConf) -> create(BridgeType, BridgeName, RawConf) ->
?SLOG(debug, #{ ?SLOG(debug, #{
brige_action => create, brige_action => create,
@ -433,6 +486,141 @@ create(BridgeType, BridgeName, RawConf) ->
#{override_to => cluster} #{override_to => cluster}
). ).
split_bridge_v1_config_and_create(BridgeV1Type, BridgeName, RawConf) ->
#{
connector_type := ConnectorType,
connector_name := NewConnectorName,
connector_conf := NewConnectorRawConf,
bridge_v2_type := BridgeType,
bridge_v2_name := BridgeName,
bridge_v2_conf := NewBridgeV2RawConf
} =
split_and_validate_bridge_v1_config(BridgeV1Type, BridgeName, RawConf),
%% TODO should we really create an atom here?
ConnectorNameAtom = binary_to_atom(NewConnectorName),
case emqx_connector:create(ConnectorType, ConnectorNameAtom, NewConnectorRawConf) of
{ok, _} ->
case create(BridgeType, BridgeName, NewBridgeV2RawConf) of
{ok, _} = Result ->
Result;
Error ->
emqx_connector:remove(ConnectorType, ConnectorNameAtom),
Error
end;
Error ->
Error
end.
split_and_validate_bridge_v1_config(BridgeType, BridgeName, RawConf) ->
%% Create fake global config for the transformation and then call
%% emqx_connector_schema:transform_bridges_v1_to_connectors_and_bridges_v2/1
ConnectorType = bridge_v2_type_to_connector_type(BridgeType),
%% Needed so name confligts will ba avoided
CurrentConnectorsConfig = emqx:get_raw_config([connectors], #{}),
FakeGlobalConfig = #{
<<"connectors">> => CurrentConnectorsConfig,
<<"bridges">> => #{
bin(BridgeType) => #{
bin(BridgeName) => RawConf
}
}
},
Output = emqx_connector_schema:transform_bridges_v1_to_connectors_and_bridges_v2(
FakeGlobalConfig
),
NewBridgeV2RawConf =
emqx_utils_maps:deep_get(
[
bin(?ROOT_KEY),
bin(BridgeType),
bin(BridgeName)
],
Output
),
ConnectorsBefore =
maps:keys(
emqx_utils_maps:deep_get(
[
<<"connectors">>,
bin(ConnectorType)
],
FakeGlobalConfig,
#{}
)
),
ConnectorsAfter =
maps:keys(
emqx_utils_maps:deep_get(
[
<<"connectors">>,
bin(ConnectorType)
],
Output
)
),
[NewConnectorName] = ConnectorsAfter -- ConnectorsBefore,
NewConnectorRawConf =
emqx_utils_maps:deep_get(
[
<<"connectors">>,
bin(ConnectorType),
bin(NewConnectorName)
],
Output
),
%% Validate the connector config and the bridge_v2 config
NewFakeGlobalConfig = #{
<<"connectors">> => #{
bin(ConnectorType) => #{
bin(NewConnectorName) => NewConnectorRawConf
}
},
<<"bridges_v2">> => #{
bin(BridgeType) => #{
bin(BridgeName) => NewBridgeV2RawConf
}
}
},
try
hocon_tconf:check_plain(
emqx_schema,
NewFakeGlobalConfig,
#{atom_key => false, required => false}
)
of
_ ->
#{
connector_type => ConnectorType,
connector_name => NewConnectorName,
connector_conf => NewConnectorRawConf,
bridge_v2_type => BridgeType,
bridge_v2_name => BridgeName,
bridge_v2_conf => NewBridgeV2RawConf
}
catch
%% validation errors
throw:Reason1 ->
{error, Reason1}
end.
bridge_v1_create_dry_run(BridgeType, RawConfig0) ->
RawConf = maps:without(<<"name">>, RawConfig0),
TmpName = iolist_to_binary([?TEST_ID_PREFIX, emqx_utils:gen_id(8)]),
#{
connector_type := _ConnectorType,
connector_name := _NewConnectorName,
connector_conf := _NewConnectorRawConf,
bridge_v2_type := _BridgeType,
bridge_v2_name := _BridgeName,
bridge_v2_conf := _NewBridgeV2RawConf
} =
split_and_validate_bridge_v1_config(BridgeType, TmpName, RawConf),
% TODO once we have implemented the dry-run for channels we should use it here
ok.
%% NOTE: This function can cause broken references but it is only called from
%% test cases.
remove(BridgeType, BridgeName) -> remove(BridgeType, BridgeName) ->
?SLOG(debug, #{ ?SLOG(debug, #{
brige_action => remove, brige_action => remove,
@ -445,6 +633,47 @@ remove(BridgeType, BridgeName) ->
#{override_to => cluster} #{override_to => cluster}
). ).
check_deps_and_remove(BridgeType, BridgeName, RemoveDeps) ->
BridgeId = external_id(BridgeType, BridgeName),
%% NOTE: This violates the design: Rule depends on data-bridge but not vice versa.
case emqx_rule_engine:get_rule_ids_by_action(BridgeId) of
[] ->
remove(BridgeType, BridgeName);
RuleIds when RemoveDeps =:= false ->
{error, {rules_deps_on_this_bridge, RuleIds}};
RuleIds when RemoveDeps =:= true ->
lists:foreach(
fun(R) ->
emqx_rule_engine:ensure_action_removed(R, BridgeId)
end,
RuleIds
),
remove(BridgeType, BridgeName)
end.
check_deps_and_remove_transform_to_bridge_v1(BridgeType, BridgeName, RemoveDeps) ->
case check_deps_and_remove(BridgeType, BridgeName, RemoveDeps) of
{error, _} = Error ->
Error;
Result ->
%% TODO: We should call emqx_connector:check_deps_and_remove here
%% to remain as backward compatible as possible.
Result
end.
%% NOTE: We depends on the `emqx_bridge:pre_config_update/3` to restart/stop the
%% underlying resources.
pre_config_update(_, {_Oper, _, _}, undefined) ->
{error, bridge_not_found};
pre_config_update(_, {Oper, _Type, _Name}, OldConfig) ->
%% to save the 'enable' to the config files
{ok, OldConfig#{<<"enable">> => operation_to_enable(Oper)}};
pre_config_update(_Path, Conf, _OldConfig) when is_map(Conf) ->
{ok, Conf}.
operation_to_enable(disable) -> false;
operation_to_enable(enable) -> true.
%% This top level handler will be triggered when the bridges_v2 path is updated %% This top level handler will be triggered when the bridges_v2 path is updated
%% with calls to emqx_conf:update([bridges_v2], BridgesConf, #{}). %% with calls to emqx_conf:update([bridges_v2], BridgesConf, #{}).
%% %%
@ -469,20 +698,32 @@ post_config_update([?ROOT_KEY], _Req, NewConf, OldConf, _AppEnv) ->
}, },
#{action => UpdateFun, data => Updated} #{action => UpdateFun, data => Updated}
]), ]),
ok = unload_message_publish_hook(),
ok = load_message_publish_hook(NewConf),
?tp(bridge_post_config_update_done, #{}), ?tp(bridge_post_config_update_done, #{}),
Result; Result;
post_config_update([?ROOT_KEY, BridgeType, BridgeName], '$remove', _, _OldConf, _AppEnvs) -> post_config_update([?ROOT_KEY, BridgeType, BridgeName], '$remove', _, _OldConf, _AppEnvs) ->
Conf = emqx:get_config([?ROOT_KEY, BridgeType, BridgeName]), Conf = emqx:get_config([?ROOT_KEY, BridgeType, BridgeName]),
ok = uninstall_bridge_v2(BridgeType, BridgeName, Conf), ok = uninstall_bridge_v2(BridgeType, BridgeName, Conf),
Bridges = emqx_utils_maps:deep_remove([BridgeType, BridgeName], emqx:get_config([?ROOT_KEY])),
reload_message_publish_hook(Bridges),
?tp(bridge_post_config_update_done, #{}), ?tp(bridge_post_config_update_done, #{}),
ok; ok;
post_config_update([?ROOT_KEY, BridgeType, BridgeName], _Req, NewConf, undefined, _AppEnvs) -> post_config_update([?ROOT_KEY, BridgeType, BridgeName], _Req, NewConf, undefined, _AppEnvs) ->
ok = install_bridge_v2(BridgeType, BridgeName, NewConf), ok = install_bridge_v2(BridgeType, BridgeName, NewConf),
Bridges = emqx_utils_maps:deep_put(
[BridgeType, BridgeName], emqx:get_config([?ROOT_KEY]), NewConf
),
reload_message_publish_hook(Bridges),
?tp(bridge_post_config_update_done, #{}), ?tp(bridge_post_config_update_done, #{}),
ok; ok;
post_config_update([?ROOT_KEY, BridgeType, BridgeName], _Req, NewConf, OldConf, _AppEnvs) -> post_config_update([?ROOT_KEY, BridgeType, BridgeName], _Req, NewConf, OldConf, _AppEnvs) ->
ok = uninstall_bridge_v2(BridgeType, BridgeName, OldConf), ok = uninstall_bridge_v2(BridgeType, BridgeName, OldConf),
ok = install_bridge_v2(BridgeType, BridgeName, NewConf), ok = install_bridge_v2(BridgeType, BridgeName, NewConf),
Bridges = emqx_utils_maps:deep_put(
[BridgeType, BridgeName], emqx:get_config([?ROOT_KEY]), NewConf
),
reload_message_publish_hook(Bridges),
?tp(bridge_post_config_update_done, #{}), ?tp(bridge_post_config_update_done, #{}),
ok. ok.
@ -564,3 +805,120 @@ unpack_bridge_conf(Type, PackedConf, TopLevelConf) ->
#{TopLevelConf := Bridges} = PackedConf, #{TopLevelConf := Bridges} = PackedConf,
#{<<"foo">> := RawConf} = maps:get(TypeBin, Bridges), #{<<"foo">> := RawConf} = maps:get(TypeBin, Bridges),
RawConf. RawConf.
bridge_v1_id_to_connector_resource_id(BridgeId) ->
case binary:split(BridgeId, <<":">>) of
[Type, Name] ->
BridgeV2Type = bin(bridge_v1_type_to_bridge_v2_type(Type)),
ConnectorName =
case lookup_raw_conf(BridgeV2Type, Name) of
#{connector := Con} ->
Con;
Error ->
throw(Error)
end,
ConnectorType = bin(bridge_v2_type_to_connector_type(BridgeV2Type)),
<<"connector:", ConnectorType/binary, ":", ConnectorName/binary>>
end.
%% The following functions are copied from emqx_bridge.erl
reload_message_publish_hook(Bridges) ->
ok = unload_message_publish_hook(),
ok = load_message_publish_hook(Bridges).
load_message_publish_hook() ->
Bridges = emqx:get_config([?ROOT_KEY], #{}),
load_message_publish_hook(Bridges).
load_message_publish_hook(Bridges) ->
lists:foreach(
fun({Type, Bridge}) ->
lists:foreach(
fun({_Name, BridgeConf}) ->
do_load_message_publish_hook(Type, BridgeConf)
end,
maps:to_list(Bridge)
)
end,
maps:to_list(Bridges)
).
do_load_message_publish_hook(_Type, #{local_topic := LocalTopic}) when is_binary(LocalTopic) ->
emqx_hooks:put('message.publish', {?MODULE, on_message_publish, []}, ?HP_BRIDGE);
do_load_message_publish_hook(_Type, _Conf) ->
ok.
unload_message_publish_hook() ->
ok = emqx_hooks:del('message.publish', {?MODULE, on_message_publish}).
on_message_publish(Message = #message{topic = Topic, flags = Flags}) ->
case maps:get(sys, Flags, false) of
false ->
{Msg, _} = emqx_rule_events:eventmsg_publish(Message),
send_to_matched_egress_bridges(Topic, Msg);
true ->
ok
end,
{ok, Message}.
send_to_matched_egress_bridges(Topic, Msg) ->
MatchedBridgeIds = get_matched_egress_bridges(Topic),
lists:foreach(
fun({Type, Name}) ->
try send_message(Type, Name, Msg, #{}) of
{error, Reason} ->
?SLOG(error, #{
msg => "send_message_to_bridge_failed",
bridge_type => Type,
bridge_name => Name,
error => Reason
});
_ ->
ok
catch
Err:Reason:ST ->
?SLOG(error, #{
msg => "send_message_to_bridge_exception",
bridge_type => Type,
bridge_name => Name,
error => Err,
reason => Reason,
stacktrace => ST
})
end
end,
MatchedBridgeIds
).
get_matched_egress_bridges(Topic) ->
Bridges = emqx:get_config([?ROOT_KEY], #{}),
maps:fold(
fun(BType, Conf, Acc0) ->
maps:fold(
fun(BName, BConf, Acc1) ->
get_matched_bridge_id(BType, BConf, Topic, BName, Acc1)
end,
Acc0,
Conf
)
end,
[],
Bridges
).
get_matched_bridge_id(_BType, #{enable := false}, _Topic, _BName, Acc) ->
Acc;
get_matched_bridge_id(BType, Conf, Topic, BName, Acc) ->
case maps:get(local_topic, Conf, undefined) of
undefined ->
Acc;
Filter ->
do_get_matched_bridge_id(Topic, Filter, BType, BName, Acc)
end.
do_get_matched_bridge_id(Topic, Filter, BType, BName, Acc) ->
case emqx_topic:match(Topic, Filter) of
true -> [{BType, BName} | Acc];
false -> Acc
end.

17
apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl
View File

@ -71,6 +71,19 @@ on_start(<<"connector:", _/binary>> = InstId, Config) ->
}, },
case wolff:ensure_supervised_client(ClientId, Hosts, ClientConfig) of case wolff:ensure_supervised_client(ClientId, Hosts, ClientConfig) of
{ok, _} -> {ok, _} ->
case wolff_client_sup:find_client(ClientId) of
{ok, Pid} ->
case wolff_client:check_connectivity(Pid) of
ok ->
ok;
{error, Error} ->
deallocate_client(ClientId),
throw({failed_to_connect, Error})
end;
{error, Reason} ->
deallocate_client(ClientId),
throw({failed_to_find_created_client, Reason})
end,
?SLOG(info, #{ ?SLOG(info, #{
msg => "kafka_client_started", msg => "kafka_client_started",
instance_id => InstId, instance_id => InstId,
@ -455,13 +468,13 @@ on_kafka_ack(_Partition, buffer_overflow_discarded, _Callback) ->
%% `emqx_resource_manager' will kill the wolff producers and messages might be lost. %% `emqx_resource_manager' will kill the wolff producers and messages might be lost.
on_get_status( on_get_status(
<<"connector:", _/binary>> = _InstId, <<"connector:", _/binary>> = _InstId,
#{client_id := ClientId} = _State #{client_id := ClientId} = State
) -> ) ->
case wolff_client_sup:find_client(ClientId) of case wolff_client_sup:find_client(ClientId) of
{ok, Pid} -> {ok, Pid} ->
case wolff_client:check_connectivity(Pid) of case wolff_client:check_connectivity(Pid) of
ok -> connected; ok -> connected;
{error, Error} -> {connecting, Error} {error, Error} -> {connecting, State, Error}
end; end;
{error, _Reason} -> {error, _Reason} ->
connecting connecting

497
apps/emqx_bridge_kafka/test/emqx_bridge_kafka_impl_producer_SUITE.erl
View File

@ -50,11 +50,18 @@
all() -> all() ->
[ [
{group, on_query}, {group, on_query}
{group, on_query_async} % {group, on_query_async}
]. ].
groups() -> groups() ->
case code:get_object_code(cthr) of
{Module, Code, Filename} ->
{module, Module} = code:load_binary(Module, Filename, Code),
ok;
error ->
error
end,
All = emqx_common_test_helpers:all(?MODULE), All = emqx_common_test_helpers:all(?MODULE),
[{on_query, All}, {on_query_async, All}]. [{on_query, All}, {on_query_async, All}].
@ -86,7 +93,7 @@ init_per_suite(Config) ->
wait_until_kafka_is_up(), wait_until_kafka_is_up(),
%% Wait until bridges API is up %% Wait until bridges API is up
(fun WaitUntilRestApiUp() -> (fun WaitUntilRestApiUp() ->
case show(http_get(["bridges"])) of case http_get(["bridges"]) of
{ok, 200, _Res} -> {ok, 200, _Res} ->
ok; ok;
Val -> Val ->
@ -127,6 +134,7 @@ set_special_configs(_) ->
%% Test case for the query_mode parameter %% Test case for the query_mode parameter
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
%% DONE
t_query_mode(CtConfig) -> t_query_mode(CtConfig) ->
%% We need this because on_query_async is in a different group %% We need this because on_query_async is in a different group
CtConfig1 = [{query_api, none} | CtConfig], CtConfig1 = [{query_api, none} | CtConfig],
@ -154,38 +162,35 @@ t_query_mode(CtConfig) ->
%% Test cases for all combinations of SSL, no SSL and authentication types %% Test cases for all combinations of SSL, no SSL and authentication types
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
%% OK
t_publish_no_auth(CtConfig) -> t_publish_no_auth(CtConfig) ->
publish_with_and_without_ssl(CtConfig, "none"). publish_with_and_without_ssl(CtConfig, "none").
%% OK
t_publish_no_auth_key_dispatch(CtConfig) -> t_publish_no_auth_key_dispatch(CtConfig) ->
publish_with_and_without_ssl(CtConfig, "none", #{"partition_strategy" => "key_dispatch"}). publish_with_and_without_ssl(CtConfig, "none", #{"partition_strategy" => "key_dispatch"}).
t_publish_sasl_plain(CtConfig) -> % t_publish_sasl_plain(CtConfig) ->
publish_with_and_without_ssl(CtConfig, valid_sasl_plain_settings()). % publish_with_and_without_ssl(CtConfig, valid_sasl_plain_settings()).
t_publish_sasl_scram256(CtConfig) -> % t_publish_sasl_scram256(CtConfig) ->
publish_with_and_without_ssl(CtConfig, valid_sasl_scram256_settings()). % publish_with_and_without_ssl(CtConfig, valid_sasl_scram256_settings()).
t_publish_sasl_scram512(CtConfig) -> % t_publish_sasl_scram512(CtConfig) ->
publish_with_and_without_ssl(CtConfig, valid_sasl_scram512_settings()). % publish_with_and_without_ssl(CtConfig, valid_sasl_scram512_settings()).
t_publish_sasl_kerberos(CtConfig) -> % t_publish_sasl_kerberos(CtConfig) ->
publish_with_and_without_ssl(CtConfig, valid_sasl_kerberos_settings()). % publish_with_and_without_ssl(CtConfig, valid_sasl_kerberos_settings()).
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
%% Test cases for REST api %% Test cases for REST api
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
show(X) -> % t_kafka_bridge_rest_api_plain_text(_CtConfig) ->
% erlang:display('______________ SHOW ______________:'), % kafka_bridge_rest_api_all_auth_methods(false).
% erlang:display(X),
X.
t_kafka_bridge_rest_api_plain_text(_CtConfig) -> % t_kafka_bridge_rest_api_ssl(_CtConfig) ->
kafka_bridge_rest_api_all_auth_methods(false). % kafka_bridge_rest_api_all_auth_methods(true).
t_kafka_bridge_rest_api_ssl(_CtConfig) ->
kafka_bridge_rest_api_all_auth_methods(true).
kafka_bridge_rest_api_all_auth_methods(UseSSL) -> kafka_bridge_rest_api_all_auth_methods(UseSSL) ->
NormalHostsString = NormalHostsString =
@ -280,8 +285,8 @@ kafka_bridge_rest_api_helper(Config) ->
BridgesPartsOpStop = OpUrlFun("stop"), BridgesPartsOpStop = OpUrlFun("stop"),
%% List bridges %% List bridges
MyKafkaBridgeExists = fun() -> MyKafkaBridgeExists = fun() ->
{ok, _Code, BridgesData} = show(http_get(BridgesParts)), {ok, _Code, BridgesData} = http_get(BridgesParts),
Bridges = show(json(BridgesData)), Bridges = json(BridgesData),
lists:any( lists:any(
fun fun
(#{<<"name">> := <<"my_kafka_bridge">>}) -> true; (#{<<"name">> := <<"my_kafka_bridge">>}) -> true;
@ -294,7 +299,7 @@ kafka_bridge_rest_api_helper(Config) ->
case MyKafkaBridgeExists() of case MyKafkaBridgeExists() of
true -> true ->
%% Delete the bridge my_kafka_bridge %% Delete the bridge my_kafka_bridge
{ok, 204, <<>>} = show(http_delete(BridgesPartsIdDeleteAlsoActions)); {ok, 204, <<>>} = http_delete(BridgesPartsIdDeleteAlsoActions);
false -> false ->
ok ok
end, end,
@ -322,7 +327,7 @@ kafka_bridge_rest_api_helper(Config) ->
true -> CreateBodyTmp#{<<"ssl">> => maps:get(<<"ssl">>, Config)}; true -> CreateBodyTmp#{<<"ssl">> => maps:get(<<"ssl">>, Config)};
false -> CreateBodyTmp false -> CreateBodyTmp
end, end,
{ok, 201, _Data} = show(http_post(BridgesParts, show(CreateBody))), {ok, 201, _Data} = http_post(BridgesParts, CreateBody),
%% Check that the new bridge is in the list of bridges %% Check that the new bridge is in the list of bridges
true = MyKafkaBridgeExists(), true = MyKafkaBridgeExists(),
%% Probe should work %% Probe should work
@ -366,7 +371,7 @@ kafka_bridge_rest_api_helper(Config) ->
timer:sleep(100), timer:sleep(100),
%% Check that Kafka got message %% Check that Kafka got message
BrodOut = brod:fetch(kafka_hosts(), KafkaTopic, 0, Offset), BrodOut = brod:fetch(kafka_hosts(), KafkaTopic, 0, Offset),
{ok, {_, [KafkaMsg]}} = show(BrodOut), {ok, {_, [KafkaMsg]}} = BrodOut,
Body = KafkaMsg#kafka_message.value, Body = KafkaMsg#kafka_message.value,
%% Check crucial counters and gauges %% Check crucial counters and gauges
?assertEqual(1, emqx_resource_metrics:matched_get(ResourceId)), ?assertEqual(1, emqx_resource_metrics:matched_get(ResourceId)),
@ -385,15 +390,15 @@ kafka_bridge_rest_api_helper(Config) ->
?assertEqual(0, emqx_resource_metrics:retried_failed_get(ResourceId)), ?assertEqual(0, emqx_resource_metrics:retried_failed_get(ResourceId)),
?assertEqual(0, emqx_resource_metrics:retried_success_get(ResourceId)), ?assertEqual(0, emqx_resource_metrics:retried_success_get(ResourceId)),
%% Perform operations %% Perform operations
{ok, 204, _} = show(http_put(show(BridgesPartsOpDisable), #{})), {ok, 204, _} = http_put(BridgesPartsOpDisable, #{}),
{ok, 204, _} = show(http_put(show(BridgesPartsOpDisable), #{})), {ok, 204, _} = http_put(BridgesPartsOpDisable, #{}),
{ok, 204, _} = show(http_put(show(BridgesPartsOpEnable), #{})), {ok, 204, _} = http_put(BridgesPartsOpEnable, #{}),
{ok, 204, _} = show(http_put(show(BridgesPartsOpEnable), #{})), {ok, 204, _} = http_put(BridgesPartsOpEnable, #{}),
{ok, 204, _} = show(http_post(show(BridgesPartsOpStop), #{})), {ok, 204, _} = http_post(BridgesPartsOpStop, #{}),
{ok, 204, _} = show(http_post(show(BridgesPartsOpStop), #{})), {ok, 204, _} = http_post(BridgesPartsOpStop, #{}),
{ok, 204, _} = show(http_post(show(BridgesPartsOpRestart), #{})), {ok, 204, _} = http_post(BridgesPartsOpRestart, #{}),
%% Cleanup %% Cleanup
{ok, 204, _} = show(http_delete(BridgesPartsIdDeleteAlsoActions)), {ok, 204, _} = http_delete(BridgesPartsIdDeleteAlsoActions),
false = MyKafkaBridgeExists(), false = MyKafkaBridgeExists(),
delete_all_bridges(), delete_all_bridges(),
ok. ok.
@ -407,28 +412,29 @@ kafka_bridge_rest_api_helper(Config) ->
%% exists and it will. This is specially bad if the %% exists and it will. This is specially bad if the
%% original crash was due to misconfiguration and we are %% original crash was due to misconfiguration and we are
%% trying to fix it... %% trying to fix it...
%% DONE
t_failed_creation_then_fix(Config) -> t_failed_creation_then_fix(Config) ->
HostsString = kafka_hosts_string_sasl(), %% TODO change this back to SASL_PLAINTEXT when we have figured out why that is not working
ValidAuthSettings = valid_sasl_plain_settings(), HostsString = kafka_hosts_string(),
WrongAuthSettings = ValidAuthSettings#{"password" := "wrong"}, %% valid_sasl_plain_settings()
ValidAuthSettings = "none",
WrongAuthSettings = (valid_sasl_plain_settings())#{"password" := "wrong"},
Hash = erlang:phash2([HostsString, ?FUNCTION_NAME]), Hash = erlang:phash2([HostsString, ?FUNCTION_NAME]),
Type = ?BRIDGE_TYPE, Type = ?BRIDGE_TYPE,
Name = "kafka_bridge_name_" ++ erlang:integer_to_list(Hash), Name = "kafka_bridge_name_" ++ erlang:integer_to_list(Hash),
ResourceId = emqx_bridge_resource:resource_id(Type, Name),
BridgeId = emqx_bridge_resource:bridge_id(Type, Name),
KafkaTopic = test_topic_one_partition(), KafkaTopic = test_topic_one_partition(),
WrongConf = config(#{ WrongConf = config(#{
"authentication" => WrongAuthSettings, "authentication" => WrongAuthSettings,
"kafka_hosts_string" => HostsString, "kafka_hosts_string" => HostsString,
"kafka_topic" => KafkaTopic, "kafka_topic" => KafkaTopic,
"instance_id" => ResourceId, "bridge_name" => Name,
"ssl" => #{} "ssl" => #{}
}), }),
ValidConf = config(#{ ValidConf = config(#{
"authentication" => ValidAuthSettings, "authentication" => ValidAuthSettings,
"kafka_hosts_string" => HostsString, "kafka_hosts_string" => HostsString,
"kafka_topic" => KafkaTopic, "kafka_topic" => KafkaTopic,
"instance_id" => ResourceId, "bridge_name" => Name,
"producer" => #{ "producer" => #{
"kafka" => #{ "kafka" => #{
"buffer" => #{ "buffer" => #{
@ -439,21 +445,17 @@ t_failed_creation_then_fix(Config) ->
"ssl" => #{} "ssl" => #{}
}), }),
%% creates, but fails to start producers %% creates, but fails to start producers
{ok, #{config := WrongConfigAtom1}} = emqx_bridge:create( {ok, #{config := _WrongConfigAtom1}} = emqx_bridge:create(
Type, erlang:list_to_atom(Name), WrongConf list_to_atom(Type), list_to_atom(Name), WrongConf
), ),
WrongConfigAtom = WrongConfigAtom1#{bridge_name => Name, bridge_type => ?BRIDGE_TYPE_BIN},
?assertThrow(Reason when is_list(Reason), ?PRODUCER:on_start(ResourceId, WrongConfigAtom)),
%% before throwing, it should cleanup the client process. we %% before throwing, it should cleanup the client process. we
%% retry because the supervisor might need some time to really %% retry because the supervisor might need some time to really
%% remove it from its tree. %% remove it from its tree.
?retry(50, 10, ?assertEqual([], supervisor:which_children(wolff_client_sup))), ?retry(50, 10, ?assertEqual([], supervisor:which_children(wolff_client_sup))),
%% must succeed with correct config %% must succeed with correct config
{ok, #{config := ValidConfigAtom1}} = emqx_bridge:create( {ok, #{config := _ValidConfigAtom1}} = emqx_bridge:create(
Type, erlang:list_to_atom(Name), ValidConf list_to_atom(Type), list_to_atom(Name), ValidConf
), ),
ValidConfigAtom = ValidConfigAtom1#{bridge_name => Name, bridge_type => ?BRIDGE_TYPE_BIN},
{ok, State} = ?PRODUCER:on_start(ResourceId, ValidConfigAtom),
Time = erlang:unique_integer(), Time = erlang:unique_integer(),
BinTime = integer_to_binary(Time), BinTime = integer_to_binary(Time),
Msg = #{ Msg = #{
@ -463,107 +465,112 @@ t_failed_creation_then_fix(Config) ->
}, },
{ok, Offset} = resolve_kafka_offset(kafka_hosts(), KafkaTopic, 0), {ok, Offset} = resolve_kafka_offset(kafka_hosts(), KafkaTopic, 0),
ct:pal("base offset before testing ~p", [Offset]), ct:pal("base offset before testing ~p", [Offset]),
ok = send(Config, ResourceId, Msg, State), BridgeV2Id = emqx_bridge_v2:id(bin(Type), bin(Name)),
ResourceId = emqx_bridge_v2:extract_connector_id_from_bridge_v2_id(BridgeV2Id),
{ok, _Group, #{state := State}} = emqx_resource:get_instance(ResourceId),
ok = send(Config, ResourceId, Msg, State, BridgeV2Id),
{ok, {_, [KafkaMsg]}} = brod:fetch(kafka_hosts(), KafkaTopic, 0, Offset), {ok, {_, [KafkaMsg]}} = brod:fetch(kafka_hosts(), KafkaTopic, 0, Offset),
?assertMatch(#kafka_message{key = BinTime}, KafkaMsg), ?assertMatch(#kafka_message{key = BinTime}, KafkaMsg),
%% TODO: refactor those into init/end per testcase % %% TODO: refactor those into init/end per testcase
ok = ?PRODUCER:on_stop(ResourceId, State), ok = ?PRODUCER:on_stop(ResourceId, State),
?assertEqual([], supervisor:which_children(wolff_client_sup)), ?assertEqual([], supervisor:which_children(wolff_client_sup)),
?assertEqual([], supervisor:which_children(wolff_producers_sup)), ?assertEqual([], supervisor:which_children(wolff_producers_sup)),
ok = emqx_bridge_resource:remove(BridgeId), {ok, _} = emqx_bridge:remove(list_to_atom(Type), list_to_atom(Name)),
delete_all_bridges(), delete_all_bridges(),
ok. ok.
t_custom_timestamp(_Config) -> % t_custom_timestamp(_Config) ->
HostsString = kafka_hosts_string_sasl(), % HostsString = kafka_hosts_string_sasl(),
AuthSettings = valid_sasl_plain_settings(), % AuthSettings = valid_sasl_plain_settings(),
Hash = erlang:phash2([HostsString, ?FUNCTION_NAME]), % Hash = erlang:phash2([HostsString, ?FUNCTION_NAME]),
Type = ?BRIDGE_TYPE, % Type = ?BRIDGE_TYPE,
Name = "kafka_bridge_name_" ++ erlang:integer_to_list(Hash), % Name = "kafka_bridge_name_" ++ erlang:integer_to_list(Hash),
ResourceId = emqx_bridge_resource:resource_id(Type, Name), % ResourceId = emqx_bridge_resource:resource_id(Type, Name),
KafkaTopic = test_topic_one_partition(), % KafkaTopic = test_topic_one_partition(),
MQTTTopic = <<"t/local/kafka">>, % MQTTTopic = <<"t/local/kafka">>,
emqx:subscribe(MQTTTopic), % emqx:subscribe(MQTTTopic),
Conf0 = config(#{ % Conf0 = config(#{
"authentication" => AuthSettings, % "authentication" => AuthSettings,
"kafka_hosts_string" => HostsString, % "kafka_hosts_string" => HostsString,
"local_topic" => MQTTTopic, % "local_topic" => MQTTTopic,
"kafka_topic" => KafkaTopic, % "kafka_topic" => KafkaTopic,
"instance_id" => ResourceId, % "instance_id" => ResourceId,
"ssl" => #{} % "ssl" => #{}
}), % }),
Conf = emqx_utils_maps:deep_put( % Conf = emqx_utils_maps:deep_put(
[<<"kafka">>, <<"message">>, <<"timestamp">>], % [<<"kafka">>, <<"message">>, <<"timestamp">>],
Conf0, % Conf0,
<<"123">> % <<"123">>
), % ),
{ok, _} = emqx_bridge:create(Type, erlang:list_to_atom(Name), Conf), % {ok, _} = emqx_bridge:create(Type, erlang:list_to_atom(Name), Conf),
{ok, Offset} = resolve_kafka_offset(kafka_hosts(), KafkaTopic, 0), % {ok, Offset} = resolve_kafka_offset(kafka_hosts(), KafkaTopic, 0),
ct:pal("base offset before testing ~p", [Offset]), % ct:pal("base offset before testing ~p", [Offset]),
Time = erlang:unique_integer(), % Time = erlang:unique_integer(),
BinTime = integer_to_binary(Time), % BinTime = integer_to_binary(Time),
Msg = #{ % Msg = #{
clientid => BinTime, % clientid => BinTime,
payload => <<"payload">>, % payload => <<"payload">>,
timestamp => Time % timestamp => Time
}, % },
emqx:publish(emqx_message:make(MQTTTopic, emqx_utils_json:encode(Msg))), % emqx:publish(emqx_message:make(MQTTTopic, emqx_utils_json:encode(Msg))),
{ok, {_, [KafkaMsg]}} = % {ok, {_, [KafkaMsg]}} =
?retry( % ?retry(
_Interval = 500, % _Interval = 500,
_NAttempts = 20, % _NAttempts = 20,
{ok, {_, [_]}} = brod:fetch(kafka_hosts(), KafkaTopic, _Partition = 0, Offset) % {ok, {_, [_]}} = brod:fetch(kafka_hosts(), KafkaTopic, _Partition = 0, Offset)
), % ),
?assertMatch(#kafka_message{ts = 123, ts_type = create}, KafkaMsg), % ?assertMatch(#kafka_message{ts = 123, ts_type = create}, KafkaMsg),
delete_all_bridges(), % delete_all_bridges(),
ok. % ok.
t_nonexistent_topic(_Config) -> % t_nonexistent_topic(_Config) ->
HostsString = kafka_hosts_string_sasl(), % HostsString = kafka_hosts_string_sasl(),
AuthSettings = valid_sasl_plain_settings(), % AuthSettings = valid_sasl_plain_settings(),
Hash = erlang:phash2([HostsString, ?FUNCTION_NAME]), % Hash = erlang:phash2([HostsString, ?FUNCTION_NAME]),
Type = ?BRIDGE_TYPE, % Type = ?BRIDGE_TYPE,
Name = "kafka_bridge_name_" ++ erlang:integer_to_list(Hash), % Name = "kafka_bridge_name_" ++ erlang:integer_to_list(Hash),
ResourceId = emqx_bridge_resource:resource_id(Type, Name), % ResourceId = emqx_bridge_resource:resource_id(Type, Name),
BridgeId = emqx_bridge_resource:bridge_id(Type, Name), % BridgeId = emqx_bridge_resource:bridge_id(Type, Name),
KafkaTopic = "undefined-test-topic", % KafkaTopic = "undefined-test-topic",
Conf = config(#{ % Conf = config(#{
"authentication" => AuthSettings, % "authentication" => AuthSettings,
"kafka_hosts_string" => HostsString, % "kafka_hosts_string" => HostsString,
"kafka_topic" => KafkaTopic, % "kafka_topic" => KafkaTopic,
"instance_id" => ResourceId, % "instance_id" => ResourceId,
"producer" => #{ % "producer" => #{
"kafka" => #{ % "kafka" => #{
"buffer" => #{ % "buffer" => #{
"memory_overload_protection" => false % "memory_overload_protection" => false
} % }
} % }
}, % },
"ssl" => #{} % "ssl" => #{}
}), % }),
{ok, #{config := ValidConfigAtom1}} = emqx_bridge:create( % {ok, #{config := ValidConfigAtom1}} = emqx_bridge:create(
Type, erlang:list_to_atom(Name), Conf % Type, erlang:list_to_atom(Name), Conf
), % ),
ValidConfigAtom = ValidConfigAtom1#{bridge_name => Name, bridge_type => ?BRIDGE_TYPE_BIN}, % ValidConfigAtom = ValidConfigAtom1#{bridge_name => Name},
?assertThrow(_, ?PRODUCER:on_start(ResourceId, ValidConfigAtom)), % ?assertThrow(_, ?PRODUCER:on_start(ResourceId, ValidConfigAtom)),
ok = emqx_bridge_resource:remove(BridgeId), % ok = emqx_bridge_resource:remove(BridgeId),
delete_all_bridges(), % delete_all_bridges(),
ok. % ok.
%% DONE
t_send_message_with_headers(Config) -> t_send_message_with_headers(Config) ->
HostsString = kafka_hosts_string_sasl(), %% TODO Change this back to SASL plain once we figure out why it is not working
AuthSettings = valid_sasl_plain_settings(), HostsString = kafka_hosts_string(),
AuthSettings = "none",
Hash = erlang:phash2([HostsString, ?FUNCTION_NAME]), Hash = erlang:phash2([HostsString, ?FUNCTION_NAME]),
Type = ?BRIDGE_TYPE, Type = ?BRIDGE_TYPE,
Name = "kafka_bridge_name_" ++ erlang:integer_to_list(Hash), Name = "kafka_bridge_name_" ++ erlang:integer_to_list(Hash),
ResourceId = emqx_bridge_resource:resource_id(Type, Name), %ResourceId = emqx_bridge_resource:resource_id(Type, Name),
BridgeId = emqx_bridge_resource:bridge_id(Type, Name), %BridgeId = emqx_bridge_resource:bridge_id(Type, Name),
KafkaTopic = test_topic_one_partition(), KafkaTopic = test_topic_one_partition(),
Conf = config_with_headers(#{ Conf = config_with_headers(#{
"authentication" => AuthSettings, "authentication" => AuthSettings,
"kafka_hosts_string" => HostsString, "kafka_hosts_string" => HostsString,
"kafka_topic" => KafkaTopic, "kafka_topic" => KafkaTopic,
"instance_id" => ResourceId, "bridge_name" => Name,
"kafka_headers" => <<"${payload.header}">>, "kafka_headers" => <<"${payload.header}">>,
"kafka_ext_headers" => emqx_utils_json:encode( "kafka_ext_headers" => emqx_utils_json:encode(
[ [
@ -586,11 +593,13 @@ t_send_message_with_headers(Config) ->
}, },
"ssl" => #{} "ssl" => #{}
}), }),
{ok, #{config := ConfigAtom1}} = emqx_bridge:create( {ok, _} = emqx_bridge:create(
Type, erlang:list_to_atom(Name), Conf list_to_atom(Type), list_to_atom(Name), Conf
), ),
ConfigAtom = ConfigAtom1#{bridge_name => Name, bridge_type => ?BRIDGE_TYPE_BIN}, % ConfigAtom = ConfigAtom1#{bridge_name => Name},
{ok, State} = ?PRODUCER:on_start(ResourceId, ConfigAtom), ResourceId = emqx_bridge_resource:resource_id(bin(Type), bin(Name)),
BridgeV2Id = emqx_bridge_v2:id(bin(Type), bin(Name)),
{ok, _Group, #{state := State}} = emqx_resource:get_instance(ResourceId),
Time1 = erlang:unique_integer(), Time1 = erlang:unique_integer(),
BinTime1 = integer_to_binary(Time1), BinTime1 = integer_to_binary(Time1),
Payload1 = emqx_utils_json:encode( Payload1 = emqx_utils_json:encode(
@ -637,8 +646,8 @@ t_send_message_with_headers(Config) ->
end, end,
?check_trace( ?check_trace(
begin begin
ok = send(Config, ResourceId, Msg1, State), ok = send(Config, ResourceId, Msg1, State, BridgeV2Id),
ok = send(Config, ResourceId, Msg2, State) ok = send(Config, ResourceId, Msg2, State, BridgeV2Id)
end, end,
fun(Trace) -> fun(Trace) ->
?assertMatch( ?assertMatch(
@ -707,17 +716,17 @@ t_send_message_with_headers(Config) ->
ok = ?PRODUCER:on_stop(ResourceId, State), ok = ?PRODUCER:on_stop(ResourceId, State),
?assertEqual([], supervisor:which_children(wolff_client_sup)), ?assertEqual([], supervisor:which_children(wolff_client_sup)),
?assertEqual([], supervisor:which_children(wolff_producers_sup)), ?assertEqual([], supervisor:which_children(wolff_producers_sup)),
ok = emqx_bridge_resource:remove(BridgeId), {ok, _} = emqx_bridge:remove(list_to_atom(Name), list_to_atom(Type)),
delete_all_bridges(), delete_all_bridges(),
ok. ok.
%% DONE
t_wrong_headers(_Config) -> t_wrong_headers(_Config) ->
HostsString = kafka_hosts_string_sasl(), HostsString = kafka_hosts_string_sasl(),
AuthSettings = valid_sasl_plain_settings(), AuthSettings = valid_sasl_plain_settings(),
Hash = erlang:phash2([HostsString, ?FUNCTION_NAME]), Hash = erlang:phash2([HostsString, ?FUNCTION_NAME]),
Type = ?BRIDGE_TYPE, % Type = ?BRIDGE_TYPE,
Name = "kafka_bridge_name_" ++ erlang:integer_to_list(Hash), Name = "kafka_bridge_name_" ++ erlang:integer_to_list(Hash),
ResourceId = emqx_bridge_resource:resource_id(Type, Name),
KafkaTopic = test_topic_one_partition(), KafkaTopic = test_topic_one_partition(),
?assertThrow( ?assertThrow(
{ {
@ -733,7 +742,7 @@ t_wrong_headers(_Config) ->
"authentication" => AuthSettings, "authentication" => AuthSettings,
"kafka_hosts_string" => HostsString, "kafka_hosts_string" => HostsString,
"kafka_topic" => KafkaTopic, "kafka_topic" => KafkaTopic,
"instance_id" => ResourceId, "bridge_name" => Name,
"kafka_headers" => <<"wrong_header">>, "kafka_headers" => <<"wrong_header">>,
"kafka_ext_headers" => <<"[]">>, "kafka_ext_headers" => <<"[]">>,
"producer" => #{ "producer" => #{
@ -762,7 +771,7 @@ t_wrong_headers(_Config) ->
"authentication" => AuthSettings, "authentication" => AuthSettings,
"kafka_hosts_string" => HostsString, "kafka_hosts_string" => HostsString,
"kafka_topic" => KafkaTopic, "kafka_topic" => KafkaTopic,
"instance_id" => ResourceId, "bridge_name" => Name,
"kafka_headers" => <<"${pub_props}">>, "kafka_headers" => <<"${pub_props}">>,
"kafka_ext_headers" => emqx_utils_json:encode( "kafka_ext_headers" => emqx_utils_json:encode(
[ [
@ -784,83 +793,97 @@ t_wrong_headers(_Config) ->
), ),
ok. ok.
t_wrong_headers_from_message(Config) -> % t_wrong_headers_from_message(Config) ->
HostsString = kafka_hosts_string_sasl(), % HostsString = kafka_hosts_string_sasl(),
AuthSettings = valid_sasl_plain_settings(), % AuthSettings = valid_sasl_plain_settings(),
Hash = erlang:phash2([HostsString, ?FUNCTION_NAME]), % Hash = erlang:phash2([HostsString, ?FUNCTION_NAME]),
Type = ?BRIDGE_TYPE, % Type = ?BRIDGE_TYPE,
Name = "kafka_bridge_name_" ++ erlang:integer_to_list(Hash), % Name = "kafka_bridge_name_" ++ erlang:integer_to_list(Hash),
ResourceId = emqx_bridge_resource:resource_id(Type, Name), % ResourceId = emqx_bridge_resource:resource_id(Type, Name),
BridgeId = emqx_bridge_resource:bridge_id(Type, Name), % BridgeId = emqx_bridge_resource:bridge_id(Type, Name),
KafkaTopic = test_topic_one_partition(), % KafkaTopic = test_topic_one_partition(),
Conf = config_with_headers(#{ % Conf = config_with_headers(#{
"authentication" => AuthSettings, % "authentication" => AuthSettings,
"kafka_hosts_string" => HostsString, % "kafka_hosts_string" => HostsString,
"kafka_topic" => KafkaTopic, % "kafka_topic" => KafkaTopic,
"instance_id" => ResourceId, % "instance_id" => ResourceId,
"kafka_headers" => <<"${payload}">>, % "kafka_headers" => <<"${payload}">>,
"producer" => #{ % "producer" => #{
"kafka" => #{ % "kafka" => #{
"buffer" => #{ % "buffer" => #{
"memory_overload_protection" => false % "memory_overload_protection" => false
} % }
} % }
}, % },
"ssl" => #{} % "ssl" => #{}
}), % }),
{ok, #{config := ConfigAtom1}} = emqx_bridge:create( % {ok, #{config := ConfigAtom1}} = emqx_bridge:create(
Type, erlang:list_to_atom(Name), Conf % Type, erlang:list_to_atom(Name), Conf
), % ),
ConfigAtom = ConfigAtom1#{bridge_name => Name, bridge_type => ?BRIDGE_TYPE_BIN}, % ConfigAtom = ConfigAtom1#{bridge_name => Name},
{ok, State} = ?PRODUCER:on_start(ResourceId, ConfigAtom), % {ok, State} = ?PRODUCER:on_start(ResourceId, ConfigAtom),
Time1 = erlang:unique_integer(), % Time1 = erlang:unique_integer(),
Payload1 = <<"wrong_header">>, % Payload1 = <<"wrong_header">>,
Msg1 = #{ % Msg1 = #{
clientid => integer_to_binary(Time1), % clientid => integer_to_binary(Time1),
payload => Payload1, % payload => Payload1,
timestamp => Time1 % timestamp => Time1
}, % },
?assertError( % ?assertError(
{badmatch, {error, {unrecoverable_error, {bad_kafka_headers, Payload1}}}}, % {badmatch, {error, {unrecoverable_error, {bad_kafka_headers, Payload1}}}},
send(Config, ResourceId, Msg1, State) % send(Config, ResourceId, Msg1, State)
), % ),
Time2 = erlang:unique_integer(), % Time2 = erlang:unique_integer(),
Payload2 = <<"[{\"foo\":\"bar\"}, {\"foo2\":\"bar2\"}]">>, % Payload2 = <<"[{\"foo\":\"bar\"}, {\"foo2\":\"bar2\"}]">>,
Msg2 = #{ % Msg2 = #{
clientid => integer_to_binary(Time2), % clientid => integer_to_binary(Time2),
payload => Payload2, % payload => Payload2,
timestamp => Time2 % timestamp => Time2
}, % },
?assertError( % ?assertError(
{badmatch, {error, {unrecoverable_error, {bad_kafka_header, #{<<"foo">> := <<"bar">>}}}}}, % {badmatch, {error, {unrecoverable_error, {bad_kafka_header, [{<<"foo">>, <<"bar">>}]}}}},
send(Config, ResourceId, Msg2, State) % send(Config, ResourceId, Msg2, State)
), % ),
Time3 = erlang:unique_integer(), % Time3 = erlang:unique_integer(),
Payload3 = <<"[{\"key\":\"foo\"}, {\"value\":\"bar\"}]">>, % Payload3 = <<"[{\"key\":\"foo\"}, {\"value\":\"bar\"}]">>,
Msg3 = #{ % Msg3 = #{
clientid => integer_to_binary(Time3), % clientid => integer_to_binary(Time3),
payload => Payload3, % payload => Payload3,
timestamp => Time3 % timestamp => Time3
}, % },
?assertError( % ?assertError(
{badmatch, {error, {unrecoverable_error, {bad_kafka_header, #{<<"key">> := <<"foo">>}}}}}, % {badmatch, {error, {unrecoverable_error, {bad_kafka_header, [{<<"key">>, <<"foo">>}]}}}},
send(Config, ResourceId, Msg3, State) % send(Config, ResourceId, Msg3, State)
), % ),
%% TODO: refactor those into init/end per testcase % Time4 = erlang:unique_integer(),
ok = ?PRODUCER:on_stop(ResourceId, State), % Payload4 = <<"[{\"key\":\"foo\", \"value\":\"bar\"}]">>,
?assertEqual([], supervisor:which_children(wolff_client_sup)), % Msg4 = #{
?assertEqual([], supervisor:which_children(wolff_producers_sup)), % clientid => integer_to_binary(Time4),
ok = emqx_bridge_resource:remove(BridgeId), % payload => Payload4,
delete_all_bridges(), % timestamp => Time4
ok. % },
% ?assertError(
% {badmatch,
% {error,
% {unrecoverable_error,
% {bad_kafka_header, [{<<"key">>, <<"foo">>}, {<<"value">>, <<"bar">>}]}}}},
% send(Config, ResourceId, Msg4, State)
% ),
% %% TODO: refactor those into init/end per testcase
% ok = ?PRODUCER:on_stop(ResourceId, State),
% ?assertEqual([], supervisor:which_children(wolff_client_sup)),
% ?assertEqual([], supervisor:which_children(wolff_producers_sup)),
% ok = emqx_bridge_resource:remove(BridgeId),
% delete_all_bridges(),
% ok.
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
%% Helper functions %% Helper functions
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
send(Config, ResourceId, Msg, State) when is_list(Config) -> send(Config, ResourceId, Msg, State, BridgeV2Id) when is_list(Config) ->
Ref = make_ref(), Ref = make_ref(),
ok = do_send(Ref, Config, ResourceId, Msg, State), ok = do_send(Ref, Config, ResourceId, Msg, State, BridgeV2Id),
receive receive
{ack, Ref} -> {ack, Ref} ->
ok ok
@ -868,7 +891,7 @@ send(Config, ResourceId, Msg, State) when is_list(Config) ->
error(timeout) error(timeout)
end. end.
do_send(Ref, Config, ResourceId, Msg, State) when is_list(Config) -> do_send(Ref, Config, ResourceId, Msg, State, BridgeV2Id) when is_list(Config) ->
Caller = self(), Caller = self(),
F = fun(ok) -> F = fun(ok) ->
Caller ! {ack, Ref}, Caller ! {ack, Ref},
@ -876,10 +899,10 @@ do_send(Ref, Config, ResourceId, Msg, State) when is_list(Config) ->
end, end,
case proplists:get_value(query_api, Config) of case proplists:get_value(query_api, Config) of
on_query -> on_query ->
ok = ?PRODUCER:on_query(ResourceId, {send_message, Msg}, State), ok = ?PRODUCER:on_query(ResourceId, {BridgeV2Id, Msg}, State),
F(ok); F(ok);
on_query_async -> on_query_async ->
{ok, _} = ?PRODUCER:on_query_async(ResourceId, {send_message, Msg}, {F, []}, State), {ok, _} = ?PRODUCER:on_query_async(ResourceId, {BridgeV2Id, Msg}, {F, []}, State),
ok ok
end. end.
@ -905,14 +928,14 @@ publish_with_and_without_ssl(CtConfig, AuthSettings, Config) ->
}, },
Config Config
), ),
publish_helper( % publish_helper(
CtConfig, % CtConfig,
#{ % #{
auth_settings => AuthSettings, % auth_settings => AuthSettings,
ssl_settings => valid_ssl_settings() % ssl_settings => valid_ssl_settings()
}, % },
Config % Config
), % ),
ok. ok.
publish_helper(CtConfig, AuthSettings) -> publish_helper(CtConfig, AuthSettings) ->
@ -941,14 +964,14 @@ publish_helper(
Hash = erlang:phash2([HostsString, AuthSettings, SSLSettings]), Hash = erlang:phash2([HostsString, AuthSettings, SSLSettings]),
Name = "kafka_bridge_name_" ++ erlang:integer_to_list(Hash), Name = "kafka_bridge_name_" ++ erlang:integer_to_list(Hash),
Type = ?BRIDGE_TYPE, Type = ?BRIDGE_TYPE,
InstId = emqx_bridge_resource:resource_id(Type, Name), %InstId = <<"connector:", (bin(Type))/binary, ":", (bin(Name))/binary>>,
KafkaTopic = test_topic_one_partition(), KafkaTopic = test_topic_one_partition(),
Conf = config( Conf = config(
#{ #{
"bridge_name" => Name,
"authentication" => AuthSettings, "authentication" => AuthSettings,
"kafka_hosts_string" => HostsString, "kafka_hosts_string" => HostsString,
"kafka_topic" => KafkaTopic, "kafka_topic" => KafkaTopic,
"instance_id" => InstId,
"local_topic" => <<"mqtt/local">>, "local_topic" => <<"mqtt/local">>,
"ssl" => SSLSettings "ssl" => SSLSettings
}, },
@ -971,12 +994,16 @@ publish_helper(
}, },
{ok, Offset0} = resolve_kafka_offset(kafka_hosts(), KafkaTopic, Partition), {ok, Offset0} = resolve_kafka_offset(kafka_hosts(), KafkaTopic, Partition),
ct:pal("base offset before testing ~p", [Offset0]), ct:pal("base offset before testing ~p", [Offset0]),
InstId = emqx_bridge_resource:resource_id(bin(Type), bin(Name)),
BridgeV2Id = emqx_bridge_v2:id(bin(Type), bin(Name)),
{ok, _Group, #{state := State}} = emqx_resource:get_instance(InstId), {ok, _Group, #{state := State}} = emqx_resource:get_instance(InstId),
ok = send(CtConfig, InstId, Msg, State), ok = send(CtConfig, InstId, Msg, State, BridgeV2Id),
{ok, {_, [KafkaMsg0]}} = brod:fetch(kafka_hosts(), KafkaTopic, Partition, Offset0), {ok, {_, [KafkaMsg0]}} = brod:fetch(kafka_hosts(), KafkaTopic, Partition, Offset0),
?assertMatch(#kafka_message{key = BinTime}, KafkaMsg0) ?assertMatch(#kafka_message{key = BinTime}, KafkaMsg0),
ok
end, end,
%% test that it forwards from local mqtt topic as well %% test that it forwards from local mqtt topic as well
%% TODO Make sure that local topic works for bridge_v2
{ok, Offset1} = resolve_kafka_offset(kafka_hosts(), KafkaTopic, Partition), {ok, Offset1} = resolve_kafka_offset(kafka_hosts(), KafkaTopic, Partition),
ct:pal("base offset before testing (2) ~p", [Offset1]), ct:pal("base offset before testing (2) ~p", [Offset1]),
emqx:publish(emqx_message:make(<<"mqtt/local">>, <<"payload">>)), emqx:publish(emqx_message:make(<<"mqtt/local">>, <<"payload">>)),
@ -1004,11 +1031,13 @@ config(Args0, More, ConfigTemplateFun) ->
Args = maps:merge(Args1, More), Args = maps:merge(Args1, More),
ConfText = hocon_config(Args, ConfigTemplateFun), ConfText = hocon_config(Args, ConfigTemplateFun),
{ok, Conf} = hocon:binary(ConfText, #{format => map}), {ok, Conf} = hocon:binary(ConfText, #{format => map}),
Name = bin(maps:get("bridge_name", Args)),
%% TODO can we skip this old check?
ct:pal("Running tests with conf:\n~p", [Conf]), ct:pal("Running tests with conf:\n~p", [Conf]),
InstId = maps:get("instance_id", Args), % % InstId = maps:get("instance_id", Args),
<<"bridge:", BridgeId/binary>> = InstId, TypeBin = list_to_binary(?BRIDGE_TYPE),
{Type, Name} = emqx_bridge_resource:parse_bridge_id(BridgeId, #{atom_name => false}), % <<"connector:", BridgeId/binary>> = InstId,
TypeBin = atom_to_binary(Type), % {Type, Name} = emqx_bridge_resource:parse_bridge_id(BridgeId, #{atom_name => false}),
hocon_tconf:check_plain( hocon_tconf:check_plain(
emqx_bridge_schema, emqx_bridge_schema,
Conf, Conf,
@ -1018,9 +1047,7 @@ config(Args0, More, ConfigTemplateFun) ->
Parsed. Parsed.
hocon_config(Args, ConfigTemplateFun) -> hocon_config(Args, ConfigTemplateFun) ->
InstId = maps:get("instance_id", Args), BridgeName = maps:get("bridge_name", Args),
<<"bridge:", BridgeId/binary>> = InstId,
{_Type, Name} = emqx_bridge_resource:parse_bridge_id(BridgeId, #{atom_name => false}),
AuthConf = maps:get("authentication", Args), AuthConf = maps:get("authentication", Args),
AuthTemplate = iolist_to_binary(hocon_config_template_authentication(AuthConf)), AuthTemplate = iolist_to_binary(hocon_config_template_authentication(AuthConf)),
AuthConfRendered = bbmustache:render(AuthTemplate, AuthConf), AuthConfRendered = bbmustache:render(AuthTemplate, AuthConf),
@ -1034,7 +1061,7 @@ hocon_config(Args, ConfigTemplateFun) ->
iolist_to_binary(ConfigTemplateFun()), iolist_to_binary(ConfigTemplateFun()),
Args#{ Args#{
"authentication" => AuthConfRendered, "authentication" => AuthConfRendered,
"bridge_name" => Name, "bridge_name" => BridgeName,
"ssl" => SSLConfRendered, "ssl" => SSLConfRendered,
"query_mode" => QueryMode, "query_mode" => QueryMode,
"kafka_headers" => KafkaHeaders, "kafka_headers" => KafkaHeaders,

46
apps/emqx_bridge_kafka/test/emqx_bridge_v2_kafka_producer_SUITE.erl
View File

@ -22,6 +22,7 @@
-include_lib("common_test/include/ct.hrl"). -include_lib("common_test/include/ct.hrl").
-include_lib("snabbkaffe/include/snabbkaffe.hrl"). -include_lib("snabbkaffe/include/snabbkaffe.hrl").
-include_lib("brod/include/brod.hrl"). -include_lib("brod/include/brod.hrl").
all() -> all() ->
emqx_common_test_helpers:all(?MODULE). emqx_common_test_helpers:all(?MODULE).
@ -123,23 +124,33 @@ t_health_check(_) ->
{ok, _} = emqx_connector:remove(kafka, test_connector3), {ok, _} = emqx_connector:remove(kafka, test_connector3),
ok. ok.
t_local_topic(_) ->
BridgeV2Config = bridge_v2_config(<<"test_connector">>),
ConnectorConfig = connector_config(),
{ok, _} = emqx_connector:create(kafka, test_connector, ConnectorConfig),
{ok, _} = emqx_bridge_v2:create(kafka, test_bridge, BridgeV2Config),
%% Send a message to the local topic
Payload = <<"local_topic_payload">>,
Offset = resolve_kafka_offset(),
emqx:publish(emqx_message:make(<<"kafka_t/hej">>, Payload)),
check_kafka_message_payload(Offset, Payload),
{ok, _} = emqx_bridge_v2:remove(kafka, test_bridge),
{ok, _} = emqx_connector:remove(kafka, test_connector),
ok.
check_send_message_with_bridge(BridgeName) -> check_send_message_with_bridge(BridgeName) ->
%% ###################################### %% ######################################
%% Create Kafka message %% Create Kafka message
%% ###################################### %% ######################################
KafkaTopic = emqx_bridge_kafka_impl_producer_SUITE:test_topic_one_partition(),
Partition = 0,
Time = erlang:unique_integer(), Time = erlang:unique_integer(),
BinTime = integer_to_binary(Time), BinTime = integer_to_binary(Time),
Payload = list_to_binary("payload" ++ integer_to_list(Time)),
Msg = #{ Msg = #{
clientid => BinTime, clientid => BinTime,
payload => <<"payload">>, payload => Payload,
timestamp => Time timestamp => Time
}, },
Hosts = emqx_bridge_kafka_impl_producer_SUITE:kafka_hosts(), Offset = resolve_kafka_offset(),
{ok, Offset0} = emqx_bridge_kafka_impl_producer_SUITE:resolve_kafka_offset(
Hosts, KafkaTopic, Partition
),
%% ###################################### %% ######################################
%% Send message %% Send message
%% ###################################### %% ######################################
@ -147,8 +158,23 @@ check_send_message_with_bridge(BridgeName) ->
%% ###################################### %% ######################################
%% Check if message is sent to Kafka %% Check if message is sent to Kafka
%% ###################################### %% ######################################
{ok, {_, [KafkaMsg0]}} = brod:fetch(Hosts, KafkaTopic, Partition, Offset0), check_kafka_message_payload(Offset, Payload).
?assertMatch(#kafka_message{key = BinTime}, KafkaMsg0).
resolve_kafka_offset() ->
KafkaTopic = emqx_bridge_kafka_impl_producer_SUITE:test_topic_one_partition(),
Partition = 0,
Hosts = emqx_bridge_kafka_impl_producer_SUITE:kafka_hosts(),
{ok, Offset0} = emqx_bridge_kafka_impl_producer_SUITE:resolve_kafka_offset(
Hosts, KafkaTopic, Partition
),
Offset0.
check_kafka_message_payload(Offset, ExpectedPayload) ->
KafkaTopic = emqx_bridge_kafka_impl_producer_SUITE:test_topic_one_partition(),
Partition = 0,
Hosts = emqx_bridge_kafka_impl_producer_SUITE:kafka_hosts(),
{ok, {_, [KafkaMsg0]}} = brod:fetch(Hosts, KafkaTopic, Partition, Offset),
?assertMatch(#kafka_message{value = ExpectedPayload}, KafkaMsg0).
bridge_v2_config(ConnectorName) -> bridge_v2_config(ConnectorName) ->
#{ #{
@ -168,7 +194,7 @@ bridge_v2_config(ConnectorName) ->
<<"message">> => #{ <<"message">> => #{
<<"key">> => <<"${.clientid}">>, <<"key">> => <<"${.clientid}">>,
<<"timestamp">> => <<"${.timestamp}">>, <<"timestamp">> => <<"${.timestamp}">>,
<<"value">> => <<"${.}">> <<"value">> => <<"${.payload}">>
}, },
<<"partition_count_refresh_interval">> => <<"60s">>, <<"partition_count_refresh_interval">> => <<"60s">>,
<<"partition_strategy">> => <<"random">>, <<"partition_strategy">> => <<"random">>,

7
apps/emqx_connector/src/schema/emqx_connector_schema.erl
View File

@ -21,7 +21,7 @@
-import(hoconsc, [mk/2, ref/2]). -import(hoconsc, [mk/2, ref/2]).
-export([transform_old_style_bridges_to_connector_and_actions/1]). -export([transform_bridges_v1_to_connectors_and_bridges_v2/1]).
-export([roots/0, fields/1, desc/1, namespace/0, tags/0]). -export([roots/0, fields/1, desc/1, namespace/0, tags/0]).
@ -151,8 +151,9 @@ transform_old_style_bridges_to_connector_and_actions_of_type(
BridgeTypes BridgeTypes
), ),
BridgeConfigsToTransform = lists:flatten(BridgeConfigsToTransform1), BridgeConfigsToTransform = lists:flatten(BridgeConfigsToTransform1),
ConnectorsWithTypeMap = maps:get(to_bin(ConnectorType), ConnectorsConfMap, #{}),
BridgeConfigsToTransformWithConnectorConf = lists:zip( BridgeConfigsToTransformWithConnectorConf = lists:zip(
lists:duplicate(length(BridgeConfigsToTransform), ConnectorsConfMap), lists:duplicate(length(BridgeConfigsToTransform), ConnectorsWithTypeMap),
BridgeConfigsToTransform BridgeConfigsToTransform
), ),
ActionConnectorTuples = lists:map( ActionConnectorTuples = lists:map(
@ -185,7 +186,7 @@ transform_old_style_bridges_to_connector_and_actions_of_type(
ActionConnectorTuples ActionConnectorTuples
). ).
transform_old_style_bridges_to_connector_and_actions(RawConfig) -> transform_bridges_v1_to_connectors_and_bridges_v2(RawConfig) ->
ConnectorFields = fields(connectors), ConnectorFields = fields(connectors),
NewRawConf = lists:foldl( NewRawConf = lists:foldl(
fun transform_old_style_bridges_to_connector_and_actions_of_type/2, fun transform_old_style_bridges_to_connector_and_actions_of_type/2,

2
apps/emqx_resource/src/emqx_resource.erl
View File

@ -233,7 +233,7 @@
-callback on_get_channels( -callback on_get_channels(
ResId :: term() ResId :: term()
) -> {ok, NewState :: term()}. ) -> {ok, [term()]}.
-spec list_types() -> [module()]. -spec list_types() -> [module()].
list_types() -> list_types() ->