602 lines
19 KiB
Erlang
602 lines
19 KiB
Erlang
%%--------------------------------------------------------------------
|
|
%% Copyright (c) 2020-2022 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
%%
|
|
%% Licensed under the Apache License, Version 2.0 (the "License");
|
|
%% you may not use this file except in compliance with the License.
|
|
%% You may obtain a copy of the License at
|
|
%%
|
|
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
%%
|
|
%% Unless required by applicable law or agreed to in writing, software
|
|
%% distributed under the License is distributed on an "AS IS" BASIS,
|
|
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
%% See the License for the specific language governing permissions and
|
|
%% limitations under the License.
|
|
%%--------------------------------------------------------------------
|
|
|
|
-module(emqx_rule_engine_api).
|
|
|
|
-include("rule_engine.hrl").
|
|
-include_lib("emqx/include/logger.hrl").
|
|
-include_lib("hocon/include/hoconsc.hrl").
|
|
-include_lib("typerefl/include/types.hrl").
|
|
|
|
-behaviour(minirest_api).
|
|
|
|
-import(hoconsc, [mk/2, ref/2, array/1]).
|
|
|
|
-export([printable_function_name/2]).
|
|
|
|
%% Swagger specs from hocon schema
|
|
-export([api_spec/0, paths/0, schema/1, namespace/0]).
|
|
|
|
%% API callbacks
|
|
-export(['/rule_events'/2, '/rule_test'/2, '/rules'/2, '/rules/:id'/2, '/rules/:id/reset_metrics'/2]).
|
|
|
|
%% query callback
|
|
-export([query/4]).
|
|
|
|
-define(ERR_NO_RULE(ID), list_to_binary(io_lib:format("Rule ~ts Not Found", [(ID)]))).
|
|
-define(ERR_BADARGS(REASON), begin
|
|
R0 = err_msg(REASON),
|
|
<<"Bad Arguments: ", R0/binary>>
|
|
end).
|
|
-define(CHECK_PARAMS(PARAMS, TAG, EXPR),
|
|
case emqx_rule_api_schema:check_params(PARAMS, TAG) of
|
|
{ok, CheckedParams} ->
|
|
EXPR;
|
|
{error, REASON} ->
|
|
{400, #{code => 'BAD_REQUEST', message => ?ERR_BADARGS(REASON)}}
|
|
end
|
|
).
|
|
-define(METRICS(
|
|
MATCH,
|
|
PASS,
|
|
FAIL,
|
|
FAIL_EX,
|
|
FAIL_NORES,
|
|
O_TOTAL,
|
|
O_FAIL,
|
|
O_FAIL_OOS,
|
|
O_FAIL_UNKNOWN,
|
|
O_SUCC,
|
|
RATE,
|
|
RATE_MAX,
|
|
RATE_5
|
|
),
|
|
#{
|
|
'matched' => MATCH,
|
|
'passed' => PASS,
|
|
'failed' => FAIL,
|
|
'failed.exception' => FAIL_EX,
|
|
'failed.no_result' => FAIL_NORES,
|
|
'actions.total' => O_TOTAL,
|
|
'actions.failed' => O_FAIL,
|
|
'actions.failed.out_of_service' => O_FAIL_OOS,
|
|
'actions.failed.unknown' => O_FAIL_UNKNOWN,
|
|
'actions.success' => O_SUCC,
|
|
'matched.rate' => RATE,
|
|
'matched.rate.max' => RATE_MAX,
|
|
'matched.rate.last5m' => RATE_5
|
|
}
|
|
).
|
|
-define(metrics(
|
|
MATCH,
|
|
PASS,
|
|
FAIL,
|
|
FAIL_EX,
|
|
FAIL_NORES,
|
|
O_TOTAL,
|
|
O_FAIL,
|
|
O_FAIL_OOS,
|
|
O_FAIL_UNKNOWN,
|
|
O_SUCC,
|
|
RATE,
|
|
RATE_MAX,
|
|
RATE_5
|
|
),
|
|
#{
|
|
'matched' := MATCH,
|
|
'passed' := PASS,
|
|
'failed' := FAIL,
|
|
'failed.exception' := FAIL_EX,
|
|
'failed.no_result' := FAIL_NORES,
|
|
'actions.total' := O_TOTAL,
|
|
'actions.failed' := O_FAIL,
|
|
'actions.failed.out_of_service' := O_FAIL_OOS,
|
|
'actions.failed.unknown' := O_FAIL_UNKNOWN,
|
|
'actions.success' := O_SUCC,
|
|
'matched.rate' := RATE,
|
|
'matched.rate.max' := RATE_MAX,
|
|
'matched.rate.last5m' := RATE_5
|
|
}
|
|
).
|
|
|
|
-define(RULE_QS_SCHEMA, [
|
|
{<<"enable">>, atom},
|
|
{<<"from">>, binary},
|
|
{<<"like_id">>, binary},
|
|
{<<"like_from">>, binary},
|
|
{<<"match_from">>, binary},
|
|
{<<"like_description">>, binary}
|
|
]).
|
|
|
|
namespace() -> "rule".
|
|
|
|
api_spec() ->
|
|
emqx_dashboard_swagger:spec(?MODULE, #{check_schema => false}).
|
|
|
|
paths() -> ["/rule_events", "/rule_test", "/rules", "/rules/:id", "/rules/:id/reset_metrics"].
|
|
|
|
error_schema(Code, Message) when is_atom(Code) ->
|
|
emqx_dashboard_swagger:error_codes([Code], list_to_binary(Message)).
|
|
|
|
rule_creation_schema() ->
|
|
ref(emqx_rule_api_schema, "rule_creation").
|
|
|
|
rule_test_schema() ->
|
|
ref(emqx_rule_api_schema, "rule_test").
|
|
|
|
rule_info_schema() ->
|
|
ref(emqx_rule_api_schema, "rule_info").
|
|
|
|
schema("/rules") ->
|
|
#{
|
|
'operationId' => '/rules',
|
|
get => #{
|
|
tags => [<<"rules">>],
|
|
description => ?DESC("api1"),
|
|
parameters => [
|
|
{enable,
|
|
mk(boolean(), #{desc => ?DESC("api1_enable"), in => query, required => false})},
|
|
{from, mk(binary(), #{desc => ?DESC("api1_from"), in => query, required => false})},
|
|
{like_id,
|
|
mk(binary(), #{desc => ?DESC("api1_like_id"), in => query, required => false})},
|
|
{like_from,
|
|
mk(binary(), #{desc => ?DESC("api1_like_from"), in => query, required => false})},
|
|
{like_description,
|
|
mk(binary(), #{
|
|
desc => ?DESC("api1_like_description"), in => query, required => false
|
|
})},
|
|
{match_from,
|
|
mk(binary(), #{desc => ?DESC("api1_match_from"), in => query, required => false})},
|
|
ref(emqx_dashboard_swagger, page),
|
|
ref(emqx_dashboard_swagger, limit)
|
|
],
|
|
summary => <<"List Rules">>,
|
|
responses => #{
|
|
200 =>
|
|
[
|
|
{data, mk(array(rule_info_schema()), #{desc => ?DESC("desc9")})},
|
|
{meta, mk(ref(emqx_dashboard_swagger, meta), #{})}
|
|
],
|
|
400 => error_schema('BAD_REQUEST', "Invalid Parameters")
|
|
}
|
|
},
|
|
post => #{
|
|
tags => [<<"rules">>],
|
|
description => ?DESC("api2"),
|
|
summary => <<"Create a Rule">>,
|
|
'requestBody' => rule_creation_schema(),
|
|
responses => #{
|
|
400 => error_schema('BAD_REQUEST', "Invalid Parameters"),
|
|
201 => rule_info_schema()
|
|
}
|
|
}
|
|
};
|
|
schema("/rule_events") ->
|
|
#{
|
|
'operationId' => '/rule_events',
|
|
get => #{
|
|
tags => [<<"rules">>],
|
|
description => ?DESC("api3"),
|
|
summary => <<"List Events">>,
|
|
responses => #{
|
|
200 => mk(ref(emqx_rule_api_schema, "rule_events"), #{})
|
|
}
|
|
}
|
|
};
|
|
schema("/rules/:id") ->
|
|
#{
|
|
'operationId' => '/rules/:id',
|
|
get => #{
|
|
tags => [<<"rules">>],
|
|
description => ?DESC("api4"),
|
|
summary => <<"Get a Rule">>,
|
|
parameters => param_path_id(),
|
|
responses => #{
|
|
404 => error_schema('NOT_FOUND', "Rule not found"),
|
|
200 => rule_info_schema()
|
|
}
|
|
},
|
|
put => #{
|
|
tags => [<<"rules">>],
|
|
description => ?DESC("api5"),
|
|
summary => <<"Update a Rule">>,
|
|
parameters => param_path_id(),
|
|
'requestBody' => rule_creation_schema(),
|
|
responses => #{
|
|
400 => error_schema('BAD_REQUEST', "Invalid Parameters"),
|
|
200 => rule_info_schema()
|
|
}
|
|
},
|
|
delete => #{
|
|
tags => [<<"rules">>],
|
|
description => ?DESC("api6"),
|
|
summary => <<"Delete a Rule">>,
|
|
parameters => param_path_id(),
|
|
responses => #{
|
|
204 => <<"Delete rule successfully">>
|
|
}
|
|
}
|
|
};
|
|
schema("/rules/:id/reset_metrics") ->
|
|
#{
|
|
'operationId' => '/rules/:id/reset_metrics',
|
|
put => #{
|
|
tags => [<<"rules">>],
|
|
description => ?DESC("api7"),
|
|
summary => <<"Reset a Rule Metrics">>,
|
|
parameters => param_path_id(),
|
|
responses => #{
|
|
400 => error_schema('BAD_REQUEST', "RPC Call Failed"),
|
|
200 => <<"Reset Success">>
|
|
}
|
|
}
|
|
};
|
|
schema("/rule_test") ->
|
|
#{
|
|
'operationId' => '/rule_test',
|
|
post => #{
|
|
tags => [<<"rules">>],
|
|
description => ?DESC("api8"),
|
|
summary => <<"Test a Rule">>,
|
|
'requestBody' => rule_test_schema(),
|
|
responses => #{
|
|
400 => error_schema('BAD_REQUEST', "Invalid Parameters"),
|
|
412 => error_schema('NOT_MATCH', "SQL Not Match"),
|
|
200 => <<"Rule Test Pass">>
|
|
}
|
|
}
|
|
}.
|
|
|
|
param_path_id() ->
|
|
[{id, mk(binary(), #{in => path, example => <<"my_rule_id">>})}].
|
|
|
|
%%------------------------------------------------------------------------------
|
|
%% Rules API
|
|
%%------------------------------------------------------------------------------
|
|
|
|
'/rule_events'(get, _Params) ->
|
|
{200, emqx_rule_events:event_info()}.
|
|
|
|
'/rules'(get, #{query_string := QueryString}) ->
|
|
case
|
|
emqx_mgmt_api:node_query(
|
|
node(),
|
|
QueryString,
|
|
?RULE_TAB,
|
|
?RULE_QS_SCHEMA,
|
|
{?MODULE, query}
|
|
)
|
|
of
|
|
{error, page_limit_invalid} ->
|
|
{400, #{code => 'BAD_REQUEST', message => <<"page_limit_invalid">>}};
|
|
Result ->
|
|
{200, Result}
|
|
end;
|
|
'/rules'(post, #{body := Params0}) ->
|
|
case maps:get(<<"id">>, Params0, list_to_binary(emqx_misc:gen_id(8))) of
|
|
<<>> ->
|
|
{400, #{code => 'BAD_REQUEST', message => <<"empty rule id is not allowed">>}};
|
|
Id ->
|
|
Params = filter_out_request_body(add_metadata(Params0)),
|
|
ConfPath = emqx_rule_engine:config_key_path() ++ [Id],
|
|
case emqx_rule_engine:get_rule(Id) of
|
|
{ok, _Rule} ->
|
|
{400, #{code => 'BAD_REQUEST', message => <<"rule id already exists">>}};
|
|
not_found ->
|
|
case emqx_conf:update(ConfPath, Params, #{override_to => cluster}) of
|
|
{ok, #{post_config_update := #{emqx_rule_engine := AllRules}}} ->
|
|
[Rule] = get_one_rule(AllRules, Id),
|
|
{201, format_rule_resp(Rule)};
|
|
{error, Reason} ->
|
|
?SLOG(error, #{
|
|
msg => "create_rule_failed",
|
|
id => Id,
|
|
reason => Reason
|
|
}),
|
|
{400, #{code => 'BAD_REQUEST', message => ?ERR_BADARGS(Reason)}}
|
|
end
|
|
end
|
|
end.
|
|
|
|
'/rule_test'(post, #{body := Params}) ->
|
|
?CHECK_PARAMS(
|
|
Params,
|
|
rule_test,
|
|
case emqx_rule_sqltester:test(CheckedParams) of
|
|
{ok, Result} ->
|
|
{200, Result};
|
|
{error, {parse_error, Reason}} ->
|
|
{400, #{code => 'BAD_REQUEST', message => err_msg(Reason)}};
|
|
{error, nomatch} ->
|
|
{412, #{code => 'NOT_MATCH', message => <<"SQL Not Match">>}};
|
|
{error, Reason} ->
|
|
{400, #{code => 'BAD_REQUEST', message => err_msg(Reason)}}
|
|
end
|
|
).
|
|
|
|
'/rules/:id'(get, #{bindings := #{id := Id}}) ->
|
|
case emqx_rule_engine:get_rule(Id) of
|
|
{ok, Rule} ->
|
|
{200, format_rule_resp(Rule)};
|
|
not_found ->
|
|
{404, #{code => 'NOT_FOUND', message => <<"Rule Id Not Found">>}}
|
|
end;
|
|
'/rules/:id'(put, #{bindings := #{id := Id}, body := Params0}) ->
|
|
Params = filter_out_request_body(Params0),
|
|
ConfPath = emqx_rule_engine:config_key_path() ++ [Id],
|
|
case emqx_conf:update(ConfPath, Params, #{override_to => cluster}) of
|
|
{ok, #{post_config_update := #{emqx_rule_engine := AllRules}}} ->
|
|
[Rule] = get_one_rule(AllRules, Id),
|
|
{200, format_rule_resp(Rule)};
|
|
{error, Reason} ->
|
|
?SLOG(error, #{
|
|
msg => "update_rule_failed",
|
|
id => Id,
|
|
reason => Reason
|
|
}),
|
|
{400, #{code => 'BAD_REQUEST', message => ?ERR_BADARGS(Reason)}}
|
|
end;
|
|
'/rules/:id'(delete, #{bindings := #{id := Id}}) ->
|
|
ConfPath = emqx_rule_engine:config_key_path() ++ [Id],
|
|
case emqx_conf:remove(ConfPath, #{override_to => cluster}) of
|
|
{ok, _} ->
|
|
{204};
|
|
{error, Reason} ->
|
|
?SLOG(error, #{
|
|
msg => "delete_rule_failed",
|
|
id => Id,
|
|
reason => Reason
|
|
}),
|
|
{500, #{code => 'INTERNAL_ERROR', message => ?ERR_BADARGS(Reason)}}
|
|
end.
|
|
'/rules/:id/reset_metrics'(put, #{bindings := #{id := RuleId}}) ->
|
|
case emqx_rule_engine_proto_v1:reset_metrics(RuleId) of
|
|
ok ->
|
|
{200, <<"Reset Success">>};
|
|
Failed ->
|
|
{400, #{
|
|
code => 'BAD_REQUEST',
|
|
message => err_msg(Failed)
|
|
}}
|
|
end.
|
|
|
|
%%------------------------------------------------------------------------------
|
|
%% Internal functions
|
|
%%------------------------------------------------------------------------------
|
|
|
|
err_msg(Msg) -> emqx_misc:readable_error_msg(Msg).
|
|
|
|
format_rule_resp(Rules) when is_list(Rules) ->
|
|
[format_rule_resp(R) || R <- Rules];
|
|
format_rule_resp({Id, Rule}) ->
|
|
format_rule_resp(Rule#{id => Id});
|
|
format_rule_resp(#{
|
|
id := Id,
|
|
name := Name,
|
|
created_at := CreatedAt,
|
|
from := Topics,
|
|
actions := Action,
|
|
sql := SQL,
|
|
enable := Enable,
|
|
description := Descr
|
|
}) ->
|
|
NodeMetrics = get_rule_metrics(Id),
|
|
#{
|
|
id => Id,
|
|
name => Name,
|
|
from => Topics,
|
|
actions => format_action(Action),
|
|
sql => SQL,
|
|
metrics => aggregate_metrics(NodeMetrics),
|
|
node_metrics => NodeMetrics,
|
|
enable => Enable,
|
|
created_at => format_datetime(CreatedAt, millisecond),
|
|
description => Descr
|
|
}.
|
|
|
|
format_datetime(Timestamp, Unit) ->
|
|
list_to_binary(calendar:system_time_to_rfc3339(Timestamp, [{unit, Unit}])).
|
|
|
|
format_action(Actions) ->
|
|
[do_format_action(Act) || Act <- Actions].
|
|
|
|
do_format_action(#{mod := Mod, func := Func, args := Args}) ->
|
|
#{
|
|
function => printable_function_name(Mod, Func),
|
|
args => maps:remove(preprocessed_tmpl, Args)
|
|
};
|
|
do_format_action(BridgeChannelId) when is_binary(BridgeChannelId) ->
|
|
BridgeChannelId.
|
|
|
|
printable_function_name(emqx_rule_actions, Func) ->
|
|
Func;
|
|
printable_function_name(Mod, Func) ->
|
|
list_to_binary(lists:concat([Mod, ":", Func])).
|
|
|
|
get_rule_metrics(Id) ->
|
|
Format = fun(
|
|
Node,
|
|
#{
|
|
counters :=
|
|
#{
|
|
'matched' := Matched,
|
|
'passed' := Passed,
|
|
'failed' := Failed,
|
|
'failed.exception' := FailedEx,
|
|
'failed.no_result' := FailedNoRes,
|
|
'actions.total' := OTotal,
|
|
'actions.failed' := OFailed,
|
|
'actions.failed.out_of_service' := OFailedOOS,
|
|
'actions.failed.unknown' := OFailedUnknown,
|
|
'actions.success' := OFailedSucc
|
|
},
|
|
rate :=
|
|
#{
|
|
'matched' :=
|
|
#{current := Current, max := Max, last5m := Last5M}
|
|
}
|
|
}
|
|
) ->
|
|
#{
|
|
metrics => ?METRICS(
|
|
Matched,
|
|
Passed,
|
|
Failed,
|
|
FailedEx,
|
|
FailedNoRes,
|
|
OTotal,
|
|
OFailed,
|
|
OFailedOOS,
|
|
OFailedUnknown,
|
|
OFailedSucc,
|
|
Current,
|
|
Max,
|
|
Last5M
|
|
),
|
|
node => Node
|
|
}
|
|
end,
|
|
[
|
|
Format(Node, emqx_plugin_libs_proto_v1:get_metrics(Node, rule_metrics, Id))
|
|
|| Node <- mria_mnesia:running_nodes()
|
|
].
|
|
|
|
aggregate_metrics(AllMetrics) ->
|
|
InitMetrics = ?METRICS(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0),
|
|
lists:foldl(
|
|
fun(
|
|
#{
|
|
metrics := ?metrics(
|
|
Match1,
|
|
Passed1,
|
|
Failed1,
|
|
FailedEx1,
|
|
FailedNoRes1,
|
|
OTotal1,
|
|
OFailed1,
|
|
OFailedOOS1,
|
|
OFailedUnknown1,
|
|
OFailedSucc1,
|
|
Rate1,
|
|
RateMax1,
|
|
Rate5m1
|
|
)
|
|
},
|
|
?metrics(
|
|
Match0,
|
|
Passed0,
|
|
Failed0,
|
|
FailedEx0,
|
|
FailedNoRes0,
|
|
OTotal0,
|
|
OFailed0,
|
|
OFailedOOS0,
|
|
OFailedUnknown0,
|
|
OFailedSucc0,
|
|
Rate0,
|
|
RateMax0,
|
|
Rate5m0
|
|
)
|
|
) ->
|
|
?METRICS(
|
|
Match1 + Match0,
|
|
Passed1 + Passed0,
|
|
Failed1 + Failed0,
|
|
FailedEx1 + FailedEx0,
|
|
FailedNoRes1 + FailedNoRes0,
|
|
OTotal1 + OTotal0,
|
|
OFailed1 + OFailed0,
|
|
OFailedOOS1 + OFailedOOS0,
|
|
OFailedUnknown1 + OFailedUnknown0,
|
|
OFailedSucc1 + OFailedSucc0,
|
|
Rate1 + Rate0,
|
|
RateMax1 + RateMax0,
|
|
Rate5m1 + Rate5m0
|
|
)
|
|
end,
|
|
InitMetrics,
|
|
AllMetrics
|
|
).
|
|
|
|
get_one_rule(AllRules, Id) ->
|
|
[R || R = #{id := Id0} <- AllRules, Id0 == Id].
|
|
|
|
add_metadata(Params) ->
|
|
Params#{
|
|
<<"metadata">> => #{
|
|
<<"created_at">> => emqx_rule_engine:now_ms()
|
|
}
|
|
}.
|
|
|
|
filter_out_request_body(Conf) ->
|
|
ExtraConfs = [
|
|
<<"id">>,
|
|
<<"status">>,
|
|
<<"node_status">>,
|
|
<<"node_metrics">>,
|
|
<<"metrics">>,
|
|
<<"node">>
|
|
],
|
|
maps:without(ExtraConfs, Conf).
|
|
|
|
query(Tab, {Qs, Fuzzy}, Start, Limit) ->
|
|
Ms = qs2ms(),
|
|
FuzzyFun = fuzzy_match_fun(Qs, Ms, Fuzzy),
|
|
emqx_mgmt_api:select_table_with_count(
|
|
Tab, {Ms, FuzzyFun}, Start, Limit, fun format_rule_resp/1
|
|
).
|
|
|
|
%% rule is not a record, so everything is fuzzy filter.
|
|
qs2ms() ->
|
|
[{'_', [], ['$_']}].
|
|
|
|
fuzzy_match_fun(Qs, Ms, Fuzzy) ->
|
|
MsC = ets:match_spec_compile(Ms),
|
|
fun(Rows) ->
|
|
Ls = ets:match_spec_run(Rows, MsC),
|
|
lists:filter(
|
|
fun(E) ->
|
|
run_qs_match(E, Qs) andalso
|
|
run_fuzzy_match(E, Fuzzy)
|
|
end,
|
|
Ls
|
|
)
|
|
end.
|
|
|
|
run_qs_match(_, []) ->
|
|
true;
|
|
run_qs_match(E = {_Id, #{enable := Enable}}, [{enable, '=:=', Pattern} | Qs]) ->
|
|
Enable =:= Pattern andalso run_qs_match(E, Qs);
|
|
run_qs_match(E = {_Id, #{from := From}}, [{from, '=:=', Pattern} | Qs]) ->
|
|
lists:member(Pattern, From) andalso run_qs_match(E, Qs);
|
|
run_qs_match(E, [_ | Qs]) ->
|
|
run_qs_match(E, Qs).
|
|
|
|
run_fuzzy_match(_, []) ->
|
|
true;
|
|
run_fuzzy_match(E = {Id, _}, [{id, like, Pattern} | Fuzzy]) ->
|
|
binary:match(Id, Pattern) /= nomatch andalso run_fuzzy_match(E, Fuzzy);
|
|
run_fuzzy_match(E = {_Id, #{description := Desc}}, [{description, like, Pattern} | Fuzzy]) ->
|
|
binary:match(Desc, Pattern) /= nomatch andalso run_fuzzy_match(E, Fuzzy);
|
|
run_fuzzy_match(E = {_Id, #{from := Topics}}, [{from, match, Pattern} | Fuzzy]) ->
|
|
lists:any(fun(For) -> emqx_topic:match(For, Pattern) end, Topics) andalso
|
|
run_fuzzy_match(E, Fuzzy);
|
|
run_fuzzy_match(E = {_Id, #{from := Topics}}, [{from, like, Pattern} | Fuzzy]) ->
|
|
lists:any(fun(For) -> binary:match(For, Pattern) /= nomatch end, Topics) andalso
|
|
run_fuzzy_match(E, Fuzzy);
|
|
run_fuzzy_match(E, [_ | Fuzzy]) ->
|
|
run_fuzzy_match(E, Fuzzy).
|