Merge pull request #9125 from emqx/cannot_reset_metrics_for_fallback_actions

Cannot reset metrics for fallback actions

CHANGES-4.3.md

@@ -71,6 +71,8 @@ File format:
   subscriber from another node in the cluster.
   Fixed in [#9122](https://github.com/emqx/emqx/pull/9122)
 
+- Fix cannot reset metrics for fallback actions. [#9125](https://github.com/emqx/emqx/pull/9125)
+
 ## v4.3.20
 
 ### Bug fixes

apps/emqx_rule_engine/src/emqx_rule_engine.appup.src

@@ -5,12 +5,14 @@
     [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]}]},
    {"4.3.14",
     [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]}]},
    {"4.3.13",
     [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
@@ -19,6 +21,7 @@
      {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
    {"4.3.12",
     [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
@@ -27,6 +30,7 @@
      {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
    {"4.3.11",
     [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
@@ -36,6 +40,7 @@
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_validator,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]}]},
    {"4.3.10",
     [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
@@ -45,6 +50,7 @@
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]}]},
    {"4.3.9",
     [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
@@ -56,6 +62,7 @@
      {load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
      {add_module,emqx_rule_date},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
@@ -202,12 +209,14 @@
     [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]}]},
    {"4.3.14",
     [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]}]},
    {"4.3.13",
     [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
@@ -216,6 +225,7 @@
      {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
    {"4.3.12",
     [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
@@ -224,6 +234,7 @@
      {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
    {"4.3.11",
     [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
@@ -233,6 +244,7 @@
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_validator,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]}]},
    {"4.3.10",
     [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
@@ -242,6 +254,7 @@
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]}]},
    {"4.3.9",
     [{load_module,emqx_rule_sqltester,brutal_purge,soft_purge,[]},
@@ -254,6 +267,7 @@
      {load_module,emqx_rule_events,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
      {delete_module,emqx_rule_date}]},
    {"4.3.8",

apps/emqx_rule_engine/src/emqx_rule_metrics.erl

@@ -131,16 +131,15 @@ clear_metrics(Id) ->
 -spec(reset_metrics(rule_id()) -> ok).
 reset_metrics(Id) ->
     reset_speeds(Id),
-    reset_metrics(Id, rule_metrics()),
+    do_reset_metrics(Id, rule_metrics()),
     case emqx_rule_registry:get_rule(Id) of
         not_found -> ok;
         {ok, #rule{actions = Actions}} ->
-            [ reset_metrics(ActionId, action_metrics())
+            reset_action_metrics(Actions),
-              || #action_instance{ id = ActionId} <- Actions],
             ok
     end.
 
-reset_metrics(Id, Metrics) ->
+do_reset_metrics(Id, Metrics) ->
     case couters_ref(Id) of
         not_found -> ok;
         Ref -> [counters:put(Ref, metrics_idx(Idx), 0)
@@ -148,6 +147,12 @@ reset_metrics(Id, Metrics) ->
                ok
     end.
 
+reset_action_metrics(Actions) ->
+    lists:foreach(fun(#action_instance{id = ActionId, fallbacks = FallbackActions}) ->
+            do_reset_metrics(ActionId, action_metrics()),
+            reset_action_metrics(FallbackActions)
+        end, Actions).
+
 reset_speeds(Id) ->
     gen_server:call(?MODULE, {reset_speeds, Id}).
 

apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl

@@ -50,6 +50,7 @@ groups() ->
        t_unregister_provider,
        t_create_rule,
        t_reset_metrics,
+       t_reset_metrics_fallbacks,
        t_create_resource
       ]},
      {actions, [],
@@ -379,15 +380,11 @@ t_inspect_action(_Config) ->
 
 t_reset_metrics(_Config) ->
     ok = emqx_rule_engine:load_providers(),
-    {ok, #resource{id = ResId}} = emqx_rule_engine:create_resource(
+    {ok, #rule{id = Id, actions = [#action_instance{id = ActId0}]}} =
-                                    #{type => built_in,
+        emqx_rule_engine:create_rule(
-                                      config => #{},
-                                      description => <<"debug resource">>}),
-    {ok, #rule{id = Id}} = emqx_rule_engine:create_rule(
             #{rawsql => "select clientid as c, username as u "
             "from \"t1\" ",
-                               actions => [#{name => 'inspect',
+            actions => [#{name => 'inspect', args => #{a=>1, b=>2}}],
-                                             args => #{'$resource' => ResId, a=>1, b=>2}}],
             type => built_in,
             description => <<"Inspect rule">>
             }),
@@ -398,16 +395,68 @@ t_reset_metrics(_Config) ->
           timer:sleep(100)
       end
       || _ <- lists:seq(1,10)],
+    ?assertMatch(#{exception := 0, failed := 0,
+                   matched := 10, no_result := 0, passed := 10},
+                 emqx_rule_metrics:get_rule_metrics(Id)),
+    ?assertMatch(#{failed := 0, success := 10, taken := 10},
+                   emqx_rule_metrics:get_action_metrics(ActId0)),
     emqx_rule_metrics:reset_metrics(Id),
     ?assertEqual(#{exception => 0,failed => 0,
                    matched => 0,no_result => 0,passed => 0,
                    speed => 0.0,speed_last5m => 0.0,speed_max => 0.0},
                  emqx_rule_metrics:get_rule_metrics(Id)),
     ?assertEqual(#{failed => 0, success => 0, taken => 0},
-                   emqx_rule_metrics:get_action_metrics(ResId)),
+                   emqx_rule_metrics:get_action_metrics(ActId0)),
     emqtt:stop(Client),
     emqx_rule_registry:remove_rule(Id),
-    emqx_rule_registry:remove_resource(ResId),
+    ok.
+
+t_reset_metrics_fallbacks(_Config) ->
+    ok = emqx_rule_engine:load_providers(),
+    ok = emqx_rule_registry:add_action(
+            #action{name = 'crash_action', app = ?APP,
+                    module = ?MODULE, on_create = crash_action,
+                    types=[], params_spec = #{},
+                    title = #{en => <<"Crash Action">>},
+                    description = #{en => <<"This action will always fail!">>}}),
+    {ok, #rule{id = Id, actions = [#action_instance{id = ActId0, fallbacks = [
+            #action_instance{id = ActId1},
+            #action_instance{id = ActId2}
+        ]}]}} =
+        emqx_rule_engine:create_rule(
+            #{rawsql => "select clientid as c, username as u "
+            "from \"t1\" ",
+            actions => [#{name => 'crash_action', args => #{a=>1, b=>2}, fallbacks => [
+                              #{name => 'inspect', args => #{}, fallbacks => []},
+                              #{name => 'inspect', args => #{}, fallbacks => []}
+                          ]}],
+            type => built_in,
+            description => <<"Inspect rule">>
+            }),
+    {ok, Client} = emqtt:start_link([{username, <<"emqx">>}]),
+    {ok, _} = emqtt:connect(Client),
+    [ begin
+          emqtt:publish(Client, <<"t1">>, <<"{\"id\": 1, \"name\": \"ha\"}">>, 0),
+          timer:sleep(100)
+      end
+      || _ <- lists:seq(1,10)],
+    ?assertMatch(#{exception := 0, failed := 0,
+                   matched := 10, no_result := 0, passed := 10},
+                 emqx_rule_metrics:get_rule_metrics(Id)),
+    [?assertMatch(#{failed := 10, success := 0, taken := 10},
+                   emqx_rule_metrics:get_action_metrics(AId)) || AId <- [ActId0]],
+    [?assertMatch(#{failed := 0, success := 10, taken := 10},
+                   emqx_rule_metrics:get_action_metrics(AId)) || AId <- [ ActId1, ActId2]],
+    emqx_rule_metrics:reset_metrics(Id),
+    ?assertEqual(#{exception => 0,failed => 0,
+                   matched => 0,no_result => 0,passed => 0,
+                   speed => 0.0,speed_last5m => 0.0,speed_max => 0.0},
+                 emqx_rule_metrics:get_rule_metrics(Id)),
+    [?assertEqual(#{failed => 0, success => 0, taken => 0},
+                   emqx_rule_metrics:get_action_metrics(AId)) || AId <- [ActId0, ActId1, ActId2]],
+    emqtt:stop(Client),
+    emqx_rule_registry:remove_rule(Id),
+    ok = emqx_rule_registry:remove_action('crash_action'),
     ok.
 
 t_republish_action(_Config) ->