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

Revert "feat(index): add topic index facility "

This commit is contained in:
JianBo He 2023-07-27 13:42:43 +08:00 committed by GitHub
parent 21bdbe5876
commit 951a96457b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 21 additions and 359 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
apps/emqx_rule_engine/include/rule_engine.hrl
View File

@ -109,7 +109,6 @@
%% Tables
-define(RULE_TAB, emqx_rule_engine).
-define(RULE_TOPIC_INDEX, emqx_rule_engine_topic_index).
%% Allowed sql function provider modules
-define(DEFAULT_SQL_FUNC_PROVIDER, emqx_rule_funcs).

71
apps/emqx_rule_engine/src/emqx_rule_engine.erl
View File

@ -176,7 +176,7 @@ create_rule(Params) ->
create_rule(Params = #{id := RuleId}, CreatedAt) when is_binary(RuleId) ->
case get_rule(RuleId) of
not_found -> with_parsed_rule(Params, CreatedAt, fun insert_rule/1);
not_found -> parse_and_insert(Params, CreatedAt);
{ok, _} -> {error, already_exists}
end.
@ -185,27 +185,18 @@ update_rule(Params = #{id := RuleId}) when is_binary(RuleId) ->
case get_rule(RuleId) of
not_found ->
{error, not_found};
{ok, RulePrev = #{created_at := CreatedAt}} ->
with_parsed_rule(Params, CreatedAt, fun(Rule) -> update_rule(Rule, RulePrev) end)
{ok, #{created_at := CreatedAt}} ->
parse_and_insert(Params, CreatedAt)
end.
-spec delete_rule(RuleId :: rule_id()) -> ok.
delete_rule(RuleId) when is_binary(RuleId) ->
case get_rule(RuleId) of
not_found ->
ok;
{ok, Rule} ->
gen_server:call(?RULE_ENGINE, {delete_rule, Rule}, ?T_CALL)
end.
gen_server:call(?RULE_ENGINE, {delete_rule, RuleId}, ?T_CALL).
-spec insert_rule(Rule :: rule()) -> ok.
insert_rule(Rule) ->
gen_server:call(?RULE_ENGINE, {insert_rule, Rule}, ?T_CALL).
-spec update_rule(Rule :: rule(), RulePrev :: rule()) -> ok.
update_rule(Rule, RulePrev) ->
gen_server:call(?RULE_ENGINE, {update_rule, Rule, RulePrev}, ?T_CALL).
%%----------------------------------------------------------------------------------------
%% Rule Management
%%----------------------------------------------------------------------------------------
@ -225,13 +216,13 @@ get_rules_ordered_by_ts() ->
-spec get_rules_for_topic(Topic :: binary()) -> [rule()].
get_rules_for_topic(Topic) ->
[
emqx_rule_index:get_record(M, ?RULE_TOPIC_INDEX)
|| M <- emqx_rule_index:matches(Topic, ?RULE_TOPIC_INDEX)
Rule
|| Rule = #{from := From} <- get_rules(),
emqx_topic:match_any(Topic, From)
].
-spec get_rules_with_same_event(Topic :: binary()) -> [rule()].
get_rules_with_same_event(Topic) ->
%% TODO: event matching index not implemented yet
EventName = emqx_rule_events:event_name(Topic),
[
Rule
@ -241,7 +232,6 @@ get_rules_with_same_event(Topic) ->
-spec get_rule_ids_by_action(action_name()) -> [rule_id()].
get_rule_ids_by_action(BridgeId) when is_binary(BridgeId) ->
%% TODO: bridge matching index not implemented yet
[
Id
|| #{actions := Acts, id := Id, from := Froms} <- get_rules(),
@ -249,7 +239,6 @@ get_rule_ids_by_action(BridgeId) when is_binary(BridgeId) ->
references_ingress_bridge(Froms, BridgeId)
];
get_rule_ids_by_action(#{function := FuncName}) when is_binary(FuncName) ->
%% TODO: action id matching index not implemented yet
{Mod, Fun} =
case string:split(FuncName, ":", leading) of
[M, F] -> {binary_to_module(M), F};
@ -422,17 +411,10 @@ init([]) ->
{ok, #{}}.
handle_call({insert_rule, Rule}, _From, State) ->
ok = do_insert_rule(Rule),
ok = do_update_rule_index(Rule),
{reply, ok, State};
handle_call({update_rule, Rule, RulePrev}, _From, State) ->
ok = do_delete_rule_index(RulePrev),
ok = do_insert_rule(Rule),
ok = do_update_rule_index(Rule),
do_insert_rule(Rule),
{reply, ok, State};
handle_call({delete_rule, Rule}, _From, State) ->
ok = do_delete_rule_index(Rule),
ok = do_delete_rule(Rule),
do_delete_rule(Rule),
{reply, ok, State};
handle_call(Req, _From, State) ->
?SLOG(error, #{msg => "unexpected_call", request => Req}),
@ -456,7 +438,7 @@ code_change(_OldVsn, State, _Extra) ->
%% Internal Functions
%%----------------------------------------------------------------------------------------
with_parsed_rule(Params = #{id := RuleId, sql := Sql, actions := Actions}, CreatedAt, Fun) ->
parse_and_insert(Params = #{id := RuleId, sql := Sql, actions := Actions}, CreatedAt) ->
case emqx_rule_sqlparser:parse(Sql) of
{ok, Select} ->
Rule = #{
@ -477,7 +459,7 @@ with_parsed_rule(Params = #{id := RuleId, sql := Sql, actions := Actions}, Creat
conditions => emqx_rule_sqlparser:select_where(Select)
%% -- calculated fields end
},
ok = Fun(Rule),
ok = insert_rule(Rule),
{ok, Rule};
{error, Reason} ->
{error, Reason}
@ -489,27 +471,16 @@ do_insert_rule(#{id := Id} = Rule) ->
true = ets:insert(?RULE_TAB, {Id, maps:remove(id, Rule)}),
ok.
do_delete_rule(#{id := Id} = Rule) ->
ok = unload_hooks_for_rule(Rule),
ok = clear_metrics_for_rule(Id),
true = ets:delete(?RULE_TAB, Id),
ok.
do_update_rule_index(#{id := Id, from := From} = Rule) ->
ok = lists:foreach(
fun(Topic) ->
true = emqx_rule_index:insert(Topic, Id, Rule, ?RULE_TOPIC_INDEX)
end,
From
).
do_delete_rule_index(#{id := Id, from := From}) ->
ok = lists:foreach(
fun(Topic) ->
true = emqx_rule_index:delete(Topic, Id, ?RULE_TOPIC_INDEX)
end,
From
).
do_delete_rule(RuleId) ->
case get_rule(RuleId) of
{ok, Rule} ->
ok = unload_hooks_for_rule(Rule),
ok = clear_metrics_for_rule(RuleId),
true = ets:delete(?RULE_TAB, RuleId),
ok;
not_found ->
ok
end.
parse_actions(Actions) ->
[do_parse_action(Act) || Act <- Actions].

1
apps/emqx_rule_engine/src/emqx_rule_engine_app.erl
View File

@ -26,7 +26,6 @@
start(_Type, _Args) ->
_ = ets:new(?RULE_TAB, [named_table, public, ordered_set, {read_concurrency, true}]),
_ = ets:new(?RULE_TOPIC_INDEX, [named_table, public, ordered_set, {read_concurrency, true}]),
ok = emqx_rule_events:reload(),
SupRet = emqx_rule_engine_sup:start_link(),
ok = emqx_rule_engine:load_rules(),

140
apps/emqx_rule_engine/src/emqx_rule_index.erl
View File

@ -1,140 +0,0 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2023 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.
%%--------------------------------------------------------------------
%% @doc Topic index for matching topics to topic filters.
%%
%% Works on top of ETS ordered_set table. Keys are parsed topic filters
%% with record ID appended to the end, wrapped in a tuple to disambiguate from
%% topic filter words. Existing table may be used if existing keys will not
%% collide with index keys.
%%
%% Designed to effectively answer questions like:
%% 1. Does any topic filter match given topic?
%% 2. Which records are associated with topic filters matching given topic?
%%
%% Questions like these are _only slightly_ less effective:
%% 1. Which topic filters match given topic?
%% 2. Which record IDs are associated with topic filters matching given topic?
-module(emqx_rule_index).
-export([insert/4]).
-export([delete/3]).
-export([match/2]).
-export([matches/2]).
-export([get_record/2]).
-type key(ID) :: [binary() | '+' | '#' | {ID}].
-type match(ID) :: key(ID).
-ifdef(TEST).
-export_type([match/1]).
-endif.
insert(Filter, ID, Record, Tab) ->
%% TODO: topic compact. see also in emqx_trie.erl
ets:insert(Tab, {emqx_topic:words(Filter) ++ [{ID}], Record}).
delete(Filter, ID, Tab) ->
ets:delete(Tab, emqx_topic:words(Filter) ++ [{ID}]).
-spec match(emqx_types:topic(), ets:table()) -> match(_ID) | false.
match(Topic, Tab) ->
{Words, RPrefix} = match_init(Topic),
match(Words, RPrefix, Tab).
match(Words, RPrefix, Tab) ->
Prefix = lists:reverse(RPrefix),
K = ets:next(Tab, Prefix),
case match_filter(Prefix, K, Words =/= []) of
true ->
K;
stop ->
false;
Matched ->
match_rest(Matched, Words, RPrefix, Tab)
end.
match_rest(false, [W | Rest], RPrefix, Tab) ->
match(Rest, [W | RPrefix], Tab);
match_rest(plus, [W | Rest], RPrefix, Tab) ->
case match(Rest, ['+' | RPrefix], Tab) of
Match when is_list(Match) ->
Match;
false ->
match(Rest, [W | RPrefix], Tab)
end;
match_rest(_, [], _RPrefix, _Tab) ->
false.
-spec matches(emqx_types:topic(), ets:table()) -> [match(_ID)].
matches(Topic, Tab) ->
{Words, RPrefix} = match_init(Topic),
matches(Words, RPrefix, Tab).
matches(Words, RPrefix, Tab) ->
Prefix = lists:reverse(RPrefix),
matches(ets:next(Tab, Prefix), Prefix, Words, RPrefix, Tab).
matches(K, Prefix, Words, RPrefix, Tab) ->
case match_filter(Prefix, K, Words =/= []) of
true ->
[K | matches(ets:next(Tab, K), Prefix, Words, RPrefix, Tab)];
stop ->
[];
Matched ->
matches_rest(Matched, Words, RPrefix, Tab)
end.
matches_rest(false, [W | Rest], RPrefix, Tab) ->
matches(Rest, [W | RPrefix], Tab);
matches_rest(plus, [W | Rest], RPrefix, Tab) ->
matches(Rest, ['+' | RPrefix], Tab) ++ matches(Rest, [W | RPrefix], Tab);
matches_rest(_, [], _RPrefix, _Tab) ->
[].
match_filter([], [{_ID}], _IsPrefix = false) ->
% NOTE: exact match is `true` only if we match whole topic, not prefix
true;
match_filter([], ['#', {_ID}], _IsPrefix) ->
% NOTE: naturally, '#' < '+', so this is already optimal for `match/2`
true;
match_filter([], ['+' | _], _) ->
plus;
match_filter([], [_H | _], _) ->
false;
match_filter([H | T1], [H | T2], IsPrefix) ->
match_filter(T1, T2, IsPrefix);
match_filter([H1 | _], [H2 | _], _) when H2 > H1 ->
% NOTE: we're strictly past the prefix, no need to continue
stop;
match_filter(_, '$end_of_table', _) ->
stop.
match_init(Topic) ->
case emqx_topic:words(Topic) of
[W = <<"$", _/bytes>> | Rest] ->
% NOTE
% This will effectively skip attempts to match special topics to `#` or `+/...`.
{Rest, [W]};
Words ->
{Words, []}
end.
-spec get_record(match(_ID), ets:table()) -> _Record.
get_record(K, Tab) ->
ets:lookup_element(Tab, K, 2).

166
apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl
View File

@ -1,166 +0,0 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2023 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_index_SUITE).
-compile(export_all).
-compile(nowarn_export_all).
-include_lib("eunit/include/eunit.hrl").
all() ->
emqx_common_test_helpers:all(?MODULE).
t_insert(_) ->
Tab = new(),
true = emqx_rule_index:insert(<<"sensor/1/metric/2">>, t_insert_1, <<>>, Tab),
true = emqx_rule_index:insert(<<"sensor/+/#">>, t_insert_2, <<>>, Tab),
true = emqx_rule_index:insert(<<"sensor/#">>, t_insert_3, <<>>, Tab),
?assertEqual(<<"sensor/#">>, get_topic(match(<<"sensor">>, Tab))),
?assertEqual(t_insert_3, get_id(match(<<"sensor">>, Tab))),
true = ets:delete(Tab).
t_match(_) ->
Tab = new(),
true = emqx_rule_index:insert(<<"sensor/1/metric/2">>, t_match_1, <<>>, Tab),
true = emqx_rule_index:insert(<<"sensor/+/#">>, t_match_2, <<>>, Tab),
true = emqx_rule_index:insert(<<"sensor/#">>, t_match_3, <<>>, Tab),
?assertMatch(
[<<"sensor/#">>, <<"sensor/+/#">>],
[get_topic(M) || M <- matches(<<"sensor/1">>, Tab)]
),
true = ets:delete(Tab).
t_match2(_) ->
Tab = new(),
true = emqx_rule_index:insert(<<"#">>, t_match2_1, <<>>, Tab),
true = emqx_rule_index:insert(<<"+/#">>, t_match2_2, <<>>, Tab),
true = emqx_rule_index:insert(<<"+/+/#">>, t_match2_3, <<>>, Tab),
?assertEqual(
[<<"#">>, <<"+/#">>, <<"+/+/#">>],
[get_topic(M) || M <- matches(<<"a/b/c">>, Tab)]
),
?assertEqual(
false,
emqx_rule_index:match(<<"$SYS/broker/zenmq">>, Tab)
),
true = ets:delete(Tab).
t_match3(_) ->
Tab = new(),
Records = [
{<<"d/#">>, t_match3_1},
{<<"a/b/+">>, t_match3_2},
{<<"a/#">>, t_match3_3},
{<<"#">>, t_match3_4},
{<<"$SYS/#">>, t_match3_sys}
],
lists:foreach(
fun({Topic, ID}) -> emqx_rule_index:insert(Topic, ID, <<>>, Tab) end,
Records
),
Matched = matches(<<"a/b/c">>, Tab),
case length(Matched) of
3 -> ok;
_ -> error({unexpected, Matched})
end,
?assertEqual(
t_match3_sys,
get_id(match(<<"$SYS/a/b/c">>, Tab))
),
true = ets:delete(Tab).
t_match4(_) ->
Tab = new(),
Records = [{<<"/#">>, t_match4_1}, {<<"/+">>, t_match4_2}, {<<"/+/a/b/c">>, t_match4_3}],
lists:foreach(
fun({Topic, ID}) -> emqx_rule_index:insert(Topic, ID, <<>>, Tab) end,
Records
),
?assertEqual(
[<<"/#">>, <<"/+">>],
[get_topic(M) || M <- matches(<<"/">>, Tab)]
),
?assertEqual(
[<<"/#">>, <<"/+/a/b/c">>],
[get_topic(M) || M <- matches(<<"/0/a/b/c">>, Tab)]
),
true = ets:delete(Tab).
t_match5(_) ->
Tab = new(),
T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>,
Records = [
{<<"#">>, t_match5_1},
{<<T/binary, "/#">>, t_match5_2},
{<<T/binary, "/+">>, t_match5_3}
],
lists:foreach(
fun({Topic, ID}) -> emqx_rule_index:insert(Topic, ID, <<>>, Tab) end,
Records
),
?assertEqual(
[<<"#">>, <<T/binary, "/#">>],
[get_topic(M) || M <- matches(T, Tab)]
),
?assertEqual(
[<<"#">>, <<T/binary, "/#">>, <<T/binary, "/+">>],
[get_topic(M) || M <- matches(<<T/binary, "/1">>, Tab)]
),
true = ets:delete(Tab).
t_match6(_) ->
Tab = new(),
T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>,
W = <<"+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/#">>,
emqx_rule_index:insert(W, ID = t_match6, <<>>, Tab),
?assertEqual(ID, get_id(match(T, Tab))),
true = ets:delete(Tab).
t_match7(_) ->
Tab = new(),
T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>,
W = <<"a/+/c/+/e/+/g/+/i/+/k/+/m/+/o/+/q/+/s/+/u/+/w/+/y/+/#">>,
emqx_rule_index:insert(W, t_match7, <<>>, Tab),
?assertEqual(W, get_topic(match(T, Tab))),
true = ets:delete(Tab).
new() ->
ets:new(?MODULE, [public, ordered_set, {write_concurrency, true}]).
match(T, Tab) ->
emqx_rule_index:match(T, Tab).
matches(T, Tab) ->
lists:sort(emqx_rule_index:matches(T, Tab)).
-spec get_id(emqx_rule_index:match(ID)) -> ID.
get_id([{ID}]) ->
ID;
get_id([_ | Rest]) ->
get_id(Rest).
-spec get_topic(emqx_rule_index:match(_ID)) -> emqx_types:topic().
get_topic(K) ->
emqx_topic:join(cut_topic(K)).
cut_topic(K) ->
cut_topic(K, []).
cut_topic([{_ID}], Acc) ->
lists:reverse(Acc);
cut_topic([W | Rest], Acc) ->
cut_topic(Rest, [W | Acc]).

1
changes/ce/perf-11282.en.md
View File

@ -1 +0,0 @@
Added indexing to the rule engine's topic matching to improve rule search performance.