Merge pull request #6503 from tigercl/feat/authz-config

feat(authz): support sync configuration in the cluster

apps/emqx_authz/src/emqx_authz.erl

@@ -31,9 +31,7 @@
         , lookup/0
         , lookup/1
         , move/2
-        , move/3
         , update/2
-        , update/3
         , authorize/5
         ]).
 
@@ -110,28 +108,19 @@ lookup(Type) ->
     {Source, _Front, _Rear} = take(Type),
     Source.
 
-move(Type, Cmd) ->
+move(Type, #{<<"before">> := Before}) ->
-    move(Type, Cmd, #{}).
+    emqx_authz_utils:update_config(?CONF_KEY_PATH, {?CMD_MOVE, type(Type), ?CMD_MOVE_BEFORE(type(Before))});
-
+move(Type, #{<<"after">> := After}) ->
-move(Type, #{<<"before">> := Before}, Opts) ->
+    emqx_authz_utils:update_config(?CONF_KEY_PATH, {?CMD_MOVE, type(Type), ?CMD_MOVE_AFTER(type(After))});
-    emqx:update_config( ?CONF_KEY_PATH
+move(Type, Position) ->
-                      , {?CMD_MOVE, type(Type), ?CMD_MOVE_BEFORE(type(Before))}, Opts);
+    emqx_authz_utils:update_config(?CONF_KEY_PATH, {?CMD_MOVE, type(Type), Position}).
-move(Type, #{<<"after">> := After}, Opts) ->
-    emqx:update_config( ?CONF_KEY_PATH
-                      , {?CMD_MOVE, type(Type), ?CMD_MOVE_AFTER(type(After))}, Opts);
-move(Type, Position, Opts) ->
-    emqx:update_config( ?CONF_KEY_PATH
-                      , {?CMD_MOVE, type(Type), Position}, Opts).
 
+update({?CMD_REPLACE, Type}, Sources) ->
+    emqx_authz_utils:update_config(?CONF_KEY_PATH, {{?CMD_REPLACE, type(Type)}, Sources});
+update({?CMD_DELETE, Type}, Sources) ->
+    emqx_authz_utils:update_config(?CONF_KEY_PATH, {{?CMD_DELETE, type(Type)}, Sources});
 update(Cmd, Sources) ->
-    update(Cmd, Sources, #{}).
+    emqx_authz_utils:update_config(?CONF_KEY_PATH, {Cmd, Sources}).
-
-update({?CMD_REPLACE, Type}, Sources, Opts) ->
-    emqx:update_config(?CONF_KEY_PATH, {{?CMD_REPLACE, type(Type)}, Sources}, Opts);
-update({?CMD_DELETE, Type}, Sources, Opts) ->
-    emqx:update_config(?CONF_KEY_PATH, {{?CMD_DELETE, type(Type)}, Sources}, Opts);
-update(Cmd, Sources, Opts) ->
-    emqx:update_config(?CONF_KEY_PATH, {Cmd, Sources}, Opts).
 
 do_update({?CMD_MOVE, Type, ?CMD_MOVE_TOP}, Conf) when is_list(Conf) ->
     {Source, Front, Rear} = take(Type, Conf),
@@ -155,8 +144,8 @@ do_update({?CMD_APPEND, Sources}, Conf) when is_list(Sources), is_list(Conf) ->
     NConf = Conf ++ Sources,
     ok = check_dup_types(NConf),
     NConf;
-do_update({{?CMD_REPLACE, Type}, #{<<"enable">> := true} = Source}, Conf) when is_map(Source),
+do_update({{?CMD_REPLACE, Type}, #{<<"enable">> := true} = Source}, Conf)
-                                                                               is_list(Conf) ->
+  when is_map(Source), is_list(Conf) ->
     case create_dry_run(Type, Source)  of
         ok ->
             {_Old, Front, Rear} = take(Type, Conf),
@@ -165,7 +154,8 @@ do_update({{?CMD_REPLACE, Type}, #{<<"enable">> := true} = Source}, Conf) when i
             NConf;
         {error, _} = Error -> Error
     end;
-do_update({{?CMD_REPLACE, Type}, Source}, Conf) when is_map(Source), is_list(Conf) ->
+do_update({{?CMD_REPLACE, Type}, Source}, Conf)
+  when is_map(Source), is_list(Conf) ->
     {_Old, Front, Rear} = take(Type, Conf),
     NConf = Front ++ [Source | Rear],
     ok = check_dup_types(NConf),

apps/emqx_authz/src/emqx_authz_api_settings.erl

@@ -54,8 +54,9 @@ settings(get, _Params) ->
 settings(put, #{body := #{<<"no_match">> := NoMatch,
                           <<"deny_action">> := DenyAction,
                           <<"cache">> := Cache}}) ->
-    {ok, _} = emqx:update_config([authorization, no_match], NoMatch),
+    {ok, _} = emqx_authz_utils:update_config([authorization, no_match], NoMatch),
-    {ok, _} = emqx:update_config([authorization, deny_action], DenyAction),
+    {ok, _} = emqx_authz_utils:update_config(
-    {ok, _} = emqx:update_config([authorization, cache], Cache),
+                [authorization, deny_action], DenyAction),
+    {ok, _} = emqx_authz_utils:update_config([authorization, cache], Cache),
     ok = emqx_authz_cache:drain_cache(),
     {200, authorization_settings()}.

apps/emqx_authz/src/emqx_authz_http.erl

@@ -48,12 +48,12 @@ init(#{url := Url} = Source) ->
     end.
 
 destroy(#{annotations := #{id := Id}}) ->
-    ok = emqx_resource:remove(Id).
+    ok = emqx_resource:remove_local(Id).
 
 dry_run(Source) ->
     URIMap = maps:get(url, Source),
     NSource = maps:put(base_url, maps:remove(query, URIMap), Source),
-    emqx_resource:create_dry_run(emqx_connector_http, NSource).
+    emqx_resource:create_dry_run_local(emqx_connector_http, NSource).
 
 authorize(Client, PubSub, Topic,
             #{type := http,

apps/emqx_authz/src/emqx_authz_mongodb.erl

@@ -46,10 +46,10 @@ init(Source) ->
     end.
 
 dry_run(Source) ->
-    emqx_resource:create_dry_run(emqx_connector_mongo, Source).
+    emqx_resource:create_dry_run_local(emqx_connector_mongo, Source).
 
 destroy(#{annotations := #{id := Id}}) ->
-    ok = emqx_resource:remove(Id).
+    ok = emqx_resource:remove_local(Id).
 
 authorize(Client, PubSub, Topic,
             #{collection := Collection,

apps/emqx_authz/src/emqx_authz_mysql.erl

@@ -48,10 +48,10 @@ init(#{query := SQL} = Source) ->
     end.
 
 dry_run(Source) ->
-    emqx_resource:create_dry_run(emqx_connector_mysql, Source).
+    emqx_resource:create_dry_run_local(emqx_connector_mysql, Source).
 
 destroy(#{annotations := #{id := Id}}) ->
-    ok = emqx_resource:remove(Id).
+    ok = emqx_resource:remove_local(Id).
 
 authorize(Client, PubSub, Topic,
             #{annotations := #{id := ResourceID,

apps/emqx_authz/src/emqx_authz_postgresql.erl

@@ -48,10 +48,10 @@ init(#{query := SQL} = Source) ->
     end.
 
 destroy(#{annotations := #{id := Id}}) ->
-    ok = emqx_resource:remove(Id).
+    ok = emqx_resource:remove_local(Id).
 
 dry_run(Source) ->
-    emqx_resource:create_dry_run(emqx_connector_pgsql, Source).
+    emqx_resource:create_dry_run_local(emqx_connector_pgsql, Source).
 
 parse_query(Sql) ->
     case re:run(Sql, ?RE_PLACEHOLDER, [global, {capture, all, list}]) of

apps/emqx_authz/src/emqx_authz_redis.erl

@@ -46,10 +46,10 @@ init(Source) ->
     end.
 
 destroy(#{annotations := #{id := Id}}) ->
-    ok = emqx_resource:remove(Id).
+    ok = emqx_resource:remove_local(Id).
 
 dry_run(Source) ->
-    emqx_resource:create_dry_run(emqx_connector_redis, Source).
+    emqx_resource:create_dry_run_local(emqx_connector_redis, Source).
 
 authorize(Client, PubSub, Topic,
             #{cmd := CMD,

apps/emqx_authz/src/emqx_authz_utils.erl

@@ -18,9 +18,11 @@
 
 -include_lib("emqx/include/emqx_placeholder.hrl").
 
--export([cleanup_resources/0,
+-export([ cleanup_resources/0
-         make_resource_id/1,
+        , make_resource_id/1
-         create_resource/2]).
+        , create_resource/2
+        , update_config/2
+        ]).
 
 -define(RESOURCE_GROUP, <<"emqx_authz">>).
 
@@ -30,7 +32,7 @@
 
 create_resource(Module, Config) ->
     ResourceID = make_resource_id(Module),
-    case emqx_resource:create(ResourceID, Module, Config) of
+    case emqx_resource:create_local(ResourceID, Module, Config) of
         {ok, already_created} -> {ok, ResourceID};
         {ok, _} -> {ok, ResourceID};
         {error, Reason} -> {error, Reason}
@@ -38,13 +40,17 @@ create_resource(Module, Config) ->
 
 cleanup_resources() ->
     lists:foreach(
-      fun emqx_resource:remove/1,
+      fun emqx_resource:remove_local/1,
       emqx_resource:list_group_instances(?RESOURCE_GROUP)).
 
 make_resource_id(Name) ->
     NameBin = bin(Name),
     emqx_resource:generate_id(?RESOURCE_GROUP, NameBin).
 
+update_config(Path, ConfigRequest) ->
+    emqx_conf:update(Path, ConfigRequest, #{rawconf_with_defaults => true,
+                                            override_to => cluster}).
+
 %%------------------------------------------------------------------------------
 %% Internal functions
 %%------------------------------------------------------------------------------

apps/emqx_authz/test/emqx_authz_SUITE.erl

@@ -31,10 +31,9 @@ groups() ->
 
 init_per_suite(Config) ->
     meck:new(emqx_resource, [non_strict, passthrough, no_history, no_link]),
-    meck:expect(emqx_resource, create, fun(_, _, _) -> {ok, meck_data} end),
+    meck:expect(emqx_resource, create_local, fun(_, _, _) -> {ok, meck_data} end),
-    meck:expect(emqx_resource, update, fun(_, _, _, _) -> {ok, meck_data} end),
+    meck:expect(emqx_resource, remove_local, fun(_) -> ok end),
-    meck:expect(emqx_resource, remove, fun(_) -> ok end),
+    meck:expect(emqx_resource, create_dry_run_local, fun(_, _) -> ok end),
-    meck:expect(emqx_resource, create_dry_run, fun(_, _) -> ok end),
 
     ok = emqx_common_test_helpers:start_apps(
            [emqx_connector, emqx_conf, emqx_authz],

apps/emqx_authz/test/emqx_authz_api_sources_SUITE.erl

@@ -96,14 +96,13 @@ groups() ->
 
 init_per_suite(Config) ->
     meck:new(emqx_resource, [non_strict, passthrough, no_history, no_link]),
-    meck:expect(emqx_resource, create, fun(_, _, _) -> {ok, meck_data} end),
+    meck:expect(emqx_resource, create_local, fun(_, _, _) -> {ok, meck_data} end),
-    meck:expect(emqx_resource, create_dry_run,
+    meck:expect(emqx_resource, create_dry_run_local,
                 fun(emqx_connector_mysql, _) -> {ok, meck_data};
                    (T, C) -> meck:passthrough([T, C])
                 end),
-    meck:expect(emqx_resource, update, fun(_, _, _, _) -> {ok, meck_data} end),
     meck:expect(emqx_resource, health_check, fun(_) -> ok end),
-    meck:expect(emqx_resource, remove, fun(_) -> ok end ),
+    meck:expect(emqx_resource, remove_local, fun(_) -> ok end ),
 
     ok = emqx_common_test_helpers:start_apps(
            [emqx_conf, emqx_authz, emqx_dashboard],

apps/emqx_authz/test/emqx_authz_http_SUITE.erl

@@ -30,8 +30,8 @@ groups() ->
 
 init_per_suite(Config) ->
     meck:new(emqx_resource, [non_strict, passthrough, no_history, no_link]),
-    meck:expect(emqx_resource, create, fun(_, _, _) -> {ok, meck_data} end),
+    meck:expect(emqx_resource, create_local, fun(_, _, _) -> {ok, meck_data} end),
-    meck:expect(emqx_resource, remove, fun(_) -> ok end ),
+    meck:expect(emqx_resource, remove_local, fun(_) -> ok end ),
 
     ok = emqx_common_test_helpers:start_apps(
            [emqx_conf, emqx_authz],