Add new machine module for SAC coordinator v5

Author: acogoluegnes

deps/rabbit/Makefile

@@ -268,7 +268,7 @@ PARALLEL_CT_SET_2_B = clustering_recovery crashing_queues deprecated_features di
 PARALLEL_CT_SET_2_C = disk_monitor dynamic_qq unit_disk_monitor unit_file_handle_cache unit_log_management unit_operator_policy
 PARALLEL_CT_SET_2_D = queue_length_limits queue_parallel quorum_queue_member_reconciliation rabbit_fifo rabbit_fifo_dlx rabbit_stream_coordinator
 
-PARALLEL_CT_SET_3_A = definition_import per_user_connection_channel_limit_partitions per_vhost_connection_limit_partitions policy priority_queue_recovery rabbit_fifo_prop rabbit_fifo_v0 rabbit_stream_sac_coordinator unit_credit_flow unit_queue_consumers unit_queue_location unit_quorum_queue
+PARALLEL_CT_SET_3_A = definition_import per_user_connection_channel_limit_partitions per_vhost_connection_limit_partitions policy priority_queue_recovery rabbit_fifo_prop rabbit_fifo_v0 rabbit_stream_sac_coordinator_v4 rabbit_stream_sac_coordinator unit_credit_flow unit_queue_consumers unit_queue_location unit_quorum_queue
 PARALLEL_CT_SET_3_B = cluster_upgrade list_consumers_sanity_check list_queues_online_and_offline logging lqueue maintenance_mode rabbit_fifo_q
 PARALLEL_CT_SET_3_C = cli_forget_cluster_node feature_flags_v2 mc_unit message_containers_deaths_v2 message_size_limit metadata_store_migration
 PARALLEL_CT_SET_3_D = metadata_store_phase1 metrics mirrored_supervisor peer_discovery_classic_config proxy_protocol runtime_parameters unit_stats_and_metrics unit_supervisor2 unit_vm_memory_monitor

deps/rabbit/ct.test.spec

@@ -117,6 +117,7 @@
 ,   rabbit_local_random_exchange_SUITE
 ,   rabbit_message_interceptor_SUITE
 ,   rabbit_stream_coordinator_SUITE
+,   rabbit_stream_sac_coordinator_v4_SUITE
 ,   rabbit_stream_sac_coordinator_SUITE
 ,   rabbitmq_4_0_deprecations_SUITE
 ,   rabbitmq_queues_cli_integration_SUITE

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -520,13 +520,13 @@ reachable_coord_members() ->
     Nodes = rabbit_nodes:list_reachable(),
     [{?MODULE, Node} || Node <- Nodes].
 
-version() -> 4.
+version() -> 5.
 
 which_module(_) ->
     ?MODULE.
 
 init(_Conf) ->
-    #?MODULE{single_active_consumer = rabbit_stream_sac_coordinator:init_state()}.
+    #?MODULE{single_active_consumer = rabbit_stream_sac_coordinator_v4:init_state()}.
 
 -spec apply(ra_machine:command_meta_data(), command(), state()) ->
     {state(), term(), ra_machine:effects()}.
@@ -564,11 +564,12 @@ apply(#{index := _Idx, machine_version := MachineVersion} = Meta0,
     end;
 apply(Meta, {sac, SacCommand}, #?MODULE{single_active_consumer = SacState0,
                                         monitors = Monitors0} = State0) ->
-    {SacState1, Reply, Effects0} = rabbit_stream_sac_coordinator:apply(SacCommand, SacState0),
+    Mod = sac_module(Meta),
+    {SacState1, Reply, Effects0} = Mod:apply(SacCommand, SacState0),
     {SacState2, Monitors1, Effects1} =
-         rabbit_stream_sac_coordinator:ensure_monitors(SacCommand, SacState1, Monitors0, Effects0),
+         Mod:ensure_monitors(SacCommand, SacState1, Monitors0, Effects0),
     return(Meta, State0#?MODULE{single_active_consumer = SacState2,
-                                 monitors = Monitors1}, Reply, Effects1);
+                                monitors = Monitors1}, Reply, Effects1);
 apply(#{machine_version := MachineVersion} = Meta, {down, Pid, Reason} = Cmd,
       #?MODULE{streams = Streams0,
                monitors = Monitors0,
@@ -629,7 +630,8 @@ apply(#{machine_version := MachineVersion} = Meta, {down, Pid, Reason} = Cmd,
                                                monitors = Monitors1}, ok, Effects0)
             end;
         {sac, Monitors1} ->
-            {SacState1, Effects} = rabbit_stream_sac_coordinator:handle_connection_down(Pid, SacState0),
+            Mod = sac_module(Meta),
+            {SacState1, Effects} = Mod:handle_connection_down(Pid, SacState0),
             return(Meta, State#?MODULE{single_active_consumer = SacState1,
                                        monitors = Monitors1}, ok, Effects);
         error ->
@@ -747,6 +749,11 @@ state_enter(leader, #?MODULE{streams = Streams,
 state_enter(_S, _) ->
     [].
 
+sac_module(#{machine_version := MachineVersion}) when MachineVersion =< 4 ->
+    rabbit_stream_sac_coordinator_v4;
+sac_module(_) ->
+    rabbit_stream_sac_coordinator.
+
 all_member_nodes(Streams) ->
     maps:keys(
       maps:fold(
@@ -2214,6 +2221,11 @@ machine_version(3, 4, #?MODULE{streams = Streams0} = State) ->
                                              end, Members)}
                 end, Streams0),
     {State#?MODULE{streams = Streams}, []};
+machine_version(4 = From, 5, #?MODULE{single_active_consumer = Sac0} = State) ->
+    rabbit_log:info("Stream coordinator machine version changes from 4 to 5, updating state."),
+    SacExport = rabbit_stream_sac_coordinator_v4:state_to_map(Sac0),
+    Sac1 = rabbit_stream_sac_coordinator:import_state(From, SacExport),
+    {State#?MODULE{single_active_consumer = Sac1}, []};
 machine_version(From, To, State) ->
     rabbit_log:info("Stream coordinator machine version changes from ~tp to ~tp, no state changes required.",
                     [From, To]),

deps/rabbit/src/rabbit_stream_coordinator.hrl

@@ -68,6 +68,7 @@
          listeners = #{} :: undefined | #{stream_id() =>
                                           #{pid() := queue_ref()}},
          single_active_consumer = undefined :: undefined |
+                                               rabbit_stream_sac_coordinator_v4:state() |
                                                rabbit_stream_sac_coordinator:state(),
          %% future extensibility
          reserved_2}).

deps/rabbit/src/rabbit_stream_sac_coordinator.erl

@@ -39,10 +39,15 @@
          handle_connection_down/2,
          consumer_groups/3,
          group_consumers/5,
-         overview/1]).
+         overview/1,
+         import_state/2]).
 
 -import(rabbit_stream_coordinator, [ra_local_query/1]).
 
+-define(ACTIVE, active).
+-define(WAITING, waiting).
+-define(DEACTIVATING, deactivating).
+
 %% Single Active Consumer API
 -spec register_consumer(binary(),
                         binary(),
@@ -231,7 +236,7 @@ apply(#command_unregister_consumer{vhost = VirtualHost,
                             G1 = remove_from_group(Consumer, Group0),
                             handle_consumer_removal(
                               G1, Stream, ConsumerName,
-                              is_active(Consumer#consumer.active));
+                              is_active(Consumer#consumer.status));
                         false ->
                             {Group0, []}
                     end,
@@ -255,10 +260,10 @@ apply(#command_activate_consumer{vhost = VirtualHost,
                                    [{VirtualHost, Stream, ConsumerName}]),
                 {undefined, []};
             Group0 ->
-                Group1 = update_consumers(Group0, waiting),
+                Group1 = update_consumers(Group0, ?WAITING),
                 #consumer{pid = Pid, subscription_id = SubId} =
                     evaluate_active_consumer(Group1),
-                    Group2 = update_consumer_state_in_group(Group1, Pid, SubId, active),
+                    Group2 = update_consumer_state_in_group(Group1, Pid, SubId, ?ACTIVE),
                 {Group2, [notify_consumer_effect(Pid, SubId, Stream, ConsumerName, true)]}
         end,
     StreamGroups1 =
@@ -314,7 +319,7 @@ group_consumers(VirtualHost,
         #{GroupId := #group{consumers = Consumers}} ->
             Cs = lists:foldr(fun(#consumer{subscription_id = SubId,
                                            owner = Owner,
-                                           active = Active},
+                                           status = Status},
                                  Acc) ->
                                 Record =
                                     lists:foldr(fun (subscription_id, RecAcc) ->
@@ -326,7 +331,7 @@ group_consumers(VirtualHost,
                                                           Owner}
                                                          | RecAcc];
                                                     (state, RecAcc) ->
-                                                        [{state, Active}
+                                                        [{state, cli_consumer_status_label(Status)}
                                                          | RecAcc];
                                                     (Unknown, RecAcc) ->
                                                         [{Unknown,
@@ -342,6 +347,11 @@ group_consumers(VirtualHost,
             {error, not_found}
     end.
 
+cli_consumer_status_label(?ACTIVE) ->
+   active;
+cli_consumer_status_label(_) ->
+   inactive.
+
 -spec ensure_monitors(command(),
                       state(),
                       map(),
@@ -432,7 +442,7 @@ handle_group_after_connection_down(Pid,
             %% keep flags to know what happened
             {Consumers1, ActiveRemoved, AnyRemoved} =
                 lists:foldl(
-                  fun(#consumer{pid = P, active = S}, {L, ActiveFlag, _})
+                  fun(#consumer{pid = P, status = S}, {L, ActiveFlag, _})
                         when P == Pid ->
                           {L, is_active(S) or ActiveFlag, true};
                      (C, {L, ActiveFlag, AnyFlag}) ->
@@ -454,6 +464,42 @@ handle_group_after_connection_down(Pid,
             end
     end.
 
+-spec import_state(ra_machine:version(), map()) -> state().
+import_state(4, #{<<"groups">> := Groups, <<"pids_groups">> := PidsGroups}) ->
+    #?MODULE{groups = map_to_groups(Groups),
+             pids_groups = map_to_pids_groups(PidsGroups)}.
+
+map_to_groups(Groups) when is_map(Groups) ->
+    maps:fold(fun(K, V, Acc) ->
+                      Acc#{K => map_to_group(V)}
+              end, #{}, Groups);
+map_to_groups(_) ->
+    #{}.
+
+map_to_pids_groups(PidsGroups) when is_map(PidsGroups) ->
+    PidsGroups;
+map_to_pids_groups(_) ->
+    #{}.
+
+map_to_group(#{<<"consumers">> := Consumers, <<"partition_index">> := Index}) ->
+    C = lists:foldl(fun(V, Acc) ->
+                            Acc ++ [map_to_consumer(V)]
+                    end, [], Consumers),
+    #group{consumers = C,
+           partition_index = Index}.
+
+map_to_consumer(#{<<"pid">> := Pid, <<"subscription_id">> := SubId,
+                  <<"owner">> := Owner, <<"active">> := Active}) ->
+    #consumer{pid = Pid,
+              subscription_id = SubId,
+              owner = Owner,
+              status = active_to_status(Active)}.
+
+active_to_status(true) ->
+    ?ACTIVE;
+active_to_status(false) ->
+    ?WAITING.
+
 is_active(waiting) ->
     false;
 is_active(_) ->
@@ -476,12 +522,12 @@ do_register_consumer(VirtualHost,
                 #consumer{pid = ConnectionPid,
                           owner = Owner,
                           subscription_id = SubscriptionId,
-                          active = waiting};
+                          status = ?WAITING};
             false ->
                 #consumer{pid = ConnectionPid,
                           subscription_id = SubscriptionId,
                           owner = Owner,
-                          active = active}
+                          status = ?ACTIVE}
         end,
     Group1 = add_to_group(Consumer, Group0),
     StreamGroups1 =
@@ -491,17 +537,17 @@ do_register_consumer(VirtualHost,
                       Group1,
                       StreamGroups0),
 
-    #consumer{active = Active} = Consumer,
+    #consumer{status = Status} = Consumer,
     Effects =
-        case Active of
-            active ->
+        case Status of
+            ?ACTIVE ->
                 [notify_consumer_effect(ConnectionPid, SubscriptionId,
-                                        Stream, ConsumerName, is_active(Active))];
+                                        Stream, ConsumerName, is_active(Status))];
             _ ->
                 []
         end,
 
-    {State#?MODULE{groups = StreamGroups1}, {ok, is_active(Active)}, Effects};
+    {State#?MODULE{groups = StreamGroups1}, {ok, is_active(Status)}, Effects};
 do_register_consumer(VirtualHost,
                      Stream,
                      _PartitionIndex,
@@ -521,7 +567,7 @@ do_register_consumer(VirtualHost,
                     #consumer{pid = ConnectionPid,
                               owner = Owner,
                               subscription_id = SubscriptionId,
-                              active = active},
+                              status = ?ACTIVE},
                 G1 = add_to_group(Consumer0, Group0),
                 {G1,
                  [notify_consumer_effect(ConnectionPid, SubscriptionId,
@@ -532,7 +578,7 @@ do_register_consumer(VirtualHost,
                     #consumer{pid = ConnectionPid,
                               owner = Owner,
                               subscription_id = SubscriptionId,
-                              active = waiting},
+                              status = ?WAITING},
                 G1 = add_to_group(Consumer0, Group0),
 
                 case lookup_active_consumer(G1) of
@@ -548,7 +594,7 @@ do_register_consumer(VirtualHost,
                                 {update_consumer_state_in_group(G1,
                                                                 ActPid,
                                                                 ActSubId,
-                                                                deactivating),
+                                                                ?DEACTIVATING),
                                  [notify_consumer_effect(ActPid,
                                                          ActSubId,
                                                          Stream,
@@ -568,9 +614,9 @@ do_register_consumer(VirtualHost,
                       ConsumerName,
                       Group1,
                       StreamGroups0),
-    {value, #consumer{active = Active}} =
+    {value, #consumer{status = Status}} =
         lookup_consumer(ConnectionPid, SubscriptionId, Group1),
-    {State#?MODULE{groups = StreamGroups1}, {ok, is_active(Active)}, Effects}.
+    {State#?MODULE{groups = StreamGroups1}, {ok, is_active(Status)}, Effects}.
 
 handle_consumer_removal(#group{consumers = []} = G, _, _, _) ->
     {G, []};
@@ -606,7 +652,7 @@ handle_consumer_removal(Group0, Stream, ConsumerName, ActiveRemoved) ->
                     {update_consumer_state_in_group(Group0,
                                                     ActPid,
                                                     ActSubId,
-                                                    deactivating),
+                                                    ?DEACTIVATING),
                      [notify_consumer_effect(ActPid, ActSubId,
                                              Stream, ConsumerName, false, true)]}
             end;
@@ -616,7 +662,7 @@ handle_consumer_removal(Group0, Stream, ConsumerName, ActiveRemoved) ->
                     %% the active one is going away, picking a new one
                     #consumer{pid = P, subscription_id = SID} =
                         evaluate_active_consumer(Group0),
-                    {update_consumer_state_in_group(Group0, P, SID, active),
+                    {update_consumer_state_in_group(Group0, P, SID, ?ACTIVE),
                      [notify_consumer_effect(P, SID,
                                              Stream, ConsumerName, true)]};
                 false ->
@@ -683,13 +729,13 @@ compute_active_consumer(#group{consumers = Crs,
 compute_active_consumer(#group{partition_index = -1,
                                consumers = [Consumer0]} =
                             Group0) ->
-    Consumer1 = Consumer0#consumer{active = active},
+    Consumer1 = Consumer0#consumer{status = ?ACTIVE},
     Group0#group{consumers = [Consumer1]};
 compute_active_consumer(#group{partition_index = -1,
                                consumers = [Consumer0 | T]} =
                             Group0) ->
-    Consumer1 = Consumer0#consumer{active = active},
-    Consumers = lists:map(fun(C) -> C#consumer{active = waiting} end, T),
+    Consumer1 = Consumer0#consumer{status = ?ACTIVE},
+    Consumers = lists:map(fun(C) -> C#consumer{status = ?WAITING} end, T),
     Group0#group{consumers = [Consumer1] ++ Consumers}.
 
 evaluate_active_consumer(#group{partition_index = PartitionIndex,
@@ -706,7 +752,7 @@ lookup_consumer(ConnectionPid, SubscriptionId,
                  Consumers).
 
 lookup_active_consumer(#group{consumers = Consumers}) ->
-    lists:search(fun(#consumer{active = Active}) -> is_active(Active) end,
+    lists:search(fun(#consumer{status = Status}) -> is_active(Status) end,
                  Consumers).
 
 update_groups(_VirtualHost,
@@ -732,20 +778,20 @@ update_groups(VirtualHost,
 update_consumer_state_in_group(#group{consumers = Consumers0} = G,
                                Pid,
                                SubId,
-                               NewState) ->
+                               NewStatus) ->
     CS1 = lists:map(fun(C0) ->
                        case C0 of
                            #consumer{pid = Pid, subscription_id = SubId} ->
-                               C0#consumer{active = NewState};
+                               C0#consumer{status = NewStatus};
                            C -> C
                        end
                     end,
                     Consumers0),
     G#group{consumers = CS1}.
 
-update_consumers(#group{consumers = Consumers0} = G, NewState) ->
+update_consumers(#group{consumers = Consumers0} = G, NewStatus) ->
     Consumers1 = lists:map(fun(C) ->
-                                   C#consumer{active = NewState}
+                                   C#consumer{status = NewStatus}
                            end, Consumers0),
     G#group{consumers = Consumers1}.
 

deps/rabbit/src/rabbit_stream_sac_coordinator.hrl

@@ -22,12 +22,13 @@
 -type subscription_id() :: byte().
 -type group_id() :: {vhost(), stream(), consumer_name()}.
 -type owner() :: binary().
+-type consumer_status() :: active | waiting | deactivating.
 
 -record(consumer,
         {pid :: pid(),
          subscription_id :: subscription_id(),
          owner :: owner(), %% just a label
-         active :: active | waiting | deactivating}).
+         status :: consumer_status()}).
 -record(group,
         {consumers :: [#consumer{}], partition_index :: integer()}).
 -record(rabbit_stream_sac_coordinator,