Merge pull request #9002 from zhongwencool/fix-listener-crash

fix: /listeners API crash when there are nodes still in initial conf

CHANGES-5.0.md

@@ -7,6 +7,7 @@
 ## Bug fixes
 
 * Check ACLs for last will testament topic before publishing the message. [#8930](https://github.com/emqx/emqx/pull/8930)
+* Fix GET /listeners API crash When some nodes still in initial configuration. [#9002](https://github.com/emqx/emqx/pull/9002)
 
 # 5.0.8
 

apps/emqx/src/emqx_listeners.erl

@@ -87,12 +87,18 @@ format_list(Listener) ->
     ].
 
 do_list_raw() ->
+    %% GET /listeners from other nodes returns [] when init config is not loaded.
+    case emqx_app:get_init_config_load_done() of
+        true ->
             Key = <<"listeners">>,
             Raw = emqx_config:get_raw([Key], #{}),
             SchemaMod = emqx_config:get_schema_mod(Key),
             #{Key := RawWithDefault} = emqx_config:fill_defaults(SchemaMod, #{Key => Raw}, #{}),
             Listeners = maps:to_list(RawWithDefault),
-    lists:flatmap(fun format_raw_listeners/1, Listeners).
+            lists:flatmap(fun format_raw_listeners/1, Listeners);
+        false ->
+            []
+    end.
 
 format_raw_listeners({Type0, Conf}) ->
     Type = binary_to_atom(Type0),

apps/emqx/test/emqx_common_test_helpers.erl

@@ -632,9 +632,9 @@ setup_node(Node, Opts) when is_map(Opts) ->
     %% Here we start the apps
     EnvHandlerForRpc =
         fun(App) ->
-            %% We load configuration, and than set the special enviroment variable
+            %% We load configuration, and than set the special environment variable
             %% which says that emqx shouldn't load configuration at startup
-            %% Otherwise, configuration get's loaded and all preset env in envhandler is lost
+            %% Otherwise, configuration gets loaded and all preset env in EnvHandler is lost
             LoadSchema andalso
                 begin
                     emqx_config:init_load(SchemaMod),

apps/emqx_conf/src/emqx_conf_app.erl

@@ -152,11 +152,17 @@ copy_override_conf_from_core_node() ->
                 _ ->
                     [{ok, Info} | _] = lists:sort(fun conf_sort/2, Ready),
                     #{node := Node, conf := RawOverrideConf, tnx_id := TnxId} = Info,
-                    Msg = #{
+                    ?SLOG(debug, #{
                         msg => "copy_overide_conf_from_core_node_success",
-                        node => Node
-                    },
-                    ?SLOG(debug, Msg),
+                        node => Node,
+                        cluster_override_conf_file => application:get_env(
+                            emqx, cluster_override_conf_file
+                        ),
+                        local_override_conf_file => application:get_env(
+                            emqx, local_override_conf_file
+                        ),
+                        data_dir => emqx:data_dir()
+                    }),
                     ok = emqx_config:save_to_override_conf(
                         RawOverrideConf,
                         #{override_to => cluster}

apps/emqx_management/test/emqx_mgmt_api_listeners_SUITE.erl

@@ -100,6 +100,68 @@ t_wss_crud_listeners_by_id(_) ->
     Type = <<"wss">>,
     crud_listeners_by_id(ListenerId, NewListenerId, MinListenerId, BadId, Type).
 
+t_api_listeners_list_not_ready(_Config) ->
+    net_kernel:start(['listeners@127.0.0.1', longnames]),
+    ct:timetrap({seconds, 120}),
+    snabbkaffe:fix_ct_logging(),
+    Cluster = [{Name, Opts}, {Name1, Opts1}] = cluster([core, core]),
+    ct:pal("Starting ~p", [Cluster]),
+    Node1 = emqx_common_test_helpers:start_slave(Name, Opts),
+    Node2 = emqx_common_test_helpers:start_slave(Name1, Opts1),
+    try
+        L1 = get_tcp_listeners(Node1),
+
+        %% test init_config not ready.
+        _ = rpc:call(Node1, application, set_env, [emqx, init_config_load_done, false]),
+        assert_config_load_not_done(Node1),
+
+        L2 = get_tcp_listeners(Node1),
+        L3 = get_tcp_listeners(Node2),
+
+        Comment = #{
+            node1 => rpc:call(Node1, mria_mnesia, running_nodes, []),
+            node2 => rpc:call(Node2, mria_mnesia, running_nodes, [])
+        },
+
+        ?assert(length(L1) > length(L2), Comment),
+        ?assertEqual(length(L2), length(L3), Comment)
+    after
+        emqx_common_test_helpers:stop_slave(Node1),
+        emqx_common_test_helpers:stop_slave(Node2)
+    end.
+
+get_tcp_listeners(Node) ->
+    Query = #{query_string => #{<<"type">> => tcp}},
+    {200, L} = rpc:call(Node, emqx_mgmt_api_listeners, list_listeners, [get, Query]),
+    [#{node_status := NodeStatus}] = L,
+    ct:pal("Node:~p:~p", [Node, L]),
+    NodeStatus.
+
+assert_config_load_not_done(Node) ->
+    Done = rpc:call(Node, emqx_app, get_init_config_load_done, []),
+    ?assertNot(Done, #{node => Node}).
+
+cluster(Specs) ->
+    Env = [
+        {emqx, init_config_load_done, false},
+        {emqx, boot_modules, []}
+    ],
+    emqx_common_test_helpers:emqx_cluster(Specs, [
+        {env, Env},
+        {apps, [emqx_conf]},
+        {load_schema, false},
+        {join_to, true},
+        {env_handler, fun
+            (emqx) ->
+                application:set_env(emqx, boot_modules, []),
+                %% test init_config not ready.
+                application:set_env(emqx, init_config_load_done, false),
+                ok;
+            (_) ->
+                ok
+        end}
+    ]).
+
 crud_listeners_by_id(ListenerId, NewListenerId, MinListenerId, BadId, Type) ->
     OriginPath = emqx_mgmt_api_test_util:api_path(["listeners", ListenerId]),
     NewPath = emqx_mgmt_api_test_util:api_path(["listeners", NewListenerId]),