refactor(test): use a linked janitor for test teardown

apps/emqx/test/emqx_common_test_helpers.erl

@@ -67,8 +67,7 @@
 -export([clear_screen/0]).
 -export([with_mock/4]).
 -export([
-    on_exit/2,
-    run_on_exit_callbacks/1
+    on_exit/1
 ]).
 
 %% Toxiproxy API
@@ -939,19 +938,16 @@ latency_up_proxy(off, Name, ProxyHost, ProxyPort) ->
 %% Testcase teardown utilities
 %%-------------------------------------------------------------------------------
 
-get_on_exit_callbacks(Id) ->
-    persistent_term:get({?MODULE, on_exit, Id}, []).
+get_or_spawn_janitor() ->
+    case get({?MODULE, janitor_proc}) of
+        undefined ->
+            {ok, Janitor} = emqx_test_janitor:start_link(),
+            put({?MODULE, janitor_proc}, Janitor),
+            Janitor;
+        Janitor ->
+            Janitor
+    end.
 
-put_on_exit_callbacks(Id, Funs) ->
-    persistent_term:put({?MODULE, on_exit, Id}, Funs).
-
-on_exit(Id, Fun) ->
-    Callbacks = get_on_exit_callbacks(Id),
-    put_on_exit_callbacks(Id, [Fun | Callbacks]).
-
-%% should be called at `end_per_testcase'.
-%% TODO: scope per group and suite as well?
-run_on_exit_callbacks(Id) ->
-    Callbacks = get_on_exit_callbacks(Id),
-    persistent_term:erase({?MODULE, on_exit, Id}),
-    lists:foreach(fun(Fun) -> Fun() end, Callbacks).
+on_exit(Fun) ->
+    Janitor = get_or_spawn_janitor(),
+    ok = emqx_test_janitor:push_on_exit_callback(Janitor, Fun).

apps/emqx/test/emqx_test_janitor.erl

@@ -0,0 +1,69 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 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_test_janitor).
+
+-behaviour(gen_server).
+
+%% `gen_server' API
+-export([
+    init/1,
+    handle_call/3,
+    handle_cast/2,
+    handle_info/2,
+    terminate/2
+]).
+
+%% API
+-export([
+    start_link/0,
+    push_on_exit_callback/2
+]).
+
+%%----------------------------------------------------------------------------------
+%% API
+%%----------------------------------------------------------------------------------
+
+start_link() ->
+    gen_server:start_link(?MODULE, self(), []).
+
+push_on_exit_callback(Server, Callback) when is_function(Callback, 0) ->
+    gen_server:call(Server, {push, Callback}).
+
+%%----------------------------------------------------------------------------------
+%% `gen_server' API
+%%----------------------------------------------------------------------------------
+
+init(Parent) ->
+    process_flag(trap_exit, true),
+    Ref = monitor(process, Parent),
+    {ok, #{callbacks => [], owner => {Ref, Parent}}}.
+
+terminate(_Reason, #{callbacks := Callbacks}) ->
+    lists:foreach(fun(Fun) -> Fun() end, Callbacks).
+
+handle_call({push, Callback}, _From, State = #{callbacks := Callbacks}) ->
+    {reply, ok, State#{callbacks := [Callback | Callbacks]}};
+handle_call(_Req, _From, State) ->
+    {reply, error, State}.
+
+handle_cast(_Req, State) ->
+    {noreply, State}.
+
+handle_info({'DOWN', Ref, process, Parent, _Reason}, State = #{owner := {Ref, Parent}}) ->
+    {stop, normal, State};
+handle_info(_Msg, State) ->
+    {noreply, State}.

lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_gcp_pubsub_SUITE.erl

@@ -16,10 +16,7 @@
 -define(BRIDGE_TYPE, gcp_pubsub).
 -define(BRIDGE_TYPE_BIN, <<"gcp_pubsub">>).
 
--import(emqx_common_test_helpers, [on_exit/2, run_on_exit_callbacks/1]).
-
--define(on_exit_key(TESTCASE), {?MODULE, TESTCASE}).
--define(on_exit(FUN), on_exit({?MODULE, ?FUNCTION_NAME}, FUN)).
+-import(emqx_common_test_helpers, [on_exit/1]).
 
 %%------------------------------------------------------------------------------
 %% CT boilerplate
@@ -138,9 +135,8 @@ init_per_testcase(TestCase, Config0) ->
     Config = generate_config(Config0),
     [{telemetry_table, Tid} | Config].
 
-end_per_testcase(TestCase, _Config) ->
+end_per_testcase(_TestCase, _Config) ->
     ok = snabbkaffe:stop(),
-    run_on_exit_callbacks(?on_exit_key(TestCase)),
     delete_all_bridges(),
     ok = emqx_connector_web_hook_server:stop(),
     ok.
@@ -515,7 +511,7 @@ install_telemetry_handler(TestCase) ->
         end,
         unused_config
     ),
-    on_exit(?on_exit_key(TestCase), fun() ->
+    on_exit(fun() ->
         telemetry:detach(HandlerId),
         ets:delete(Tid)
     end),
@@ -567,7 +563,7 @@ t_publish_success(Config) ->
         end
     ),
     {ok, #{<<"id">> := RuleId}} = create_rule_and_action_http(Config),
-    ?on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
+    on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
     assert_empty_metrics(ResourceId),
     Payload = <<"payload">>,
     Message = emqx_message:make(Topic, Payload),
@@ -669,7 +665,7 @@ t_publish_templated(Config) ->
         end
     ),
     {ok, #{<<"id">> := RuleId}} = create_rule_and_action_http(Config),
-    ?on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
+    on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
     assert_empty_metrics(ResourceId),
     Payload = <<"payload">>,
     Message =
@@ -734,7 +730,7 @@ t_publish_success_batch(Config) ->
         )
     ),
     {ok, #{<<"id">> := RuleId}} = create_rule_and_action_http(Config),
-    ?on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
+    on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
     assert_empty_metrics(ResourceId),
     NumMessages = BatchSize * 2,
     Messages = [emqx_message:make(Topic, integer_to_binary(N)) || N <- lists:seq(1, NumMessages)],
@@ -916,7 +912,7 @@ t_publish_econnrefused(Config) ->
     %% in ehttpc.
     {ok, _} = create_bridge(Config, #{<<"pipelining">> => 1}),
     {ok, #{<<"id">> := RuleId}} = create_rule_and_action_http(Config),
-    ?on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
+    on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
     assert_empty_metrics(ResourceId),
     ok = emqx_connector_web_hook_server:stop(),
     do_econnrefused_or_timeout_test(Config, econnrefused).
@@ -931,7 +927,7 @@ t_publish_timeout(Config) ->
         <<"resource_opts">> => #{<<"batch_size">> => 1}
     }),
     {ok, #{<<"id">> := RuleId}} = create_rule_and_action_http(Config),
-    ?on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
+    on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
     assert_empty_metrics(ResourceId),
     TestPid = self(),
     TimeoutHandler =
@@ -1132,7 +1128,7 @@ t_success_no_body(Config) ->
     Topic = <<"t/topic">>,
     {ok, _} = create_bridge(Config),
     {ok, #{<<"id">> := RuleId}} = create_rule_and_action_http(Config),
-    ?on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
+    on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
     Payload = <<"payload">>,
     Message = emqx_message:make(Topic, Payload),
     ?check_trace(
@@ -1170,7 +1166,7 @@ t_failure_with_body(Config) ->
     Topic = <<"t/topic">>,
     {ok, _} = create_bridge(Config),
     {ok, #{<<"id">> := RuleId}} = create_rule_and_action_http(Config),
-    ?on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
+    on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
     Payload = <<"payload">>,
     Message = emqx_message:make(Topic, Payload),
     ?check_trace(
@@ -1208,7 +1204,7 @@ t_failure_no_body(Config) ->
     Topic = <<"t/topic">>,
     {ok, _} = create_bridge(Config),
     {ok, #{<<"id">> := RuleId}} = create_rule_and_action_http(Config),
-    ?on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
+    on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
     Payload = <<"payload">>,
     Message = emqx_message:make(Topic, Payload),
     ?check_trace(
@@ -1257,7 +1253,7 @@ t_unrecoverable_error(Config) ->
     {ok, _} = create_bridge(Config),
     assert_empty_metrics(ResourceId),
     {ok, #{<<"id">> := RuleId}} = create_rule_and_action_http(Config),
-    ?on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
+    on_exit(fun() -> ok = emqx_rule_engine:delete_rule(RuleId) end),
     Payload = <<"payload">>,
     Message = emqx_message:make(Topic, Payload),
     ?check_trace(