Refactor definition import to allow for arbitrary sources

The classic local filesystem source is still supported
using the same traditional configuration key, load_definitions.

Configuration schema follows peer discovery in spirit:

 * definitions.import_backend configures the mechanism to use,
   which can be a module provided by a plugin
 * definitions.* keys can be defined by plugins and contain any
   keys a specific mechanism needs

For example, the classic local filesystem source can now be
configured like this:

``` ini
definitions.import_backend = local_filesystem
definitions.local.path = /path/to/definitions.d/definition.json
```

``` ini
definitions.import_backend = https
definitions.https.url = https://hostname/path/to/definitions.json
```

HTTPS may require additional configuration keys related to TLS/x.509
peer verification. Such extra keys will be added as the need for them
becomes evident.

References #3249

7 files changed, 391 insertions, 58 deletions

diff --git a/deps/rabbit/priv/schema/rabbit.schema b/deps/rabbit/priv/schema/rabbit.schema
index a8b2930467..c4688b64a5 100644
--- a/deps/rabbit/priv/schema/rabbit.schema
+++ b/deps/rabbit/priv/schema/rabbit.schema
@@ -121,15 +121,44 @@ end}.
 %% Definition import
 %%
 
-%% Load definitions from a JSON file or directory of files. See
+%% Original key for definition loading from a JSON file or directory of files. See
 %% https://www.rabbitmq.com/management.html#load-definitions
-%%
-%% {load_definitions, "/path/to/schema.json"},
-%% {load_definitions, "/path/to/schemas"},
 {mapping, "load_definitions", "rabbit.load_definitions",
     [{datatype, string},
      {validators, ["file_accessible"]}]}.
 
+%% Newer syntax for definition loading from a JSON file or directory of files. See
+%% https://www.rabbitmq.com/management.html#load-definitions
+{mapping, "definitions.local.path", "rabbit.definitions.local_path",
+    [{datatype, string},
+     {validators, ["file_accessible"]}]}.
+
+%% Extensive mechanism for loading definitions from a remote source
+{mapping, "definitions.import_backend", "rabbit.definitions.import_backend", [
+    {datatype, atom}
+]}.
+
+{translation, "rabbit.definitions.import_backend",
+fun(Conf) ->
+    case cuttlefish:conf_get("definitions.import_backend", Conf, rabbit_definitions_import_local_filesystem) of
+        %% short aliases for known backends
+        local_filesystem -> rabbit_definitions_import_local_filesystem;
+        local            -> rabbit_definitions_import_local_filesystem;
+        https            -> rabbit_definitions_import_https;
+        http             -> rabbit_definitions_import_https;
+        %% accept both rabbitmq_ and rabbit_ (typical core module prefix)
+        rabbitmq_definitions_import_local_filesystem -> rabbit_definitions_import_local_filesystem;
+        rabbitmq_definitions_import_local_filesystem -> rabbit_definitions_import_https;
+        %% any other value is used as is
+        Module           -> Module
+    end
+end}.
+
+%% Load definitions from a remote URL over HTTPS. See
+%% https://www.rabbitmq.com/management.html#load-definitions
+{mapping, "definitions.https.url", "rabbit.definitions.url",
+    [{datatype, string}]}.
+
 %%
 %% Security / AAA
 %% ==============
diff --git a/deps/rabbit/src/rabbit_definitions.erl b/deps/rabbit/src/rabbit_definitions.erl
index 0a07d8f175..2edc233d74 100644
--- a/deps/rabbit/src/rabbit_definitions.erl
+++ b/deps/rabbit/src/rabbit_definitions.erl
@@ -10,8 +10,13 @@
 
 -export([boot/0]).
 %% automatic import on boot
--export([maybe_load_definitions/0, maybe_load_definitions/2, maybe_load_definitions_from/2,
-         has_configured_definitions_to_load/0]).
+-export([
+    maybe_load_definitions/0,
+    maybe_load_definitions/2,
+    maybe_load_definitions_from/2,
+
+    has_configured_definitions_to_load/0
+]).
 %% import
 -export([import_raw/1, import_raw/2, import_parsed/1, import_parsed/2,
          apply_defs/2, apply_defs/3, apply_defs/4, apply_defs/5]).
@@ -25,7 +30,7 @@
 ]).
 -export([decode/1, decode/2, args/1]).
 
--import(rabbit_misc, [pget/2]).
+-import(rabbit_misc, [pget/2, pget/3]).
 -import(rabbit_data_coercion, [to_binary/1]).
 
 %%
@@ -59,9 +64,10 @@ boot() ->
     rabbit_sup:start_supervisor_child(definition_import_pool_sup, worker_pool_sup, [PoolSize, ?IMPORT_WORK_POOL]).
 
 maybe_load_definitions() ->
-    %% Note that management.load_definitions is handled in the plugin for backwards compatibility.
-    %% This executes the "core" version of load_definitions.
-    maybe_load_definitions(rabbit, load_definitions).
+    %% Classic source: local file or data directory
+    maybe_load_definitions_from_local_filesystem(rabbit, load_definitions),
+    %% Extensible sources
+    maybe_load_definitions_from_pluggable_source(rabbit, definitions).
 
 -spec import_raw(Body :: binary() | iolist()) -> ok | {error, term()}.
 import_raw(Body) ->
@@ -126,63 +132,76 @@ all_definitions() ->
         exchanges         => Xs
     }.
 
+-spec has_configured_definitions_to_load() -> boolean().
+has_configured_definitions_to_load() ->
+    has_configured_definitions_to_load_via_classic_option() or has_configured_definitions_to_load_via_modern_option().
+
+%% Retained for backwards compatibility, implicitly assumes the local filesystem source
+maybe_load_definitions(App, Key) ->
+    maybe_load_definitions_from_local_filesystem(App, Key).
+
+maybe_load_definitions_from(IsDir, Path) ->
+    rabbit_definitions_import_local_filesystem:load(IsDir, Path).
+
 %%
 %% Implementation
 %%
 
--spec has_configured_definitions_to_load() -> boolean().
-has_configured_definitions_to_load() ->
+-spec has_configured_definitions_to_load_via_modern_option() -> boolean().
+has_configured_definitions_to_load_via_modern_option() ->
+    case application:get_env(rabbit, definitions) of
+        undefined  -> false;
+        {ok, none} -> false;
+        {ok, []}   -> false;
+        {ok, _Options} -> true
+    end.
+
+has_configured_definitions_to_load_via_classic_option() ->
     case application:get_env(rabbit, load_definitions) of
         undefined   -> false;
         {ok, none}  -> false;
         {ok, _Path} -> true
     end.
 
-maybe_load_definitions(App, Key) ->
+maybe_load_definitions_from_local_filesystem(App, Key) ->
     case application:get_env(App, Key) of
-        undefined  ->
-            rabbit_log:debug("No definition file configured to import via load_definitions"),
-            ok;
-        {ok, none} ->
-            rabbit_log:debug("No definition file configured to import via load_definitions"),
-            ok;
-        {ok, FileOrDir} ->
-            rabbit_log:debug("Will import definitions file from load_definitions"),
-            IsDir = filelib:is_dir(FileOrDir),
-            maybe_load_definitions_from(IsDir, FileOrDir)
+        undefined  -> ok;
+        {ok, none} -> ok;
+        {ok, Path} ->
+            IsDir = filelib:is_dir(Path),
+            rabbit_definitions_import_local_filesystem:load(IsDir, Path)
     end.
 
-maybe_load_definitions_from(true, Dir) ->
-    rabbit_log:info("Applying definitions from directory ~s", [Dir]),
-    load_definitions_from_files(file:list_dir(Dir), Dir);
-maybe_load_definitions_from(false, File) ->
-    load_definitions_from_file(File).
-
-load_definitions_from_files({ok, Filenames0}, Dir) ->
-    Filenames1 = lists:sort(Filenames0),
-    Filenames2 = [filename:join(Dir, F) || F <- Filenames1],
-    load_definitions_from_filenames(Filenames2);
-load_definitions_from_files({error, E}, Dir) ->
-    rabbit_log:error("Could not read definitions from directory ~s, Error: ~p", [Dir, E]),
-    {error, {could_not_read_defs, E}}.
-
-load_definitions_from_filenames([]) ->
-    ok;
-load_definitions_from_filenames([File|Rest]) ->
-    case load_definitions_from_file(File) of
-        ok         -> load_definitions_from_filenames(Rest);
-        {error, E} -> {error, {failed_to_import_definitions, File, E}}
+maybe_load_definitions_from_pluggable_source(App, Key) ->
+    case application:get_env(App, Key) of
+        undefined  -> ok;
+        {ok, none} -> ok;
+        {ok, []}   -> ok;
+        {ok, Proplist} ->
+            case pget(import_backend, Proplist, undefined) of
+                undefined  ->
+                    {error, "definition import source is configured but definitions.import_backend is not set"};
+                ModOrAlias ->
+                    Mod = normalize_backend_module(ModOrAlias),
+                    rabbit_log:debug("Will use module ~s to import definitions", [Mod]),
+                    Mod:load(Proplist)
+            end
     end.
 
-load_definitions_from_file(File) ->
-    case file:read_file(File) of
-        {ok, Body} ->
-            rabbit_log:info("Applying definitions from file at '~s'", [File]),
-            import_raw(Body);
-        {error, E} ->
-            rabbit_log:error("Could not read definitions from file at '~s', error: ~p", [File, E]),
-            {error, {could_not_read_defs, {File, E}}}
-    end.
+normalize_backend_module(local_filesystem) ->
+    rabbit_definitions_import_local_filesystem;
+normalize_backend_module(local) ->
+    rabbit_definitions_import_local_filesystem;
+normalize_backend_module(https) ->
+    rabbit_definitions_import_https;
+normalize_backend_module(http)  ->
+    rabbit_definitions_import_https;
+normalize_backend_module(rabbitmq_definitions_import_local_filesystem) ->
+    rabbit_definitions_import_local_filesystem;
+normalize_backend_module(rabbitmq_definitions_import_https) ->
+    rabbit_definitions_import_https;
+normalize_backend_module(Other) ->
+    Other.
 
 decode(Keys, Body) ->
     case decode(Body) of
diff --git a/deps/rabbit/src/rabbit_definitions_import_https.erl b/deps/rabbit/src/rabbit_definitions_import_https.erl
new file mode 100644
index 0000000000..7115f2d2e8
--- /dev/null
+++ b/deps/rabbit/src/rabbit_definitions_import_https.erl
@@ -0,0 +1,72 @@
+%% 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_definitions_import_https).
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-export([
+    is_enabled/0,
+    load/1
+]).
+
+
+
+-import(rabbit_misc, [pget/2]).
+-import(rabbit_data_coercion, [to_binary/1]).
+-import(rabbit_definitions, [import_raw/1]).
+
+%%
+%% API
+%%
+
+-spec is_enabled() -> boolean().
+is_enabled() ->
+    case application:get_env(rabbit, definitions) of
+        undefined   -> false;
+        {ok, none}  -> false;
+        {ok, []}    -> false;
+        {ok, Proplist} ->
+            case proplists:get_value(import_backend, Proplist, undefined) of
+                undefined -> false;
+                ?MODULE   -> true;
+                _         -> false
+            end
+    end.
+
+load(Proplist) ->
+    URL = pget(url, Proplist),
+    %% TODO
+    HTTPOptions = [],
+    load_from_url(URL, HTTPOptions).
+
+
+%%
+%% Implementation
+%%
+
+load_from_url(URL, HTTPOptions0) ->
+    inets:start(),
+    Options = [
+        {body_format, binary}
+    ],
+    HTTPOptions = HTTPOptions0 ++ [
+        {autoredirect, true}
+    ],
+    rabbit_log:info("Applying definitions from remote URL"),
+    case httpc:request(get, {URL, []}, lists:usort(HTTPOptions), Options) of
+        %% 2XX
+        {ok, {{_, Code, _}, _Headers, Body}} when Code div 100 == 2 ->
+            rabbit_log:debug("Requested definitions from remote URL '~s', response code: ~b", [URL, Code]),
+            rabbit_log:debug("Requested definitions from remote URL '~s', body: ~p", [URL, Body]),
+            import_raw(Body);
+        {ok, {{_, Code, _}, _Headers, _Body}} when Code >= 400 ->
+            rabbit_log:debug("Requested definitions from remote URL '~s', response code: ~b", [URL, Code]),
+            {error, {could_not_read_defs, {URL, rabbit_misc:format("URL request failed with response code ~b", [Code])}}};
+        {error, Reason} ->
+            rabbit_log:error("Requested definitions from remote URL '~s', error: ~p", [URL, Reason]),
+            {error, {could_not_read_defs, {URL, Reason}}}
+    end.
diff --git a/deps/rabbit/src/rabbit_definitions_import_local_filesystem.erl b/deps/rabbit/src/rabbit_definitions_import_local_filesystem.erl
new file mode 100644
index 0000000000..896fcaa40e
--- /dev/null
+++ b/deps/rabbit/src/rabbit_definitions_import_local_filesystem.erl
@@ -0,0 +1,140 @@
+%% 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_definitions_import_local_filesystem).
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-export([
+    is_enabled/0,
+    %% definition source options
+    load/1,
+    %% classic arguments specific to this source
+    load/2,
+    location/0
+]).
+
+
+
+-import(rabbit_misc, [pget/2, pget/3]).
+-import(rabbit_data_coercion, [to_binary/1]).
+-import(rabbit_definitions, [import_raw/1]).
+
+%%
+%% API
+%%
+
+-spec is_enabled() -> boolean().
+is_enabled() ->
+    is_enabled_via_classic_option() or is_enabled_via_modern_option().
+
+load(Proplist) when is_list(Proplist) ->
+    case pget(local_path, Proplist, undefined) of
+        undefined -> {error, "local definition file path is not configured: local_path is not set"};
+        Path      ->
+            rabbit_log:debug("Asked to import definitions from a local file or directory at '~s'", [Path]),
+            case file:read_file_info(Path) of
+                {ok, FileInfo} ->
+                    %% same check is used by Cuttlefish validation, this is to be extra defensive
+                    IsReadable = (element(4, FileInfo) == read) or (element(4, FileInfo) == read_write),
+                    case IsReadable of
+                        true ->
+                            load_from_single_file(Path);
+                        false ->
+                            Msg = rabbit_misc:format("local definition file '~s' does not exist or cannot be read by the node", [Path]),
+                            {error, Msg}
+                    end;
+                _ ->
+                    Msg = rabbit_misc:format("local definition file '~s' does not exist or cannot be read by the node", [Path]),
+                    {error, {could_not_read_defs, Msg}}
+            end
+    end.
+
+load(IsDir, Path) ->
+    load_from_local_path(IsDir, Path).
+
+location() ->
+    case location_from_classic_option() of
+        undefined -> location_from_modern_option();
+        Value     -> Value
+    end.
+
+load_from_local_path(true, Dir) ->
+    rabbit_log:info("Applying definitions from directory ~s", [Dir]),
+    load_from_files(file:list_dir(Dir), Dir);
+load_from_local_path(false, File) ->
+    load_from_single_file(File).
+
+%%
+%% Implementation
+%%
+
+-spec is_enabled_via_classic_option() -> boolean().
+is_enabled_via_classic_option() ->
+    %% Classic way of defining a local filesystem definition source
+    case application:get_env(rabbit, load_definitions) of
+        undefined   -> false;
+        {ok, none}  -> false;
+        {ok, _Path} -> true
+    end.
+
+-spec is_enabled_via_modern_option() -> boolean().
+is_enabled_via_modern_option() ->
+    %% Modern way of defining a local filesystem definition source
+    case application:get_env(rabbit, definitions) of
+        undefined   -> false;
+        {ok, none}  -> false;
+        {ok, []}    -> false;
+        {ok, Proplist} ->
+            case pget(import_backend, Proplist, undefined) of
+                undefined -> false;
+                ?MODULE   -> true;
+                _         -> false
+            end
+    end.
+
+location_from_classic_option() ->
+    case application:get_env(rabbit, load_definitions) of
+        undefined  -> undefined;
+        {ok, none} -> undefined;
+        {ok, Path} -> Path
+    end.
+
+location_from_modern_option() ->
+    case application:get_env(rabbit, definitions) of
+        undefined  -> undefined;
+        {ok, none} -> undefined;
+        {ok, Proplist} ->
+            pget(local_path, Proplist)
+    end.
+
+
+load_from_files({ok, Filenames0}, Dir) ->
+    Filenames1 = lists:sort(Filenames0),
+    Filenames2 = [filename:join(Dir, F) || F <- Filenames1],
+    load_from_multiple_files(Filenames2);
+load_from_files({error, E}, Dir) ->
+    rabbit_log:error("Could not read definitions from directory ~s, Error: ~p", [Dir, E]),
+    {error, {could_not_read_defs, E}}.
+
+load_from_multiple_files([]) ->
+    ok;
+load_from_multiple_files([File|Rest]) ->
+    case load_from_single_file(File) of
+        ok         -> load_from_multiple_files(Rest);
+        {error, E} -> {error, {failed_to_import_definitions, File, E}}
+    end.
+
+load_from_single_file(Path) ->
+    rabbit_log:debug("Will try to load definitions from a local file or directory at '~s'", [Path]),
+    case file:read_file(Path) of
+        {ok, Body} ->
+            rabbit_log:info("Applying definitions from file at '~s'", [Path]),
+            import_raw(Body);
+        {error, E} ->
+            rabbit_log:error("Could not read definitions from file at '~s', error: ~p", [Path, E]),
+            {error, {could_not_read_defs, {Path, E}}}
+    end.
diff --git a/deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets b/deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets
index f54212c2e5..93fd8d855b 100644
--- a/deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets
+++ b/deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets
@@ -666,11 +666,23 @@ credential_validator.regexp = ^abc\\d+",
   %% Definitions
   %%
 
+ %% classic configuration key, implies a local filesystem path
  {definition_files, "load_definitions = test/definition_import_SUITE_data/case1.json",
   [{rabbit,
        [{load_definitions, "test/definition_import_SUITE_data/case1.json"}]}],
   []},
 
+ %% modern configuration key, local filesystem source
+ {definition_files, "definitions.import_backend = local_filesystem
+                     definitions.local.path = test/definition_import_SUITE_data/case1.json",
+  [{rabbit, [
+    {definitions, [
+      {import_backend, rabbit_definitions_import_local_filesystem},
+      {local_path, "test/definition_import_SUITE_data/case1.json"}
+    ]}
+  ]}],
+  []},
+
   %%
   %% Raft
   %%
diff --git a/deps/rabbit/test/definition_import_SUITE.erl b/deps/rabbit/test/definition_import_SUITE.erl
index d424d86dc4..f4bf62fcdb 100644
--- a/deps/rabbit/test/definition_import_SUITE.erl
+++ b/deps/rabbit/test/definition_import_SUITE.erl
@@ -15,7 +15,10 @@
 
 all() ->
     [
-     {group, boot_time_import},
+     %% uses rabbit.load_definitions
+     {group, boot_time_import_using_classic_source},
+     %% uses rabbit.definitions with import_backend set to local_filesystem
+     {group, boot_time_import_using_modern_local_filesystem_source},
      {group, roundtrip},
      {group, import_on_a_running_node}
     ].
@@ -43,8 +46,13 @@ groups() ->
                                import_case14,
                                import_case15
                               ]},
-        {boot_time_import, [], [
-            import_on_a_booting_node
+        
+        {boot_time_import_using_classic_source, [], [
+            import_on_a_booting_node_using_classic_local_source
+        ]},
+
+        {boot_time_import_using_modern_local_filesystem_source, [], [
+            import_on_a_booting_node_using_modern_local_filesystem_source
         ]},
 
         {roundtrip, [], [
@@ -64,7 +72,7 @@ init_per_suite(Config) ->
 end_per_suite(Config) ->
     Config.
 
-init_per_group(boot_time_import = Group, Config) ->
+init_per_group(boot_time_import_using_classic_source = Group, Config) ->
     CasePath = filename:join(?config(data_dir, Config), "case5.json"),
     Config1 = rabbit_ct_helpers:set_config(Config, [
         {rmq_nodename_suffix, Group},
@@ -75,6 +83,21 @@ init_per_group(boot_time_import = Group, Config) ->
           {load_definitions, CasePath}
       ]}),
     rabbit_ct_helpers:run_setup_steps(Config2, rabbit_ct_broker_helpers:setup_steps());
+%% same as the classic source semantically but uses a different configuration structure
+init_per_group(boot_time_import_using_modern_local_filesystem_source = Group, Config) ->
+    CasePath = filename:join(?config(data_dir, Config), "case5.json"),
+    Config1 = rabbit_ct_helpers:set_config(Config, [
+        {rmq_nodename_suffix, Group},
+        {rmq_nodes_count, 1}
+      ]),
+    Config2 = rabbit_ct_helpers:merge_app_env(Config1,
+      {rabbit, [
+          {definitions, [
+              {import_backend, rabbit_definitions_import_local_filesystem},
+              {local_path,     CasePath}
+          ]}
+      ]}),
+    rabbit_ct_helpers:run_setup_steps(Config2, rabbit_ct_broker_helpers:setup_steps());
 init_per_group(Group, Config) ->
     Config1 = rabbit_ct_helpers:set_config(Config, [
                                                     {rmq_nodename_suffix, Group}
@@ -158,7 +181,16 @@ export_import_round_trip_case2(Config) ->
     Defs = export(Config),
     import_parsed(Config, Defs).
 
-import_on_a_booting_node(Config) ->
+import_on_a_booting_node_using_classic_local_source(Config) ->
+    %% see case5.json
+    VHost = <<"vhost2">>,
+    %% verify that vhost2 eventually starts
+    case rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_vhost, await_running_on_all_nodes, [VHost, 3000]) of
+        ok -> ok;
+        {error, timeout} -> ct:fail("virtual host ~p was not imported on boot", [VHost])
+    end.
+
+import_on_a_booting_node_using_modern_local_filesystem_source(Config) ->
     %% see case5.json
     VHost = <<"vhost2">>,
     %% verify that vhost2 eventually starts
diff --git a/release-notes/3.9.4.md b/release-notes/3.9.4.md
index f6b33dae78..6b445fea39 100644
--- a/release-notes/3.9.4.md
+++ b/release-notes/3.9.4.md
@@ -21,7 +21,36 @@ consistent release schedule.
 
    GitHub issue: [#3299](https://github.com/rabbitmq/rabbitmq-server/pull/3299)
 
-
+ * Definitions now can be imported from different sources, including those provided by plugins. Local filesystem imports are still supported.
+
+   The following `rabbitmq.conf` example uses a local file as the source:
+   
+   ``` ini
+   # equivalent to the classic load_definitions configuration key   
+   definitions.import_backend = local_filesystem
+
+   definitions.local.path = /path/to/definitions.json
+   ```
+   
+   This `rabbitmq.conf` example uses a local directory with definition files:
+   
+   ``` ini
+   # equivalent to the classic load_definitions configuration key   
+   definitions.import_backend = local_filesystem
+
+   definitions.local.path = /path/to/rabbitmq/definitions.d
+   ```
+
+   In this example config file, definitions are loaded from a URL accessible over HTTPS:
+   
+   ``` ini
+   # downloads definitions over HTTPS
+   definitions.import_backend = https
+
+   definitions.https.url = https://rabbitmq.eng.megacorp.local/env-1/definitions.json
+   ```
+
+   GitHub issue: [#3249](https://github.com/rabbitmq/rabbitmq-server/issues/3249)
 
 ## Dependency Upgrades