Merge pull request #14425 from rabbitmq/mk-shovel-library-mode

Shovel: introduce operating modes

Author: michaelklishin

deps/rabbitmq_shovel/include/rabbit_shovel.hrl

@@ -5,6 +5,8 @@
 %% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
 %%
 
+-define(SHOVEL_APP, rabbitmq_shovel).
+
 -record(endpoint,
         {uris,
          resource_declaration

deps/rabbitmq_shovel/priv/schema/rabbitmq_shovel.schema

@@ -2,6 +2,10 @@
 %% RabbitMQ Shovel plugin
 %% ----------------------------------------------------------------------------
 
+{mapping, "shovel.operating_mode", "rabbitmq_shovel.operating_mode", [
+     [{datatype, {enum, [standard, alternative, library]}}]
+]}.
+
 {mapping, "shovel.topology.predeclared", "rabbitmq_shovel.topology.predeclared", [
      [{datatype, {enum, [true, false]}}]
 ]}.

deps/rabbitmq_shovel/src/rabbit_shovel_dyn_worker_sup_sup.erl

@@ -26,7 +26,14 @@ start_link() ->
             {ok, Pid0}                       -> Pid0;
             {error, {already_started, Pid0}} -> Pid0
           end,
-    Shovels = rabbit_runtime_parameters:list_component(<<"shovel">>),
+    OpMode = rabbit_shovel_operating_mode:operating_mode(),
+    Shovels = case OpMode of
+        standard ->
+            rabbit_runtime_parameters:list_component(<<"shovel">>);
+        _Other ->
+            %% when operating in a non-standard mode, do not start any shovels
+            []
+    end,
     [start_child({pget(vhost, Shovel), pget(name, Shovel)},
                  pget(value, Shovel)) || Shovel <- Shovels],
     {ok, Pid}.

deps/rabbitmq_shovel/src/rabbit_shovel_operating_mode.erl

@@ -0,0 +1,24 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%%
+
+-module(rabbit_shovel_operating_mode).
+
+-include("rabbit_shovel.hrl").
+
+-export([
+  operating_mode/0
+]).
+
+-type operating_mode() :: 'standard' | atom().
+
+%%
+%% API
+%%
+
+-spec operating_mode() -> operating_mode().
+operating_mode() ->
+    application:get_env(?SHOVEL_APP, operating_mode, standard).

deps/rabbitmq_shovel/src/rabbit_shovel_parameters.erl

@@ -39,10 +39,22 @@
                     {enables, recovery}]}).
 
 register() ->
-    rabbit_registry:register(runtime_parameter, <<"shovel">>, ?MODULE).
+    OpMode = rabbit_shovel_operating_mode:operating_mode(),
+    case OpMode of
+        standard ->
+            rabbit_registry:register(runtime_parameter, <<"shovel">>, ?MODULE);
+        _Other ->
+            ?LOG_DEBUG("Shovel: skipping runtime parameter registration, operating mode: ~ts", [OpMode])
+    end.
 
 unregister() ->
-    rabbit_registry:unregister(runtime_parameter, <<"shovel">>).
+    OpMode = rabbit_shovel_operating_mode:operating_mode(),
+    case OpMode of
+        standard ->
+            rabbit_registry:unregister(runtime_parameter, <<"shovel">>);
+        _Other ->
+            ?LOG_DEBUG("Shovel: skipping runtime parameter deregistration, operating mode: ~ts", [OpMode])
+    end.
 
 validate(_VHost, <<"shovel">>, Name, Def0, User) ->
     Def = rabbit_data_coercion:to_proplist(Def0),
@@ -65,10 +77,22 @@ pget2(K1, K2, Defs) -> case {pget(K1, Defs), pget(K2, Defs)} of
                        end.
 
 notify(VHost, <<"shovel">>, Name, Definition, _Username) ->
-    rabbit_shovel_dyn_worker_sup_sup:adjust({VHost, Name}, Definition).
+    OpMode = rabbit_shovel_operating_mode:operating_mode(),
+    case OpMode of
+        standard ->
+            rabbit_shovel_dyn_worker_sup_sup:adjust({VHost, Name}, Definition);
+        _Other ->
+            ?LOG_DEBUG("Shovel: ignoring a runtime parameter update, operating mode: ~ts", [OpMode])
+    end.
 
 notify_clear(VHost, <<"shovel">>, Name, _Username) ->
-    rabbit_shovel_dyn_worker_sup_sup:stop_child({VHost, Name}).
+    OpMode = rabbit_shovel_operating_mode:operating_mode(),
+    case OpMode of
+        standard ->
+            rabbit_shovel_dyn_worker_sup_sup:stop_child({VHost, Name});
+        _Other ->
+            ?LOG_DEBUG("Shovel: ignoring a cleared runtime parameter, operating mode: ~ts", [OpMode])
+    end.
 
 %%----------------------------------------------------------------------------
 

deps/rabbitmq_shovel/src/rabbit_shovel_sup.erl

@@ -8,6 +8,9 @@
 -module(rabbit_shovel_sup).
 -behaviour(supervisor).
 
+-include_lib("kernel/include/logger.hrl").
+-include_lib("logging.hrl").
+
 -export([start_link/0, init/1]).
 
 -import(rabbit_shovel_config, []).
@@ -21,6 +24,9 @@ start_link() ->
     end.
 
 init([Configurations]) ->
+    OpMode = rabbit_shovel_operating_mode:operating_mode(),
+    ?LOG_DEBUG("Shovel: operating mode set to ~ts", [OpMode]),
+    StaticShovelSpecs = make_child_specs(OpMode, Configurations),
     Len = dict:size(Configurations),
     ChildSpecs = [
         #{
@@ -39,11 +45,13 @@ init([Configurations]) ->
             type => supervisor,
             modules => [rabbit_shovel_dyn_worker_sup_sup]
         }
-        | make_child_specs(Configurations)
+        | StaticShovelSpecs
     ],
-    {ok, {#{strategy => one_for_one, intensity => 2 * Len, period => 2}, ChildSpecs}}.
+    Opts = #{strategy => one_for_one, intensity => 2 * Len, period => 2},
+    {ok, {Opts, ChildSpecs}}.
+
 
-make_child_specs(Configurations) ->
+make_child_specs(standard, Configurations) ->
     dict:fold(
       fun (ShovelName, ShovelConfig, Acc) ->
             [
@@ -57,7 +65,10 @@ make_child_specs(Configurations) ->
                 }
                 | Acc
             ]
-        end, [], Configurations).
+        end, [], Configurations);
+make_child_specs(_NonStandardOpMode, _Configurations) ->
+    %% when operating in a non-standard mode, do not start any shovels
+    [].
 
 parse_configuration(undefined) ->
     {ok, dict:new()};