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

Improve global runtime parameters support

2 files changed, 95 insertions, 10 deletions

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) ->