diff --git a/apps/emqx_authn/test/emqx_authn_http_SUITE.erl b/apps/emqx_authn/test/emqx_authn_http_SUITE.erl index 2c0716e8b..146e9ef32 100644 --- a/apps/emqx_authn/test/emqx_authn_http_SUITE.erl +++ b/apps/emqx_authn/test/emqx_authn_http_SUITE.erl @@ -153,9 +153,8 @@ t_destroy(_Config) -> ?GLOBAL), % Authenticator should not be usable anymore - ?assertException( - error, - _, + ?assertMatch( + ignore, emqx_authn_http:authenticate( Credentials, State)). diff --git a/apps/emqx_authn/test/emqx_authn_mongo_SUITE.erl b/apps/emqx_authn/test/emqx_authn_mongo_SUITE.erl index edd91be55..855a2226d 100644 --- a/apps/emqx_authn/test/emqx_authn_mongo_SUITE.erl +++ b/apps/emqx_authn/test/emqx_authn_mongo_SUITE.erl @@ -146,9 +146,8 @@ t_destroy(_Config) -> ?GLOBAL), % Authenticator should not be usable anymore - ?assertException( - error, - _, + ?assertMatch( + ignore, emqx_authn_mongodb:authenticate( #{username => <<"plain">>, password => <<"plain">> diff --git a/apps/emqx_authn/test/emqx_authn_mysql_SUITE.erl b/apps/emqx_authn/test/emqx_authn_mysql_SUITE.erl index 95eecdead..ffe98be65 100644 --- a/apps/emqx_authn/test/emqx_authn_mysql_SUITE.erl +++ b/apps/emqx_authn/test/emqx_authn_mysql_SUITE.erl @@ -159,9 +159,8 @@ t_destroy(_Config) -> ?GLOBAL), % Authenticator should not be usable anymore - ?assertException( - error, - _, + ?assertMatch( + ignore, emqx_authn_mysql:authenticate( #{username => <<"plain">>, password => <<"plain">> diff --git a/apps/emqx_authn/test/emqx_authn_pgsql_SUITE.erl b/apps/emqx_authn/test/emqx_authn_pgsql_SUITE.erl index e33f5c100..56044faf4 100644 --- a/apps/emqx_authn/test/emqx_authn_pgsql_SUITE.erl +++ b/apps/emqx_authn/test/emqx_authn_pgsql_SUITE.erl @@ -159,9 +159,8 @@ t_destroy(_Config) -> ?GLOBAL), % Authenticator should not be usable anymore - ?assertException( - error, - _, + ?assertMatch( + ignore, emqx_authn_pgsql:authenticate( #{username => <<"plain">>, password => <<"plain">> diff --git a/apps/emqx_authn/test/emqx_authn_redis_SUITE.erl b/apps/emqx_authn/test/emqx_authn_redis_SUITE.erl index de556a7bd..eeb674f86 100644 --- a/apps/emqx_authn/test/emqx_authn_redis_SUITE.erl +++ b/apps/emqx_authn/test/emqx_authn_redis_SUITE.erl @@ -164,9 +164,8 @@ t_destroy(_Config) -> ?GLOBAL), % Authenticator should not be usable anymore - ?assertException( - error, - _, + ?assertMatch( + ignore, emqx_authn_redis:authenticate( #{username => <<"plain">>, password => <<"plain">> diff --git a/apps/emqx_bridge/src/emqx_bridge.erl b/apps/emqx_bridge/src/emqx_bridge.erl index d46ce217e..3518190c7 100644 --- a/apps/emqx_bridge/src/emqx_bridge.erl +++ b/apps/emqx_bridge/src/emqx_bridge.erl @@ -80,17 +80,36 @@ unload_hook() -> on_message_publish(Message = #message{topic = Topic, flags = Flags}) -> case maps:get(sys, Flags, false) of false -> - lists:foreach(fun (Id) -> - send_message(Id, emqx_rule_events:eventmsg_publish(Message)) - end, get_matched_bridges(Topic)); + Msg = emqx_rule_events:eventmsg_publish(Message), + send_to_egress_matched_bridges(Topic, Msg); true -> ok end, {ok, Message}. +send_to_egress_matched_bridges(Topic, Msg) -> + lists:foreach(fun (Id) -> + try send_message(Id, Msg) of + ok -> ok; + Error -> ?SLOG(error, #{msg => "send_message_to_bridge_failed", + bridge => Id, error => Error}) + catch Err:Reason:ST -> + ?SLOG(error, #{msg => "send_message_to_bridge_crash", + bridge => Id, error => Err, reason => Reason, + stacktrace => ST}) + end + end, get_matched_bridges(Topic)). + send_message(BridgeId, Message) -> {BridgeType, BridgeName} = parse_bridge_id(BridgeId), ResId = emqx_bridge:resource_id(BridgeType, BridgeName), - emqx_resource:query(ResId, {send_message, Message}). + case emqx:get_config([bridges, BridgeType, BridgeName], not_found) of + not_found -> + {error, {bridge_not_found, BridgeId}}; + #{enable := true} -> + emqx_resource:query(ResId, {send_message, Message}); + #{enable := false} -> + {error, {bridge_stopped, BridgeId}} + end. config_key_path() -> [bridges]. @@ -279,6 +298,8 @@ get_matched_bridges(Topic) -> end, Acc0, Conf) end, [], Bridges). +get_matched_bridge_id(#{enable := false}, _Topic, _BType, _BName, Acc) -> + Acc; get_matched_bridge_id(#{local_topic := Filter}, Topic, BType, BName, Acc) -> case emqx_topic:match(Topic, Filter) of true -> [bridge_id(BType, BName) | Acc]; @@ -309,21 +330,21 @@ parse_confs(Type, Name, #{connector := ConnId, direction := Direction} = Conf) {Type, ConnName} -> ConnectorConfs = emqx:get_config([connectors, Type, ConnName]), make_resource_confs(Direction, ConnectorConfs, - maps:without([connector, direction], Conf), Name); + maps:without([connector, direction], Conf), Type, Name); {_ConnType, _ConnName} -> error({cannot_use_connector_with_different_type, ConnId}) end; -parse_confs(_Type, Name, #{connector := ConnectorConfs, direction := Direction} = Conf) +parse_confs(Type, Name, #{connector := ConnectorConfs, direction := Direction} = Conf) when is_map(ConnectorConfs) -> make_resource_confs(Direction, ConnectorConfs, - maps:without([connector, direction], Conf), Name). + maps:without([connector, direction], Conf), Type, Name). -make_resource_confs(ingress, ConnectorConfs, BridgeConf, Name) -> - BName = bin(Name), +make_resource_confs(ingress, ConnectorConfs, BridgeConf, Type, Name) -> + BName = bridge_id(Type, Name), ConnectorConfs#{ ingress => BridgeConf#{hookpoint => <<"$bridges/", BName/binary>>} }; -make_resource_confs(egress, ConnectorConfs, BridgeConf, _Name) -> +make_resource_confs(egress, ConnectorConfs, BridgeConf, _Type, _Name) -> ConnectorConfs#{ egress => BridgeConf }. diff --git a/apps/emqx_bridge/src/emqx_bridge_http_schema.erl b/apps/emqx_bridge/src/emqx_bridge_http_schema.erl index 494911d21..152289cf1 100644 --- a/apps/emqx_bridge/src/emqx_bridge_http_schema.erl +++ b/apps/emqx_bridge/src/emqx_bridge_http_schema.erl @@ -68,7 +68,6 @@ How long will the HTTP request timeout. fields("post") -> [ type_field() - , name_field() ] ++ fields("bridge"); fields("put") -> @@ -103,8 +102,5 @@ id_field() -> type_field() -> {type, mk(http, #{desc => "The Bridge Type"})}. -name_field() -> - {name, mk(binary(), #{desc => "The Bridge Name"})}. - method() -> enum([post, put, get, delete]). diff --git a/apps/emqx_bridge/src/emqx_bridge_mqtt_schema.erl b/apps/emqx_bridge/src/emqx_bridge_mqtt_schema.erl index 3de011b4c..96c9a1d38 100644 --- a/apps/emqx_bridge/src/emqx_bridge_mqtt_schema.erl +++ b/apps/emqx_bridge/src/emqx_bridge_mqtt_schema.erl @@ -24,11 +24,9 @@ fields("egress") -> fields("post_ingress") -> [ type_field() - , name_field() ] ++ proplists:delete(enable, fields("ingress")); fields("post_egress") -> [ type_field() - , name_field() ] ++ proplists:delete(enable, fields("egress")); fields("put_ingress") -> @@ -49,9 +47,3 @@ id_field() -> type_field() -> {type, mk(mqtt, #{desc => "The Bridge Type"})}. - -name_field() -> - {name, mk(binary(), - #{ desc => "The Bridge Name" - , example => "some_bridge_name" - })}. diff --git a/apps/emqx_connector/src/emqx_connector_mqtt.erl b/apps/emqx_connector/src/emqx_connector_mqtt.erl index c216e905c..beeff6d3e 100644 --- a/apps/emqx_connector/src/emqx_connector_mqtt.erl +++ b/apps/emqx_connector/src/emqx_connector_mqtt.erl @@ -68,10 +68,6 @@ fields("put") -> fields("post") -> [ {type, mk(mqtt, #{desc => "The Connector Type"})} - , {name, mk(binary(), - #{ desc => "The Connector Name" - , example => <<"my_mqtt_connector">> - })} ] ++ fields("put"). %% =================================================================== diff --git a/apps/emqx_connector/src/mqtt/emqx_connector_mqtt_mod.erl b/apps/emqx_connector/src/mqtt/emqx_connector_mqtt_mod.erl index d7abcda84..30a1ccb30 100644 --- a/apps/emqx_connector/src/mqtt/emqx_connector_mqtt_mod.erl +++ b/apps/emqx_connector/src/mqtt/emqx_connector_mqtt_mod.erl @@ -173,15 +173,27 @@ handle_publish(Msg, Vars) -> _ = erlang:apply(Mod, Func, [Msg | Args]); _ -> ok end, - case maps:get(local_topic, Vars, undefined) of - undefined -> ok; - _Topic -> - emqx_broker:publish(emqx_connector_mqtt_msg:to_broker_msg(Msg, Vars)) - end. + maybe_publish_to_local_broker(Msg, Vars). handle_disconnected(Reason, Parent) -> Parent ! {disconnected, self(), Reason}. +maybe_publish_to_local_broker(#{topic := Topic} = Msg, #{remote_topic := SubTopic} = Vars) -> + case maps:get(local_topic, Vars, undefined) of + undefined -> + %% local topic is not set, discard it + ok; + _ -> + case emqx_topic:match(Topic, SubTopic) of + true -> + _ = emqx_broker:publish(emqx_connector_mqtt_msg:to_broker_msg(Msg, Vars)), + ok; + false -> + ?SLOG(warning, #{msg => "discard_message_as_topic_not_matched", + message => Msg, subscribed => SubTopic, got_topic => Topic}) + end + end. + make_hdlr(Parent, Vars) -> #{puback => {fun ?MODULE:handle_puback/2, [Parent]}, publish => {fun ?MODULE:handle_publish/2, [Vars]}, diff --git a/apps/emqx_resource/include/emqx_resource.hrl b/apps/emqx_resource/include/emqx_resource.hrl index 08c230401..ed1de18cf 100644 --- a/apps/emqx_resource/include/emqx_resource.hrl +++ b/apps/emqx_resource/include/emqx_resource.hrl @@ -25,7 +25,7 @@ mod := module(), config := resource_config(), state := resource_state(), - status := started | stopped, + status := started | stopped | starting, metrics := emqx_plugin_libs_metrics:metrics() }. -type resource_group() :: binary(). @@ -41,3 +41,5 @@ %% the `after_query_fun()` is mainly for callbacks that increment counters or do some fallback %% actions upon query failure -type after_query_fun() :: {fun((...) -> ok), Args :: [term()]}. + +-define(TEST_ID_PREFIX, "_test_:"). diff --git a/apps/emqx_resource/src/emqx_resource.erl b/apps/emqx_resource/src/emqx_resource.erl index 37c4caa2e..5ec5fd92a 100644 --- a/apps/emqx_resource/src/emqx_resource.erl +++ b/apps/emqx_resource/src/emqx_resource.erl @@ -82,7 +82,6 @@ ]). -define(HOCON_CHECK_OPTS, #{atom_key => true, nullable => true}). - -define(DEFAULT_RESOURCE_GROUP, <<"default">>). -optional_callbacks([ on_query/4 @@ -170,7 +169,7 @@ create_dry_run(ResourceType, Config) -> -spec create_dry_run_local(resource_type(), resource_config()) -> ok | {error, Reason :: term()}. create_dry_run_local(ResourceType, Config) -> - InstId = iolist_to_binary(emqx_misc:gen_id(16)), + InstId = emqx_resource_instance:make_test_id(), call_instance(InstId, {create_dry_run, InstId, ResourceType, Config}). -spec recreate(instance_id(), resource_type(), resource_config(), term()) -> @@ -201,14 +200,18 @@ query(InstId, Request) -> -spec query(instance_id(), Request :: term(), after_query()) -> Result :: term(). query(InstId, Request, AfterQuery) -> case get_instance(InstId) of + {ok, #{status := starting}} -> + query_error(starting, <<"cannot serve query when the resource " + "instance is still starting">>); {ok, #{status := stopped}} -> - error({resource_stopped, InstId}); + query_error(stopped, <<"cannot serve query when the resource " + "instance is stopped">>); {ok, #{mod := Mod, state := ResourceState, status := started}} -> %% the resource state is readonly to Module:on_query/4 %% and the `after_query()` functions should be thread safe Mod:on_query(InstId, Request, AfterQuery, ResourceState); - {error, Reason} -> - error({get_instance, {InstId, Reason}}) + {error, not_found} -> + query_error(not_found, <<"the resource id not exists">>) end. -spec restart(instance_id()) -> ok | {error, Reason :: term()}. @@ -368,3 +371,6 @@ cluster_call(Func, Args) -> {ok, _TxnId, Result} -> Result; Failed -> Failed end. + +query_error(Reason, Msg) -> + {error, {?MODULE, #{reason => Reason, msg => Msg}}}. diff --git a/apps/emqx_resource/src/emqx_resource_instance.erl b/apps/emqx_resource/src/emqx_resource_instance.erl index 745b8b684..c413691d9 100644 --- a/apps/emqx_resource/src/emqx_resource_instance.erl +++ b/apps/emqx_resource/src/emqx_resource_instance.erl @@ -26,6 +26,7 @@ -export([ lookup/1 , get_metrics/1 , list_all/0 + , make_test_id/0 ]). -export([ hash_call/2 @@ -61,7 +62,7 @@ hash_call(InstId, Request) -> hash_call(InstId, Request, Timeout) -> gen_server:call(pick(InstId), Request, Timeout). --spec lookup(instance_id()) -> {ok, resource_data()} | {error, Reason :: term()}. +-spec lookup(instance_id()) -> {ok, resource_data()} | {error, not_found}. lookup(InstId) -> case ets:lookup(emqx_resource_instance, InstId) of [] -> {error, not_found}; @@ -69,6 +70,10 @@ lookup(InstId) -> {ok, Data#{id => InstId, metrics => get_metrics(InstId)}} end. +make_test_id() -> + RandId = iolist_to_binary(emqx_misc:gen_id(16)), + <>. + get_metrics(InstId) -> emqx_plugin_libs_metrics:get_metrics(resource_metrics, InstId). @@ -146,7 +151,7 @@ do_recreate(InstId, ResourceType, NewConfig, Params) -> {ok, #{mod := ResourceType, state := ResourceState, config := OldConfig}} -> Config = emqx_resource:call_config_merge(ResourceType, OldConfig, NewConfig, Params), - TestInstId = iolist_to_binary(emqx_misc:gen_id(16)), + TestInstId = make_test_id(), case do_create_dry_run(TestInstId, ResourceType, Config) of ok -> do_remove(ResourceType, InstId, ResourceState, false), @@ -166,7 +171,9 @@ do_create(InstId, ResourceType, Config, Opts) -> {ok, _} -> {ok, already_created}; _ -> Res0 = #{id => InstId, mod => ResourceType, config => Config, - status => stopped, state => undefined}, + status => starting, state => undefined}, + %% The `emqx_resource:call_start/3` need the instance exist beforehand + ets:insert(emqx_resource_instance, {InstId, Res0}), case emqx_resource:call_start(InstId, ResourceType, Config) of {ok, ResourceState} -> ok = emqx_plugin_libs_metrics:create_metrics(resource_metrics, InstId), @@ -181,6 +188,7 @@ do_create(InstId, ResourceType, Config, Opts) -> ets:insert(emqx_resource_instance, {InstId, Res0}), {ok, Res0}; {error, Reason} when ForceCreate == false -> + ets:delete(emqx_resource_instance, InstId), {error, Reason} end end. diff --git a/apps/emqx_resource/test/emqx_resource_SUITE.erl b/apps/emqx_resource/test/emqx_resource_SUITE.erl index 6b2e5903e..4e9c35efc 100644 --- a/apps/emqx_resource/test/emqx_resource_SUITE.erl +++ b/apps/emqx_resource/test/emqx_resource_SUITE.erl @@ -96,9 +96,7 @@ t_query(_) -> ?assert(false) end, - ?assertException( - error, - {get_instance, _Reason}, + ?assertMatch({error, {emqx_resource, #{reason := not_found}}}, emqx_resource:query(<<"unknown">>, get_state)), ok = emqx_resource:remove_local(?ID). @@ -142,7 +140,8 @@ t_stop_start(_) -> ?assertNot(is_process_alive(Pid0)), - ?assertException(error, {resource_stopped, ?ID}, emqx_resource:query(?ID, get_state)), + ?assertMatch({error, {emqx_resource, #{reason := stopped}}}, + emqx_resource:query(?ID, get_state)), ok = emqx_resource:restart(?ID), diff --git a/apps/emqx_rule_engine/src/emqx_rule_runtime.erl b/apps/emqx_rule_engine/src/emqx_rule_runtime.erl index 60a7cbaad..049829c59 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_runtime.erl +++ b/apps/emqx_rule_engine/src/emqx_rule_runtime.erl @@ -101,7 +101,7 @@ do_apply_rule(#{ true -> ok = emqx_plugin_libs_metrics:inc_matched(rule_metrics, RuleId), Collection2 = filter_collection(Input, InCase, DoEach, Collection), - {ok, [handle_output_list(Outputs, Coll, Input) || Coll <- Collection2]}; + {ok, [handle_output_list(RuleId, Outputs, Coll, Input) || Coll <- Collection2]}; false -> {error, nomatch} end; @@ -118,7 +118,7 @@ do_apply_rule(#{id := RuleId, {match_conditions_error, {_EXCLASS_,_EXCPTION_,_ST_}}) of true -> ok = emqx_plugin_libs_metrics:inc_matched(rule_metrics, RuleId), - {ok, handle_output_list(Outputs, Selected, Input)}; + {ok, handle_output_list(RuleId, Outputs, Selected, Input)}; false -> {error, nomatch} end. @@ -231,15 +231,17 @@ number(Bin) -> catch error:badarg -> binary_to_float(Bin) end. -handle_output_list(Outputs, Selected, Envs) -> - [handle_output(Out, Selected, Envs) || Out <- Outputs]. +handle_output_list(RuleId, Outputs, Selected, Envs) -> + [handle_output(RuleId, Out, Selected, Envs) || Out <- Outputs]. -handle_output(OutId, Selected, Envs) -> +handle_output(RuleId, OutId, Selected, Envs) -> try do_handle_output(OutId, Selected, Envs) catch Err:Reason:ST -> - ?SLOG(error, #{msg => "output_failed", + ok = emqx_plugin_libs_metrics:inc_failed(rule_metrics, RuleId), + Level = case Err of throw -> debug; _ -> error end, + ?SLOG(Level, #{msg => "output_failed", output => OutId, exception => Err, reason => Reason,