Merge remote-tracking branch 'emqx/main-v4.3' into sync-from-v4.3

.gitignore

@@ -47,3 +47,15 @@ dist.zip
 scripts/git-token
 etc/*.seg
 _upgrade_base/
+erlang_ls.config
+.els_cache/
+# VSCode files
+.vs/
+.vscode/
+# Emacs Backup files
+*~
+# Emacs temporary files
+.#*
+*#
+# For direnv
+.envrc

apps/emqx_sn/src/emqx_sn.appup.src

@@ -5,7 +5,8 @@
       {load_module, emqx_sn_registry, brutal_purge, soft_purge, []}
     ]},
     {"4.3.2", [
-      {load_module, emqx_sn_gateway, brutal_purge, soft_purge, []}
+      {load_module, emqx_sn_gateway, brutal_purge, soft_purge, []},
+      {load_module, emqx_sn_registry, brutal_purge, soft_purge, []}
     ]},
    {<<"4\\.3\\.[0-1]">>, [
      {restart_application, emqx_sn}
@@ -16,7 +17,8 @@
       {load_module, emqx_sn_registry, brutal_purge, soft_purge, []}
    ]},
    {"4.3.2", [
-      {load_module, emqx_sn_gateway, brutal_purge, soft_purge, []}
+      {load_module, emqx_sn_gateway, brutal_purge, soft_purge, []},
+      {load_module, emqx_sn_registry, brutal_purge, soft_purge, []}
     ]},
    {<<"4\\.3\\.[0-1]">>, [
      {restart_application, emqx_sn}

apps/emqx_web_hook/src/emqx_web_hook.appup.src

@@ -1,34 +1,22 @@
 %% -*- mode: erlang -*-
 {VSN,
-  [{"4.3.7",
+  [{<<"4\\.3\\.[0-2]">>,
-    [{apply,{application,stop,[emqx_web_hook]}},
-     {load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]},
-     {load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]},
-   {"4.3.5",
-    [{load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]},
-     {load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]},
-   {<<"4.3.[0-2]">>,
     [{apply,{application,stop,[emqx_web_hook]}},
      {load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_web_hook,brutal_purge,soft_purge,[]},
      {load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]},
-   {<<"4.3.[3-4]">>,
+   {<<"4\\.3\\.[3-7]">>,
-    [{load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]},
+    [{apply,{application,stop,[emqx_web_hook]}},
+     {load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
-  [{"4.3.7",
+  [{<<"4\\.3\\.[0-2]">>,
-    [{apply,{application,stop,[emqx_web_hook]}},
-     {load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]},
-     {load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]},
-   {"4.3.5",
-    [{load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]},
-     {load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]},
-   {<<"4.3.[0-2]">>,
     [{apply,{application,stop,[emqx_web_hook]}},
      {load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_web_hook,brutal_purge,soft_purge,[]},
      {load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]},
-   {<<"4.3.[3-4]">>,
+   {<<"4\\.3\\.[3-7]">>,
-    [{load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]},
+    [{apply,{application,stop,[emqx_web_hook]}},
+     {load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}]}.

deploy/charts/emqx/templates/ingress.yaml

@@ -1,5 +1,7 @@
 {{- if .Values.ingress.dashboard.enabled -}}
-{{- if semverCompare ">=1.14-0" .Capabilities.KubeVersion.GitVersion -}}
+{{- if semverCompare ">=1.19-0" .Capabilities.KubeVersion.GitVersion -}}
+apiVersion: networking.k8s.io/v1
+{{- else if semverCompare ">=1.14-0" .Capabilities.KubeVersion.GitVersion -}}
 apiVersion: networking.k8s.io/v1beta1
 {{- else -}}
 apiVersion: extensions/v1beta1
@@ -17,15 +19,28 @@ metadata:
     {{- toYaml .Values.ingress.dashboard.annotations | nindent 4 }}
   {{- end }}
 spec:
+{{- if and .Values.ingress.dashboard.ingressClassName (semverCompare ">=1.18-0" .Capabilities.KubeVersion.GitVersion) }}
+  ingressClassName: {{ .Values.ingress.dashboard.ingressClassName }}
+{{- end }}
   rules:
   {{- range $host := .Values.ingress.dashboard.hosts }}
   - host: {{ $host }}
     http:
       paths:
       - path: /
+        {{- if (semverCompare ">=1.18-0" $.Capabilities.KubeVersion.GitVersion) }}
+        pathType: ImplementationSpecific
+        {{- end }}
         backend:
+          {{- if semverCompare ">=1.19-0" $.Capabilities.KubeVersion.GitVersion }}
+          service:
+            name: {{ include "emqx.fullname" $ }}
+            port:
+              number: {{ $.Values.service.dashboard }}
+          {{- else }}
           serviceName: {{ include "emqx.fullname" $ }}
           servicePort: {{ $.Values.service.dashboard }}
+          {{- end }}
   {{- end -}}
   {{- if .Values.ingress.dashboard.tls }}
   tls:
@@ -34,7 +49,9 @@ spec:
 ---
 {{- end }}
 {{- if .Values.ingress.mgmt.enabled -}}
-{{- if semverCompare ">=1.14-0" .Capabilities.KubeVersion.GitVersion -}}
+{{- if semverCompare ">=1.19-0" .Capabilities.KubeVersion.GitVersion -}}
+apiVersion: networking.k8s.io/v1
+{{- else if semverCompare ">=1.14-0" .Capabilities.KubeVersion.GitVersion -}}
 apiVersion: networking.k8s.io/v1beta1
 {{- else -}}
 apiVersion: extensions/v1beta1
@@ -52,15 +69,28 @@ metadata:
     {{- toYaml .Values.ingress.mgmt.annotations | nindent 4 }}
   {{- end }}
 spec:
+{{- if and .Values.ingress.mgmt.ingressClassName (semverCompare ">=1.18-0" .Capabilities.KubeVersion.GitVersion) }}
+  ingressClassName: {{ .Values.ingress.mgmt.ingressClassName }}
+{{- end }}
   rules:
   {{- range $host := .Values.ingress.mgmt.hosts }}
   - host: {{ $host }}
     http:
       paths:
       - path: /
+        {{- if (semverCompare ">=1.18-0" $.Capabilities.KubeVersion.GitVersion) }}
+        pathType: ImplementationSpecific
+        {{- end }}
         backend:
+          {{- if semverCompare ">=1.19-0" $.Capabilities.KubeVersion.GitVersion }}
+          service:
+            name: {{ include "emqx.fullname" $ }}
+            port:
+              number: {{ $.Values.service.mgmt }}
+          {{- else }}
           serviceName: {{ include "emqx.fullname" $ }}
           servicePort: {{ $.Values.service.mgmt }}
+          {{- end }}
   {{- end -}}
   {{- if .Values.ingress.mgmt.tls }}
   tls:

etc/emqx.conf

@@ -777,8 +777,7 @@ zone.external.force_gc_policy = 16000|16MB
 ## Maximum topic levels allowed. 0 means no limit.
 ##
 ## Value: Number [0-65535]
-## Default: 7
+## zone.external.max_topic_levels = 7
-zone.external.max_topic_levels = 7
 
 ## Maximum QoS allowed.
 ##

src/emqx.appup.src

@@ -3,12 +3,14 @@
   [{"4.3.11",
     [{load_module,emqx_vm,brutal_purge,soft_purge,[]},
      {load_module,emqx_sys_mon,brutal_purge,soft_purge,[]},
+     {load_module,emqx_frame,brutal_purge,soft_purge,[]},
      {load_module,emqx_http_lib,brutal_purge,soft_purge,[]}]},
    {"4.3.10",
     [{load_module,emqx_vm,brutal_purge,soft_purge,[]},
      {load_module,emqx_sys_mon,brutal_purge,soft_purge,[]},
      {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
      {load_module,emqx_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_frame,brutal_purge,soft_purge,[]},
      {load_module,emqx_connection,brutal_purge,soft_purge,[]}]},
    {"4.3.9",
     [{load_module,emqx_vm,brutal_purge,soft_purge,[]},
@@ -194,12 +196,14 @@
   [{"4.3.11",
     [{load_module,emqx_vm,brutal_purge,soft_purge,[]},
      {load_module,emqx_sys_mon,brutal_purge,soft_purge,[]},
+     {load_module,emqx_frame,brutal_purge,soft_purge,[]},
      {load_module,emqx_http_lib,brutal_purge,soft_purge,[]}]},
    {"4.3.10",
     [{load_module,emqx_vm,brutal_purge,soft_purge,[]},
      {load_module,emqx_sys_mon,brutal_purge,soft_purge,[]},
      {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
      {load_module,emqx_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_frame,brutal_purge,soft_purge,[]},
      {load_module,emqx_connection,brutal_purge,soft_purge,[]}]},
    {"4.3.9",
     [{load_module,emqx_vm,brutal_purge,soft_purge,[]},

src/emqx_connection.erl

@@ -629,10 +629,15 @@ parse_incoming(Data, Packets, State = #state{parse_state = ParseState}) ->
             NState = State#state{parse_state = NParseState},
             parse_incoming(Rest, [Packet|Packets], NState)
     catch
+        error:proxy_protocol_config_disabled:_Stk ->
+            ?LOG(error,
+                 "~nMalformed packet, "
+                 "please check proxy_protocol config for specific listeners and zones~n"),
+            {[{frame_error, proxy_protocol_config_disabled} | Packets], State};
         error:Reason:Stk ->
             ?LOG(error, "~nParse failed for ~0p~n~0p~nFrame data:~0p",
                  [Reason, Stk, Data]),
-            {[{frame_error, Reason}|Packets], State}
+            {[{frame_error, Reason} | Packets], State}
     end.
 
 -compile({inline, [next_incoming_msgs/1]}).

src/emqx_frame.erl

@@ -71,8 +71,19 @@
 
 -define(MULTIPLIER_MAX, 16#200000).
 
+%% proxy_protocol v1 header human readable
+-define(PPV1_PROXY, "PROXY ").
+-define(PPV1_PROXY_UNKNOWN, "PROXY UNKNOWN").
+%% proxy_protocol v2 header signature:
+%% 16#0D,16#0A, 16#0D,16#0A,16#00,16#0D,16#0A,16#51,16#55,16#49,16#54,16#0A
+-define(PPV2_HEADER_SIG, "\r\n\r\n\0\r\nQUIT\n").
+
 -dialyzer({no_match, [serialize_utf8_string/2]}).
 
+-ifdef(TEST).
+-export([parse_variable_byte_integer/1]).
+-endif.
+
 %%--------------------------------------------------------------------
 %% Init Parse State
 %%--------------------------------------------------------------------
@@ -100,6 +111,13 @@ parse(Bin) ->
 -spec(parse(binary(), parse_state()) -> parse_result()).
 parse(<<>>, {none, Options}) ->
     {more, {none, Options}};
+parse(<<?PPV1_PROXY, IPVer:5/binary, _Rest/binary>>, {none, _Options})
+  when IPVer =:= <<"TCP4 ">> orelse IPVer =:= <<"TCP6 ">> ->
+    error(proxy_protocol_config_disabled);
+parse(<<?PPV1_PROXY_UNKNOWN, _Rest/binary>>, {none, _Options}) ->
+    error(proxy_protocol_config_disabled);
+parse(<<?PPV2_HEADER_SIG, _Rest/binary>>, {none, _Options}) ->
+    error(proxy_protocol_config_disabled);
 parse(<<Type:4, Dup:1, QoS:2, Retain:1, Rest/binary>>,
       {none, Options = #{strict_mode := StrictMode}}) ->
     %% Validate header if strict mode.

test/emqx_frame_SUITE.erl

@@ -43,7 +43,9 @@ groups() ->
     [{parse, [parallel],
       [t_parse_cont,
        t_parse_frame_too_large,
-       t_parse_frame_malformed_variable_byte_integer
+       t_parse_frame_malformed_variable_byte_integer,
+       t_parse_frame_variable_byte_integer,
+       t_parse_frame_proxy_protocol %% proxy_protocol_config_disabled packet.
       ]},
      {connect, [parallel],
       [t_serialize_parse_v3_connect,
@@ -142,6 +144,13 @@ t_parse_frame_variable_byte_integer(_) ->
     ?catch_error(malformed_variable_byte_integer,
         emqx_frame:parse_variable_byte_integer(Bin)).
 
+t_parse_frame_proxy_protocol(_) ->
+    BinList = [ <<"PROXY TCP4 ">>, <<"PROXY TCP6 ">>, <<"PROXY UNKNOWN">>
+              , <<"\r\n\r\n\0\r\nQUIT\n">>],
+    [?assertError( proxy_protocol_config_disabled
+                 , emqx_frame:parse(Bin))
+     || Bin <- BinList].
+
 t_serialize_parse_v3_connect(_) ->
     Bin = <<16,37,0,6,77,81,73,115,100,112,3,2,0,60,0,23,109,111,115,
             113,112,117, 98,47,49,48,52,53,49,45,105,77,97,99,46,108,
@@ -555,4 +564,3 @@ parse_to_packet(Bin, Opts) ->
     Packet.
 
 payload(Len) -> iolist_to_binary(lists:duplicate(Len, 1)).
-