feat: configurable TCP keepalive

changes/v4.4.19-en.md

@@ -1,5 +1,9 @@
 # v4.4.19
 
+- Added support for TCP keep-alive in MQTT/TCP and MQTT/SSL listeners [#10854](https://github.com/emqx/emqx/pull/10854).
+
+  A new configuration option has been added: `zone.<zone-name>.tcp_keepalive = Idle,Interval,Probes`. Users can enable the TCP layer's Keep Alive feature and specify time parameters using this configuration. This configuration is only effective on Linux and MacOS systems.
+
 ## Enhancements
 
 - Improving error logs related to Proxy Protocol [emqx/esockd#177](https://github.com/emqx/esockd/pull/177).

changes/v4.4.19-zh.md

@@ -2,6 +2,10 @@
 
 ## 增强
 
+- 为 MQTT/TCP 和 MQTT/SSL 监听器增加 TCP Keep Alive 的支持 [#10854](https://github.com/emqx/emqx/pull/10854)。
+
+  现在增加了一个配置项：`zone.<zone-name>.tcp_keepalive = Idle,Interval,Probes`，用户可以通过此配置来启用 TCP 层的 Keep Alive 功能并指定时间参数。此配置仅在 Linux 和 MacOS 系统上生效。
+
 - 改进 Proxy Protocol 相关的错误日志 [emqx/esockd#177](https://github.com/emqx/esockd/pull/177)。
 
   改进之前的日志样例:
@@ -18,7 +22,7 @@
   ```
 
 - 增加了一个新功能，为 TLS 监听器启用部分证书链验证[#10553](https://github.com/emqx/emqx/pull/10553)。
-  
+
 - 增加了一个新功能，为 TLS 监听器启用客户端证书扩展密钥使用验证 [#10669](https://github.com/emqx/emqx/pull/10669)。
 
 ## 修复

etc/emqx.conf

@@ -1072,6 +1072,29 @@ zone.external.strict_mode = false
 ## Value: String
 ## zone.external.response_information = example
 
+## Enable TCP keepalive for MQTT connections over TCP or SSL.
+##
+## The value is three comma separated numbers in the format of
+## 'Idle,Interval,Probes'
+##
+## Idle: The number of seconds a connection needs to be idle before
+##       the server begins to send out keep-alive probes (Linux default 7200).
+##
+## Interval: The number of seconds between TCP keep-alive probes (Linux default 75).
+##
+## Probes: The maximum number of TCP keep-alive probes to send before
+##         giving up and killing the connection if no response is
+##         obtained from the other end (Linux default 9).
+##
+## For example "240,30,5" means: EMQX should start sending TCP keepalive probes
+## after the connection is in idel for 240 seconds,
+## and the probes are sent every 30 seconds until a response is received from the MQTT
+## client, if it misses 5 consecutive responses, EMQX should close the connection.
+##
+## Value: string
+## Default: none
+## zone.external.tcp_keepalive = none
+
 ##--------------------------------------------------------------------
 ## Internal Zone
 
@@ -1177,6 +1200,29 @@ zone.internal.strict_mode = false
 ## Value: true | false
 zone.internal.bypass_auth_plugins = true
 
+## Enable TCP keepalive for MQTT connections over TCP or SSL.
+##
+## The value is three comma separated numbers in the format of
+## 'Idle,Interval,Probes'
+##
+## Idle: The number of seconds a connection needs to be idle before
+##       the server begins to send out keep-alive probes (Linux default 7200).
+##
+## Interval: The number of seconds between TCP keep-alive probes (Linux default 75).
+##
+## Probes: The maximum number of TCP keep-alive probes to send before
+##         giving up and killing the connection if no response is
+##         obtained from the other end (Linux default 9).
+##
+## For example "240,30,5" means: EMQX should start sending TCP keepalive probes
+## after the connection is in idel for 240 seconds,
+## and the probes are sent every 30 seconds until a response is received from the MQTT
+## client, if it misses 5 consecutive responses, EMQX should close the connection.
+##
+## Value: string
+## Default: none
+## zone.internal.tcp_keepalive = none
+
 ## CONFIG_SECTION_END=zones ====================================================
 
 ## CONFIG_SECTION_BGN=listeners ================================================

priv/emqx.schema

@@ -166,11 +166,11 @@
              {Ip, Port}
            end,
   Options = fun(static) ->
-                 [{seeds, [list_to_atom(S) || S <- string:tokens(cuttlefish:conf_get("cluster.static.seeds", Conf, ""), ",")]}];
+                 [{seeds, [list_to_atom(S) || S <- string:tokens(cuttlefish:conf_get("cluster.static.seeds", Conf, ""), ", ")]}];
                (mcast) ->
                  {ok, Addr} = inet:parse_address(cuttlefish:conf_get("cluster.mcast.addr", Conf)),
                  {ok, Iface} = inet:parse_address(cuttlefish:conf_get("cluster.mcast.iface", Conf)),
-                 Ports = [list_to_integer(S) || S <- string:tokens(cuttlefish:conf_get("cluster.mcast.ports", Conf), ",")],
+                 Ports = [list_to_integer(S) || S <- string:tokens(cuttlefish:conf_get("cluster.mcast.ports", Conf), ", ")],
                  [{addr, Addr}, {ports, Ports}, {iface, Iface},
                   {ttl, cuttlefish:conf_get("cluster.mcast.ttl", Conf, 1)},
                   {loop, cuttlefish:conf_get("cluster.mcast.loop", Conf, true)}];
@@ -185,7 +185,7 @@
                                             {list_to_atom(Name), Value}
                                         end, Options)
                             end,
-                 [{server, string:tokens(cuttlefish:conf_get("cluster.etcd.server", Conf), ",")},
+                 [{server, string:tokens(cuttlefish:conf_get("cluster.etcd.server", Conf), ", ")},
                   {prefix, cuttlefish:conf_get("cluster.etcd.prefix", Conf, "emqcl")},
                   {node_ttl, cuttlefish:conf_get("cluster.etcd.node_ttl", Conf, 60)},
                   {ssl_options, SslOpts(Conf)}];
@@ -1286,6 +1286,12 @@ end}.
     {datatype, {enum, [true, false]}}
 ]}.
 
+%% @doc Set TCP-keepalive options.
+{mapping, "zone.$name.tcp_keepalive", "emqx.zones", [
+    {default, "none"},
+    {datatype, string}
+]}.
+
 {translation, "emqx.zones", fun(Conf) ->
   Ratelimit = fun(Val) ->
                 [L, D] = string:tokens(Val, ", "),
@@ -1371,6 +1377,15 @@ end}.
                    {quota, {conn_messages_routing, Ratelimit(Val)}};
                (["quota", "overall_messages_routing"], Val) ->
                    {quota, {overall_messages_routing, Ratelimit(Val)}};
+               (["tcp_keepalive"], Val) ->
+                    V = case Val of
+                            "none" ->
+                                false;
+                            _ ->
+                                [Idle, Interval, Probes] = string:tokens(Val, ", "),
+                                {list_to_integer(Idle), list_to_integer(Interval), list_to_integer(Probes)}
+                        end,
+                    {tcp_keepalive, V};
                ([Opt], Val) ->
                     {list_to_atom(Opt), Val}
             end,
@@ -2227,7 +2242,7 @@ end}.
                 end,
 
         CheckOrigin = fun(S) ->
-                        Origins = string:tokens(S, ","),
+                        Origins = string:tokens(S, ", "),
                         [ list_to_binary(string:trim(O)) || O <- Origins]
                       end,
 
@@ -2873,7 +2888,7 @@ end}.
 ]}.
 
 {translation, "emqx.alarm", fun(Conf) ->
-    [{actions, [list_to_atom(Action) || Action <- string:tokens(cuttlefish:conf_get("alarm.actions", Conf), ",")]},
+    [{actions, [list_to_atom(Action) || Action <- string:tokens(cuttlefish:conf_get("alarm.actions", Conf), ", ")]},
      {size_limit, cuttlefish:conf_get("alarm.size_limit", Conf)},
      {validity_period, cuttlefish:conf_get("alarm.validity_period", Conf)}]
 end}.

scripts/macos-sign-binaries.sh

@@ -68,6 +68,7 @@ for f in \
         sasl_auth.so \
         snappyer.so \
         odbcserver \
+        ezstd_nif.so \
     ; do
     find "${REL_DIR}"/lib/ -name "$f" -exec codesign -s "${APPLE_DEVELOPER_IDENTITY}" -f --verbose=4 --timestamp --options=runtime {} \;
 done

scripts/relup-base-packages.sh

@@ -64,17 +64,24 @@ mkdir -p _upgrade_base
 pushd _upgrade_base
 
 otp_vsn_for() {
-    ../scripts/relup-base-vsns.escript otp-vsn-for "${1#[e|v]}" ../data/relup-paths.eterm
+    case "$PROFILE" in
+        *-ee-*)
+            ../scripts/relup-base-vsns.escript otp-vsn-for "${1#[e|v]}" ../data/relup-paths-ee.eterm
+            ;;
+        *)
+            ../scripts/relup-base-vsns.escript otp-vsn-for "${1#[e|v]}" ../data/relup-paths.eterm
+            ;;
+    esac
 }
 
 for tag in $(../scripts/relup-base-vsns.sh $EDITION | xargs echo -n); do
     filename="$PROFILE-${tag#[e|v]}-otp$(otp_vsn_for "$tag")-$SYSTEM-$ARCH.zip"
     url="https://packages.emqx.io/$DIR/$tag/$filename"
-    echo "downloading base package from ${url} ..."
     if [ -f "$filename" ]; then
-        echo "file $filename already downloaded; skikpped"
+        echo "file $filename already downloaded; skipped"
         continue
     fi
+    curl -L -I -m 10 -o /dev/null -s -w "%{http_code}" "${url}" | grep -q -oE "^[23]+"
     echo "downloading base package from ${url} ..."
     curl -L -o "${filename}" "${url}"
     if [ "$SYSTEM" != "centos6" ]; then

scripts/split-config.escript

@@ -57,7 +57,15 @@ parse_sections([Line | Lines], Parse, Section, Sections) ->
 
 dump_sections([]) -> ok;
 dump_sections([{Name, Lines0} | Rest]) ->
-    Filename = filename:join(["etc", iolist_to_binary([Name, ".conf.seg"])]),
-    Lines = [[L, "\n"] || L <- Lines0],
-    ok = file:write_file(Filename, Lines),
-    dump_sections(Rest).
+    case is_skipped(Name) of
+        true ->
+            dump_sections(Rest);
+        false ->
+            Filename = filename:join(["etc", iolist_to_binary([Name, ".conf.seg"])]),
+            Lines = [[L, "\n"] || L <- Lines0],
+            ok = file:write_file(Filename, Lines),
+            dump_sections(Rest)
+    end.
+
+is_skipped(Name) ->
+    Name =:= <<"modules">>.