Merge pull request #3167 from rabbitmq/mk-duplicate-shovels-imported-from-definitions

Acquire locks before starting or restarting dynamic Shovels

(cherry picked from commit f08c895)

Author: michaelklishin

deps/rabbitmq_shovel/src/rabbit_shovel_dyn_worker_sup_sup.erl

@@ -35,28 +35,38 @@ adjust(Name, Def) ->
     end,
     start_child(Name, Def).
 
-start_child(Name, Def) ->
+start_child({VHost, ShovelName} = Name, Def) ->
+    rabbit_log_shovel:debug("Asked to start a dynamic Shovel named '~s' in virtual host '~s'", [ShovelName, VHost]),
+    LockId = rabbit_shovel_locks:lock(Name),
     cleanup_specs(),
-    case mirrored_supervisor:start_child(
+    rabbit_log_shovel:debug("Starting a mirrored supervisor named '~s' in virtual host '~s'", [ShovelName, VHost]),
+    Result = case mirrored_supervisor:start_child(
            ?SUPERVISOR,
            {Name, {rabbit_shovel_dyn_worker_sup, start_link, [Name, Def]},
             transient, ?WORKER_WAIT, worker, [rabbit_shovel_dyn_worker_sup]}) of
         {ok,                      _Pid}  -> ok;
         {error, {already_started, _Pid}} -> ok
-    end.
+    end,
+    %% release the lock if we managed to acquire one
+    rabbit_shovel_locks:unlock(LockId),
+    Result.
 
 child_exists(Name) ->
     lists:any(fun ({N, _, _, _}) -> N =:= Name end,
               mirrored_supervisor:which_children(?SUPERVISOR)).
 
-stop_child(Name) ->
+stop_child({VHost, ShovelName} = Name) ->
+    rabbit_log_shovel:debug("Asked to stop a dynamic Shovel named '~s' in virtual host '~s'", [ShovelName, VHost]),
+    LockId = rabbit_shovel_locks:lock(Name),
     case get({shovel_worker_autodelete, Name}) of
         true -> ok; %% [1]
         _ ->
             ok = mirrored_supervisor:terminate_child(?SUPERVISOR, Name),
             ok = mirrored_supervisor:delete_child(?SUPERVISOR, Name),
             rabbit_shovel_status:remove(Name)
-    end.
+    end,
+    rabbit_shovel_locks:unlock(LockId),
+    ok.
 
 %% [1] An autodeleting worker removes its own parameter, and thus ends
 %% up here via the parameter callback. It is a transient worker that

deps/rabbitmq_shovel/src/rabbit_shovel_locks.erl

@@ -0,0 +1,32 @@
+%% 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-2021 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_shovel_locks).
+
+-export([lock/1, unlock/1]).
+
+%%
+%% API
+%%
+
+lock(Name) ->
+    Nodes   = rabbit_nodes:all_running(),
+    Retries = rabbit_nodes:lock_retries(),
+    %% try to acquire a lock to avoid duplicate starts
+    LockId = case global:set_lock({dynamic_shovel, Name}, Nodes, Retries) of
+        true  -> Name;
+        false -> undefined
+    end,
+    LockId.
+
+unlock(LockId) ->
+    Nodes = rabbit_nodes:all_running(),
+    case LockId of
+        undefined -> ok;
+        Value     -> global:del_lock({dynamic_shovel, Value}, Nodes)
+    end,
+    ok.