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

fix: discard rpc call failed data

This commit is contained in:
EMQ-YangM 2022-04-08 10:12:40 +08:00
parent 16dc0d6555
commit 96b26bf4ce
2 changed files with 38 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
apps/emqx_rule_engine/src/emqx_rule_engine_api.erl
View File

@ -323,13 +323,13 @@ show_resource(#{id := Id}, _Params) ->
case emqx_rule_registry:find_resource(Id) of
{ok, R} ->
Status =
[begin
St = case rpc:call(Node, emqx_rule_engine, get_resource_status, [Id]) of
{ok, St0} -> St0;
{error, _} -> #{is_alive => false}
end,
maps:put(node, Node, St)
end || Node <- ekka_mnesia:running_nodes()],
lists:flatten(
[ case rpc:call(Node, emqx_rule_engine, get_resource_status, [Id]) of
{badrpc, _} -> [];
{ok, St} -> [maps:put(node, Node, St)];
{error, _} -> [maps:put(node, Node, #{is_alive => false})]
end
|| Node <- ekka_mnesia:running_nodes()]),
return({ok, maps:put(status, Status, record_to_map(R))});
not_found ->
return({error, 404, <<"Not Found">>})
@ -575,9 +575,17 @@ sort_by(Pos, TplList) ->
end, TplList).
get_rule_metrics(Id) ->
[maps:put(node, Node, rpc:call(Node, emqx_rule_metrics, get_rule_metrics, [Id]))
|| Node <- ekka_mnesia:running_nodes()].
lists:flatten(
[ case rpc:call(Node, emqx_rule_metrics, get_rule_metrics, [Id]) of
{badrpc, _} -> [];
Res -> [maps:put(node, Node, Res)]
end
|| Node <- ekka_mnesia:running_nodes()]).
get_action_metrics(Id) ->
[maps:put(node, Node, rpc:call(Node, emqx_rule_metrics, get_action_metrics, [Id]))
|| Node <- ekka_mnesia:running_nodes()].
lists:flatten(
[ case rpc:call(Node, emqx_rule_metrics, get_action_metrics, [Id]) of
{badrpc, _} -> [];
Res -> [maps:put(node, Node, Res)]
end
|| Node <- ekka_mnesia:running_nodes()]).

27
apps/emqx_rule_engine/src/emqx_rule_engine_cli.erl
View File

@ -273,10 +273,13 @@ format(#resource{id = Id,
config = Config,
description = Descr}) ->
Status =
[begin
{ok, St} = rpc:call(Node, emqx_rule_engine, get_resource_status, [Id]),
maps:put(node, Node, St)
end || Node <- [node()| nodes()]],
lists:flatten(
[ case rpc:call(Node, emqx_rule_engine, get_resource_status, [Id]) of
{badrpc, _} -> [];
{ok, St} -> [maps:put(node, Node, St)];
{error, _} -> [maps:put(node, Node, #{is_alive => false})]
end
|| Node <- ekka_mnesia:running_nodes()]),
lists:flatten(io_lib:format("resource(id='~s', type='~s', config=~0p, status=~0p, description='~s')~n", [Id, Type, Config, Status, Descr]));
format(#resource_type{name = Name,
@ -369,12 +372,20 @@ get_actions() ->
emqx_rule_registry:get_actions().
get_rule_metrics(Id) ->
[maps:put(node, Node, rpc:call(Node, emqx_rule_metrics, get_rule_metrics, [Id]))
|| Node <- [node()| nodes()]].
lists:flatten(
[ case rpc:call(Node, emqx_rule_metrics, get_rule_metrics, [Id]) of
{badrpc, _} -> [];
Res -> [maps:put(node, Node, Res)]
end
|| Node <- ekka_mnesia:running_nodes()]).
get_action_metrics(Id) ->
[maps:put(node, Node, rpc:call(Node, emqx_rule_metrics, get_action_metrics, [Id]))
|| Node <- [node()| nodes()]].
lists:flatten(
[ case rpc:call(Node, emqx_rule_metrics, get_action_metrics, [Id]) of
{badrpc, _} -> [];
Res -> [maps:put(node, Node, Res)]
end
|| Node <- ekka_mnesia:running_nodes()]).
on_failed(continue) -> continue;
on_failed(stop) -> stop;