Merge branch 'emqx:master' into feat_round_robin_per_node

CHANGES-5.0.md

@@ -46,6 +46,7 @@
 * Improve authentication tracing. [#8554](https://github.com/emqx/emqx/pull/8554)
 * Standardize the '/listeners' and `/gateway/<name>/listeners` API fields.
   It will introduce some incompatible updates, see [#8571](https://github.com/emqx/emqx/pull/8571)
+* Add option to perform GC on connection process after TLS/SSL handshake is performed. [#8637](https://github.com/emqx/emqx/pull/8637)
 
 # 5.0.3
 

apps/emqx/i18n/emqx_schema_i18n.conf

@@ -1843,6 +1843,23 @@ Maximum time duration allowed for the handshake to complete
     }
 }
 
+server_ssl_opts_schema_gc_after_handshake {
+    desc {
+        en: """
+Memory usage tuning. If enabled, will immediately perform a garbage collection after
+the TLS/SSL handshake.
+"""
+        zh: """
+内存使用调优。如果启用，将在TLS/SSL握手完成后立即执行垃圾回收。
+TLS/SSL握手建立后立即进行GC。
+"""
+    }
+    label: {
+        en: "Perform GC after handshake"
+        zh: "握手后执行GC"
+    }
+}
+
 fields_listeners_tcp {
     desc {
         en: """

apps/emqx/rebar.config

@@ -26,7 +26,7 @@
     {gproc, {git, "https://github.com/uwiger/gproc", {tag, "0.8.0"}}},
     {jiffy, {git, "https://github.com/emqx/jiffy", {tag, "1.0.5"}}},
     {cowboy, {git, "https://github.com/emqx/cowboy", {tag, "2.9.0"}}},
-    {esockd, {git, "https://github.com/emqx/esockd", {tag, "5.9.3"}}},
+    {esockd, {git, "https://github.com/emqx/esockd", {tag, "5.9.4"}}},
     {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.13.3"}}},
     {gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "2.8.1"}}},
     {hocon, {git, "https://github.com/emqx/hocon.git", {tag, "0.29.0"}}},

apps/emqx/src/emqx_schema.erl

@@ -1969,7 +1969,15 @@ server_ssl_opts_schema(Defaults, IsRanchListener) ->
                         }
                     )}
              || IsRanchListener
-            ]
+            ] ++
+                [
+                    {"gc_after_handshake",
+                        sc(boolean(), #{
+                            default => false,
+                            desc => ?DESC(server_ssl_opts_schema_gc_after_handshake)
+                        })}
+                 || not IsRanchListener
+                ]
         ].
 
 %% @doc Make schema for SSL client.

apps/emqx/test/emqx_schema_tests.erl

@@ -141,3 +141,38 @@ bad_tls_version_test() ->
         validate(Sc, #{<<"versions">> => [<<"foo">>]})
     ),
     ok.
+
+ssl_opts_gc_after_handshake_test_rancher_listener_test() ->
+    Sc = emqx_schema:server_ssl_opts_schema(
+        #{
+            gc_after_handshake => false
+        },
+        _IsRanchListener = true
+    ),
+    ?assertThrow(
+        {_Sc, [
+            #{
+                kind := validation_error,
+                reason := unknown_fields,
+                unknown := <<"gc_after_handshake">>
+            }
+        ]},
+        validate(Sc, #{<<"gc_after_handshake">> => true})
+    ),
+    ok.
+
+ssl_opts_gc_after_handshake_test_not_rancher_listener_test() ->
+    Sc = emqx_schema:server_ssl_opts_schema(
+        #{
+            gc_after_handshake => false
+        },
+        _IsRanchListener = false
+    ),
+    Checked = validate(Sc, #{<<"gc_after_handshake">> => <<"true">>}),
+    ?assertMatch(
+        #{
+            gc_after_handshake := true
+        },
+        Checked
+    ),
+    ok.

apps/emqx_gateway/test/emqx_gateway_auth_ct.erl

@@ -143,7 +143,9 @@ on_start_auth(authn_http) ->
     Setup = fun(Gateway) ->
         Path = io_lib:format("/gateway/~ts/authentication", [Gateway]),
         {204, _} = request(delete, Path),
-        {201, _} = request(post, Path, http_authn_config())
+        timer:sleep(200),
+        {201, _} = request(post, Path, http_authn_config()),
+        timer:sleep(200)
     end,
     lists:foreach(Setup, ?GATEWAYS),
 

apps/emqx_gateway/test/emqx_sn_protocol_SUITE.erl

@@ -103,7 +103,7 @@ init_per_suite(Config) ->
 
 end_per_suite(_) ->
     {ok, _} = emqx:remove_config([gateway, mqttsn]),
-    emqx_mgmt_api_test_util:end_suite([emqx_gateway, emqx_auhtn, emqx_conf]).
+    emqx_mgmt_api_test_util:end_suite([emqx_gateway, emqx_authn, emqx_conf]).
 
 restart_mqttsn_with_subs_resume_on() ->
     Conf = emqx:get_raw_config([gateway, mqttsn]),

lib-ee/emqx_license/i18n/emqx_license_http_api.conf

@@ -10,10 +10,21 @@ emqx_license_http_api {
     }
   }
 
-  desc_license_upload_api {
+  desc_license_file_api {
     desc {
-      en: "Upload a license file or key"
+      en: "Upload a license file"
-      zh: "上传许可证文件或密钥"
+      zh: "上传一个许可证文件"
+    }
+    label: {
+      en: "Update license"
+      zh: "更新许可证"
+    }
+  }
+
+  desc_license_key_api {
+    desc {
+      en: "Update a license key"
+      zh: "更新一个许可证密钥"
     }
     label: {
       en: "Update license"

lib-ee/emqx_license/src/emqx_license.erl

@@ -22,6 +22,7 @@
     read_license/0,
     read_license/1,
     update_file/1,
+    update_file_contents/1,
     update_key/1,
     license_dir/0,
     save_and_backup_license/1
@@ -70,16 +71,21 @@ relative_license_path() ->
 update_file(Filename) when is_binary(Filename); is_list(Filename) ->
     case file:read_file(Filename) of
         {ok, Contents} ->
-            Result = emqx_conf:update(
+            update_file_contents(Contents);
-                ?CONF_KEY_PATH,
-                {file, Contents},
-                #{rawconf_with_defaults => true, override_to => local}
-            ),
-            handle_config_update_result(Result);
         {error, Error} ->
             {error, Error}
     end.
 
+-spec update_file_contents(binary() | string()) ->
+    {ok, emqx_config:update_result()} | {error, emqx_config:update_error()}.
+update_file_contents(Contents) when is_binary(Contents) ->
+    Result = emqx_conf:update(
+        ?CONF_KEY_PATH,
+        {file, Contents},
+        #{rawconf_with_defaults => true, override_to => local}
+    ),
+    handle_config_update_result(Result).
+
 -spec update_key(binary() | string()) ->
     {ok, emqx_config:update_result()} | {error, emqx_config:update_error()}.
 update_key(Value) when is_binary(Value); is_list(Value) ->

lib-ee/emqx_license/src/emqx_license_http_api.erl

@@ -18,11 +18,11 @@
 
 -export([
     '/license'/2,
-    '/license/upload'/2
+    '/license/key'/2,
+    '/license/file'/2
 ]).
 
 -define(BAD_REQUEST, 'BAD_REQUEST').
--define(NOT_FOUND, 'NOT_FOUND').
 
 namespace() -> "license_http_api".
 
@@ -32,7 +32,8 @@ api_spec() ->
 paths() ->
     [
         "/license",
-        "/license/upload"
+        "/license/key",
+        "/license/file"
     ].
 
 schema("/license") ->
@@ -54,15 +55,36 @@ schema("/license") ->
             }
         }
     };
-schema("/license/upload") ->
+schema("/license/file") ->
     #{
-        'operationId' => '/license/upload',
+        'operationId' => '/license/file',
         post => #{
             tags => [<<"license">>],
-            summary => <<"Upload license">>,
+            summary => <<"Upload license file">>,
-            description => ?DESC("desc_license_upload_api"),
+            description => ?DESC("desc_license_file_api"),
+            'requestBody' => emqx_dashboard_swagger:file_schema(filename),
+            responses => #{
+                200 => emqx_dashboard_swagger:schema_with_examples(
+                    map(),
+                    #{
+                        sample_license_info => #{
+                            value => sample_license_info_response()
+                        }
+                    }
+                ),
+                400 => emqx_dashboard_swagger:error_codes([?BAD_REQUEST], <<"Bad license file">>)
+            }
+        }
+    };
+schema("/license/key") ->
+    #{
+        'operationId' => '/license/key',
+        post => #{
+            tags => [<<"license">>],
+            summary => <<"Update license key">>,
+            description => ?DESC("desc_license_key_api"),
             'requestBody' => emqx_dashboard_swagger:schema_with_examples(
-                emqx_license_schema:license_type(),
+                emqx_license_schema:key_license(),
                 #{
                     license_key => #{
                         summary => <<"License key string">>,
@@ -71,14 +93,6 @@ schema("/license/upload") ->
                             <<"connection_low_watermark">> => "75%",
                             <<"connection_high_watermark">> => "80%"
                         }
-                    },
-                    license_file => #{
-                        summary => <<"Path to a license file">>,
-                        value => #{
-                            <<"file">> => <<"/path/to/license">>,
-                            <<"connection_low_watermark">> => "75%",
-                            <<"connection_high_watermark">> => "80%"
-                        }
                     }
                 }
             ),
@@ -91,8 +105,7 @@ schema("/license/upload") ->
                         }
                     }
                 ),
-                400 => emqx_dashboard_swagger:error_codes([?BAD_REQUEST], <<"Bad license key">>),
+                400 => emqx_dashboard_swagger:error_codes([?BAD_REQUEST], <<"Bad license file">>)
-                404 => emqx_dashboard_swagger:error_codes([?NOT_FOUND], <<"File not found">>)
             }
         }
     }.
@@ -117,37 +130,26 @@ error_msg(Code, Msg) ->
     License = maps:from_list(emqx_license_checker:dump()),
     {200, License}.
 
-'/license/upload'(post, #{body := #{<<"file">> := Filepath}}) ->
+'/license/file'(post, #{body := #{<<"filename">> := #{type := _} = File}}) ->
-    case emqx_license:update_file(Filepath) of
+    [{_Filename, Contents}] = maps:to_list(maps:without([type], File)),
-        {error, enoent} ->
+    case emqx_license:update_file_contents(Contents) of
-            ?SLOG(error, #{
-                msg => "license_file_not_found",
-                path => Filepath
-            }),
-            {404, error_msg(?NOT_FOUND, <<"File not found">>)};
-        {error, Error} when is_atom(Error) ->
-            ?SLOG(error, #{
-                msg => "bad_license_file",
-                reason => Error,
-                path => Filepath
-            }),
-            {400, error_msg(?BAD_REQUEST, emqx_misc:explain_posix(Error))};
         {error, Error} ->
             ?SLOG(error, #{
                 msg => "bad_license_file",
-                reason => Error,
+                reason => Error
-                path => Filepath
             }),
             {400, error_msg(?BAD_REQUEST, <<"Bad license file">>)};
         {ok, _} ->
             ?SLOG(info, #{
-                msg => "updated_license_file",
+                msg => "updated_license_file"
-                path => Filepath
             }),
             License = maps:from_list(emqx_license_checker:dump()),
             {200, License}
     end;
-'/license/upload'(post, #{body := #{<<"key">> := Key}}) ->
+'/license/file'(post, _Params) ->
+    {400, error_msg(?BAD_REQUEST, <<"Invalid request params">>)}.
+
+'/license/key'(post, #{body := #{<<"key">> := Key}}) ->
     case emqx_license:update_key(Key) of
         {error, Error} ->
             ?SLOG(error, #{
@@ -160,5 +162,5 @@ error_msg(Code, Msg) ->
             License = maps:from_list(emqx_license_checker:dump()),
             {200, License}
     end;
-'/license/upload'(post, _Params) ->
+'/license/key'(post, _Params) ->
     {400, error_msg(?BAD_REQUEST, <<"Invalid request params">>)}.

lib-ee/emqx_license/src/emqx_license_parser.erl

@@ -72,9 +72,16 @@
 %% API
 %%--------------------------------------------------------------------
 
+-ifdef(TEST).
+-spec parse(string() | binary()) -> {ok, license()} | {error, term()}.
+parse(Content) ->
+    PubKey = persistent_term:get(emqx_license_test_pubkey, ?PUBKEY),
+    parse(Content, PubKey).
+-else.
 -spec parse(string() | binary()) -> {ok, license()} | {error, term()}.
 parse(Content) ->
     parse(Content, ?PUBKEY).
+-endif.
 
 parse(Content, Pem) ->
     [PemEntry] = public_key:pem_decode(Pem),

lib-ee/emqx_license/src/emqx_license_schema.erl

@@ -15,7 +15,9 @@
 -export([roots/0, fields/1, validations/0, desc/1]).
 
 -export([
-    license_type/0
+    license_type/0,
+    key_license/0,
+    file_license/0
 ]).
 
 roots() ->
@@ -99,10 +101,16 @@ validations() ->
 
 license_type() ->
     hoconsc:union([
-        hoconsc:ref(?MODULE, key_license),
+        key_license(),
-        hoconsc:ref(?MODULE, file_license)
+        file_license()
     ]).
 
+key_license() ->
+    hoconsc:ref(?MODULE, key_license).
+
+file_license() ->
+    hoconsc:ref(?MODULE, file_license).
+
 check_license_watermark(Conf) ->
     case hocon_maps:get("license.connection_low_watermark", Conf) of
         undefined ->

lib-ee/emqx_license/test/emqx_license_SUITE.erl

@@ -141,16 +141,9 @@ setup_test(TestCase, Config) when
                     emqx_config:put([license], LicConfig),
                     RawConfig = #{<<"type">> => file, <<"file">> => LicensePath},
                     emqx_config:put_raw([<<"license">>], RawConfig),
-                    ok = meck:new(emqx_license, [non_strict, passthrough, no_history, no_link]),
+                    ok = persistent_term:put(
-                    meck:expect(
+                        emqx_license_test_pubkey,
-                        emqx_license_parser,
+                        emqx_license_test_lib:public_key_pem()
-                        parse,
-                        fun(X) ->
-                            emqx_license_parser:parse(
-                                X,
-                                emqx_license_test_lib:public_key_pem()
-                            )
-                        end
                     ),
                     ok;
                 (_) ->

lib-ee/emqx_license/test/emqx_license_http_api_SUITE.erl

@@ -21,27 +21,16 @@ all() ->
 init_per_suite(Config) ->
     _ = application:load(emqx_conf),
     emqx_config:save_schema_mod_and_names(emqx_license_schema),
-    ok = meck:new(emqx_license_parser, [non_strict, passthrough, no_history, no_link]),
-    ok = meck:expect(
-        emqx_license_parser,
-        parse,
-        fun(X) ->
-            emqx_license_parser:parse(
-                X,
-                emqx_license_test_lib:public_key_pem()
-            )
-        end
-    ),
     emqx_common_test_helpers:start_apps([emqx_license, emqx_dashboard], fun set_special_configs/1),
     Config.
 
 end_per_suite(_) ->
     emqx_common_test_helpers:stop_apps([emqx_license, emqx_dashboard]),
-    ok = meck:unload([emqx_license_parser]),
     Config = #{type => file, file => emqx_license_test_lib:default_license()},
     emqx_config:put([license], Config),
     RawConfig = #{<<"type">> => file, <<"file">> => emqx_license_test_lib:default_license()},
     emqx_config:put_raw([<<"license">>], RawConfig),
+    persistent_term:erase(emqx_license_test_pubkey),
     ok.
 
 set_special_configs(emqx_dashboard) ->
@@ -51,7 +40,12 @@ set_special_configs(emqx_license) ->
     Config = #{type => key, key => LicenseKey},
     emqx_config:put([license], Config),
     RawConfig = #{<<"type">> => key, <<"key">> => LicenseKey},
-    emqx_config:put_raw([<<"license">>], RawConfig);
+    emqx_config:put_raw([<<"license">>], RawConfig),
+    ok = persistent_term:put(
+        emqx_license_test_pubkey,
+        emqx_license_test_lib:public_key_pem()
+    ),
+    ok;
 set_special_configs(_) ->
     ok.
 
@@ -88,6 +82,14 @@ assert_untouched_license() ->
         get_license()
     ).
 
+multipart_formdata_request(Uri, File) ->
+    emqx_dashboard_api_test_helpers:multipart_formdata_request(
+        Uri,
+        _Username = <<"license_admin">>,
+        _Fields = [],
+        [File]
+    ).
+
 %%------------------------------------------------------------------------------
 %% Testcases
 %%------------------------------------------------------------------------------
@@ -114,109 +116,72 @@ t_license_info(_Config) ->
 
 t_license_upload_file_success(_Config) ->
     NewKey = emqx_license_test_lib:make_license(#{max_connections => "999"}),
-    Path = "/tmp/new.lic",
+    Res = multipart_formdata_request(
-    ok = file:write_file(Path, NewKey),
+        uri(["license", "file"]),
-    try
+        {filename, "emqx.lic", NewKey}
-        Res = request(
-            post,
-            uri(["license", "upload"]),
-            #{file => Path}
-        ),
-        ?assertMatch({ok, 200, _}, Res),
-        {ok, 200, Payload} = Res,
-        ?assertEqual(
-            #{
-                <<"customer">> => <<"Foo">>,
-                <<"customer_type">> => 10,
-                <<"deployment">> => <<"bar-deployment">>,
-                <<"email">> => <<"contact@foo.com">>,
-                <<"expiry">> => false,
-                <<"expiry_at">> => <<"2295-10-27">>,
-                <<"max_connections">> => 999,
-                <<"start_at">> => <<"2022-01-11">>,
-                <<"type">> => <<"trial">>
-            },
-            emqx_json:decode(Payload, [return_maps])
-        ),
-        ?assertMatch(
-            #{max_connections := 999},
-            get_license()
-        ),
-        ok
-    after
-        ok = file:delete(Path),
-        ok
-    end.
-
-t_license_upload_file_not_found(_Config) ->
-    Res = request(
-        post,
-        uri(["license", "upload"]),
-        #{file => "/tmp/inexistent.lic"}
     ),
-
+    ?assertMatch({ok, 200, _}, Res),
-    ?assertMatch({ok, 404, _}, Res),
+    {ok, 200, Payload} = Res,
-    {ok, 404, Payload} = Res,
     ?assertEqual(
         #{
-            <<"code">> => <<"NOT_FOUND">>,
+            <<"customer">> => <<"Foo">>,
-            <<"message">> => <<"File not found">>
+            <<"customer_type">> => 10,
+            <<"deployment">> => <<"bar-deployment">>,
+            <<"email">> => <<"contact@foo.com">>,
+            <<"expiry">> => false,
+            <<"expiry_at">> => <<"2295-10-27">>,
+            <<"max_connections">> => 999,
+            <<"start_at">> => <<"2022-01-11">>,
+            <<"type">> => <<"trial">>
         },
         emqx_json:decode(Payload, [return_maps])
     ),
-    assert_untouched_license(),
+    ?assertMatch(
+        #{max_connections := 999},
+        get_license()
+    ),
     ok.
 
-t_license_upload_file_reading_error(_Config) ->
+t_license_upload_file_bad_license(_Config) ->
-    %% eisdir
+    Res = multipart_formdata_request(
-    Path = "/tmp/",
+        uri(["license", "file"]),
-    Res = request(
+        {filename, "bad.lic", <<"bad key">>}
-        post,
-        uri(["license", "upload"]),
-        #{file => Path}
     ),
     ?assertMatch({ok, 400, _}, Res),
     {ok, 400, Payload} = Res,
     ?assertEqual(
         #{
             <<"code">> => <<"BAD_REQUEST">>,
-            <<"message">> => <<"Illegal operation on a directory">>
+            <<"message">> => <<"Bad license file">>
         },
         emqx_json:decode(Payload, [return_maps])
     ),
     assert_untouched_license(),
     ok.
 
-t_license_upload_file_bad_license(_Config) ->
+t_license_upload_file_not_json(_Config) ->
-    Path = "/tmp/bad.lic",
+    Res = request(
-    ok = file:write_file(Path, <<"bad key">>),
+        post,
-    try
+        uri(["license", "file"]),
-        Res = request(
+        <<"">>
-            post,
+    ),
-            uri(["license", "upload"]),
+    ?assertMatch({ok, 400, _}, Res),
-            #{file => Path}
+    {ok, 400, Payload} = Res,
-        ),
+    ?assertEqual(
-        ?assertMatch({ok, 400, _}, Res),
+        #{
-        {ok, 400, Payload} = Res,
+            <<"code">> => <<"BAD_REQUEST">>,
-        ?assertEqual(
+            <<"message">> => <<"Invalid request params">>
-            #{
+        },
-                <<"code">> => <<"BAD_REQUEST">>,
+        emqx_json:decode(Payload, [return_maps])
-                <<"message">> => <<"Bad license file">>
+    ),
-            },
+    assert_untouched_license(),
-            emqx_json:decode(Payload, [return_maps])
+    ok.
-        ),
-        assert_untouched_license(),
-        ok
-    after
-        ok = file:delete(Path),
-        ok
-    end.
 
 t_license_upload_key_success(_Config) ->
     NewKey = emqx_license_test_lib:make_license(#{max_connections => "999"}),
     Res = request(
         post,
-        uri(["license", "upload"]),
+        uri(["license", "key"]),
         #{key => NewKey}
     ),
     ?assertMatch({ok, 200, _}, Res),
@@ -245,7 +210,7 @@ t_license_upload_key_bad_key(_Config) ->
     BadKey = <<"bad key">>,
     Res = request(
         post,
-        uri(["license", "upload"]),
+        uri(["license", "key"]),
         #{key => BadKey}
     ),
     ?assertMatch({ok, 400, _}, Res),
@@ -259,3 +224,21 @@ t_license_upload_key_bad_key(_Config) ->
     ),
     assert_untouched_license(),
     ok.
+
+t_license_upload_key_not_json(_Config) ->
+    Res = request(
+        post,
+        uri(["license", "key"]),
+        <<"">>
+    ),
+    ?assertMatch({ok, 400, _}, Res),
+    {ok, 400, Payload} = Res,
+    ?assertEqual(
+        #{
+            <<"code">> => <<"BAD_REQUEST">>,
+            <<"message">> => <<"Invalid request params">>
+        },
+        emqx_json:decode(Payload, [return_maps])
+    ),
+    assert_untouched_license(),
+    ok.

mix.exs

@@ -51,7 +51,7 @@ defmodule EMQXUmbrella.MixProject do
       {:gproc, github: "uwiger/gproc", tag: "0.8.0", override: true},
       {:jiffy, github: "emqx/jiffy", tag: "1.0.5", override: true},
       {:cowboy, github: "emqx/cowboy", tag: "2.9.0", override: true},
-      {:esockd, github: "emqx/esockd", tag: "5.9.3", override: true},
+      {:esockd, github: "emqx/esockd", tag: "5.9.4", override: true},
       {:ekka, github: "emqx/ekka", tag: "0.13.3", override: true},
       {:gen_rpc, github: "emqx/gen_rpc", tag: "2.8.1", override: true},
       {:grpc, github: "emqx/grpc-erl", tag: "0.6.6", override: true},

rebar.config

@@ -53,7 +53,7 @@
     , {gproc, {git, "https://github.com/uwiger/gproc", {tag, "0.8.0"}}}
     , {jiffy, {git, "https://github.com/emqx/jiffy", {tag, "1.0.5"}}}
     , {cowboy, {git, "https://github.com/emqx/cowboy", {tag, "2.9.0"}}}
-    , {esockd, {git, "https://github.com/emqx/esockd", {tag, "5.9.3"}}}
+    , {esockd, {git, "https://github.com/emqx/esockd", {tag, "5.9.4"}}}
     , {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.13.3"}}}
     , {gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "2.8.1"}}}
     , {grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.6"}}}