feat: support clean pem_cache cli

apps/emqx/src/proto/emqx_proto_v1.erl

@@ -33,7 +33,8 @@
     delete_all_deactivated_alarms/1,
 
     clean_authz_cache/1,
-    clean_authz_cache/2
+    clean_authz_cache/2,
+    clean_pem_cache/1
 ]).
 
 introduced_in() ->
@@ -66,6 +67,10 @@ clean_authz_cache(Node, ClientId) ->
 clean_authz_cache(Node) ->
     rpc:call(Node, emqx_authz_cache, drain_cache, []).
 
+-spec clean_pem_cache(node()) -> ok | {badrpc, _}.
+clean_pem_cache(Node) ->
+    rpc:call(Node, ssl_pem_cache, clear, []).
+
 -spec deactivate_alarm(node(), binary() | atom()) ->
     ok | {error, not_found} | {badrpc, _}.
 deactivate_alarm(Node, Name) ->

apps/emqx_management/src/emqx_mgmt.erl

@@ -56,12 +56,14 @@
     clean_authz_cache/2,
     clean_authz_cache_all/0,
     clean_authz_cache_all/1,
+    clean_pem_cache_all/0,
+    clean_pem_cache_all/1,
     set_ratelimit_policy/2,
     set_quota_policy/2,
     set_keepalive/2
 ]).
 
-%% Internal funcs
+%% Internal functions
 -export([do_call_client/2]).
 
 %% Subscriptions
@@ -283,6 +285,13 @@ clean_authz_cache(Node, ClientId) ->
 
 clean_authz_cache_all() ->
     Results = [{Node, clean_authz_cache_all(Node)} || Node <- mria_mnesia:running_nodes()],
+    wrap_results(Results).
+
+clean_pem_cache_all() ->
+    Results = [{Node, clean_pem_cache_all(Node)} || Node <- mria_mnesia:running_nodes()],
+    wrap_results(Results).
+
+wrap_results(Results) ->
     case lists:filter(fun({_Node, Item}) -> Item =/= ok end, Results) of
         [] -> ok;
         BadNodes -> {error, BadNodes}
@@ -291,6 +300,9 @@ clean_authz_cache_all() ->
 clean_authz_cache_all(Node) ->
     wrap_rpc(emqx_proto_v1:clean_authz_cache(Node)).
 
+clean_pem_cache_all(Node) ->
+    wrap_rpc(emqx_proto_v1:clean_pem_cache(Node)).
+
 set_ratelimit_policy(ClientId, Policy) ->
     call_client(ClientId, {ratelimit, Policy}).
 

apps/emqx_management/src/emqx_mgmt_cli.erl

@@ -41,6 +41,7 @@
     traces/1,
     log/1,
     authz/1,
+    pem_cache/1,
     olp/1
 ]).
 
@@ -601,21 +602,14 @@ listeners(_) ->
 %% @doc authz Command
 
 authz(["cache-clean", "node", Node]) ->
-    case emqx_mgmt:clean_authz_cache_all(erlang:list_to_existing_atom(Node)) of
+    Msg = io_lib:format("Authorization cache drain started on node ~ts", [Node]),
-        ok ->
+    with_log(fun() -> for_node(fun emqx_mgmt:clean_authz_cache_all/1, Node) end, Msg);
-            emqx_ctl:print("Authorization cache drain started on node ~ts.~n", [Node]);
-        {error, Reason} ->
-            emqx_ctl:print("Authorization drain failed on node ~ts: ~0p.~n", [Node, Reason])
-    end;
 authz(["cache-clean", "all"]) ->
-    case emqx_mgmt:clean_authz_cache_all() of
+    Msg = "Authorization cache drain started on all nodes",
-        ok ->
+    with_log(fun emqx_mgmt:clean_authz_cache_all/0, Msg);
-            emqx_ctl:print("Started Authorization cache drain in all nodes~n");
-        {error, Reason} ->
-            emqx_ctl:print("Authorization cache-clean failed: ~p.~n", [Reason])
-    end;
 authz(["cache-clean", ClientId]) ->
-    emqx_mgmt:clean_authz_cache(ClientId);
+    Msg = io_lib:format("Drain ~ts authz cache", [ClientId]),
+    with_log(fun() -> emqx_mgmt:clean_authz_cache(ClientId) end, Msg);
 authz(_) ->
     emqx_ctl:usage(
         [
@@ -625,6 +619,17 @@ authz(_) ->
         ]
     ).
 
+pem_cache(["clean", "all"]) ->
+    with_log(fun emqx_mgmt:clean_pem_cache_all/0, "PEM cache clean");
+pem_cache(["clean", "node", Node]) ->
+    Msg = io_lib:format("~ts PEM cache clean", [Node]),
+    with_log(fun() -> for_node(fun emqx_mgmt:clean_pem_cache_all/1, Node) end, Msg);
+pem_cache(_) ->
+    emqx_ctl:usage([
+        {"pem_cache clean all", "Clears x509 certificate cache on all nodes"},
+        {"pem_cache clean node <Node>", "Clears x509 certificate cache on given node"}
+    ]).
+
 %%--------------------------------------------------------------------
 %% @doc OLP (Overload Protection related)
 olp(["status"]) ->
@@ -786,3 +791,20 @@ format_listen_on({Addr, Port}) when is_tuple(Addr) ->
 
 name(Filter) ->
     iolist_to_binary(["CLI-", Filter]).
+
+for_node(Fun, Node) ->
+    try list_to_existing_atom(Node) of
+        NodeAtom ->
+            Fun(NodeAtom)
+    catch
+        error:badarg ->
+            {error, unknown_node}
+    end.
+
+with_log(Fun, Msg) ->
+    case Fun() of
+        ok ->
+            emqx_ctl:print("~s OK~n", [Msg]);
+        {error, Reason} ->
+            emqx_ctl:print("~s FAILED~n~p~n", [Msg, Reason])
+    end.