Add cluster support for rules and resource API

src/emqx_rule_engine_api.erl

@@ -360,18 +360,11 @@ record_to_map(#rule{id = Id,
                     actions = Actions,
                     enabled = Enabled,
                     description = Descr}) ->
-    #{max := Max, current := Current, last5m := Last5M} = emqx_rule_metrics:get_rule_speed(Id),
-    Metrics = #{
-        matched => emqx_rule_metrics:get(Id, 'rules.matched'),
-        speed => Current,
-        speed_max => Max,
-        speed_last5m => Last5M
-    },
     #{id => Id,
       for => Hook,
       rawsql => RawSQL,
       actions => printable_actions(Actions),
-      metrics => Metrics,
+      metrics => get_rule_metrics(Id),
       enabled => Enabled,
       description => Descr
      };
@@ -398,7 +391,11 @@ record_to_map(#resource{id = Id,
                         type = Type,
                         config = Config,
                         description = Descr}) ->
-    {ok, Status} = emqx_rule_engine:get_resource_status(Id),
+    Status =
+        [begin
+            {ok, St} = rpc:call(Node, emqx_rule_engine, get_resource_status, [Id]),
+            maps:put(node, Node, St)
+        end || Node <- [node()| nodes()]],
     #{id => Id,
       type => Type,
       config => Config,
@@ -419,13 +416,8 @@ record_to_map(#resource_type{name = Name,
      }.
 
 printable_actions(Actions) ->
-    [begin
-        Metrics = #{
-            success => emqx_rule_metrics:get(Id, 'actions.success'),
-            failed => emqx_rule_metrics:get(Id, 'actions.failure')
-        },
-        #{name => Name, params => Args, metrics => Metrics}
-     end || #action_instance{id = Id, name = Name, args = Args} <- Actions].
+    [#{id => Id, name => Name, params => Args, metrics => get_action_metrics(Id)}
+     || #action_instance{id = Id, name = Name, args = Args} <- Actions].
 
 parse_rule_params(Params) ->
     parse_rule_params(Params, #{description => <<"">>}).
@@ -558,6 +550,14 @@ sort_spec(Spec) when is_list(Spec) ->
                 _ -> {Key, Spec0}
               end || {Key, Spec0} <- Spec]).
 
+get_rule_metrics(Id) ->
+    [maps:put(node, Node, rpc:call(Node, emqx_rule_metrics, get_rule_metrics, [Id]))
+     || Node <- [node()| nodes()]].
+
+get_action_metrics(Id) ->
+    [maps:put(node, Node, rpc:call(Node, emqx_rule_metrics, get_action_metrics, [Id]))
+     || Node <- [node()| nodes()]].
+
 %% TEST
 -ifdef(EUNIT).
 -include_lib("eunit/include/eunit.hrl").

src/emqx_rule_engine_cli.erl

@@ -200,7 +200,7 @@ resource_types(_usage) ->
 %%------------------------------------------------------------------------------
 
 print(Data) ->
-    emqx_cli:print(format(Data)).
+    emqx_cli:print(untilde(format(Data))).
 
 print_all(DataList) ->
     lists:map(fun(Data) ->
@@ -221,14 +221,7 @@ format(#rule{id = Id,
              actions = Actions,
              enabled = Enabled,
              description = Descr}) ->
-    #{max := Max, current := Current, last5m := Last5M} = emqx_rule_metrics:get_rule_speed(Id),
-    Metrics = #{
-        matched => emqx_rule_metrics:get(Id, 'rules.matched'),
-        speed => Current,
-        speed_max => Max,
-        speed_last5m => Last5M
-    },
-    lists:flatten(io_lib:format("rule(id='~s', for='~0p', rawsql='~s', actions=~s, metrics=~0p, enabled='~s', description='~s')~n", [Id, Hook, Sql, printable_actions(Actions), Metrics, Enabled, Descr]));
+    lists:flatten(io_lib:format("rule(id='~s', for='~0p', rawsql='~s', actions=~0p, metrics=~0p, enabled='~s', description='~s')~n", [Id, Hook, Sql, printable_actions(Actions), get_rule_metrics(Id), Enabled, Descr]));
 
 format(#action{hidden = true}) ->
     ok;
@@ -266,13 +259,8 @@ make_resource(Opts) ->
       description => get_value(descr, Opts)}.
 
 printable_actions(Actions) when is_list(Actions) ->
-    jsx:encode([begin
-                    Metrics = #{
-                        success => emqx_rule_metrics:get(Id, 'actions.success'),
-                        failed => emqx_rule_metrics:get(Id, 'actions.failure')
-                    },
-                    #{name => Name, params => Args, metrics => Metrics}
-                end || #action_instance{id = Id, name = Name, args = Args} <- Actions]).
+    jsx:encode([#{id => Id, name => Name, params => Args, metrics => get_action_metrics(Id)}
+                || #action_instance{id = Id, name = Name, args = Args} <- Actions]).
 
 with_opts(Action, RawParams, OptSpecList, {CmdObject, CmdName}) ->
     case getopt:parse_and_check(OptSpecList, RawParams) of
@@ -300,3 +288,13 @@ get_actions(undefined) ->
 get_actions(Hook) ->
     emqx_rule_registry:get_actions_for(Hook).
 
+get_rule_metrics(Id) ->
+    [maps:put(node, Node, rpc:call(Node, emqx_rule_metrics, get_rule_metrics, [Id]))
+     || Node <- [node()| nodes()]].
+
+get_action_metrics(Id) ->
+    [maps:put(node, Node, rpc:call(Node, emqx_rule_metrics, get_action_metrics, [Id]))
+     || Node <- [node()| nodes()]].
+
+untilde(Str) ->
+    re:replace(Str,"~","&&",[{return,list}, global]).

src/emqx_rule_metrics.erl

@@ -34,6 +34,10 @@
         , overall_metrics/0
         ]).
 
+-export([ get_rule_metrics/1
+        , get_action_metrics/1
+        ]).
+
 %% gen_server callbacks
 -export([ init/1
         , handle_call/3
@@ -106,6 +110,21 @@ get_rule_speed(Id) ->
 get_overall_rule_speed() ->
     gen_server:call(?MODULE, get_overall_rule_speed).
 
+-spec(get_rule_metrics(rule_id()) -> map()).
+get_rule_metrics(Id) ->
+    #{max := Max, current := Current, last5m := Last5M} = get_rule_speed(Id),
+    #{matched => get(Id, 'rules.matched'),
+      speed => Current,
+      speed_max => Max,
+      speed_last5m => Last5M
+    }.
+
+-spec(get_action_metrics(action_instance_id()) -> map()).
+get_action_metrics(Id) ->
+    #{success => get(Id, 'actions.success'),
+      failed => get(Id, 'actions.failure')
+     }.
+
 -spec(inc(rule_id(), atom()) -> ok).
 inc(Id, Metric) ->
     inc(Id, Metric, 1).