Merge pull request #10443 from rabbitmq/use-local-khepri-queries

Use local Khepri queries only

Author: dumbbell

MODULE.bazel

@@ -49,7 +49,7 @@ bazel_dep(
 
 bazel_dep(
     name = "rabbitmq_ra",
-    version = "2.7.1",
+    version = "2.7.3",
     repo_name = "ra",
 )
 
@@ -295,15 +295,15 @@ erlang_package.hex_package(
 erlang_package.hex_package(
     name = "khepri",
     build_file = "@rabbitmq-server//bazel:BUILD.khepri",
-    sha256 = "da618ec76abd188620c8fbdc16cb8eb94b09055a657c488da7fab160d49abcbe",
-    version = "0.10.1",
+    sha256 = "0f90ad4d163493abf8bc03ab161a8c65795fe5578fe16a5a1d27846185e68aaf",
+    version = "0.11.0",
 )
 
-erlang_package.git_package(
+erlang_package.hex_package(
     name = "khepri_mnesia_migration",
     build_file = "@rabbitmq-server//bazel:BUILD.khepri_mnesia_migration",
-    commit = "56d7d123cc3e9dce309582b4d45ea58f3580ee71",
-    repository = "rabbitmq/khepri_mnesia_migration",
+    sha256 = "996f7db1f094a3d5e72ff9e3a2db4fd1895872a1fcbc223b91671bcbc047b03b",
+    version = "0.3.0",
 )
 
 erlang_package.hex_package(
@@ -338,8 +338,8 @@ erlang_package.hex_package(
     name = "ra_alt",
     build_file = "@rabbitmq-server//bazel:BUILD.ra",
     pkg = "ra",
-    sha256 = "20ba906dc4feb824f07facca227b5a4f9bf7435eb21f469028112313a64a8490",
-    version = "2.7.1",
+    sha256 = "6f2b912a779f4efa4deea762b65192ed6e87111c7d98cbbe8a29576964739147",
+    version = "2.7.3",
 )
 
 erlang_package.git_package(

deps/rabbit/src/rabbit_db_maintenance.erl

@@ -155,7 +155,12 @@ get_consistent_in_mnesia(Node) ->
 
 get_consistent_in_khepri(Node) ->
     Path = khepri_maintenance_path(Node),
-    case rabbit_khepri:get(Path, #{favor => consistency}) of
+    %% FIXME: Ra consistent queries are fragile in the sense that the query
+    %% function may run on a remote node and the function reference or MFA may
+    %% not be valid on that node. That's why we force a local query for now.
+    %Options = #{favor => consistent},
+    Options = #{favor => local},
+    case rabbit_khepri:get(Path, Options) of
         {ok, #node_maintenance_state{status = Status}} ->
             Status;
         _ ->

deps/rabbit/src/rabbit_khepri.erl

@@ -276,7 +276,7 @@ wait_for_leader(Timeout, Retries) ->
     rabbit_log:info("Waiting for Khepri leader for ~tp ms, ~tp retries left",
                     [Timeout, Retries - 1]),
     Options = #{timeout => Timeout,
-                favor => compromise},
+                favor => low_latency},
     case khepri:exists(?STORE_ID, [], Options) of
         Exists when is_boolean(Exists) ->
             rabbit_log:info("Khepri leader elected"),
@@ -524,9 +524,14 @@ ensure_ra_system_started() ->
 %% cluster.
 %%
 %% The membership is as it is known to the Ra leader in the cluster.
+%%
+%% The returned list is empty if there was an error.
 
 members() ->
-    khepri_cluster:members(?RA_CLUSTER_NAME).
+    case khepri_cluster:members(?RA_CLUSTER_NAME) of
+        {ok, Members}    -> Members;
+        {error, _Reason} -> []
+    end.
 
 -spec locally_known_members() -> Members when
       Members :: [ra:server_id()].
@@ -536,18 +541,28 @@ members() ->
 %% The membership is as it is known to the local Ra server and may be
 %% inconsistent compared to the "official" membership as seen by the Ra
 %% leader.
+%%
+%% The returned list is empty if there was an error.
 
 locally_known_members() ->
-    khepri_cluster:locally_known_members(?RA_CLUSTER_NAME).
+    case khepri_cluster:locally_known_members(?RA_CLUSTER_NAME) of
+        {ok, Members}    -> Members;
+        {error, _Reason} -> []
+    end.
 
 -spec nodes() -> Nodes when
       Nodes :: [node()].
 %% @doc Returns the list of Erlang nodes that are part of the cluster.
 %%
 %% The membership is as it is known to the Ra leader in the cluster.
+%%
+%% The returned list is empty if there was an error.
 
 nodes() ->
-    khepri_cluster:nodes(?RA_CLUSTER_NAME).
+    case khepri_cluster:nodes(?RA_CLUSTER_NAME) of
+        {ok, Nodes}      -> Nodes;
+        {error, _Reason} -> []
+    end.
 
 -spec locally_known_nodes() -> Nodes when
       Nodes :: [node()].
@@ -556,9 +571,14 @@ nodes() ->
 %% The membership is as it is known to the local Ra server and may be
 %% inconsistent compared to the "official" membership as seen by the Ra
 %% leader.
+%%
+%% The returned list is empty if there was an error.
 
 locally_known_nodes() ->
-    khepri_cluster:locally_known_nodes(?RA_CLUSTER_NAME).
+    case khepri_cluster:locally_known_nodes(?RA_CLUSTER_NAME) of
+        {ok, Nodes}      -> Nodes;
+        {error, _Reason} -> []
+    end.
 
 -spec get_ra_cluster_name() -> RaClusterName when
       RaClusterName :: ra:cluster_name().
@@ -838,20 +858,24 @@ cas(Path, Pattern, Data) ->
       ?STORE_ID, Path, Pattern, Data, ?DEFAULT_COMMAND_OPTIONS).
 
 fold(Path, Pred, Acc) ->
-    khepri:fold(?STORE_ID, Path, Pred, Acc).
+    khepri:fold(?STORE_ID, Path, Pred, Acc, #{favor => low_latency}).
 
 fold(Path, Pred, Acc, Options) ->
-    khepri:fold(?STORE_ID, Path, Pred, Acc, Options).
+    Options1 = Options#{favor => low_latency},
+    khepri:fold(?STORE_ID, Path, Pred, Acc, Options1).
 
-foreach(Path, Pred) -> khepri:foreach(?STORE_ID, Path, Pred).
+foreach(Path, Pred) ->
+    khepri:foreach(?STORE_ID, Path, Pred, #{favor => low_latency}).
 
-filter(Path, Pred) -> khepri:filter(?STORE_ID, Path, Pred).
+filter(Path, Pred) ->
+    khepri:filter(?STORE_ID, Path, Pred, #{favor => low_latency}).
 
 get(Path) ->
     khepri:get(?STORE_ID, Path, #{favor => low_latency}).
 
 get(Path, Options) ->
-    khepri:get(?STORE_ID, Path, Options).
+    Options1 = Options#{favor => low_latency},
+    khepri:get(?STORE_ID, Path, Options1).
 
 get_many(PathPattern) ->
     khepri:get_many(?STORE_ID, PathPattern, #{favor => low_latency}).
@@ -862,14 +886,19 @@ adv_get(Path) ->
 match(Path) ->
     match(Path, #{}).
 
-match(Path, Options) -> khepri:get_many(?STORE_ID, Path, Options).
+match(Path, Options) ->
+    Options1 = Options#{favor => low_latency},
+    khepri:get_many(?STORE_ID, Path, Options1).
 
 exists(Path) -> khepri:exists(?STORE_ID, Path, #{favor => low_latency}).
 
-list(Path) -> khepri:get_many(?STORE_ID, Path ++ [?KHEPRI_WILDCARD_STAR]).
+list(Path) ->
+    khepri:get_many(
+      ?STORE_ID, Path ++ [?KHEPRI_WILDCARD_STAR], #{favor => low_latency}).
 
 list_child_nodes(Path) ->
-    Options = #{props_to_return => [child_names]},
+    Options = #{props_to_return => [child_names],
+                favor => low_latency},
     case khepri_adv:get_many(?STORE_ID, Path, Options) of
         {ok, Result} ->
             case maps:values(Result) of
@@ -883,7 +912,8 @@ list_child_nodes(Path) ->
     end.
 
 count_children(Path) ->
-    Options = #{props_to_return => [child_list_length]},
+    Options = #{props_to_return => [child_list_length],
+               favor => low_latency},
     case khepri_adv:get_many(?STORE_ID, Path, Options) of
         {ok, Map} ->
             lists:sum([L || #{child_list_length := L} <- maps:values(Map)]);

deps/rabbit/src/rabbit_maintenance.erl

@@ -9,6 +9,13 @@
 
 -include_lib("rabbit_common/include/rabbit.hrl").
 
+%% FIXME: Ra consistent queries are currently fragile in the sense that the
+%% query function may run on a remote node and the function reference or MFA
+%% may not be valid on that node. That's why consistent queries in this module
+%% are in fact local queries when Khepri is enabled.
+%%
+%% See `rabbit_db_maintenance:get_consistent_in_khepri/1'.
+
 -export([
     is_enabled/0,
     drain/0,

rabbitmq-components.mk

@@ -115,11 +115,11 @@ dep_accept = hex 0.3.5
 dep_cowboy = hex 2.11.0
 dep_cowlib = hex 2.12.1
 dep_credentials_obfuscation = hex 3.4.0
-dep_khepri = hex 0.10.0
-dep_khepri_mnesia_migration = hex 0.2.1
+dep_khepri = hex 0.11.0
+dep_khepri_mnesia_migration = hex 0.3.0
 dep_looking_glass = git https://github.com/rabbitmq/looking_glass.git main
 dep_prometheus = hex 4.11.0
-dep_ra = hex 2.7.1
+dep_ra = hex 2.7.3
 dep_ranch = hex 2.1.0
 dep_recon = hex 2.5.3
 dep_redbug = hex 2.0.7