Merge pull request #12196 from SergeTupchiy/EMQX-10891-route-cleanup-optimization

EMQX-10891 route cleanup optimization

.ci/docker-compose-file/docker-compose-kafka.yaml

@@ -18,7 +18,7 @@ services:
       - /tmp/emqx-ci/emqx-shared-secret:/var/lib/secret
   kdc:
     hostname: kdc.emqx.net
-    image:  ghcr.io/emqx/emqx-builder/5.2-8:1.15.7-26.1.2-2-ubuntu22.04
+    image:  ghcr.io/emqx/emqx-builder/5.2-9:1.15.7-26.1.2-3-ubuntu22.04
     container_name: kdc.emqx.net
     expose:
       - 88 # kdc

.ci/docker-compose-file/docker-compose.yaml

@@ -3,7 +3,7 @@ version: '3.9'
 services:
   erlang:
     container_name: erlang
-    image: ${DOCKER_CT_RUNNER_IMAGE:-ghcr.io/emqx/emqx-builder/5.2-8:1.15.7-26.1.2-2-ubuntu22.04}
+    image: ${DOCKER_CT_RUNNER_IMAGE:-ghcr.io/emqx/emqx-builder/5.2-9:1.15.7-26.1.2-3-ubuntu22.04}
     env_file:
       - credentials.env
       - conf.env

.github/workflows/_pr_entrypoint.yaml

@@ -17,16 +17,16 @@ env:
 jobs:
   sanity-checks:
     runs-on: ubuntu-22.04
-    container: "ghcr.io/emqx/emqx-builder/5.2-8:1.15.7-26.1.2-2-ubuntu22.04"
+    container: "ghcr.io/emqx/emqx-builder/5.2-9:1.15.7-26.1.2-3-ubuntu22.04"
     outputs:
       ct-matrix: ${{ steps.matrix.outputs.ct-matrix }}
       ct-host: ${{ steps.matrix.outputs.ct-host }}
       ct-docker: ${{ steps.matrix.outputs.ct-docker }}
       version-emqx: ${{ steps.matrix.outputs.version-emqx }}
       version-emqx-enterprise: ${{ steps.matrix.outputs.version-emqx-enterprise }}
-      builder: "ghcr.io/emqx/emqx-builder/5.2-8:1.15.7-26.1.2-2-ubuntu22.04"
+      builder: "ghcr.io/emqx/emqx-builder/5.2-9:1.15.7-26.1.2-3-ubuntu22.04"
-      builder_vsn: "5.2-8"
+      builder_vsn: "5.2-9"
-      otp_vsn: "26.1.2-2"
+      otp_vsn: "26.1.2-3"
       elixir_vsn: "1.15.7"
 
     steps:
@@ -92,13 +92,13 @@ jobs:
           MATRIX="$(echo "${APPS}" | jq -c '
             [
               (.[] | select(.profile == "emqx") | . + {
-                builder: "5.2-8",
+                builder: "5.2-9",
-                otp: "26.1.2-2",
+                otp: "26.1.2-3",
                 elixir: "1.15.7"
               }),
               (.[] | select(.profile == "emqx-enterprise") | . + {
-                builder: "5.2-8",
+                builder: "5.2-9",
-                otp: ["26.1.2-2"][],
+                otp: ["26.1.2-3"][],
                 elixir: "1.15.7"
               })
             ]

.github/workflows/_push-entrypoint.yaml

@@ -20,7 +20,7 @@ env:
 jobs:
   prepare:
     runs-on: ubuntu-22.04
-    container: 'ghcr.io/emqx/emqx-builder/5.2-8:1.15.7-26.1.2-2-ubuntu22.04'
+    container: 'ghcr.io/emqx/emqx-builder/5.2-9:1.15.7-26.1.2-3-ubuntu22.04'
     outputs:
       profile: ${{ steps.parse-git-ref.outputs.profile }}
       release: ${{ steps.parse-git-ref.outputs.release }}
@@ -29,9 +29,9 @@ jobs:
       ct-matrix: ${{ steps.matrix.outputs.ct-matrix }}
       ct-host: ${{ steps.matrix.outputs.ct-host }}
       ct-docker: ${{ steps.matrix.outputs.ct-docker }}
-      builder: 'ghcr.io/emqx/emqx-builder/5.2-8:1.15.7-26.1.2-2-ubuntu22.04'
+      builder: 'ghcr.io/emqx/emqx-builder/5.2-9:1.15.7-26.1.2-3-ubuntu22.04'
-      builder_vsn: '5.2-8'
+      builder_vsn: '5.2-9'
-      otp_vsn: '26.1.2-2'
+      otp_vsn: '26.1.2-3'
       elixir_vsn: '1.15.7'
 
     steps:
@@ -62,13 +62,13 @@ jobs:
           MATRIX="$(echo "${APPS}" | jq -c '
             [
               (.[] | select(.profile == "emqx") | . + {
-                builder: "5.2-8",
+                builder: "5.2-9",
-                otp: "26.1.2-2",
+                otp: "26.1.2-3",
                 elixir: "1.15.7"
               }),
               (.[] | select(.profile == "emqx-enterprise") | . + {
-                builder: "5.2-8",
+                builder: "5.2-9",
-                otp: ["26.1.2-2"][],
+                otp: ["26.1.2-3"][],
                 elixir: "1.15.7"
               })
             ]

.github/workflows/build_and_push_docker_images.yaml

@@ -58,7 +58,7 @@ on:
       otp_vsn:
         required: false
         type: string
-        default: '26.1.2-2'
+        default: '26.1.2-3'
       elixir_vsn:
         required: false
         type: string
@@ -66,7 +66,7 @@ on:
       builder_vsn:
         required: false
         type: string
-        default: '5.2-8'
+        default: '5.2-9'
 
 permissions:
   contents: read

.github/workflows/build_packages.yaml

@@ -54,7 +54,7 @@ on:
       otp_vsn:
         required: false
         type: string
-        default: '26.1.2-2'
+        default: '26.1.2-3'
       elixir_vsn:
         required: false
         type: string
@@ -62,7 +62,7 @@ on:
       builder_vsn:
         required: false
         type: string
-        default: '5.2-8'
+        default: '5.2-9'
 
 jobs:
   mac:

.github/workflows/build_packages_cron.yaml

@@ -20,7 +20,7 @@ jobs:
       fail-fast: false
       matrix:
         profile:
-          - ['emqx', 'master', '5.2-8:1.15.7-26.1.2-2']
+          - ['emqx', 'master', '5.2-9:1.15.7-26.1.2-3']
           - ['emqx-enterprise', 'release-54', '5.2-3:1.14.5-25.3.2-2']
         os:
           - debian10
@@ -90,7 +90,7 @@ jobs:
         branch:
           - master
         otp:
-          - 26.1.2-2
+          - 26.1.2-3
         os:
           - macos-12-arm64
 

.github/workflows/build_slim_packages.yaml

@@ -27,15 +27,15 @@ on:
       builder:
         required: false
         type: string
-        default: 'ghcr.io/emqx/emqx-builder/5.2-8:1.15.7-26.1.2-2-ubuntu22.04'
+        default: 'ghcr.io/emqx/emqx-builder/5.2-9:1.15.7-26.1.2-3-ubuntu22.04'
       builder_vsn:
         required: false
         type: string
-        default: '5.2-8'
+        default: '5.2-9'
       otp_vsn:
         required: false
         type: string
-        default: '26.1.2-2'
+        default: '26.1.2-3'
       elixir_vsn:
         required: false
         type: string
@@ -51,9 +51,9 @@ jobs:
       fail-fast: false
       matrix:
         profile:
-          - ["emqx", "26.1.2-2", "ubuntu22.04", "elixir", "x64"]
+          - ["emqx", "26.1.2-3", "ubuntu22.04", "elixir", "x64"]
-          - ["emqx", "26.1.2-2", "ubuntu22.04", "elixir", "arm64"]
+          - ["emqx", "26.1.2-3", "ubuntu22.04", "elixir", "arm64"]
-          - ["emqx-enterprise", "26.1.2-2", "ubuntu22.04", "erlang", "x64"]
+          - ["emqx-enterprise", "26.1.2-3", "ubuntu22.04", "erlang", "x64"]
 
     container: "ghcr.io/emqx/emqx-builder/${{ inputs.builder_vsn }}:${{ inputs.elixir_vsn }}-${{ matrix.profile[1] }}-${{ matrix.profile[2] }}"
 

.github/workflows/codeql.yaml

@@ -20,7 +20,7 @@ jobs:
       actions: read
       security-events: write
     container:
-      image: ghcr.io/emqx/emqx-builder/5.2-8:1.15.7-26.1.2-2-ubuntu22.04
+      image: ghcr.io/emqx/emqx-builder/5.2-9:1.15.7-26.1.2-3-ubuntu22.04
 
     strategy:
       fail-fast: false

.github/workflows/performance_test.yaml

@@ -26,7 +26,7 @@ jobs:
   prepare:
     runs-on: ubuntu-latest
     if: github.repository_owner == 'emqx'
-    container: ghcr.io/emqx/emqx-builder/5.2-8:1.15.7-26.1.2-2-ubuntu20.04
+    container: ghcr.io/emqx/emqx-builder/5.2-9:1.15.7-26.1.2-3-ubuntu20.04
     outputs:
       BENCH_ID: ${{ steps.prepare.outputs.BENCH_ID }}
       PACKAGE_FILE: ${{ steps.package_file.outputs.PACKAGE_FILE }}

.tool-versions

@@ -1,2 +1,2 @@
-erlang 26.1.2-2
+erlang 26.1.2-3
 elixir 1.15.7-otp-26

Makefile

@@ -7,7 +7,7 @@ REBAR = $(CURDIR)/rebar3
 BUILD = $(CURDIR)/build
 SCRIPTS = $(CURDIR)/scripts
 export EMQX_RELUP ?= true
-export EMQX_DEFAULT_BUILDER = ghcr.io/emqx/emqx-builder/5.2-8:1.15.7-26.1.2-2-debian11
+export EMQX_DEFAULT_BUILDER = ghcr.io/emqx/emqx-builder/5.2-9:1.15.7-26.1.2-3-debian11
 export EMQX_DEFAULT_RUNNER = public.ecr.aws/debian/debian:11-slim
 export EMQX_REL_FORM ?= tgz
 export QUICER_DOWNLOAD_FROM_RELEASE = 1

apps/emqx/rebar.config

@@ -28,7 +28,7 @@
     {gproc, {git, "https://github.com/emqx/gproc", {tag, "0.9.0.1"}}},
     {cowboy, {git, "https://github.com/emqx/cowboy", {tag, "2.9.2"}}},
     {esockd, {git, "https://github.com/emqx/esockd", {tag, "5.11.1"}}},
-    {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.17.0"}}},
+    {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.18.1"}}},
     {gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "3.3.1"}}},
     {hocon, {git, "https://github.com/emqx/hocon.git", {tag, "0.40.4"}}},
     {emqx_http_lib, {git, "https://github.com/emqx/emqx_http_lib.git", {tag, "0.5.3"}}},

apps/emqx/src/emqx_router.erl

@@ -95,6 +95,18 @@
     unused = [] :: nil()
 }).
 
+-define(node_patterns(Node), [Node, {'_', Node}]).
+
+-define(UNSUPPORTED, unsupported).
+
+-define(with_fallback(Expr, FallbackExpr),
+    try
+        Expr
+    catch
+        throw:?UNSUPPORTED -> FallbackExpr
+    end
+).
+
 %%--------------------------------------------------------------------
 %% Mnesia bootstrap
 %%--------------------------------------------------------------------
@@ -293,8 +305,6 @@ pick(Topic) ->
 %% Schema v1
 %% --------------------------------------------------------------------
 
--dialyzer({nowarn_function, [cleanup_routes_v1/1]}).
-
 mria_insert_route_v1(Topic, Dest) ->
     Route = #route{topic = Topic, dest = Dest},
     case emqx_topic:wildcard(Topic) of
@@ -356,10 +366,18 @@ has_route_tab_entry(Topic, Dest) ->
     [] =/= ets:match(?ROUTE_TAB, #route{topic = Topic, dest = Dest}).
 
 cleanup_routes_v1(Node) ->
-    Patterns = [
+    ?with_fallback(
-        #route{_ = '_', dest = Node},
+        lists:foreach(
-        #route{_ = '_', dest = {'_', Node}}
+            fun(Pattern) ->
-    ],
+                throw_unsupported(mria:match_delete(?ROUTE_TAB, make_route_rec_pat(Pattern)))
+            end,
+            ?node_patterns(Node)
+        ),
+        cleanup_routes_v1_fallback(Node)
+    ).
+
+cleanup_routes_v1_fallback(Node) ->
+    Patterns = [make_route_rec_pat(P) || P <- ?node_patterns(Node)],
     mria:transaction(?ROUTE_SHARD, fun() ->
         [
             mnesia:delete_object(?ROUTE_TAB, Route, write)
@@ -435,8 +453,25 @@ has_route_v2(Topic, Dest) ->
     end.
 
 cleanup_routes_v2(Node) ->
-    % NOTE
+    ?with_fallback(
-    % No point in transaction here because all the operations on filters table are dirty.
+        lists:foreach(
+            fun(Pattern) ->
+                _ = throw_unsupported(
+                    mria:match_delete(
+                        ?ROUTE_TAB_FILTERS,
+                        #routeidx{entry = emqx_trie_search:make_pat('_', Pattern)}
+                    )
+                ),
+                throw_unsupported(mria:match_delete(?ROUTE_TAB, make_route_rec_pat(Pattern)))
+            end,
+            ?node_patterns(Node)
+        ),
+        cleanup_routes_v2_fallback(Node)
+    ).
+
+cleanup_routes_v2_fallback(Node) ->
+    %% NOTE
+    %% No point in transaction here because all the operations on filters table are dirty.
     ok = ets:foldl(
         fun(#routeidx{entry = K}, ok) ->
             case get_dest_node(emqx_topic_index:get_id(K)) of
@@ -467,6 +502,19 @@ get_dest_node({_, Node}) ->
 get_dest_node(Node) ->
     Node.
 
+throw_unsupported({error, unsupported_otp_version}) ->
+    throw(?UNSUPPORTED);
+throw_unsupported(Other) ->
+    Other.
+
+%% Make Dialyzer happy
+make_route_rec_pat(DestPattern) ->
+    erlang:make_tuple(
+        record_info(size, route),
+        '_',
+        [{1, route}, {#route.dest, DestPattern}]
+    ).
+
 select_v2(Spec, Limit, undefined) ->
     Stream = mk_route_stream(Spec),
     select_next(Limit, Stream);

apps/emqx/test/emqx_router_helper_SUITE.erl

@@ -35,16 +35,32 @@ all() ->
 groups() ->
     TCs = emqx_common_test_helpers:all(?MODULE),
     [
-        {routing_schema_v1, [], TCs},
+        {routing_schema_v1, [], [
-        {routing_schema_v2, [], TCs}
+            {mria_match_delete, [], TCs},
+            {fallback, [], TCs}
+        ]},
+        {routing_schema_v2, [], [
+            {mria_match_delete, [], TCs},
+            {fallback, [], TCs}
+        ]}
     ].
 
+init_per_group(fallback, Config) ->
+    ok = mock_mria_match_delete(),
+    Config;
+init_per_group(mria_match_delete, Config) ->
+    Config;
 init_per_group(GroupName, Config) ->
     WorkDir = filename:join([?config(priv_dir, Config), ?MODULE, GroupName]),
     AppSpecs = [{emqx, mk_config(GroupName)}],
     Apps = emqx_cth_suite:start(AppSpecs, #{work_dir => WorkDir}),
     [{group_name, GroupName}, {group_apps, Apps} | Config].
 
+end_per_group(fallback, _Config) ->
+    unmock_mria_match_delete(),
+    ok;
+end_per_group(mria_match_delete, _Config) ->
+    ok;
 end_per_group(_GroupName, Config) ->
     ok = emqx_cth_suite:stop(?config(group_apps, Config)).
 
@@ -59,6 +75,13 @@ mk_config(routing_schema_v2) ->
         override_env => [{boot_modules, [broker]}]
     }.
 
+mock_mria_match_delete() ->
+    ok = meck:new(mria, [no_link, passthrough]),
+    ok = meck:expect(mria, match_delete, fun(_, _) -> {error, unsupported_otp_version} end).
+
+unmock_mria_match_delete() ->
+    ok = meck:unload(mria).
+
 init_per_testcase(_TestCase, Config) ->
     ok = snabbkaffe:start_trace(),
     Config.

build

@@ -387,9 +387,9 @@ docker_cleanup() {
 
 ## Build the default docker image based on debian 11.
 make_docker() {
-    local EMQX_BUILDER_VERSION="${EMQX_BUILDER_VERSION:-5.2-8}"
+    local EMQX_BUILDER_VERSION="${EMQX_BUILDER_VERSION:-5.2-9}"
     local EMQX_BUILDER_PLATFORM="${EMQX_BUILDER_PLATFORM:-debian11}"
-    local EMQX_BUILDER_OTP="${EMQX_BUILDER_OTP:-26.1.2-2}"
+    local EMQX_BUILDER_OTP="${EMQX_BUILDER_OTP:-26.1.2-3}"
     local EMQX_BUILDER_ELIXIR="${EMQX_BUILDER_ELIXIR:-1.15.7}"
     local EMQX_BUILDER=${EMQX_BUILDER:-ghcr.io/emqx/emqx-builder/${EMQX_BUILDER_VERSION}:${EMQX_BUILDER_ELIXIR}-${EMQX_BUILDER_OTP}-${EMQX_BUILDER_PLATFORM}}
     local EMQX_RUNNER="${EMQX_RUNNER:-${EMQX_DEFAULT_RUNNER}}"

changes/ce/perf-12196.en.md

@@ -0,0 +1,5 @@
+Improve network efficiency during routes cleanup.
+
+Previously, when a node node was down, a delete operation for every route to that node must have been exchanged between all other live nodes.
+After this change, only one 'match and delete' operation is exchanged between all live nodes, meaning that much fewer packets are to be sent over inter-cluster network.
+This optimization must be especially helpful for geo-distributed EMQX deployments, when network latency can be significantly high.

deploy/docker/Dockerfile

@@ -1,4 +1,4 @@
-ARG BUILD_FROM=ghcr.io/emqx/emqx-builder/5.2-8:1.15.7-26.1.2-2-debian11
+ARG BUILD_FROM=ghcr.io/emqx/emqx-builder/5.2-9:1.15.7-26.1.2-3-debian11
 ARG RUN_FROM=public.ecr.aws/debian/debian:11-slim
 FROM ${BUILD_FROM} AS builder
 ARG DEBUG=0

mix.exs

@@ -55,7 +55,7 @@ defmodule EMQXUmbrella.MixProject do
       {:cowboy, github: "emqx/cowboy", tag: "2.9.2", override: true},
       {:esockd, github: "emqx/esockd", tag: "5.11.1", override: true},
       {:rocksdb, github: "emqx/erlang-rocksdb", tag: "1.8.0-emqx-2", override: true},
-      {:ekka, github: "emqx/ekka", tag: "0.17.0", override: true},
+      {:ekka, github: "emqx/ekka", tag: "0.18.1", override: true},
       {:gen_rpc, github: "emqx/gen_rpc", tag: "3.3.1", override: true},
       {:grpc, github: "emqx/grpc-erl", tag: "0.6.12", override: true},
       {:minirest, github: "emqx/minirest", tag: "1.3.15", override: true},

rebar.config

@@ -83,7 +83,7 @@
     {cowboy, {git, "https://github.com/emqx/cowboy", {tag, "2.9.2"}}},
     {esockd, {git, "https://github.com/emqx/esockd", {tag, "5.11.1"}}},
     {rocksdb, {git, "https://github.com/emqx/erlang-rocksdb", {tag, "1.8.0-emqx-2"}}},
-    {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.17.0"}}},
+    {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.18.1"}}},
     {gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "3.3.1"}}},
     {grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.12"}}},
     {minirest, {git, "https://github.com/emqx/minirest", {tag, "1.3.15"}}},

scripts/buildx.sh

@@ -9,7 +9,7 @@
 
 ## example:
 ## ./scripts/buildx.sh --profile emqx --pkgtype tgz --arch arm64 \
-##     --builder ghcr.io/emqx/emqx-builder/5.2-8:1.15.7-26.1.2-2-debian11
+##     --builder ghcr.io/emqx/emqx-builder/5.2-9:1.15.7-26.1.2-3-debian11
 
 set -euo pipefail
 
@@ -24,7 +24,7 @@ help() {
     echo "--arch amd64|arm64:        Target arch to build the EMQX package for"
     echo "--src_dir <SRC_DIR>:       EMQX source code in this dir, default to PWD"
     echo "--builder <BUILDER>:       Builder image to pull"
-    echo "                           E.g. ghcr.io/emqx/emqx-builder/5.2-8:1.15.7-26.1.2-2-debian11"
+    echo "                           E.g. ghcr.io/emqx/emqx-builder/5.2-9:1.15.7-26.1.2-3-debian11"
 }
 
 die() {

scripts/pr-sanity-checks.sh

@@ -12,8 +12,8 @@ if ! type "yq" > /dev/null; then
     exit 1
 fi
 
-EMQX_BUILDER_VERSION=${EMQX_BUILDER_VERSION:-5.2-8}
+EMQX_BUILDER_VERSION=${EMQX_BUILDER_VERSION:-5.2-9}
-EMQX_BUILDER_OTP=${EMQX_BUILDER_OTP:-26.1.2-2}
+EMQX_BUILDER_OTP=${EMQX_BUILDER_OTP:-26.1.2-3}
 EMQX_BUILDER_ELIXIR=${EMQX_BUILDER_ELIXIR:-1.15.7}
 EMQX_BUILDER_PLATFORM=${EMQX_BUILDER_PLATFORM:-ubuntu22.04}
 EMQX_BUILDER=${EMQX_BUILDER:-ghcr.io/emqx/emqx-builder/${EMQX_BUILDER_VERSION}:${EMQX_BUILDER_ELIXIR}-${EMQX_BUILDER_OTP}-${EMQX_BUILDER_PLATFORM}}

scripts/relup-test/start-relup-test-cluster.sh

@@ -22,7 +22,7 @@ WEBHOOK="webhook.$NET"
 BENCH="bench.$NET"
 COOKIE='this-is-a-secret'
 ## Erlang image is needed to run webhook server and emqtt-bench
-ERLANG_IMAGE="ghcr.io/emqx/emqx-builder/5.2-8:1.15.7-26.1.2-2-ubuntu22.04"
+ERLANG_IMAGE="ghcr.io/emqx/emqx-builder/5.2-9:1.15.7-26.1.2-3-ubuntu22.04"
 # builder has emqtt-bench installed
 BENCH_IMAGE="$ERLANG_IMAGE"