Merge pull request #10085 from rabbitmq/mergify/bp/v3.12.x/pr-10072

michaelklishin · web-flow · commit 12ce1ed9a766 · 2023-12-09T22:57:24.000-05:00
Usability improvements during definition import #10068 (backport #10072)
diff --git a/deps/rabbit/src/rabbit_definitions.erl b/deps/rabbit/src/rabbit_definitions.erl
@@ -80,7 +80,8 @@
                                'bindings' |
                                'exchanges'.
 
--type definition_object() :: #{binary() => any()}.
+-type definition_key() :: binary() | atom().
+-type definition_object() :: #{definition_key() => any()}.
 -type definition_list() :: [definition_object()].
 
 -type definitions() :: #{
@@ -104,18 +105,73 @@ maybe_load_definitions() ->
         {error, E} -> {error, E}
     end.
 
-validate_definitions(Defs) when is_list(Defs) ->
+-spec validate_parsing_of_doc(any()) -> boolean().
+validate_parsing_of_doc(Body) when is_binary(Body) ->
+    case decode(Body) of
+        {ok, _Map}    -> true;
+        {error, _Err} -> false
+    end.
+
+-spec validate_parsing_of_doc_collection(list(any())) -> boolean().
+validate_parsing_of_doc_collection(Defs) when is_list(Defs) ->
+    lists:foldl(fun(_Body, false) ->
+        false;
+        (Body, true) ->
+            case decode(Body) of
+                {ok, _Map}    -> true;
+                {error, _Err} -> false
+            end
+    end, true, Defs).
+
+-spec filter_orphaned_objects(definition_list()) -> definition_list().
+filter_orphaned_objects(Maps) ->
+    lists:filter(fun(M) -> maps:get(<<"vhost">>, M, undefined) =:= undefined end, Maps).
+
+-spec any_orphaned_objects(definition_list()) -> boolean().
+any_orphaned_objects(Maps) ->
+    length(filter_orphaned_objects(Maps)) > 0.
+
+-spec any_orphaned_in_doc(definitions()) -> boolean().
+any_orphaned_in_doc(DefsMap) ->
+    any_orphaned_in_category(DefsMap, <<"queues">>)
+    orelse any_orphaned_in_category(DefsMap, <<"exchanges">>)
+    orelse any_orphaned_in_category(DefsMap, <<"bindings">>).
+
+-spec any_orphaned_in_category(definitions(), definition_category() | binary()) -> boolean().
+any_orphaned_in_category(DefsMap, Category) ->
+    %% try both binary and atom keys
+    any_orphaned_objects(maps:get(Category, DefsMap,
+                            maps:get(rabbit_data_coercion:to_atom(Category), DefsMap, []))).
+
+-spec validate_orphaned_objects_in_doc_collection(list() | binary()) -> boolean().
+validate_orphaned_objects_in_doc_collection(Defs) when is_list(Defs) ->
     lists:foldl(fun(_Body, false) ->
-                     false;
-                   (Body, true) ->
-                       case decode(Body) of
-                           {ok, _Map}    -> true;
-                           {error, _Err} -> false
-                       end
-                end, true, Defs);
+        false;
+        (Body, true) ->
+            validate_parsing_of_doc(Body)
+    end, true, Defs).
+
+-spec validate_orphaned_objects_in_doc(binary()) -> boolean().
+validate_orphaned_objects_in_doc(Body) when is_binary(Body) ->
+    case decode(Body) of
+        {ok, DefsMap}    ->
+            AnyOrphaned = any_orphaned_in_doc(DefsMap),
+            case AnyOrphaned of
+                true  ->
+                    log_an_error_about_orphaned_objects();
+                false -> ok
+            end,
+            AnyOrphaned;
+        {error, _Err} -> false
+    end.
+
+-spec validate_definitions(list(any()) | binary()) -> boolean().
+validate_definitions(Defs) when is_list(Defs) ->
+    validate_parsing_of_doc_collection(Defs) andalso
+    validate_orphaned_objects_in_doc_collection(Defs);
 validate_definitions(Body) when is_binary(Body) ->
     case decode(Body) of
-        {ok, _Map}    -> true;
+        {ok, Defs}    -> validate_orphaned_objects_in_doc(Defs);
         {error, _Err} -> false
     end.
 
@@ -284,6 +340,7 @@ maybe_load_definitions_from_local_filesystem(App, Key) ->
         undefined  -> ok;
         {ok, none} -> ok;
         {ok, Path} ->
+            rabbit_log:debug("~ts.~ts is set to '~ts', will discover definition file(s) to import", [App, Key, Path]),
             IsDir = filelib:is_dir(Path),
             Mod = rabbit_definitions_import_local_filesystem,
             rabbit_log:debug("Will use module ~ts to import definitions", [Mod]),
@@ -409,6 +466,10 @@ should_skip_if_unchanged() ->
     ReachedTargetClusterSize = rabbit_nodes:reached_target_cluster_size(),
     OptedIn andalso ReachedTargetClusterSize.
 
+log_an_error_about_orphaned_objects() ->
+    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").
 
 -spec apply_defs(Map :: #{atom() => any()}, ActingUser :: rabbit_types:username()) -> 'ok' | {error, term()}.
 
@@ -424,6 +485,20 @@ 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.
+    AnyOrphaned = any_orphaned_in_doc(Map),
+
+    case AnyOrphaned of
+        true ->
+            log_an_error_about_orphaned_objects(),
+            throw({error, invalid_definitions_file});
+        false ->
+            ok
+    end,
+
     try
         concurrent_for_all(users, ActingUser, Map,
                 fun(User, _Username) ->
@@ -587,8 +662,11 @@ do_concurrent_for_all(List, WorkPoolFun) ->
            fun() ->
                    _ = try
                        WorkPoolFun(M)
-                   catch {error, E} -> gatherer:in(Gatherer, {error, E});
-                         _:E        -> gatherer:in(Gatherer, {error, E})
+                   catch {error, E}     -> gatherer:in(Gatherer, {error, E});
+                         _:E:Stacktrace ->
+                             rabbit_log:debug("Definition import: a work pool operation has thrown an exception ~st, stacktrace: ~p",
+                                              [E, Stacktrace]),
+                             gatherer:in(Gatherer, {error, E})
                    end,
                    gatherer:finish(Gatherer)
            end)
@@ -731,6 +809,10 @@ add_queue_int(_Queue, R = #resource{kind = queue,
     Name = R#resource.name,
     rabbit_log:warning("Skipping import of a queue whose name begins with 'amq.', "
                        "name: ~ts, acting user: ~ts", [Name, ActingUser]);
+add_queue_int(_Queue, R = #resource{kind = queue, virtual_host = undefined}, ActingUser) ->
+    Name = R#resource.name,
+    rabbit_log:warning("Skipping import of a queue with an unset virtual host field, "
+    "name: ~ts, acting user: ~ts", [Name, ActingUser]);
 add_queue_int(Queue, Name = #resource{virtual_host = VHostName}, ActingUser) ->
     case rabbit_amqqueue:exists(Name) of
         true ->
@@ -824,6 +906,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.
 
@@ -848,19 +931,30 @@ filter_out_existing_queues(VHost, Queues) ->
 
 build_queue_data(Queue) ->
     VHost = maps:get(<<"vhost">>, Queue, undefined),
-    Rec = rv(VHost, queue, <<"name">>, Queue),
-    {Rec, VHost}.
+    case VHost of
+        undefined -> undefined;
+        Value ->
+            Rec = rv(Value, queue, <<"name">>, Queue),
+            {Rec, VHost}
+    end.
 
 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 build_queue_data(Queue) of
+        undefined -> build_filtered_map(Rest, AccMap0);
+        {Rec, VHost} when VHost =/= undefined ->
+            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) ->
diff --git a/deps/rabbit/src/rabbit_definitions_import_local_filesystem.erl b/deps/rabbit/src/rabbit_definitions_import_local_filesystem.erl
@@ -94,7 +94,9 @@ load_with_hashing(IsDir, Path, PreviousHash, Algo) when is_boolean(IsDir) ->
                             Other
                     end;
                 false ->
-                    rabbit_log:error("Failed to parse a definition file, path: ~p", [Path]),
+                    rabbit_log:error("Definitions file at path ~p failed validation. The file must be a valid JSON document "
+                                     "and all virtual host-scoped resources must have a virtual host field to be set. "
+                                     "Definition files exported for a single virtual host CANNOT be imported at boot time", [Path]),
                     {error, not_json}
             end
     end.
diff --git a/deps/rabbit/test/definition_import_SUITE.erl b/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 ->
@@ -386,7 +389,10 @@ import_file_case(Config, Subdirectory, CaseName) ->
 
 import_invalid_file_case(Config, CaseName) ->
     CasePath = filename:join(?config(data_dir, Config), CaseName ++ ".json"),
-    rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, run_invalid_import_case, [CasePath]),
+    try
+        rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, run_invalid_import_case, [CasePath])
+    catch _:_:_ -> ok
+    end,
     ok.
 
 import_invalid_file_case_if_unchanged(Config, CaseName) ->
diff --git a/deps/rabbit/test/definition_import_SUITE_data/failing_case21.json b/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": []
+}
diff --git a/deps/rabbitmq_cli/lib/rabbitmq/cli/ctl/commands/import_definitions_command.ex b/deps/rabbitmq_cli/lib/rabbitmq/cli/ctl/commands/import_definitions_command.ex
@@ -121,7 +121,7 @@ defmodule RabbitMQ.CLI.Ctl.Commands.ImportDefinitionsCommand do
       false ->
         {:ok,
          "Successfully started definition import. " <>
-           "This process is asynchronous and can take some time.\n"}
+           "This process is asynchronous and can take some time. Watch target node logs for completion.\n"}
     end
   end
 
diff --git a/deps/rabbitmq_management/src/rabbit_mgmt_load_definitions.erl b/deps/rabbitmq_management/src/rabbit_mgmt_load_definitions.erl
@@ -15,7 +15,6 @@
 %% Definition import functionality is now a core server feature.
 
 boot() ->
-    rabbit_log:debug("Will import definitions file from management.load_definitions"),
     rabbit_definitions:maybe_load_definitions(rabbitmq_management, load_definitions).
 
 maybe_load_definitions() ->
diff --git a/deps/rabbitmq_management/src/rabbit_mgmt_wm_definitions.erl b/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),
