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

Merge pull request #11889 from thalesmg/rename-bridgev2-to-action-apis-r53-20231106 

chore: rename `bridges_v2` -> `actions` in the public facing APIs
This commit is contained in:
Thales Macedo Garitezi 2023-11-07 09:29:06 -03:00 committed by GitHub
parent ac6ad79029 2b8cf50a1d
commit 9742dae30d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 125 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -752,7 +752,7 @@ is_bridge_enabled_v1(BridgeType, BridgeName) ->
is_bridge_enabled_v2(BridgeV1Type, BridgeName) ->
BridgeV2Type = emqx_bridge_v2:bridge_v1_type_to_bridge_v2_type(BridgeV1Type),
try emqx:get_config([bridges_v2, BridgeV2Type, binary_to_existing_atom(BridgeName)]) of
try emqx:get_config([actions, BridgeV2Type, binary_to_existing_atom(BridgeName)]) of
ConfMap ->
maps:get(enable, ConfMap, true)
catch

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

@ -24,7 +24,9 @@
-include_lib("emqx_resource/include/emqx_resource.hrl").
-include_lib("snabbkaffe/include/snabbkaffe.hrl").
-define(ROOT_KEY, bridges_v2).
%% Note: this is strange right now, because it lives in `emqx_bridge_v2', but it shall be
%% refactored into a new module/application with appropriate name.
-define(ROOT_KEY, actions).
%% Loading and unloading config when EMQX starts and stops
-export([
@ -600,7 +602,7 @@ create_dry_run_helper(BridgeType, ConnectorRawConf, BridgeV2RawConf) ->
create_dry_run(Type, Conf0) ->
Conf1 = maps:without([<<"name">>], Conf0),
TypeBin = bin(Type),
RawConf = #{<<"bridges_v2">> => #{TypeBin => #{<<"temp_name">> => Conf1}}},
RawConf = #{<<"actions">> => #{TypeBin => #{<<"temp_name">> => Conf1}}},
%% Check config
try
_ =
@ -741,7 +743,7 @@ parse_id(Id) ->
case binary:split(Id, <<":">>, [global]) of
[Type, Name] ->
{Type, Name};
[<<"bridge_v2">>, Type, Name | _] ->
[<<"action">>, Type, Name | _] ->
{Type, Name};
_X ->
error({error, iolist_to_binary(io_lib:format("Invalid id: ~p", [Id]))})
@ -785,7 +787,7 @@ id(BridgeType, BridgeName) ->
id(BridgeType, BridgeName, ConnectorName) ->
ConnectorType = bin(connector_type(BridgeType)),
<<"bridge_v2:", (bin(BridgeType))/binary, ":", (bin(BridgeName))/binary, ":connector:",
<<"action:", (bin(BridgeType))/binary, ":", (bin(BridgeName))/binary, ":connector:",
(bin(ConnectorType))/binary, ":", (bin(ConnectorName))/binary>>.
connector_type(Type) ->
@ -807,8 +809,8 @@ bridge_v2_type_to_connector_type(azure_event_hub_producer) ->
%%====================================================================
import_config(RawConf) ->
%% bridges v2 structure
emqx_bridge:import_config(RawConf, <<"bridges_v2">>, ?ROOT_KEY, config_key_path()).
%% actions structure
emqx_bridge:import_config(RawConf, <<"actions">>, ?ROOT_KEY, config_key_path()).
%%====================================================================
%% Config Update Handler API
@ -833,8 +835,8 @@ pre_config_update(_Path, Conf, _OldConfig) when is_map(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
%% with calls to emqx_conf:update([bridges_v2], BridgesConf, #{}).
%% This top level handler will be triggered when the actions path is updated
%% with calls to emqx_conf:update([actions], BridgesConf, #{}).
%%
%% A public API that can trigger this is:
%% bin/emqx ctl conf load data/configs/cluster.hocon
@ -1086,7 +1088,7 @@ lookup_and_transform_to_bridge_v1_helper(
BridgeV2RawConfig2 = fill_defaults(
BridgeV2Type,
BridgeV2RawConfig1,
<<"bridges_v2">>,
<<"actions">>,
emqx_bridge_v2_schema
),
BridgeV1Config1 = maps:remove(<<"connector">>, BridgeV2RawConfig2),
@ -1246,7 +1248,7 @@ split_and_validate_bridge_v1_config(BridgeV1Type, BridgeName, RawConf, PreviousR
bin(ConnectorName) => NewConnectorRawConf
}
},
<<"bridges_v2">> => #{
<<"actions">> => #{
bin(BridgeV2Type) => #{
bin(BridgeName) => NewBridgeV2RawConf
}
@ -1439,10 +1441,10 @@ bin(Atom) when is_atom(Atom) -> atom_to_binary(Atom, utf8).
extract_connector_id_from_bridge_v2_id(Id) ->
case binary:split(Id, <<":">>, [global]) of
[<<"bridge_v2">>, _Type, _Name, <<"connector">>, ConnectorType, ConnecorName] ->
[<<"action">>, _Type, _Name, <<"connector">>, ConnectorType, ConnecorName] ->
<<"connector:", ConnectorType/binary, ":", ConnecorName/binary>>;
_X ->
error({error, iolist_to_binary(io_lib:format("Invalid bridge V2 ID: ~p", [Id]))})
error({error, iolist_to_binary(io_lib:format("Invalid action ID: ~p", [Id]))})
end.
to_existing_atom(X) ->

88
apps/emqx_bridge/src/emqx_bridge_v2_api.erl
View File

@ -35,12 +35,12 @@
%% API callbacks
-export([
'/bridges_v2'/2,
'/bridges_v2/:id'/2,
'/bridges_v2/:id/enable/:enable'/2,
'/bridges_v2/:id/:operation'/2,
'/nodes/:node/bridges_v2/:id/:operation'/2,
'/bridges_v2_probe'/2
'/actions'/2,
'/actions/:id'/2,
'/actions/:id/enable/:enable'/2,
'/actions/:id/:operation'/2,
'/nodes/:node/actions/:id/:operation'/2,
'/actions_probe'/2
]).
%% BpAPI
@ -67,19 +67,19 @@
end
).
namespace() -> "bridge_v2".
namespace() -> "actions".
api_spec() ->
emqx_dashboard_swagger:spec(?MODULE, #{check_schema => true}).
paths() ->
[
"/bridges_v2",
"/bridges_v2/:id",
"/bridges_v2/:id/enable/:enable",
"/bridges_v2/:id/:operation",
"/nodes/:node/bridges_v2/:id/:operation",
"/bridges_v2_probe"
"/actions",
"/actions/:id",
"/actions/:id/enable/:enable",
"/actions/:id/:operation",
"/nodes/:node/actions/:id/:operation",
"/actions_probe"
].
error_schema(Code, Message) when is_atom(Code) ->
@ -183,11 +183,11 @@ param_path_enable() ->
}
)}.
schema("/bridges_v2") ->
schema("/actions") ->
#{
'operationId' => '/bridges_v2',
'operationId' => '/actions',
get => #{
tags => [<<"bridges_v2">>],
tags => [<<"actions">>],
summary => <<"List bridges">>,
description => ?DESC("desc_api1"),
responses => #{
@ -198,7 +198,7 @@ schema("/bridges_v2") ->
}
},
post => #{
tags => [<<"bridges_v2">>],
tags => [<<"actions">>],
summary => <<"Create bridge">>,
description => ?DESC("desc_api2"),
'requestBody' => emqx_dashboard_swagger:schema_with_examples(
@ -211,11 +211,11 @@ schema("/bridges_v2") ->
}
}
};
schema("/bridges_v2/:id") ->
schema("/actions/:id") ->
#{
'operationId' => '/bridges_v2/:id',
'operationId' => '/actions/:id',
get => #{
tags => [<<"bridges_v2">>],
tags => [<<"actions">>],
summary => <<"Get bridge">>,
description => ?DESC("desc_api3"),
parameters => [param_path_id()],
@ -225,7 +225,7 @@ schema("/bridges_v2/:id") ->
}
},
put => #{
tags => [<<"bridges_v2">>],
tags => [<<"actions">>],
summary => <<"Update bridge">>,
description => ?DESC("desc_api4"),
parameters => [param_path_id()],
@ -240,7 +240,7 @@ schema("/bridges_v2/:id") ->
}
},
delete => #{
tags => [<<"bridges_v2">>],
tags => [<<"actions">>],
summary => <<"Delete bridge">>,
description => ?DESC("desc_api5"),
parameters => [param_path_id(), param_qs_delete_cascade()],
@ -255,12 +255,12 @@ schema("/bridges_v2/:id") ->
}
}
};
schema("/bridges_v2/:id/enable/:enable") ->
schema("/actions/:id/enable/:enable") ->
#{
'operationId' => '/bridges_v2/:id/enable/:enable',
'operationId' => '/actions/:id/enable/:enable',
put =>
#{
tags => [<<"bridges_v2">>],
tags => [<<"actions">>],
summary => <<"Enable or disable bridge">>,
desc => ?DESC("desc_enable_bridge"),
parameters => [param_path_id(), param_path_enable()],
@ -274,11 +274,11 @@ schema("/bridges_v2/:id/enable/:enable") ->
}
}
};
schema("/bridges_v2/:id/:operation") ->
schema("/actions/:id/:operation") ->
#{
'operationId' => '/bridges_v2/:id/:operation',
'operationId' => '/actions/:id/:operation',
post => #{
tags => [<<"bridges_v2">>],
tags => [<<"actions">>],
summary => <<"Manually start a bridge">>,
description => ?DESC("desc_api7"),
parameters => [
@ -296,11 +296,11 @@ schema("/bridges_v2/:id/:operation") ->
}
}
};
schema("/nodes/:node/bridges_v2/:id/:operation") ->
schema("/nodes/:node/actions/:id/:operation") ->
#{
'operationId' => '/nodes/:node/bridges_v2/:id/:operation',
'operationId' => '/nodes/:node/actions/:id/:operation',
post => #{
tags => [<<"bridges_v2">>],
tags => [<<"actions">>],
summary => <<"Manually start a bridge">>,
description => ?DESC("desc_api8"),
parameters => [
@ -322,11 +322,11 @@ schema("/nodes/:node/bridges_v2/:id/:operation") ->
}
}
};
schema("/bridges_v2_probe") ->
schema("/actions_probe") ->
#{
'operationId' => '/bridges_v2_probe',
'operationId' => '/actions_probe',
post => #{
tags => [<<"bridges_v2">>],
tags => [<<"actions">>],
desc => ?DESC("desc_api9"),
summary => <<"Test creating bridge">>,
'requestBody' => emqx_dashboard_swagger:schema_with_examples(
@ -340,7 +340,7 @@ schema("/bridges_v2_probe") ->
}
}.
'/bridges_v2'(post, #{body := #{<<"type">> := BridgeType, <<"name">> := BridgeName} = Conf0}) ->
'/actions'(post, #{body := #{<<"type">> := BridgeType, <<"name">> := BridgeName} = Conf0}) ->
case emqx_bridge_v2:lookup(BridgeType, BridgeName) of
{ok, _} ->
?BAD_REQUEST('ALREADY_EXISTS', <<"bridge already exists">>);
@ -348,7 +348,7 @@ schema("/bridges_v2_probe") ->
Conf = filter_out_request_body(Conf0),
create_bridge(BridgeType, BridgeName, Conf)
end;
'/bridges_v2'(get, _Params) ->
'/actions'(get, _Params) ->
Nodes = mria:running_nodes(),
NodeReplies = emqx_bridge_proto_v5:v2_list_bridges_on_nodes(Nodes),
case is_ok(NodeReplies) of
@ -362,9 +362,9 @@ schema("/bridges_v2_probe") ->
?INTERNAL_ERROR(Reason)
end.
'/bridges_v2/:id'(get, #{bindings := #{id := Id}}) ->
'/actions/:id'(get, #{bindings := #{id := Id}}) ->
?TRY_PARSE_ID(Id, lookup_from_all_nodes(BridgeType, BridgeName, 200));
'/bridges_v2/:id'(put, #{bindings := #{id := Id}, body := Conf0}) ->
'/actions/:id'(put, #{bindings := #{id := Id}, body := Conf0}) ->
Conf1 = filter_out_request_body(Conf0),
?TRY_PARSE_ID(
Id,
@ -377,7 +377,7 @@ schema("/bridges_v2_probe") ->
?BRIDGE_NOT_FOUND(BridgeType, BridgeName)
end
);
'/bridges_v2/:id'(delete, #{bindings := #{id := Id}, query_string := Qs}) ->
'/actions/:id'(delete, #{bindings := #{id := Id}, query_string := Qs}) ->
?TRY_PARSE_ID(
Id,
case emqx_bridge_v2:lookup(BridgeType, BridgeName) of
@ -416,7 +416,7 @@ schema("/bridges_v2_probe") ->
end
).
'/bridges_v2/:id/enable/:enable'(put, #{bindings := #{id := Id, enable := Enable}}) ->
'/actions/:id/enable/:enable'(put, #{bindings := #{id := Id, enable := Enable}}) ->
?TRY_PARSE_ID(
Id,
case emqx_bridge_v2:disable_enable(enable_func(Enable), BridgeType, BridgeName) of
@ -433,7 +433,7 @@ schema("/bridges_v2_probe") ->
end
).
'/bridges_v2/:id/:operation'(post, #{
'/actions/:id/:operation'(post, #{
bindings :=
#{id := Id, operation := Op}
}) ->
@ -446,7 +446,7 @@ schema("/bridges_v2_probe") ->
end
).
'/nodes/:node/bridges_v2/:id/:operation'(post, #{
'/nodes/:node/actions/:id/:operation'(post, #{
bindings :=
#{id := Id, operation := Op, node := Node}
}) ->
@ -461,8 +461,8 @@ schema("/bridges_v2_probe") ->
end
).
'/bridges_v2_probe'(post, Request) ->
RequestMeta = #{module => ?MODULE, method => post, path => "/bridges_v2_probe"},
'/actions_probe'(post, Request) ->
RequestMeta = #{module => ?MODULE, method => post, path => "/actions_probe"},
case emqx_dashboard_swagger:filter_check_request_and_translate_body(Request, RequestMeta) of
{ok, #{body := #{<<"type">> := ConnType} = Params}} ->
Params1 = maybe_deobfuscate_bridge_probe(Params),

12
apps/emqx_bridge/src/schema/emqx_bridge_v2_enterprise.erl
View File

@ -31,24 +31,24 @@ schema_modules() ->
emqx_bridge_azure_event_hub
].
fields(bridges_v2) ->
bridge_v2_structs().
fields(actions) ->
action_structs().
bridge_v2_structs() ->
action_structs() ->
[
{kafka_producer,
mk(
hoconsc:map(name, ref(emqx_bridge_kafka, kafka_producer_action)),
#{
desc => <<"Kafka Producer Bridge V2 Config">>,
desc => <<"Kafka Producer Actions Config">>,
required => false
}
)},
{azure_event_hub_producer,
mk(
hoconsc:map(name, ref(emqx_bridge_azure_event_hub, bridge_v2)),
hoconsc:map(name, ref(emqx_bridge_azure_event_hub, actions)),
#{
desc => <<"Azure Event Hub Bridge V2 Config">>,
desc => <<"Azure Event Hub Actions Config">>,
required => false
}
)}

18
apps/emqx_bridge/src/schema/emqx_bridge_v2_schema.erl
View File

@ -48,7 +48,7 @@ enterprise_fields_actions() ->
_ = emqx_bridge_v2_enterprise:module_info(),
case erlang:function_exported(emqx_bridge_v2_enterprise, fields, 1) of
true ->
emqx_bridge_v2_enterprise:fields(bridges_v2);
emqx_bridge_v2_enterprise:fields(actions);
false ->
[]
end.
@ -103,28 +103,28 @@ bridge_api_union(Refs) ->
%% HOCON Schema Callbacks
%%======================================================================================
namespace() -> "bridges_v2".
namespace() -> "actions".
tags() ->
[<<"Bridge V2">>].
[<<"Actions">>].
-dialyzer({nowarn_function, roots/0}).
roots() ->
case fields(bridges_v2) of
case fields(actions) of
[] ->
[
{bridges_v2,
{actions,
?HOCON(hoconsc:map(name, typerefl:map()), #{importance => ?IMPORTANCE_LOW})}
];
_ ->
[{bridges_v2, ?HOCON(?R_REF(bridges_v2), #{importance => ?IMPORTANCE_LOW})}]
[{actions, ?HOCON(?R_REF(actions), #{importance => ?IMPORTANCE_LOW})}]
end.
fields(bridges_v2) ->
fields(actions) ->
[] ++ enterprise_fields_actions().
desc(bridges_v2) ->
desc(actions) ->
?DESC("desc_bridges_v2");
desc(_) ->
undefined.
@ -137,7 +137,7 @@ schema_homogeneous_test() ->
fun({_Name, Schema}) ->
is_bad_schema(Schema)
end,
fields(bridges_v2)
fields(actions)
)
of
[] ->

18
apps/emqx_bridge/test/emqx_bridge_v1_compatibility_layer_SUITE.erl
View File

@ -234,7 +234,7 @@ unwrap_fun(UniqRefStr) ->
ets:lookup_element(fun_table_name(), UniqRefStr, 2).
update_root_config(RootConf) ->
emqx_conf:update([bridges_v2], RootConf, #{override_to => cluster}).
emqx_conf:update([actions], RootConf, #{override_to => cluster}).
delete_all_bridges() ->
lists:foreach(
@ -287,7 +287,7 @@ list_bridges_http_api_v1() ->
Res.
list_bridges_http_api_v2() ->
Path = emqx_mgmt_api_test_util:api_path(["bridges_v2"]),
Path = emqx_mgmt_api_test_util:api_path(["actions"]),
ct:pal("list bridges (http v2)"),
Res = request(get, Path, _Params = []),
ct:pal("list bridges (http v2) result:\n ~p", [Res]),
@ -310,7 +310,7 @@ get_bridge_http_api_v1(Name) ->
get_bridge_http_api_v2(Name) ->
BridgeId = emqx_bridge_resource:bridge_id(bridge_type(), Name),
Path = emqx_mgmt_api_test_util:api_path(["bridges_v2", BridgeId]),
Path = emqx_mgmt_api_test_util:api_path(["actions", BridgeId]),
ct:pal("get bridge (http v2) (~p)", [#{name => Name}]),
Res = request(get, Path, _Params = []),
ct:pal("get bridge (http v2) (~p) result:\n ~p", [#{name => Name}, Res]),
@ -341,7 +341,7 @@ create_bridge_http_api_v2(Opts) ->
Overrides = maps:get(overrides, Opts, #{}),
BridgeConfig = emqx_utils_maps:deep_merge(bridge_config(), Overrides),
Params = BridgeConfig#{<<"type">> => bridge_type_bin(), <<"name">> => Name},
Path = emqx_mgmt_api_test_util:api_path(["bridges_v2"]),
Path = emqx_mgmt_api_test_util:api_path(["actions"]),
ct:pal("creating bridge (http v2): ~p", [Params]),
Res = request(post, Path, Params),
ct:pal("bridge create (http v2) result:\n ~p", [Res]),
@ -372,7 +372,7 @@ delete_bridge_http_api_v1(Opts) ->
delete_bridge_http_api_v2(Opts) ->
Name = maps:get(name, Opts),
BridgeId = emqx_bridge_resource:bridge_id(bridge_type(), Name),
Path = emqx_mgmt_api_test_util:api_path(["bridges_v2", BridgeId]),
Path = emqx_mgmt_api_test_util:api_path(["actions", BridgeId]),
ct:pal("deleting bridge (http v2)"),
Res = request(delete, Path, _Params = []),
ct:pal("bridge delete (http v2) result:\n ~p", [Res]),
@ -388,7 +388,7 @@ enable_bridge_http_api_v1(Name) ->
enable_bridge_http_api_v2(Name) ->
BridgeId = emqx_bridge_resource:bridge_id(bridge_type(), Name),
Path = emqx_mgmt_api_test_util:api_path(["bridges_v2", BridgeId, "enable", "true"]),
Path = emqx_mgmt_api_test_util:api_path(["actions", BridgeId, "enable", "true"]),
ct:pal("enabling bridge (http v2)"),
Res = request(put, Path, _Params = []),
ct:pal("bridge enable (http v2) result:\n ~p", [Res]),
@ -404,7 +404,7 @@ disable_bridge_http_api_v1(Name) ->
disable_bridge_http_api_v2(Name) ->
BridgeId = emqx_bridge_resource:bridge_id(bridge_type(), Name),
Path = emqx_mgmt_api_test_util:api_path(["bridges_v2", BridgeId, "enable", "false"]),
Path = emqx_mgmt_api_test_util:api_path(["actions", BridgeId, "enable", "false"]),
ct:pal("disabling bridge (http v2)"),
Res = request(put, Path, _Params = []),
ct:pal("bridge disable (http v2) result:\n ~p", [Res]),
@ -422,7 +422,7 @@ bridge_operation_http_api_v1(Name, Op0) ->
bridge_operation_http_api_v2(Name, Op0) ->
Op = atom_to_list(Op0),
BridgeId = emqx_bridge_resource:bridge_id(bridge_type(), Name),
Path = emqx_mgmt_api_test_util:api_path(["bridges_v2", BridgeId, Op]),
Path = emqx_mgmt_api_test_util:api_path(["actions", BridgeId, Op]),
ct:pal("bridge op ~p (http v2)", [Op]),
Res = request(post, Path, _Params = []),
ct:pal("bridge op ~p (http v2) result:\n ~p", [Op, Res]),
@ -442,7 +442,7 @@ bridge_node_operation_http_api_v2(Name, Node0, Op0) ->
Op = atom_to_list(Op0),
Node = atom_to_list(Node0),
BridgeId = emqx_bridge_resource:bridge_id(bridge_type(), Name),
Path = emqx_mgmt_api_test_util:api_path(["nodes", Node, "bridges_v2", BridgeId, Op]),
Path = emqx_mgmt_api_test_util:api_path(["nodes", Node, "actions", BridgeId, Op]),
ct:pal("bridge node op ~p (http v2)", [{Node, Op}]),
Res = request(post, Path, _Params = []),
ct:pal("bridge node op ~p (http v2) result:\n ~p", [{Node, Op}, Res]),

6
apps/emqx_bridge/test/emqx_bridge_v2_SUITE.erl
View File

@ -207,7 +207,7 @@ unwrap_fun(UniqRefStr) ->
ets:lookup_element(fun_table_name(), UniqRefStr, 2).
update_root_config(RootConf) ->
emqx_conf:update([bridges_v2], RootConf, #{override_to => cluster}).
emqx_conf:update([actions], RootConf, #{override_to => cluster}).
update_root_connectors_config(RootConf) ->
emqx_conf:update([connectors], RootConf, #{override_to => cluster}).
@ -584,7 +584,7 @@ t_unhealthy_channel_alarm(_) ->
get_bridge_v2_alarm_cnt() ->
Alarms = emqx_alarm:get_alarms(activated),
FilterFun = fun
(#{name := S}) when is_binary(S) -> string:find(S, "bridge_v2") =/= nomatch;
(#{name := S}) when is_binary(S) -> string:find(S, "action") =/= nomatch;
(_) -> false
end,
length(lists:filter(FilterFun, Alarms)).
@ -639,7 +639,7 @@ t_load_config_success(_Config) ->
BridgeNameBin = atom_to_binary(BridgeName),
%% pre-condition
?assertEqual(#{}, emqx_config:get([bridges_v2])),
?assertEqual(#{}, emqx_config:get([actions])),
%% create
RootConf0 = #{BridgeTypeBin => #{BridgeNameBin => Conf}},

18
apps/emqx_bridge/test/emqx_bridge_v2_api_SUITE.erl
View File

@ -24,7 +24,7 @@
-include_lib("common_test/include/ct.hrl").
-include_lib("snabbkaffe/include/test_macros.hrl").
-define(ROOT, "bridges_v2").
-define(ROOT, "actions").
-define(CONNECTOR_NAME, <<"my_connector">>).
@ -153,7 +153,7 @@
emqx_auth,
emqx_management,
emqx_connector,
{emqx_bridge, "bridges_v2 {}"},
{emqx_bridge, "actions {}"},
{emqx_rule_engine, "rule_engine { rules {} }"}
]).
@ -221,8 +221,8 @@ mk_cluster(Name, Config, Opts) ->
Node2Apps = ?APPSPECS,
emqx_cth_cluster:start(
[
{emqx_bridge_api_SUITE_1, Opts#{role => core, apps => Node1Apps}},
{emqx_bridge_api_SUITE_2, Opts#{role => core, apps => Node2Apps}}
{emqx_bridge_v2_api_SUITE_1, Opts#{role => core, apps => Node1Apps}},
{emqx_bridge_v2_api_SUITE_2, Opts#{role => core, apps => Node2Apps}}
],
#{work_dir => filename:join(?config(priv_dir, Config), Name)}
).
@ -778,7 +778,7 @@ connector_operation(Config, ConnectorType, ConnectorName, OperationName) ->
t_bridges_probe(Config) ->
{ok, 204, <<>>} = request(
post,
uri(["bridges_v2_probe"]),
uri(["actions_probe"]),
?KAFKA_BRIDGE(?BRIDGE_NAME),
Config
),
@ -786,7 +786,7 @@ t_bridges_probe(Config) ->
%% second time with same name is ok since no real bridge created
{ok, 204, <<>>} = request(
post,
uri(["bridges_v2_probe"]),
uri(["actions_probe"]),
?KAFKA_BRIDGE(?BRIDGE_NAME),
Config
),
@ -800,7 +800,7 @@ t_bridges_probe(Config) ->
}},
request_json(
post,
uri(["bridges_v2_probe"]),
uri(["actions_probe"]),
?KAFKA_BRIDGE(<<"broken_bridge">>, <<"brokenhost:1234">>),
Config
)
@ -812,7 +812,7 @@ t_bridges_probe(Config) ->
{ok, 400, #{<<"code">> := <<"BAD_REQUEST">>}},
request_json(
post,
uri(["bridges_v2_probe"]),
uri(["actions_probe"]),
?RESOURCE(<<"broken_bridge">>, <<"unknown_type">>),
Config
)
@ -823,7 +823,7 @@ t_cascade_delete_actions(Config) ->
%% assert we there's no bridges at first
{ok, 200, []} = request_json(get, uri([?ROOT]), Config),
%% then we add a a bridge, using POST
%% POST /bridges_v2/ will create a bridge
%% POST /actions/ will create a bridge
BridgeID = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE, ?BRIDGE_NAME),
{ok, 201, _} = request(
post,

10
apps/emqx_bridge/test/emqx_bridge_v2_testlib.erl
View File

@ -121,7 +121,7 @@ bridge_id(Config) ->
BridgeName = ?config(bridge_name, Config),
BridgeId = emqx_bridge_resource:bridge_id(BridgeType, BridgeName),
ConnectorId = emqx_bridge_resource:resource_id(BridgeType, BridgeName),
<<"bridge_v2:", BridgeId/binary, ":", ConnectorId/binary>>.
<<"action:", BridgeId/binary, ":", ConnectorId/binary>>.
resource_id(Config) ->
BridgeType = ?config(bridge_type, Config),
@ -161,7 +161,7 @@ create_bridge_api(Config, Overrides) ->
emqx_connector:create(ConnectorType, ConnectorName, ConnectorConfig),
Params = BridgeConfig#{<<"type">> => BridgeType, <<"name">> => BridgeName},
Path = emqx_mgmt_api_test_util:api_path(["bridges_v2"]),
Path = emqx_mgmt_api_test_util:api_path(["actions"]),
AuthHeader = emqx_mgmt_api_test_util:auth_header_(),
Opts = #{return_all => true},
ct:pal("creating bridge (via http): ~p", [Params]),
@ -184,7 +184,7 @@ update_bridge_api(Config, Overrides) ->
BridgeConfig0 = ?config(bridge_config, Config),
BridgeConfig = emqx_utils_maps:deep_merge(BridgeConfig0, Overrides),
BridgeId = emqx_bridge_resource:bridge_id(BridgeType, Name),
Path = emqx_mgmt_api_test_util:api_path(["bridges_v2", BridgeId]),
Path = emqx_mgmt_api_test_util:api_path(["actions", BridgeId]),
AuthHeader = emqx_mgmt_api_test_util:auth_header_(),
Opts = #{return_all => true},
ct:pal("updating bridge (via http): ~p", [BridgeConfig]),
@ -198,7 +198,7 @@ update_bridge_api(Config, Overrides) ->
op_bridge_api(Op, BridgeType, BridgeName) ->
BridgeId = emqx_bridge_resource:bridge_id(BridgeType, BridgeName),
Path = emqx_mgmt_api_test_util:api_path(["bridges_v2", BridgeId, Op]),
Path = emqx_mgmt_api_test_util:api_path(["actions", BridgeId, Op]),
AuthHeader = emqx_mgmt_api_test_util:auth_header_(),
Opts = #{return_all => true},
ct:pal("calling bridge ~p (via http): ~p", [BridgeId, Op]),
@ -228,7 +228,7 @@ probe_bridge_api(Config, Overrides) ->
probe_bridge_api(BridgeType, BridgeName, BridgeConfig) ->
Params = BridgeConfig#{<<"type">> => BridgeType, <<"name">> => BridgeName},
Path = emqx_mgmt_api_test_util:api_path(["bridges_v2_probe"]),
Path = emqx_mgmt_api_test_util:api_path(["actions_probe"]),
AuthHeader = emqx_mgmt_api_test_util:auth_header_(),
Opts = #{return_all => true},
ct:pal("probing bridge (via http): ~p", [Params]),

6
apps/emqx_bridge_azure_event_hub/src/emqx_bridge_azure_event_hub.erl
View File

@ -79,7 +79,7 @@ fields("post_producer") ->
),
override_documentations(Fields);
fields("config_bridge_v2") ->
fields(bridge_v2);
fields(actions);
fields("config_connector") ->
Fields = override(
emqx_bridge_kafka:fields(kafka_connector),
@ -114,7 +114,7 @@ fields(kafka_message) ->
Fields0 = emqx_bridge_kafka:fields(kafka_message),
Fields = proplists:delete(timestamp, Fields0),
override_documentations(Fields);
fields(bridge_v2) ->
fields(actions) ->
Fields =
override(
emqx_bridge_kafka:fields(producer_opts),
@ -154,7 +154,7 @@ struct_names() ->
auth_username_password,
kafka_message,
producer_kafka_opts,
bridge_v2,
actions,
ssl_client_opts
].

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

@ -50,7 +50,7 @@ bridge_v2_examples(Method) ->
[
#{
<<"kafka_producer">> => #{
summary => <<"Kafka Producer Bridge v2">>,
summary => <<"Kafka Producer Action">>,
value => values({Method, bridge_v2_producer})
}
}

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

@ -62,7 +62,7 @@ enterprise_fields_connectors() -> [].
connector_type_to_bridge_types(kafka_producer) -> [kafka, kafka_producer];
connector_type_to_bridge_types(azure_event_hub_producer) -> [azure_event_hub_producer].
actions_config_name() -> <<"bridges_v2">>.
actions_config_name() -> <<"actions">>.
has_connector_field(BridgeConf, ConnectorFields) ->
lists:any(
@ -83,7 +83,7 @@ bridge_configs_to_transform(
true ->
PreviousRawConfig =
emqx_utils_maps:deep_get(
[<<"bridges_v2">>, to_bin(BridgeType), to_bin(BridgeName)],
[<<"actions">>, to_bin(BridgeType), to_bin(BridgeName)],
RawConfig,
undefined
),
@ -201,7 +201,7 @@ transform_old_style_bridges_to_connector_and_actions_of_type(
[<<"bridges">>, to_bin(BridgeType), BridgeName],
RawConfigSoFar1
),
%% Add bridge_v2
%% Add action
RawConfigSoFar3 = emqx_utils_maps:deep_put(
[actions_config_name(), to_bin(maybe_rename(BridgeType)), BridgeName],
RawConfigSoFar2,

4
apps/emqx_connector/test/emqx_connector_api_SUITE.erl
View File

@ -172,8 +172,8 @@ mk_cluster(Name, Config, Opts) ->
Node2Apps = ?APPSPECS,
emqx_cth_cluster:start(
[
{emqx_bridge_api_SUITE_1, Opts#{role => core, apps => Node1Apps}},
{emqx_bridge_api_SUITE_2, Opts#{role => core, apps => Node2Apps}}
{emqx_connector_api_SUITE_1, Opts#{role => core, apps => Node1Apps}},
{emqx_connector_api_SUITE_2, Opts#{role => core, apps => Node2Apps}}
],
#{work_dir => filename:join(?config(priv_dir, Config), Name)}
).

11
apps/emqx_dashboard/src/emqx_dashboard_schema_api.erl
View File

@ -45,7 +45,7 @@ paths() ->
%% This is a rather hidden API, so we don't need to add translations for the description.
schema("/schemas/:name") ->
Schemas = [hotconf, bridges, bridges_v2, connectors],
Schemas = [hotconf, bridges, actions, connectors],
#{
'operationId' => get_schema,
get => #{
@ -79,13 +79,14 @@ gen_schema(hotconf) ->
emqx_conf:hotconf_schema_json();
gen_schema(bridges) ->
emqx_conf:bridge_schema_json();
gen_schema(bridges_v2) ->
bridge_v2_schema_json();
gen_schema(actions) ->
actions_schema_json();
gen_schema(connectors) ->
connectors_schema_json().
bridge_v2_schema_json() ->
SchemaInfo = #{title => <<"EMQX Data Bridge V2 API Schema">>, version => <<"0.1.0">>},
actions_schema_json() ->
SchemaInfo = #{title => <<"EMQX Data Actions API Schema">>, version => <<"0.1.0">>},
%% Note: this will be moved to `emqx_actions' application in the future.
gen_api_schema_json_iodata(emqx_bridge_v2_api, SchemaInfo).
connectors_schema_json() ->

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

@ -1087,7 +1087,7 @@ call_query(QM, Id, Index, Ref, Query, QueryOpts) ->
?RESOURCE_ERROR(not_found, "resource not found")
end.
%% bridge_v2:kafka_producer:myproducer1:connector:kafka_producer:mykakfaclient1
%% action:kafka_producer:myproducer1:connector:kafka_producer:mykakfaclient1
extract_connector_id(Id) when is_binary(Id) ->
case binary:split(Id, <<":">>, [global]) of
[

10
changes/ee/feat-11581.en.md
View File

@ -1,12 +1,12 @@
Preview Feature: Support for Version 2 Bridge Design
- Introduction of Bridge v2 with a new 'connector' concept
- Introduction of Action with a new 'connector' concept
- In the original Bridge v1 design, each connector was exclusively tied to a single bridge.
This design prioritized error isolation and performance but posed challenges for users setting up multiple bridges to the same service.
For instance, setting up 10 bridges to a single Kafka cluster required the same configuration to be repeated for each bridge.
- The revamped Bridge v2 design provides more flexibility and better scalability:
- The revamped Action design provides more flexibility and better scalability:
- Users have the option to either share a connector across multiple bridges or retain it exclusively for one bridge, as in v1.
- For the majority of data bridges, sharing a connector among too many bridges might lead to performance degradation but avoids
overloading the external system with too many connections if the number of bridges is very high.
@ -18,9 +18,9 @@ Preview Feature: Support for Version 2 Bridge Design
- Management of Connectors
- Connectors can now be managed separately, bringing in more flexibility.
- New API endpoints have been introduced under the `/connectors` path for connector management.
- Version 2 bridges can be managed via the `/bridges_v2` endpoint.
- Actions can be managed via the `/actions` endpoint.
- Limitations in e5.3.1
- Currently, only the Kafka and Azure Event Hub producer bridges have been upgraded to the v2 design.
- The v2 bridge feature is accessible through config files and HTTP APIs. However, it's not yet available on the dashboard UI.
- Currently, only the Kafka and Azure Event Hub producer bridges have been upgraded to the action design.
- The action feature is accessible through config files and HTTP APIs. However, it's not yet available on the dashboard UI.

8
rel/i18n/emqx_bridge_azure_event_hub.hocon
View File

@ -195,10 +195,10 @@ bridge_v2_type.label:
bridge_v2_type.desc:
"""The type of the bridge."""
bridge_v2.label:
"""Bridge v2 Config"""
bridge_v2.desc:
"""The configuration for a bridge v2."""
actions.label:
"""Action Config"""
actions.desc:
"""The configuration for an action."""
buffer_memory_overload_protection.desc:
"""Applicable when buffer mode is set to <code>memory</code>