Merge pull request #11795 from rabbitmq/unify-khepri-paths-api

Unify Khepri paths API

Author: dumbbell

deps/rabbit/src/rabbit_db_binding.erl

@@ -32,12 +32,13 @@
          delete_transient_for_destination_in_mnesia/1,
          has_for_source_in_mnesia/1,
          has_for_source_in_khepri/1,
-         match_source_and_destination_in_khepri_tx/2
+         match_source_and_destination_in_khepri_tx/2,
+         clear_in_khepri/0
         ]).
 
 -export([
-         khepri_route_path/1,
-         khepri_routes_path/0,
+         khepri_route_path/1, khepri_route_path/5,
+         khepri_route_path_to_args/1,
          khepri_route_exchange_path/1
         ]).
 
@@ -610,9 +611,12 @@ fold_in_mnesia(Fun, Acc) ->
               end, Acc, ?MNESIA_TABLE).
 
 fold_in_khepri(Fun, Acc) ->
-    Path = khepri_routes_path() ++ [_VHost = ?KHEPRI_WILDCARD_STAR,
-                                    _SrcName = ?KHEPRI_WILDCARD_STAR,
-                                    rabbit_khepri:if_has_data_wildcard()],
+    Path = khepri_route_path(
+             _VHost = ?KHEPRI_WILDCARD_STAR,
+             _SrcName = ?KHEPRI_WILDCARD_STAR,
+             _Kind = ?KHEPRI_WILDCARD_STAR,
+             _DstName = ?KHEPRI_WILDCARD_STAR,
+             _RoutingKey = #if_has_data{}),
     {ok, Res} = rabbit_khepri:fold(
                   Path,
                   fun(_, #{data := SetOfBindings}, Acc0) ->
@@ -828,9 +832,14 @@ delete_all_for_exchange_in_khepri(X = #exchange{name = XName}, OnlyDurable, Remo
     {deleted, X, Bindings, delete_for_destination_in_khepri(XName, OnlyDurable)}.
 
 delete_for_source_in_khepri(#resource{virtual_host = VHost, name = Name}) ->
-    Path = khepri_routes_path() ++ [VHost, Name],
-    {ok, Bindings} = khepri_tx:get_many(Path ++ [rabbit_khepri:if_has_data_wildcard()]),
-    ok = khepri_tx:delete(Path),
+    Path = khepri_route_path(
+             VHost,
+             Name,
+             _Kind = ?KHEPRI_WILDCARD_STAR,
+             _DstName = ?KHEPRI_WILDCARD_STAR,
+             _RoutingKey = #if_has_data{}),
+    {ok, Bindings} = khepri_tx:get_many(Path),
+    ok = khepri_tx:delete_many(Path),
     maps:fold(fun(_P, Set, Acc) ->
                       sets:to_list(Set) ++ Acc
               end, [], Bindings).
@@ -885,7 +894,12 @@ delete_for_destination_in_khepri(DstName, OnlyDurable) ->
                                        lists:keysort(#binding.source, Bindings), OnlyDurable).
 
 match_destination_in_khepri(#resource{virtual_host = VHost, kind = Kind, name = Name}) ->
-    Path = khepri_routes_path() ++ [VHost, ?KHEPRI_WILDCARD_STAR, Kind, Name, ?KHEPRI_WILDCARD_STAR_STAR],
+    Path = khepri_route_path(
+             VHost,
+             _SrcName = ?KHEPRI_WILDCARD_STAR,
+             Kind,
+             Name,
+             _RoutingKey = ?KHEPRI_WILDCARD_STAR),
     {ok, Map} = khepri_tx:get_many(Path),
     Map.
 
@@ -926,7 +940,12 @@ has_for_source_in_mnesia(SrcName) ->
 -spec has_for_source_in_khepri(rabbit_types:binding_source()) -> boolean().
 
 has_for_source_in_khepri(#resource{virtual_host = VHost, name = Name}) ->
-    Path = khepri_routes_path() ++ [VHost, Name, rabbit_khepri:if_has_data_wildcard()],
+    Path = khepri_route_path(
+             VHost,
+             Name,
+             _Kind = ?KHEPRI_WILDCARD_STAR,
+             _DstName = ?KHEPRI_WILDCARD_STAR,
+             _RoutingKey = #if_has_data{}),
     case khepri_tx:get_many(Path) of
         {ok, Map} ->
             maps:size(Map) > 0;
@@ -945,7 +964,8 @@ has_for_source_in_khepri(#resource{virtual_host = VHost, name = Name}) ->
 
 match_source_and_destination_in_khepri_tx(#resource{virtual_host = VHost, name = Name},
                                           #resource{kind = Kind, name = DstName}) ->
-    Path = khepri_routes_path() ++ [VHost, Name, Kind, DstName, rabbit_khepri:if_has_data_wildcard()],
+    Path = khepri_route_path(
+             VHost, Name, Kind, DstName, _RoutingKey = #if_has_data{}),
     case khepri_tx:get_many(Path) of
         {ok, Map} -> maps:values(Map);
         _         -> []
@@ -974,7 +994,12 @@ clear_in_mnesia() ->
     ok.
 
 clear_in_khepri() ->
-    Path = khepri_routes_path(),
+    Path = khepri_route_path(
+             _VHost = ?KHEPRI_WILDCARD_STAR,
+             _SrcName = ?KHEPRI_WILDCARD_STAR,
+             _Kind = ?KHEPRI_WILDCARD_STAR,
+             _DstName = ?KHEPRI_WILDCARD_STAR,
+             _RoutingKey = ?KHEPRI_WILDCARD_STAR),
     case rabbit_khepri:delete(Path) of
         ok -> ok;
         Error -> throw(Error)
@@ -983,13 +1008,44 @@ clear_in_khepri() ->
 %% --------------------------------------------------------------
 %% Paths
 %% --------------------------------------------------------------
-khepri_route_path(#binding{source = #resource{virtual_host = VHost, name = SrcName},
-                           destination = #resource{kind = Kind, name = DstName},
-                           key = RoutingKey}) ->
+
+khepri_route_path(
+  #binding{source = #resource{virtual_host = VHost, name = SrcName},
+           destination = #resource{kind = Kind, name = DstName},
+           key = RoutingKey}) ->
+    khepri_route_path(VHost, SrcName, Kind, DstName, RoutingKey).
+
+khepri_route_path(VHost, SrcName, Kind, DstName, RoutingKey)
+  when ?IS_KHEPRI_PATH_CONDITION(VHost) andalso
+       ?IS_KHEPRI_PATH_CONDITION(SrcName) andalso
+       ?IS_KHEPRI_PATH_CONDITION(Kind) andalso
+       ?IS_KHEPRI_PATH_CONDITION(DstName) andalso
+       ?IS_KHEPRI_PATH_CONDITION(RoutingKey) ->
     [?MODULE, routes, VHost, SrcName, Kind, DstName, RoutingKey].
 
-khepri_routes_path() ->
-    [?MODULE, routes].
+khepri_route_path_to_args(Path) ->
+    Pattern = khepri_route_path(
+                '$VHost', '$SrcName', '$Kind', '$DstName', '$RoutingKey'),
+    khepri_route_path_to_args(Pattern, Path, #{}).
+
+khepri_route_path_to_args([Var | Pattern], [Value | Path], Result)
+  when Var =:= '$VHost' orelse
+       Var =:= '$SrcName' orelse
+       Var =:= '$Kind' orelse
+       Var =:= '$DstName' orelse
+       Var =:= '$RoutingKey' ->
+    Result1 = Result#{Var => Value},
+    khepri_route_path_to_args(Pattern, Path, Result1);
+khepri_route_path_to_args([Comp | Pattern], [Comp | Path], Result) ->
+    khepri_route_path_to_args(Pattern, Path, Result);
+khepri_route_path_to_args(
+  [], _,
+  #{'$VHost' := VHost,
+    '$SrcName' := SrcName,
+    '$Kind' := Kind,
+    '$DstName' := DstName,
+    '$RoutingKey' := RoutingKey}) ->
+    {VHost, SrcName, Kind, DstName, RoutingKey}.
 
 khepri_route_exchange_path(#resource{virtual_host = VHost, name = SrcName}) ->
     [?MODULE, routes, VHost, SrcName].

deps/rabbit/src/rabbit_db_binding_m2k_converter.erl

@@ -10,6 +10,7 @@
 -behaviour(mnesia_to_khepri_converter).
 
 -include_lib("kernel/include/logger.hrl").
+-include_lib("khepri/include/khepri.hrl").
 -include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
 
@@ -111,8 +112,4 @@ delete_from_khepri(rabbit_route = Table, Key, State) ->
       Table :: atom().
 
 clear_data_in_khepri(rabbit_route) ->
-    Path = rabbit_db_binding:khepri_routes_path(),
-    case rabbit_khepri:delete(Path) of
-        ok -> ok;
-        Error -> throw(Error)
-    end.
+    rabbit_db_binding:clear_in_khepri().

deps/rabbit/src/rabbit_db_exchange.erl

@@ -41,17 +41,16 @@
          get_in_khepri_tx/1,
          update_in_mnesia_tx/2,
          update_in_khepri_tx/2,
-         path/1
+         clear_exchanges_in_khepri/0,
+         clear_exchange_serials_in_khepri/0
          ]).
 
 %% For testing
 -export([clear/0]).
 
 -export([
-         khepri_exchange_path/1,
-         khepri_exchange_serial_path/1,
-         khepri_exchanges_path/0,
-         khepri_exchange_serials_path/0
+         khepri_exchange_path/1, khepri_exchange_path/2,
+         khepri_exchange_serial_path/1, khepri_exchange_serial_path/2
         ]).
 
 -define(MNESIA_TABLE, rabbit_exchange).
@@ -81,7 +80,8 @@ get_all_in_mnesia() ->
     rabbit_db:list_in_mnesia(?MNESIA_TABLE, #exchange{_ = '_'}).
 
 get_all_in_khepri() ->
-    rabbit_db:list_in_khepri(khepri_exchanges_path() ++ [rabbit_khepri:if_has_data_wildcard()]).
+    Path = khepri_exchange_path(?KHEPRI_WILDCARD_STAR, #if_has_data{}),
+    rabbit_db:list_in_khepri(Path).
 
 -spec get_all(VHostName) -> [Exchange] when
       VHostName :: vhost:name(),
@@ -103,7 +103,8 @@ get_all_in_mnesia(VHost) ->
     rabbit_db:list_in_mnesia(?MNESIA_TABLE, Match).
 
 get_all_in_khepri(VHost) ->
-    rabbit_db:list_in_khepri(khepri_exchanges_path() ++ [VHost, rabbit_khepri:if_has_data_wildcard()]).
+    Path = khepri_exchange_path(VHost, #if_has_data{}),
+    rabbit_db:list_in_khepri(Path).
 
 %% -------------------------------------------------------------------
 %% get_all_durable().
@@ -127,7 +128,7 @@ get_all_durable_in_mnesia() ->
     rabbit_db:list_in_mnesia(rabbit_durable_exchange, #exchange{_ = '_'}).
 
 get_all_durable_in_khepri() ->
-    rabbit_db:list_in_khepri(khepri_exchanges_path() ++ [rabbit_khepri:if_has_data_wildcard()]).
+    get_all_in_khepri().
 
 %% -------------------------------------------------------------------
 %% list().
@@ -202,7 +203,8 @@ get_in_khepri(Name) ->
       Ret :: [Exchange :: rabbit_types:exchange()].
 
 get_in_khepri_tx(Name) ->
-    case khepri_tx:get(khepri_exchange_path(Name)) of
+    Path = khepri_exchange_path(Name),
+    case khepri_tx:get(Path) of
         {ok, X} -> [X];
         _ -> []
     end.
@@ -261,7 +263,11 @@ count_in_mnesia() ->
     mnesia:table_info(?MNESIA_TABLE, size).
 
 count_in_khepri() ->
-    rabbit_khepri:count_children(khepri_exchanges_path() ++ [?KHEPRI_WILDCARD_STAR]).
+    Path = khepri_exchange_path(?KHEPRI_WILDCARD_STAR, ?KHEPRI_WILDCARD_STAR),
+    case rabbit_khepri:count(Path) of
+        {ok, Count} -> Count;
+        _           -> 0
+    end.
 
 %% -------------------------------------------------------------------
 %% update().
@@ -719,8 +725,8 @@ recover_in_khepri(VHost) ->
     %% cannot be skipped and stopping the node is not an option -
     %% the next boot most likely would behave the same way.
     %% Any other request stays with the default timeout, currently 30s.
-    Exchanges0 = rabbit_db:list_in_khepri(khepri_exchanges_path() ++ [VHost, rabbit_khepri:if_has_data_wildcard()],
-                                #{timeout => infinity}),
+    Path = khepri_exchange_path(VHost, #if_has_data{}),
+    Exchanges0 = rabbit_db:list_in_khepri(Path, #{timeout => infinity}),
     Exchanges = [rabbit_exchange_decorator:set(X) || X <- Exchanges0],
 
     rabbit_khepri:transaction(
@@ -765,7 +771,8 @@ match_in_mnesia(Pattern) ->
 
 match_in_khepri(Pattern0) ->
     Pattern = #if_data_matches{pattern = Pattern0},
-    rabbit_db:list_in_khepri(khepri_exchanges_path() ++ [?KHEPRI_WILDCARD_STAR, Pattern]).
+    Path = khepri_exchange_path(?KHEPRI_WILDCARD_STAR, Pattern),
+    rabbit_db:list_in_khepri(Path).
 
 %% -------------------------------------------------------------------
 %% exists().
@@ -814,8 +821,17 @@ clear_in_mnesia() ->
     ok.
 
 clear_in_khepri() ->
-    khepri_delete(khepri_exchanges_path()),
-    khepri_delete(khepri_exchange_serials_path()).
+    clear_exchanges_in_khepri(),
+    clear_exchange_serials_in_khepri().
+
+clear_exchanges_in_khepri() ->
+    Path = khepri_exchange_path(?KHEPRI_WILDCARD_STAR, ?KHEPRI_WILDCARD_STAR),
+    khepri_delete(Path).
+
+clear_exchange_serials_in_khepri() ->
+    Path = khepri_exchange_serial_path(
+             ?KHEPRI_WILDCARD_STAR, ?KHEPRI_WILDCARD_STAR),
+    khepri_delete(Path).
 
 khepri_delete(Path) ->
     case rabbit_khepri:delete(Path) of
@@ -875,25 +891,18 @@ maybe_auto_delete_in_khepri(XName, OnlyDurable) ->
 %% Khepri paths
 %% -------------------------------------------------------------------
 
-khepri_exchanges_path() ->
-    [?MODULE, exchanges].
-
 khepri_exchange_path(#resource{virtual_host = VHost, name = Name}) ->
-    [?MODULE, exchanges, VHost, Name].
+    khepri_exchange_path(VHost, Name).
 
-khepri_exchange_serials_path() ->
-    [?MODULE, exchange_serials].
+khepri_exchange_path(VHost, Name)
+  when ?IS_KHEPRI_PATH_CONDITION(VHost) andalso
+       ?IS_KHEPRI_PATH_CONDITION(Name) ->
+    [?MODULE, exchanges, VHost, Name].
 
 khepri_exchange_serial_path(#resource{virtual_host = VHost, name = Name}) ->
-    [?MODULE, exchange_serials, VHost, Name].
-
-%% -------------------------------------------------------------------
-%% path().
-%% -------------------------------------------------------------------
+    khepri_exchange_serial_path(VHost, Name).
 
--spec path(ExchangeName) -> Path when
-      ExchangeName :: rabbit_exchange:name(),
-      Path :: khepri_path:path().
-
-path(Name) ->
-    khepri_exchange_path(Name).
+khepri_exchange_serial_path(VHost, Name)
+  when ?IS_KHEPRI_PATH_CONDITION(VHost) andalso
+       ?IS_KHEPRI_PATH_CONDITION(Name) ->
+    [?MODULE, exchange_serials, VHost, Name].

deps/rabbit/src/rabbit_db_exchange_m2k_converter.erl

@@ -129,12 +129,6 @@ delete_from_khepri(rabbit_exchange_serial = Table, Key, State) ->
       Table :: atom().
 
 clear_data_in_khepri(rabbit_exchange) ->
-    khepri_delete(rabbit_db_exchange:khepri_exchanges_path());
+    rabbit_db_exchange:clear_exchanges_in_khepri();
 clear_data_in_khepri(rabbit_exchange_serial) ->
-    khepri_delete(rabbit_db_exchange:khepri_exchange_serials_path()).
-
-khepri_delete(Path) ->
-    case rabbit_khepri:delete(Path) of
-        ok -> ok;
-        Error -> throw(Error)
-    end.
+    rabbit_db_exchange:clear_exchange_serials_in_khepri().

deps/rabbit/src/rabbit_db_maintenance.erl

@@ -7,6 +7,7 @@
 
 -module(rabbit_db_maintenance).
 
+-include_lib("khepri/include/khepri.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
 
 -export([
@@ -17,8 +18,7 @@
         ]).
 
 -export([
-         khepri_maintenance_path/1,
-         khepri_maintenance_path/0
+         khepri_maintenance_path/1
         ]).
 
 -define(TABLE, rabbit_node_maintenance_states).
@@ -167,8 +167,5 @@ get_consistent_in_khepri(Node) ->
 %% Khepri paths
 %% -------------------------------------------------------------------
 
-khepri_maintenance_path() ->
-    [?MODULE, maintenance].
-
-khepri_maintenance_path(Node) ->
+khepri_maintenance_path(Node) when ?IS_KHEPRI_PATH_CONDITION(Node) ->
     [?MODULE, maintenance, Node].

deps/rabbit/src/rabbit_db_maintenance_m2k_converter.erl

@@ -10,6 +10,7 @@
 -behaviour(mnesia_to_khepri_converter).
 
 -include_lib("kernel/include/logger.hrl").
+-include_lib("khepri/include/khepri.hrl").
 -include_lib("khepri_mnesia_migration/src/kmm_logging.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
 
@@ -95,7 +96,7 @@ delete_from_khepri(rabbit_node_maintenance_states = Table, Key, State) ->
       Table :: atom().
 
 clear_data_in_khepri(rabbit_node_maintenance_states) ->
-    Path = rabbit_db_maintenance:khepri_maintenance_path(),
+    Path = rabbit_db_maintenance:khepri_maintenance_path(?KHEPRI_WILDCARD_STAR),
     case rabbit_khepri:delete(Path) of
         ok -> ok;
         Error -> throw(Error)