Merge pull request #10636 from rabbitmq/mergify/bp/v3.13.x/pr-10569

Configurable writer module (backport #10569)

Author: michaelklishin

MODULE.bazel

@@ -43,7 +43,7 @@ bazel_dep(
 
 bazel_dep(
     name = "rabbitmq_osiris",
-    version = "1.7.2",
+    version = "1.8.0",
     repo_name = "osiris",
 )
 

deps/rabbit/Makefile

@@ -139,7 +139,7 @@ TEST_DEPS = rabbitmq_ct_helpers rabbitmq_ct_client_helpers amqp_client meck prop
 PLT_APPS += mnesia
 
 dep_syslog = git https://github.com/schlagert/syslog 4.0.0
-dep_osiris = git https://github.com/rabbitmq/osiris v1.7.2
+dep_osiris = git https://github.com/rabbitmq/osiris v1.8.0
 dep_systemd = hex 0.6.1
 
 dep_seshat = git https://github.com/rabbitmq/seshat v0.6.1

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -20,7 +20,9 @@
          tick/2,
          version/0,
          which_module/1,
-         overview/1,
+         overview/1]).
+
+-export([update_config/2,
          policy_changed/1]).
 
 %% coordinator API
@@ -236,16 +238,38 @@ delete_replica(StreamId, Node) ->
 
 policy_changed(Q) when ?is_amqqueue(Q) ->
     StreamId = maps:get(name, amqqueue:get_type_state(Q)),
+    Config = rabbit_stream_queue:update_stream_conf(Q, #{}),
+    case update_config(Q, Config) of
+        {ok, ok, _} = Res ->
+            Res;
+        {error, feature_not_enabled} ->
+            %% backwards compatibility
+            %% TODO: remove in future
+            process_command({policy_changed, StreamId, #{queue => Q}});
+        Err ->
+            Err
+    end.
+
+-spec update_config(amqqueue:amqqueue(), #{atom() => term()}) ->
+    {ok, ok, ra:server_id()} | {error, not_supported | term()}.
+update_config(Q, Config)
+  when ?is_amqqueue(Q) andalso is_map(Config) ->
     case rabbit_feature_flags:is_enabled(stream_update_config_command) of
         true ->
-            %% there are the only two configuration keys that are safe to
+            %% there are the only a few configuration keys that are safe to
             %% update
-            Conf = maps:with([filter_size,
-                              retention],
-                             rabbit_stream_queue:update_stream_conf(Q, #{})),
-            process_command({update_config, StreamId, Conf});
+            StreamId = maps:get(name, amqqueue:get_type_state(Q)),
+            case maps:with([filter_size,
+                            retention,
+                            writer_mod,
+                            replica_mod], Config) of
+                Conf when map_size(Conf) > 0 ->
+                    process_command({update_config, StreamId, Conf});
+                _ ->
+                    {error, no_updatable_keys}
+            end;
         false ->
-            process_command({policy_changed, StreamId, #{queue => Q}})
+            {error, feature_not_enabled}
     end.
 
 sac_state(#?MODULE{single_active_consumer = SacState}) ->
@@ -1016,7 +1040,7 @@ phase_start_replica(StreamId, #{epoch := Epoch,
                 {error, already_present} ->
                     %% need to remove child record if this is the case
                     %% can it ever happen?
-                    _ = osiris_replica:stop(Node, Conf0),
+                    _ = osiris:stop_member(Node, Conf0),
                     send_action_failed(StreamId, starting, Args);
                 {error, {already_started, Pid}} ->
                     %% TODO: we need to check that the current epoch is the same
@@ -1049,7 +1073,7 @@ phase_delete_member(StreamId, #{node := Node} = Arg, Conf) ->
     fun() ->
             case rabbit_nodes:is_member(Node) of
                 true ->
-                    try osiris_server_sup:delete_child(Node, Conf) of
+                    try osiris:delete_member(Node, Conf) of
                         ok ->
                             rabbit_log:info("~ts: Member deleted for ~ts : on node ~ts",
                                             [?MODULE, StreamId, Node]),
@@ -1071,10 +1095,9 @@ phase_delete_member(StreamId, #{node := Node} = Arg, Conf) ->
             end
     end.
 
-phase_stop_member(StreamId, #{node := Node,
-                              epoch := Epoch} = Arg0, Conf) ->
+phase_stop_member(StreamId, #{node := Node, epoch := Epoch} = Arg0, Conf) ->
     fun() ->
-            try osiris_server_sup:stop_child(Node, StreamId) of
+            try osiris_member:stop(Node, Conf) of
                 ok ->
                     %% get tail
                     try get_replica_tail(Node, Conf) of
@@ -1093,13 +1116,7 @@ phase_stop_member(StreamId, #{node := Node,
                                                [?MODULE, StreamId, Node, Epoch, Err]),
                             maybe_sleep(Err),
                             send_action_failed(StreamId, stopping, Arg0)
-                    end;
-                Err ->
-                    rabbit_log:warning("~ts: failed to stop "
-                                       "member ~ts ~w Error: ~w",
-                                       [?MODULE, StreamId, Node, Err]),
-                    maybe_sleep(Err),
-                    send_action_failed(StreamId, stopping, Arg0)
+                    end
             catch _:Err ->
                       rabbit_log:warning("~ts: failed to stop member ~ts ~w Error: ~w",
                                          [?MODULE, StreamId, Node, Err]),
@@ -1108,10 +1125,9 @@ phase_stop_member(StreamId, #{node := Node,
             end
     end.
 
-phase_start_writer(StreamId, #{epoch := Epoch,
-                               node := Node} = Args0, Conf) ->
+phase_start_writer(StreamId, #{epoch := Epoch, node := Node} = Args0, Conf) ->
     fun() ->
-            try osiris_writer:start(Conf) of
+            try osiris:start_writer(Conf) of
                 {ok, Pid} ->
                     Args = Args0#{epoch => Epoch, pid => Pid},
                     rabbit_log:info("~ts: started writer ~ts on ~w in ~b",
@@ -2019,7 +2035,6 @@ make_writer_conf(Node, #stream{epoch = Epoch,
           replica_nodes => lists:delete(Node, Nodes),
           epoch => Epoch}.
 
-
 find_leader(Members) ->
     case lists:partition(
            fun ({_, #member{target = deleted}}) ->