Merge pull request #1056 from rabbitmq/rabbitmq-mqtt-73

Improve global runtime parameters support

4 files changed, 278 insertions, 21 deletions

diff --git a/docs/rabbitmqctl.1.xml b/docs/rabbitmqctl.1.xml
index 217d2d93ca..0465db02fd 100644
--- a/docs/rabbitmqctl.1.xml
+++ b/docs/rabbitmqctl.1.xml
@@ -1035,11 +1035,16 @@
       <para>
         Certain features of RabbitMQ (such as the federation plugin)
         are controlled by dynamic,
-        cluster-wide <emphasis>parameters</emphasis>. Each parameter
-        consists of a component name, a name and a value, and is
-        associated with a virtual host. The component name and name are
-        strings, and the value is an Erlang term. Parameters can be
-        set, cleared and listed. In general you should refer to the
+        cluster-wide <emphasis>parameters</emphasis>.
+        There are 2 kinds of parameters: parameters scoped to
+        a virtual host and global parameters.
+        Each vhost-scoped parameter
+        consists of a component name, a name and a value.
+        The component name and name are
+        strings, and the value is an Erlang term.
+        A global parameter consists of a name and value. The name
+        is a string and the value is an Erlang term.
+        Parameters can be set, cleared and listed. In general you should refer to the
         documentation for the feature in question to see how to set
         parameters.
       </para>
@@ -1121,6 +1126,72 @@
             </para>
           </listitem>
         </varlistentry>
+        <varlistentry>
+          <term><cmdsynopsis><command>set_global_parameter</command> <arg choice="req"><replaceable>name</replaceable></arg> <arg choice="req"><replaceable>value</replaceable></arg></cmdsynopsis></term>
+          <listitem>
+            <para>
+              Sets a global runtime parameter. This is similar to <command>set_parameter</command>
+              but the key-value pair isn't tied to a virtual host.
+            </para>
+            <variablelist>
+              <varlistentry>
+                <term>name</term>
+                <listitem><para>
+                  The name of the global runtime parameter being set.
+                </para></listitem>
+              </varlistentry>
+              <varlistentry>
+                <term>value</term>
+                <listitem><para>
+                  The value for the global runtime parameter, as a
+                  JSON term. In most shells you are very likely to
+                  need to quote this.
+                </para></listitem>
+              </varlistentry>
+            </variablelist>
+            <para role="example-prefix">For example:</para>
+            <screen role="example">rabbitmqctl set_global_parameter mqtt_default_vhosts '{"O=client,CN=guest":"/"}'</screen>
+            <para role="example">
+              This command sets the global runtime parameter <command>mqtt_default_vhosts</command> to the JSON term <command>{"O=client,CN=guest":"/"}</command>.
+            </para>
+          </listitem>
+        </varlistentry>
+        <varlistentry>
+          <term><cmdsynopsis><command>clear_global_parameter</command> <arg choice="req"><replaceable>name</replaceable></arg></cmdsynopsis></term>
+          <listitem>
+            <para>
+              Clears a global runtime parameter. This is similar to <command>clear_global_parameter</command>
+              but the key-value pair isn't tied to a virtual host.
+            </para>
+            <variablelist>
+              <varlistentry>
+                <term>name</term>
+                <listitem><para>
+                  The name of the global runtime parameter being cleared.
+                </para></listitem>
+              </varlistentry>
+            </variablelist>
+            <para role="example-prefix">For example:</para>
+            <screen role="example">rabbitmqctl clear_global_parameter mqtt_default_vhosts</screen>
+            <para role="example">
+              This command clears the global runtime parameter <command>mqtt_default_vhosts</command>.
+            </para>
+          </listitem>
+        </varlistentry>
+        <varlistentry>
+          <term><cmdsynopsis><command>list_global_parameters</command></cmdsynopsis></term>
+          <listitem>
+            <para>
+              Lists all global runtime parameters. This is similar to <command>list_parameters</command>
+              but the global runtime parameters are not tied to any virtual host.
+            </para>
+            <para role="example-prefix">For example:</para>
+            <screen role="example">rabbitmqctl list_global_parameters</screen>
+            <para role="example">
+              This command lists all global parameters.
+            </para>
+          </listitem>
+        </varlistentry>
       </variablelist>
     </refsect2>
 
diff --git a/src/rabbit_control_main.erl b/src/rabbit_control_main.erl
index e306afc198..d6bc81945e 100644
--- a/src/rabbit_control_main.erl
+++ b/src/rabbit_control_main.erl
@@ -70,6 +70,10 @@
          {clear_parameter, [?VHOST_DEF]},
          {list_parameters, [?VHOST_DEF]},
 
+         set_global_parameter,
+         clear_global_parameter,
+         list_global_parameters,
+
          {set_policy, [?VHOST_DEF, ?PRIORITY_DEF, ?APPLY_TO_DEF]},
          {clear_policy, [?VHOST_DEF]},
          {list_policies, [?VHOST_DEF]},
@@ -121,7 +125,7 @@
 -define(COMMANDS_WITH_TIMEOUT,
         [list_user_permissions, list_policies, list_queues, list_exchanges,
          list_bindings, list_connections, list_channels, list_consumers,
-         list_vhosts, list_parameters,
+         list_vhosts, list_parameters, list_global_parameters,
          purge_queue,
          {node_health_check, 70000}]).
 
@@ -527,6 +531,20 @@ action(clear_parameter, Node, [Component, Key], Opts, Inform) ->
                                                       list_to_binary(Component),
                                                       list_to_binary(Key)]);
 
+action(set_global_parameter, Node, [Key, Value], _Opts, Inform) ->
+    Inform("Setting global runtime parameter ~p to ~p", [Key, Value]),
+    rpc_call(
+        Node, rabbit_runtime_parameters, parse_set_global,
+        [rabbit_data_coercion:to_atom(Key), rabbit_data_coercion:to_binary(Value)]
+    );
+
+action(clear_global_parameter, Node, [Key], _Opts, Inform) ->
+    Inform("Clearing global runtime parameter ~p", [Key]),
+    rpc_call(
+        Node, rabbit_runtime_parameters, clear_global,
+        [rabbit_data_coercion:to_atom(Key)]
+    );
+
 action(set_policy, Node, [Key, Pattern, Defn], Opts, Inform) ->
     Msg = "Setting policy ~p for pattern ~p to ~p with priority ~p",
     VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
@@ -623,6 +641,11 @@ action(list_parameters, Node, [], Opts, Inform, Timeout) ->
     call(Node, {rabbit_runtime_parameters, list_formatted, [VHostArg]},
          rabbit_runtime_parameters:info_keys(), Timeout);
 
+action(list_global_parameters, Node, [], _Opts, Inform, Timeout) ->
+    Inform("Listing global runtime parameters", []),
+    call(Node, {rabbit_runtime_parameters, list_global_formatted, []},
+         rabbit_runtime_parameters:global_info_keys(), Timeout);
+
 action(list_policies, Node, [], Opts, Inform, Timeout) ->
     VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
     Inform("Listing policies", []),
diff --git a/src/rabbit_runtime_parameters.erl b/src/rabbit_runtime_parameters.erl
index 97f78da8ba..292aa26d32 100644
--- a/src/rabbit_runtime_parameters.erl
+++ b/src/rabbit_runtime_parameters.erl
@@ -40,6 +40,8 @@
 %% Parameters are stored in Mnesia and can be global. Their changes
 %% are broadcasted over rabbit_event.
 %%
+%% Global parameters keys are atoms and values are JSON documents.
+%%
 %% See also:
 %%
 %%  * rabbit_policies
@@ -53,7 +55,9 @@
          list_component/1, list/2, list_formatted/1, list_formatted/3,
          lookup/3, value/3, value/4, info_keys/0, clear_component/1]).
 
--export([set_global/2, value_global/1, value_global/2]).
+-export([parse_set_global/2, set_global/2, value_global/1, value_global/2,
+         list_global/0, list_global_formatted/0, list_global_formatted/2,
+         lookup_global/1, global_info_keys/0, clear_global/1]).
 
 %%----------------------------------------------------------------------------
 
@@ -108,10 +112,17 @@ set(_, <<"policy">>, _, _, _) ->
 set(VHost, Component, Name, Term, User) ->
     set_any(VHost, Component, Name, Term, User).
 
-set_global(Name, Term) ->
-    mnesia_update(Name, Term),
-    event_notify(parameter_set, none, global, [{name,  Name},
-                                               {value, Term}]),
+parse_set_global(Name, String) ->
+    case rabbit_misc:json_decode(String) of
+        {ok, JSON} -> set_global(Name, rabbit_misc:json_to_term(JSON));
+        error      -> {error_string, "JSON decoding error"}
+    end.
+
+set_global(Name, Term)  ->
+    NameAsAtom = rabbit_data_coercion:to_atom(Name),
+    mnesia_update(NameAsAtom, Term),
+    event_notify(parameter_set, none, global, [{name,  NameAsAtom},
+        {value, Term}]),
     ok.
 
 format_error(L) ->
@@ -167,6 +178,25 @@ clear(_, <<"policy">> , _) ->
 clear(VHost, Component, Name) ->
     clear_any(VHost, Component, Name).
 
+clear_global(Key) ->
+    Notify = fun() ->
+                    event_notify(parameter_set, none, global, [{name,  Key}]),
+                    ok
+             end,
+    case value_global(Key) of
+        not_found ->
+            {error_string, "Parameter does not exist"};
+        _         ->
+            F = fun () ->
+                ok = mnesia:delete(?TABLE, Key, write)
+                end,
+            ok = rabbit_misc:execute_mnesia_transaction(F),
+            case mnesia:is_transaction() of
+                true  -> Notify;
+                false -> Notify()
+            end
+    end.
+
 clear_component(Component) ->
     case rabbit_runtime_parameters:list_component(Component) of
         [] ->
@@ -234,13 +264,30 @@ list(VHost, Component) ->
                        Comp =/= <<"policy">> orelse Component =:= <<"policy">>]
       end).
 
+list_global() ->
+    %% list only atom keys
+    mnesia:async_dirty(
+        fun () ->
+            Match = #runtime_parameters{key = '_', _ = '_'},
+            [p(P) || P <- mnesia:match_object(?TABLE, Match, read),
+                is_atom(P#runtime_parameters.key)]
+        end).
+
 list_formatted(VHost) ->
     [pset(value, format(pget(value, P)), P) || P <- list(VHost)].
 
 list_formatted(VHost, Ref, AggregatorPid) ->
     rabbit_control_misc:emitting_map(
-      AggregatorPid, Ref,
-      fun(P) -> pset(value, format(pget(value, P)), P) end, list(VHost)).
+        AggregatorPid, Ref,
+        fun(P) -> pset(value, format(pget(value, P)), P) end, list(VHost)).
+
+list_global_formatted() ->
+    [pset(value, format(pget(value, P)), P) || P <- list_global()].
+
+list_global_formatted(Ref, AggregatorPid) ->
+    rabbit_control_misc:emitting_map(
+        AggregatorPid, Ref,
+        fun(P) -> pset(value, format(pget(value, P)), P) end, list_global()).
 
 lookup(VHost, Component, Name) ->
     case lookup0({VHost, Component, Name}, rabbit_misc:const(not_found)) of
@@ -248,11 +295,20 @@ lookup(VHost, Component, Name) ->
         Params    -> p(Params)
     end.
 
+lookup_global(Name)  ->
+    case lookup0(Name, rabbit_misc:const(not_found)) of
+        not_found -> not_found;
+        Params    -> p(Params)
+    end.
+
 value(VHost, Comp, Name)      -> value0({VHost, Comp, Name}).
 value(VHost, Comp, Name, Def) -> value0({VHost, Comp, Name}, Def).
 
-value_global(Key)          -> value0(Key).
-value_global(Key, Default) -> value0(Key, Default).
+value_global(Key) ->
+    value0(Key).
+
+value_global(Key, Default) ->
+    value0(Key, Default).
 
 value0(Key) ->
     case lookup0(Key, rabbit_misc:const(not_found)) of
@@ -289,10 +345,16 @@ p(#runtime_parameters{key = {VHost, Component, Name}, value = Value}) ->
     [{vhost,     VHost},
      {component, Component},
      {name,      Name},
+     {value,     Value}];
+
+p(#runtime_parameters{key = Key, value = Value}) when is_atom(Key) ->
+    [{name,      Key},
      {value,     Value}].
 
 info_keys() -> [component, name, value].
 
+global_info_keys() -> [name, value].
+
 %%---------------------------------------------------------------------------
 
 lookup_component(Component) ->
diff --git a/test/rabbitmqctl_integration_SUITE.erl b/test/rabbitmqctl_integration_SUITE.erl
index 9305781bda..ef85472f48 100644
--- a/test/rabbitmqctl_integration_SUITE.erl
+++ b/test/rabbitmqctl_integration_SUITE.erl
@@ -31,17 +31,24 @@
 -export([list_queues_local/1
         ,list_queues_offline/1
         ,list_queues_online/1
+        ,manage_global_parameters/1
         ]).
 
 all() ->
-    [{group, list_queues}].
+    [
+        {group, list_queues},
+        {group, global_parameters}
+    ].
 
 groups() ->
-    [{list_queues, [],
-      [list_queues_local
-      ,list_queues_online
-      ,list_queues_offline
-      ]}].
+    [
+        {list_queues, [],
+            [list_queues_local
+            ,list_queues_online
+            ,list_queues_offline
+            ]},
+        {global_parameters, [], [manage_global_parameters]}
+    ].
 
 init_per_suite(Config) ->
     rabbit_ct_helpers:log_environment(),
@@ -56,6 +63,13 @@ init_per_group(list_queues, Config0) ->
     Config1 = declare_some_queues(Config),
     rabbit_ct_broker_helpers:stop_node(Config1, NumNodes - 1),
     Config1;
+init_per_group(global_parameters,Config) ->
+    Config1 = rabbit_ct_helpers:set_config(Config, [
+        {rmq_nodename_suffix, ?MODULE}
+    ]),
+    rabbit_ct_helpers:run_setup_steps(Config1,
+        rabbit_ct_broker_helpers:setup_steps() ++
+        rabbit_ct_client_helpers:setup_steps());
 init_per_group(_, Config) ->
     Config.
 
@@ -92,6 +106,10 @@ end_per_group(list_queues, Config0) ->
     rabbit_ct_helpers:run_steps(Config1,
                                 rabbit_ct_client_helpers:teardown_steps() ++
                                     rabbit_ct_broker_helpers:teardown_steps());
+end_per_group(global_parameters, Config) ->
+    rabbit_ct_helpers:run_teardown_steps(Config,
+        rabbit_ct_client_helpers:teardown_steps() ++
+        rabbit_ct_broker_helpers:teardown_steps());
 end_per_group(_, Config) ->
     Config.
 
@@ -126,6 +144,75 @@ list_queues_offline(Config) ->
     assert_ctl_queues(Config, 1, ["--offline"], OfflineQueues),
     ok.
 
+manage_global_parameters(Config) ->
+    0 = length(global_parameters(Config)),
+    Parameter1Key = global_param1,
+    GlobalParameter1ValueAsString = "{\"a\":\"b\", \"c\":\"d\"}",
+    ok = control_action(Config, set_global_parameter,
+        [atom_to_list(Parameter1Key),
+            GlobalParameter1ValueAsString
+    ]),
+
+    1 = length(global_parameters(Config)),
+
+    GlobalParameter1Value = rabbit_ct_broker_helpers:rpc(
+        Config, 0,
+        rabbit_runtime_parameters, value_global,
+        [Parameter1Key]
+    ),
+
+    [{<<"a">>,<<"b">>}, {<<"c">>,<<"d">>}] = GlobalParameter1Value,
+
+    Parameter2Key = global_param2,
+    GlobalParameter2ValueAsString = "{\"e\":\"f\", \"g\":\"h\"}",
+    ok = control_action(Config, set_global_parameter,
+        [atom_to_list(Parameter2Key),
+            GlobalParameter2ValueAsString
+        ]),
+
+    2 = length(global_parameters(Config)),
+
+    GlobalParameter2Value = rabbit_ct_broker_helpers:rpc(
+        Config, 0,
+        rabbit_runtime_parameters, value_global,
+        [Parameter2Key]
+    ),
+
+    [{<<"e">>,<<"f">>}, {<<"g">>,<<"h">>}] = GlobalParameter2Value,
+
+
+    GlobalParameter1Value2AsString = "{\"a\":\"z\", \"c\":\"d\"}",
+    ok = control_action(Config, set_global_parameter,
+        [atom_to_list(Parameter1Key),
+            GlobalParameter1Value2AsString
+        ]),
+
+    2 = length(global_parameters(Config)),
+
+    GlobalParameter1Value2 = rabbit_ct_broker_helpers:rpc(
+        Config, 0,
+        rabbit_runtime_parameters, value_global,
+        [Parameter1Key]
+    ),
+
+    [{<<"a">>,<<"z">>}, {<<"c">>,<<"d">>}] = GlobalParameter1Value2,
+
+    ok = control_action(Config, clear_global_parameter,
+        [atom_to_list(Parameter1Key)]
+    ),
+
+    1 = length(global_parameters(Config)),
+
+    not_found = rabbit_ct_broker_helpers:rpc(
+        Config, 0,
+        rabbit_runtime_parameters, value_global,
+        [Parameter1Key]
+    ),
+
+    ok = control_action(Config, list_global_parameters, []),
+
+    ok.
+
 %%----------------------------------------------------------------------------
 %% Helpers
 %%----------------------------------------------------------------------------
@@ -144,3 +231,17 @@ assert_ctl_queues(Config, Node, Args, Expected0) ->
 
 run_list_queues(Config, Node, Args) ->
     rabbit_ct_broker_helpers:rabbitmqctl_list(Config, Node, ["list_queues"] ++ Args ++ ["name"]).
+
+control_action(Config, Command, Args) ->
+    Node = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    rabbit_control_main:action(
+        Command, Node, Args, [],
+        fun (Format, Args1) ->
+            io:format(Format ++ " ...~n", Args1)
+        end).
+
+global_parameters(Config) ->
+    rabbit_ct_broker_helpers:rpc(
+        Config, 0,
+        rabbit_runtime_parameters, list_global, []
+    ).
\ No newline at end of file