Another take at #10068

Scan queues, exchanges and bindings before attempting
to import anything on boot. If they miss the virtual
host field, fail early and log a sensible message.

(cherry picked from commit 62fffb6)

Author: michaelklishin

deps/rabbit/src/rabbit_definitions.erl

@@ -409,6 +409,13 @@ should_skip_if_unchanged() ->
     ReachedTargetClusterSize = rabbit_nodes:reached_target_cluster_size(),
     OptedIn andalso ReachedTargetClusterSize.
 
+-spec any_orphaned_objects(list(#{atom() => any()})) -> boolean().
+any_orphaned_objects(Maps) ->
+    Filtered = lists:filter(fun(M) ->
+                              maps:get(<<"vhost">>, M, undefined) =:= undefined
+                            end, Maps),
+    length(Filtered) > 0.
+
 
 -spec apply_defs(Map :: #{atom() => any()}, ActingUser :: rabbit_types:username()) -> 'ok' | {error, term()}.
 
@@ -424,6 +431,25 @@ apply_defs(Map, ActingUser, VHost) when is_binary(VHost) ->
     apply_defs(Map, ActingUser, fun () -> ok end, VHost);
 apply_defs(Map, ActingUser, SuccessFun) when is_function(SuccessFun) ->
     Version = maps:get(rabbitmq_version, Map, maps:get(rabbit_version, Map, undefined)),
+
+    %% If any of the queues or exchanges do not have virtual hosts set,
+    %% this definition file was a virtual-host specific import. They cannot be applied
+    %% as "complete" definition imports, most notably, imported on boot.
+    HasQueuesWithoutVirtualHostField = any_orphaned_objects(maps:get(queues, Map, [])),
+    HasExchangesWithoutVirtualHostField = any_orphaned_objects(maps:get(exchanges, Map, [])),
+    HasBindingsWithoutVirtualHostField = any_orphaned_objects(maps:get(bindings, Map, [])),
+
+    case (HasQueuesWithoutVirtualHostField and HasExchangesWithoutVirtualHostField and HasBindingsWithoutVirtualHostField) of
+        true ->
+            rabbit_log:error("Definitions import: some queues, exchanges or bindings in the definition file "
+                             "are missing the virtual host field. Such files are produced when definitions of "
+                             "a single virtual host are exported. They cannot be used to import definitions at boot time",
+                             []),
+            throw({error, invalid_definitions_file});
+        false ->
+            ok
+    end,
+
     try
         concurrent_for_all(users, ActingUser, Map,
                 fun(User, _Username) ->
@@ -824,6 +850,7 @@ validate_limits(All) ->
         undefined -> ok;
         Queues0 ->
             {ok, VHostMap} = filter_out_existing_queues(Queues0),
+            _ = rabbit_log:debug("Definition import. Virtual host map for validation: ~p", [VHostMap]),
             maps:fold(fun validate_vhost_limit/3, ok, VHostMap)
     end.
 
@@ -855,12 +882,21 @@ build_filtered_map([], AccMap) ->
     {ok, AccMap};
 build_filtered_map([Queue|Rest], AccMap0) ->
     {Rec, VHost} = build_queue_data(Queue),
-    case rabbit_amqqueue:exists(Rec) of
-        false ->
-            AccMap1 = maps:update_with(VHost, fun(V) -> V + 1 end, 1, AccMap0),
-            build_filtered_map(Rest, AccMap1);
-        true ->
-            build_filtered_map(Rest, AccMap0)
+    %% If virtual host is not specified in a queue,
+    %% this definition file is likely virtual host-specific.
+    %%
+    %% Skip such queues.
+    case VHost of
+        undefined ->
+            build_filtered_map(Rest, AccMap0);
+        _Other ->
+            case rabbit_amqqueue:exists(Rec) of
+                false ->
+                    AccMap1 = maps:update_with(VHost, fun(V) -> V + 1 end, 1, AccMap0),
+                    build_filtered_map(Rest, AccMap1);
+                true ->
+                    build_filtered_map(Rest, AccMap0)
+            end
     end.
 
 validate_vhost_limit(VHost, AddCount, ok) ->

deps/rabbit/test/definition_import_SUITE.erl

@@ -53,7 +53,8 @@ groups() ->
                                import_case17,
                                import_case18,
                                import_case19,
-                               import_case20
+                               import_case20,
+                               import_case21
                               ]},
 
         {boot_time_import_using_classic_source, [], [
@@ -305,6 +306,8 @@ import_case20(Config) ->
             {skip, "Should not run in mixed version environments"}
     end.
 
+import_case21(Config) -> import_invalid_file_case(Config, "failing_case21").
+
 export_import_round_trip_case1(Config) ->
     case rabbit_ct_helpers:is_mixed_versions() of
       false ->

deps/rabbit/test/definition_import_SUITE_data/failing_case21.json

@@ -0,0 +1,33 @@
+{
+  "rabbit_version": "3.12.8",
+  "parameters": [],
+  "policies": [],
+  "queues": [
+    {
+      "name": "qq.1",
+      "durable": true,
+      "auto_delete": false,
+      "arguments": {
+        "x-queue-type": "quorum"
+      }
+    },
+    {
+      "name": "cq.1",
+      "durable": true,
+      "auto_delete": false,
+      "arguments": {
+        "x-queue-type": "classic"
+      }
+    },
+    {
+      "name": "sq.1",
+      "durable": true,
+      "auto_delete": false,
+      "arguments": {
+        "x-queue-type": "stream"
+      }
+    }
+  ],
+  "exchanges": [],
+  "bindings": []
+}

deps/rabbitmq_management/src/rabbit_mgmt_wm_definitions.erl

@@ -95,12 +95,14 @@ accept_json(ReqData0, Context) ->
     end.
 
 vhost_definitions(ReqData, VHost, Context) ->
-    %% rabbit_mgmt_wm_<>:basic/1 filters by VHost if it is available
+    %% rabbit_mgmt_wm_<>:basic/1 filters by VHost if it is available.
+    %% TODO: should we stop stripping virtual host? Such files cannot be imported on boot, for example.
     Xs = [strip_vhost(X) || X <- rabbit_mgmt_wm_exchanges:basic(ReqData),
                export_exchange(X)],
     VQs = [Q || Q <- rabbit_mgmt_wm_queues:basic(ReqData), export_queue(Q)],
     Qs = [strip_vhost(Q) || Q <- VQs],
     QNames = [{pget(name, Q), pget(vhost, Q)} || Q <- VQs],
+    %% TODO: should we stop stripping virtual host? Such files cannot be imported on boot, for example.
     Bs = [strip_vhost(B) || B <- rabbit_mgmt_wm_bindings:basic(ReqData),
                             export_binding(B, QNames)],
     {ok, Vsn} = application:get_key(rabbit, vsn),