Merge pull request #11823 from rabbitmq/mergify/bp/v4.0.x/pr-11793

Configure shovels to wait until all sources and destinations are declared (backport #11793)

Author: michaelklishin

deps/rabbitmq_shovel/app.bzl

@@ -110,6 +110,7 @@ def all_srcs(name = "all_srcs"):
 
     filegroup(
         name = "priv",
+        srcs = ["priv/schema/rabbitmq_shovel.schema"],
     )
 
     filegroup(

deps/rabbitmq_shovel/priv/schema/rabbitmq_shovel.schema

@@ -0,0 +1,11 @@
+%% ----------------------------------------------------------------------------
+%% RabbitMQ Shovel plugin
+%%
+%% See https://github.com/rabbitmq/rabbitmq-shovel/blob/stable/README.md
+%% for details
+%% ----------------------------------------------------------------------------
+
+
+{mapping, "shovel.topology.predeclared", "rabbitmq_shovel.topology.predeclared", [
+     [{datatype, {enum, [true, false]}}]
+]}.

deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl

@@ -7,6 +7,8 @@
 
 -module(rabbit_amqp091_shovel).
 
+-define(APP, rabbitmq_shovel).
+
 -behaviour(rabbit_shovel_behaviour).
 
 -include_lib("amqp_client/include/amqp_client.hrl").
@@ -39,7 +41,7 @@
 %% from and can break with the next upgrade. It should not be used by
 %% another one that the one who created it or survive a node restart.
 %% Thus, function references have been replace by the following MFA.
--export([decl_fun/3, publish_fun/4, props_fun_timestamp_header/4,
+-export([decl_fun/3, check_fun/3, publish_fun/4, props_fun_timestamp_header/4,
          props_fun_forward_header/5]).
 
 -define(MAX_CONNECTION_CLOSE_TIMEOUT, 10000).
@@ -54,7 +56,7 @@ parse(_Name, {source, Source}) ->
     CArgs = proplists:get_value(consumer_args, Source, []),
     #{module => ?MODULE,
       uris => proplists:get_value(uris, Source),
-      resource_decl => decl_fun(Source),
+      resource_decl => decl_fun({source, Source}),
       queue => Queue,
       delete_after => proplists:get_value(delete_after, Source, never),
       prefetch_count => Prefetch,
@@ -70,7 +72,7 @@ parse(Name, {destination, Dest}) ->
     PropsFun2 = add_timestamp_header_fun(ATH, PropsFun1),
     #{module => ?MODULE,
       uris => proplists:get_value(uris, Dest),
-      resource_decl  => decl_fun(Dest),
+      resource_decl  => decl_fun({destination, Dest}),
       props_fun => PropsFun2,
       fields_fun => PubFieldsFun,
       add_forward_headers => AFH,
@@ -604,16 +606,31 @@ parse_declaration({[{Method, Props} | _Rest], _Acc}) ->
 parse_declaration({[Method | Rest], Acc}) ->
     parse_declaration({[{Method, []} | Rest], Acc}).
 
-decl_fun(Endpoint) ->
-    Decl = parse_declaration({proplists:get_value(declarations, Endpoint, []),
-                              []}),
+decl_fun({source, Endpoint}) ->
+    case parse_declaration({proplists:get_value(declarations, Endpoint, []), []}) of 
+        [] -> 
+            case proplists:get_value(predeclared, application:get_env(?APP, topology, []), false) of
+                true -> case proplists:get_value(queue, Endpoint) of 
+                            <<>> -> fail({invalid_parameter_value, declarations, {require_non_empty}});
+                            Queue -> {?MODULE, check_fun, [Queue]}
+                        end;
+                false -> {?MODULE, decl_fun, []}
+            end;
+        Decl -> {?MODULE, decl_fun, [Decl]}
+    end;
+decl_fun({destination, Endpoint}) ->
+    Decl = parse_declaration({proplists:get_value(declarations, Endpoint, []), []}),
     {?MODULE, decl_fun, [Decl]}.
-
+    
 decl_fun(Decl, _Conn, Ch) ->
     [begin
          amqp_channel:call(Ch, M)
      end || M <- lists:reverse(Decl)].
 
+check_fun(Queue, _Conn, Ch) ->
+    amqp_channel:call(Ch, #'queue.declare'{queue   = Queue,
+                                            passive = true}).
+
 parse_parameter(Param, Fun, Value) ->
     try
         Fun(Value)

deps/rabbitmq_shovel/src/rabbit_shovel_parameters.erl

@@ -8,6 +8,8 @@
 -module(rabbit_shovel_parameters).
 -behaviour(rabbit_runtime_parameter).
 
+-define(APP, rabbitmq_shovel).
+
 -include_lib("amqp_client/include/amqp_client.hrl").
 -include("rabbit_shovel.hrl").
 
@@ -20,7 +22,8 @@
 %% from and can break with the next upgrade. It should not be used by
 %% another one that the one who created it or survive a node restart.
 %% Thus, function references have been replace by the following MFA.
--export([dest_decl/4, src_decl_exchange/4, src_decl_queue/4,
+-export([dest_decl/4, dest_check/4,
+         src_decl_exchange/4, src_decl_queue/4, src_check_queue/4,
          fields_fun/5, props_fun/9]).
 
 -import(rabbit_misc, [pget/2, pget/3, pset/3]).
@@ -146,7 +149,8 @@ amqp091_src_validation(_Def, User) ->
      %% a deprecated pre-3.7 setting
      {<<"delete-after">>, fun validate_delete_after/2, optional},
      %% currently used multi-protocol friend name, introduced in 3.7
-     {<<"src-delete-after">>, fun validate_delete_after/2, optional}
+     {<<"src-delete-after">>, fun validate_delete_after/2, optional},
+     {<<"src-predeclared">>,  fun rabbit_parameter_validation:boolean/2, optional}
     ].
 
 dest_validation(Def0, User) ->
@@ -178,7 +182,8 @@ amqp091_dest_validation(_Def, User) ->
      {<<"dest-add-forward-headers">>, fun rabbit_parameter_validation:boolean/2,optional},
      {<<"dest-add-timestamp-header">>, fun rabbit_parameter_validation:boolean/2,optional},
      {<<"publish-properties">>, fun validate_properties/2,  optional},
-     {<<"dest-publish-properties">>, fun validate_properties/2,  optional}
+     {<<"dest-publish-properties">>, fun validate_properties/2,  optional},
+     {<<"dest-predeclared">>,  fun rabbit_parameter_validation:boolean/2, optional}
     ].
 
 validate_uri_fun(User) ->
@@ -329,7 +334,13 @@ parse_amqp091_dest({VHost, Name}, ClusterName, Def, SourceHeaders) ->
     DestXKey  = pget(<<"dest-exchange-key">>, Def, none),
     DestQ     = pget(<<"dest-queue">>,        Def, none),
     DestQArgs = pget(<<"dest-queue-args">>,   Def, #{}),
-    DestDeclFun = {?MODULE, dest_decl, [DestQ, DestQArgs]},
+    GlobalPredeclared = proplists:get_value(predeclared, application:get_env(?APP, topology, []), false),
+    Predeclared = pget(<<"dest-predeclared">>, Def, GlobalPredeclared),
+    DestDeclFun = case Predeclared of 
+        true -> {?MODULE, dest_check, [DestQ, DestQArgs]};
+        false -> {?MODULE, dest_decl, [DestQ, DestQArgs]}
+    end,
+    
     {X, Key} = case DestQ of
                    none -> {DestX, DestXKey};
                    _    -> {<<>>,  DestQ}
@@ -394,6 +405,11 @@ dest_decl(DestQ, DestQArgs, Conn, _Ch) ->
         none -> ok;
         _ -> ensure_queue(Conn, DestQ, rabbit_misc:to_amqp_table(DestQArgs))
     end.
+dest_check(DestQ, DestQArgs, Conn, _Ch) ->
+    case DestQ of
+        none -> ok;
+        _ -> check_queue(Conn, DestQ, rabbit_misc:to_amqp_table(DestQArgs))
+    end.
 
 parse_amqp10_source(Def) ->
     Uris = deobfuscated_uris(<<"src-uri">>, Def),
@@ -415,13 +431,21 @@ parse_amqp091_source(Def) ->
     SrcQ     = pget(<<"src-queue">>, Def, none),
     SrcQArgs = pget(<<"src-queue-args">>,   Def, #{}),
     SrcCArgs = rabbit_misc:to_amqp_table(pget(<<"src-consumer-args">>, Def, [])),
+    GlobalPredeclared = proplists:get_value(predeclared, application:get_env(?APP, topology, []), false),
+    Predeclared = pget(<<"src-predeclared">>, Def, GlobalPredeclared),
     {SrcDeclFun, Queue, DestHeaders} =
     case SrcQ of
         none -> {{?MODULE, src_decl_exchange, [SrcX, SrcXKey]}, <<>>,
                  [{<<"src-exchange">>,     SrcX},
                   {<<"src-exchange-key">>, SrcXKey}]};
-        _ -> {{?MODULE, src_decl_queue, [SrcQ, SrcQArgs]},
-              SrcQ, [{<<"src-queue">>, SrcQ}]}
+        _ -> case Predeclared of 
+                false -> 
+                    {{?MODULE, src_decl_queue, [SrcQ, SrcQArgs]},
+                        SrcQ, [{<<"src-queue">>, SrcQ}]};
+                true ->
+                    {{?MODULE, src_check_queue, [SrcQ, SrcQArgs]},
+                        SrcQ, [{<<"src-queue">>, SrcQ}]}
+            end
     end,
     DeleteAfter = pget(<<"src-delete-after">>, Def,
                        pget(<<"delete-after">>, Def, <<"never">>)),
@@ -450,6 +474,9 @@ src_decl_exchange(SrcX, SrcXKey, _Conn, Ch) ->
 src_decl_queue(SrcQ, SrcQArgs, Conn, _Ch) ->
     ensure_queue(Conn, SrcQ, rabbit_misc:to_amqp_table(SrcQArgs)).
 
+src_check_queue(SrcQ, SrcQArgs, Conn, _Ch) ->
+    check_queue(Conn, SrcQ, rabbit_misc:to_amqp_table(SrcQArgs)).
+
 get_uris(Key, Def) ->
     URIs = case pget(Key, Def) of
                B when is_binary(B) -> [B];
@@ -481,7 +508,14 @@ ensure_queue(Conn, Queue, XArgs) ->
     after
         catch amqp_channel:close(Ch)
     end.
-
+check_queue(Conn, Queue, _XArgs) ->
+    {ok, Ch} = amqp_connection:open_channel(Conn),
+    try
+        amqp_channel:call(Ch, #'queue.declare'{queue   = Queue,
+                                               passive = true})
+    after
+        catch amqp_channel:close(Ch)
+    end.
 opt_b2a(B) when is_binary(B) -> list_to_atom(binary_to_list(B));
 opt_b2a(N)                   -> N.
 

deps/rabbitmq_shovel/test/configuration_SUITE.erl

@@ -12,6 +12,7 @@
 
 -compile(export_all).
 
+-define(QUEUE,       <<"test_queue">>).
 -define(EXCHANGE,    <<"test_exchange">>).
 -define(TO_SHOVEL,   <<"to_the_shovel">>).
 -define(FROM_SHOVEL, <<"from_the_shovel">>).
@@ -21,7 +22,8 @@
 
 all() ->
     [
-      {group, non_parallel_tests}
+      {group, non_parallel_tests},
+      {group, with_predefined_topology}
     ].
 
 groups() ->
@@ -31,7 +33,10 @@ groups() ->
           invalid_legacy_configuration,
           valid_legacy_configuration,
           valid_configuration
-        ]}
+        ]},
+      {with_predefined_topology, [], [
+          valid_configuration_with_predefined_resources
+      ]}
     ].
 
 %% -------------------------------------------------------------------
@@ -41,7 +46,9 @@ groups() ->
 init_per_suite(Config) ->
     rabbit_ct_helpers:log_environment(),
     Config1 = rabbit_ct_helpers:set_config(Config, [
-        {rmq_nodename_suffix, ?MODULE}
+        {rmq_nodename_suffix, ?MODULE},
+         {ignored_crashes,
+                  ["server_initiated_close,404"]}
       ]),
     rabbit_ct_helpers:run_setup_steps(Config1,
       rabbit_ct_broker_helpers:setup_steps() ++
@@ -53,9 +60,19 @@ end_per_suite(Config) ->
       rabbit_ct_client_helpers:teardown_steps() ++
       rabbit_ct_broker_helpers:teardown_steps()).
 
+init_per_group(with_predefined_topology, Config) ->
+    ok = rabbit_ct_broker_helpers:rpc(Config, 0, application, set_env,
+        [rabbitmq_shovel, topology, [{predeclared, true}]]),
+    Config;
+
 init_per_group(_, Config) ->
     Config.
 
+end_per_group(with_predefined_topology, Config) ->
+     ok = rabbit_ct_broker_helpers:rpc(Config, 0, application, unset_env,
+        [rabbitmq_shovel, topology]),
+    Config;
+
 end_per_group(_, Config) ->
     Config.
 
@@ -209,6 +226,12 @@ valid_configuration(Config) ->
     ok = setup_shovels(Config),
     run_valid_test(Config).
 
+valid_configuration_with_predefined_resources(Config) ->
+    ok = rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, setup_shovels2, [Config]),
+    ok = rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE,  await_terminated_shovel, [test_shovel]),    
+    declare_queue(Config),
+    ok = rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE,  await_running_shovel, [test_shovel]).
+
 run_valid_test(Config) ->
     Chan = rabbit_ct_client_helpers:open_channel(Config, 0),
 
@@ -271,6 +294,12 @@ run_valid_test(Config) ->
 
     rabbit_ct_client_helpers:close_channel(Chan).
 
+declare_queue(Config) ->
+    Chan = rabbit_ct_client_helpers:open_channel(Config, 0),
+    amqp_channel:call(Chan, #'queue.declare'{queue   = ?QUEUE,
+                                             durable = true}),             
+    rabbit_ct_client_helpers:close_channel(Chan).
+
 setup_legacy_shovels(Config) ->
     ok = rabbit_ct_broker_helpers:rpc(Config, 0,
       ?MODULE, setup_legacy_shovels1, [Config]).
@@ -349,6 +378,34 @@ setup_shovels1(Config) ->
     ok = application:start(rabbitmq_shovel),
     await_running_shovel(test_shovel).
 
+setup_shovels2(Config) ->
+    _ = application:stop(rabbitmq_shovel),
+    Hostname = ?config(rmq_hostname, Config),
+    TcpPort = rabbit_ct_broker_helpers:get_node_config(Config, 0,
+      tcp_port_amqp),
+    %% a working config
+    application:set_env(
+      rabbitmq_shovel,
+      shovels,
+      [{test_shovel,
+        [{source,
+          [{uris, [rabbit_misc:format("amqp://~ts:~b/%2f?heartbeat=5",
+                                      [Hostname, TcpPort])]},
+           {queue, ?QUEUE}]},
+         {destination,
+          [{uris, [rabbit_misc:format("amqp://~ts:~b/%2f",
+                                      [Hostname, TcpPort])]},
+           {publish_fields, [{exchange, ?EXCHANGE}, {routing_key, ?FROM_SHOVEL}]},
+           {publish_properties, [{delivery_mode, 2},
+                                 {cluster_id,    <<"my-cluster">>},
+                                 {content_type,  ?SHOVELLED}]},
+           {add_forward_headers, true},
+           {add_timestamp_header, true}]},
+         {ack_mode, on_confirm}]}],
+      infinity),
+
+    ok = application:start(rabbitmq_shovel).    
+
 await_running_shovel(Name) ->
     case [N || {N, _, {running, _}, _}
                       <- rabbit_shovel_status:status(),
@@ -357,3 +414,11 @@ await_running_shovel(Name) ->
         _   -> timer:sleep(100),
                await_running_shovel(Name)
     end.
+await_terminated_shovel(Name) ->
+    case [N || {N, _, {terminated, _}, _}
+                      <- rabbit_shovel_status:status(),
+                         N =:= Name] of
+        [_] -> ok;
+        _   -> timer:sleep(100),
+               await_terminated_shovel(Name)
+    end.