Merge pull request #6597 from terry-xiaoyu/bridge_bug_fixes_2

Bridge bug fixes 2

apps/emqx_authn/test/emqx_authn_http_SUITE.erl

@@ -153,9 +153,8 @@ t_destroy(_Config) ->
       ?GLOBAL),
 
     % Authenticator should not be usable anymore
-    ?assertException(
+    ?assertMatch(
-       error,
+       ignore,
-       _,
        emqx_authn_http:authenticate(
          Credentials,
          State)).

apps/emqx_authn/test/emqx_authn_mongo_SUITE.erl

@@ -146,9 +146,8 @@ t_destroy(_Config) ->
       ?GLOBAL),
 
     % Authenticator should not be usable anymore
-    ?assertException(
+    ?assertMatch(
-       error,
+       ignore,
-       _,
        emqx_authn_mongodb:authenticate(
          #{username => <<"plain">>,
            password => <<"plain">>

apps/emqx_authn/test/emqx_authn_mysql_SUITE.erl

@@ -159,9 +159,8 @@ t_destroy(_Config) ->
       ?GLOBAL),
 
     % Authenticator should not be usable anymore
-    ?assertException(
+    ?assertMatch(
-       error,
+       ignore,
-       _,
        emqx_authn_mysql:authenticate(
          #{username => <<"plain">>,
            password => <<"plain">>

apps/emqx_authn/test/emqx_authn_pgsql_SUITE.erl

@@ -159,9 +159,8 @@ t_destroy(_Config) ->
       ?GLOBAL),
 
     % Authenticator should not be usable anymore
-    ?assertException(
+    ?assertMatch(
-       error,
+       ignore,
-       _,
        emqx_authn_pgsql:authenticate(
          #{username => <<"plain">>,
            password => <<"plain">>

apps/emqx_authn/test/emqx_authn_redis_SUITE.erl

@@ -164,9 +164,8 @@ t_destroy(_Config) ->
       ?GLOBAL),
 
     % Authenticator should not be usable anymore
-    ?assertException(
+    ?assertMatch(
-       error,
+       ignore,
-       _,
        emqx_authn_redis:authenticate(
          #{username => <<"plain">>,
            password => <<"plain">>

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) ->
+            Msg = emqx_rule_events:eventmsg_publish(Message),
-                    send_message(Id, emqx_rule_events:eventmsg_publish(Message))
+            send_to_egress_matched_bridges(Topic, Msg);
-                end, get_matched_bridges(Topic));
         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) ->
+make_resource_confs(ingress, ConnectorConfs, BridgeConf, Type, Name) ->
-    BName = bin(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
     }.

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]).

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"
-         })}.

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").
 
 %% ===================================================================

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
+    maybe_publish_to_local_broker(Msg, Vars).
-        undefined -> ok;
-        _Topic ->
-            emqx_broker:publish(emqx_connector_mqtt_msg:to_broker_msg(Msg, Vars))
-    end.
 
 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]},

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_:").

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, not_found} ->
-            error({get_instance, {InstId, Reason}})
+            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}}}.

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)),
+    <<?TEST_ID_PREFIX, RandId/binary>>.
+
 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.

apps/emqx_resource/test/emqx_resource_SUITE.erl

@@ -96,9 +96,7 @@ t_query(_) ->
         ?assert(false)
     end,
 
-    ?assertException(
+    ?assertMatch({error, {emqx_resource, #{reason := not_found}}},
-       error,
-       {get_instance, _Reason},
        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),
 

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_list(RuleId, Outputs, Selected, Envs) ->
-    [handle_output(Out, Selected, Envs) || Out <- Outputs].
+    [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,