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

feat(emqx_rule_events): add field 'is_cache'

This commit is contained in:
EMQ-YangM 2022-03-25 18:05:25 +08:00
parent 44f4dfa498
commit 059fc6e3c7
3 changed files with 16 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
apps/emqx_rule_engine/src/emqx_rule_events.erl
View File

@ -39,7 +39,7 @@
, on_message_delivered/3 , on_message_delivered/3
, on_message_acked/3 , on_message_acked/3
, on_delivery_dropped/4 , on_delivery_dropped/4
, on_client_check_acl_complete/5 , on_client_check_acl_complete/6
]). ]).
-export([ event_info/0 -export([ event_info/0
@ -114,12 +114,13 @@ on_client_connack(ConnInfo, Reason, _, Env) ->
may_publish_and_apply('client.connack', may_publish_and_apply('client.connack',
fun() -> eventmsg_connack(ConnInfo, Reason) end, Env). fun() -> eventmsg_connack(ConnInfo, Reason) end, Env).
on_client_check_acl_complete(ClientInfo, PubSub, Topic, Result, Env) -> on_client_check_acl_complete(ClientInfo, PubSub, Topic, Result, IsCache, Env) ->
may_publish_and_apply('client.check_acl_complete', may_publish_and_apply('client.check_acl_complete',
fun() -> eventmsg_check_acl_complete(ClientInfo, fun() -> eventmsg_check_acl_complete(ClientInfo,
PubSub, PubSub,
Topic, Topic,
Result) end, Env). Result,
IsCache) end, Env).
on_session_subscribed(ClientInfo, Topic, SubOpts, Env) -> on_session_subscribed(ClientInfo, Topic, SubOpts, Env) ->
may_publish_and_apply('session.subscribed', may_publish_and_apply('session.subscribed',
@ -266,13 +267,14 @@ eventmsg_check_acl_complete(_ClientInfo = #{
clientid := ClientId, clientid := ClientId,
username := Username, username := Username,
peerhost := PeerHost peerhost := PeerHost
}, PubSub, Topic, Result) -> }, PubSub, Topic, Result, IsCache) ->
with_basic_columns('client.check_acl_complete', with_basic_columns('client.check_acl_complete',
#{clientid => ClientId, #{clientid => ClientId,
username => Username, username => Username,
peerhost => ntoa(PeerHost), peerhost => ntoa(PeerHost),
topic => Topic, topic => Topic,
action => PubSub, action => PubSub,
is_cache => IsCache,
result => Result result => Result
}). }).
@ -737,6 +739,7 @@ columns_with_exam('client.check_acl_complete') ->
, {<<"peerhost">>, <<"192.168.0.10">>} , {<<"peerhost">>, <<"192.168.0.10">>}
, {<<"topic">>, <<"t/a">>} , {<<"topic">>, <<"t/a">>}
, {<<"action">>, <<"publish">>} , {<<"action">>, <<"publish">>}
, {<<"is_cache">>, <<"false">>}
, {<<"result">>, <<"allow">>} , {<<"result">>, <<"allow">>}
, {<<"timestamp">>, erlang:system_time(millisecond)} , {<<"timestamp">>, erlang:system_time(millisecond)}
, {<<"node">>, node()} , {<<"node">>, node()}

2
apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl
View File

@ -2809,11 +2809,13 @@ verify_event_fields('client.check_acl_complete', Fields) ->
action := Action, action := Action,
result := Result, result := Result,
topic := Topic, topic := Topic,
is_cache := IsCache,
username := Username username := Username
} = Fields, } = Fields,
?assertEqual(<<"t1">>, Topic), ?assertEqual(<<"t1">>, Topic),
?assert(lists:member(Action, [subscribe, publish])), ?assert(lists:member(Action, [subscribe, publish])),
?assert(lists:member(Result, [allow, deny])), ?assert(lists:member(Result, [allow, deny])),
?assert(lists:member(IsCache, [true, false])),
?assert(lists:member(ClientId, [<<"c_event">>, <<"c_event2">>])), ?assert(lists:member(ClientId, [<<"c_event">>, <<"c_event2">>])),
?assert(lists:member(Username, [<<"u_event">>, <<"u_event2">>])). ?assert(lists:member(Username, [<<"u_event">>, <<"u_event2">>])).

12
src/emqx_access_control.erl
View File

@ -50,7 +50,6 @@ check_acl(ClientInfo, PubSub, Topic) ->
false -> do_check_acl(ClientInfo, PubSub, Topic) false -> do_check_acl(ClientInfo, PubSub, Topic)
end, end,
inc_acl_metrics(Result), inc_acl_metrics(Result),
emqx:run_hook('client.check_acl_complete', [ClientInfo, PubSub, Topic, Result]),
Result. Result.
check_acl_cache(ClientInfo, PubSub, Topic) -> check_acl_cache(ClientInfo, PubSub, Topic) ->
@ -61,15 +60,18 @@ check_acl_cache(ClientInfo, PubSub, Topic) ->
AclResult; AclResult;
AclResult -> AclResult ->
inc_acl_metrics(cache_hit), inc_acl_metrics(cache_hit),
emqx:run_hook('client.check_acl_complete', [ClientInfo, PubSub, Topic, AclResult, true]),
AclResult AclResult
end. end.
do_check_acl(ClientInfo = #{zone := Zone}, PubSub, Topic) -> do_check_acl(ClientInfo = #{zone := Zone}, PubSub, Topic) ->
Default = emqx_zone:get_env(Zone, acl_nomatch, deny), Default = emqx_zone:get_env(Zone, acl_nomatch, deny),
case run_hooks('client.check_acl', [ClientInfo, PubSub, Topic], Default) of Result = case run_hooks('client.check_acl', [ClientInfo, PubSub, Topic], Default) of
allow -> allow; allow -> allow;
_Other -> deny _Other -> deny
end. end,
emqx:run_hook('client.check_acl_complete', [ClientInfo, PubSub, Topic, Result, false]),
Result.
default_auth_result(Zone) -> default_auth_result(Zone) ->
case emqx_zone:get_env(Zone, allow_anonymous, false) of case emqx_zone:get_env(Zone, allow_anonymous, false) of