Improve global runtime parameters support

References rabbitmq/rabbitmq-mqtt#73

3 files changed, 184 insertions, 10 deletions

diff --git a/src/rabbit_control_main.erl b/src/rabbit_control_main.erl
index 8c245892b7..c4106aa88a 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]},
@@ -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, set_global,
+        [evaluate_input_as_term(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,
+        [evaluate_input_as_term(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", []),
@@ -978,3 +1001,9 @@ alarms_by_node(Name) ->
             {_, As} = lists:keyfind(alarms, 1, Status),
             {Name, As}
     end.
+
+evaluate_input_as_term(Input) ->
+    {ok,Tokens,_EndLine} = erl_scan:string(Input ++ "."),
+    {ok,AbsForm} = erl_parse:parse_exprs(Tokens),
+    {value,TermValue,_Bs} = erl_eval:exprs(AbsForm, erl_eval:new_bindings()),
+    TermValue.
\ No newline at end of file
diff --git a/src/rabbit_runtime_parameters.erl b/src/rabbit_runtime_parameters.erl
index 97f78da8ba..8f5cc2c9c3 100644
--- a/src/rabbit_runtime_parameters.erl
+++ b/src/rabbit_runtime_parameters.erl
@@ -53,7 +53,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([set_global/2, value_global/1, value_global/2,
+         list_global/0, list_global_formatted/0,
+         global_info_keys/0, clear_global/1]).
 
 %%----------------------------------------------------------------------------
 
@@ -108,10 +110,17 @@ set(_, <<"policy">>, _, _, _) ->
 set(VHost, Component, Name, Term, User) ->
     set_any(VHost, Component, Name, Term, User).
 
+set_global(Name, Term) when is_atom(Name) ->
+    %% for cluster_name
+    set_global0(Name, Term);
+
 set_global(Name, Term) ->
+    set_global0({global, Name}, Term).
+
+set_global0(Name, Term) ->
     mnesia_update(Name, Term),
     event_notify(parameter_set, none, global, [{name,  Name},
-                                               {value, Term}]),
+        {value, Term}]),
     ok.
 
 format_error(L) ->
@@ -167,6 +176,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, {global, 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,9 +262,20 @@ list(VHost, Component) ->
                        Comp =/= <<"policy">> orelse Component =:= <<"policy">>]
       end).
 
+list_global() ->
+    mnesia:async_dirty(
+        fun () ->
+            Match = #runtime_parameters{key = {global, '_'},
+                _   = '_'},
+            [p(P) || P <- mnesia:match_object(?TABLE, Match, read)]
+        end).
+
 list_formatted(VHost) ->
     [pset(value, format(pget(value, P)), P) || P <- list(VHost)].
 
+list_global_formatted() ->
+    [pset(value, format(pget(value, P)), P) || P <- list_global()].
+
 list_formatted(VHost, Ref, AggregatorPid) ->
     rabbit_control_misc:emitting_map(
       AggregatorPid, Ref,
@@ -251,8 +290,17 @@ lookup(VHost, Component, Name) ->
 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) when is_atom(Key) ->
+    %% for cluster_name
+    value0(Key);
+value_global(Key) ->
+    value0({global, Key}).
+
+value_global(Key, Default) when is_atom(Key) ->
+    %% for cluster_name
+    value0(Key, Default);
+value_global(Key, Default) ->
+    value0({global, Key}, Default).
 
 value0(Key) ->
     case lookup0(Key, rabbit_misc:const(not_found)) of
@@ -289,10 +337,16 @@ p(#runtime_parameters{key = {VHost, Component, Name}, value = Value}) ->
     [{vhost,     VHost},
      {component, Component},
      {name,      Name},
+     {value,     Value}];
+
+p(#runtime_parameters{key = {global, Name}, value = Value}) ->
+    [{name,      Name},
      {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..ba72578775 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,65 @@ list_queues_offline(Config) ->
     assert_ctl_queues(Config, 1, ["--offline"], OfflineQueues),
     ok.
 
+manage_global_parameters(Config) ->
+    0 = length(global_parameters(Config)),
+    GlobalParameterValue1 = <<"vhost1">>,
+    control_action(Config, set_global_parameter,
+        ["{mqtt_cert_user_vhost, <<\"O=client,CN=dummy1\">>}",
+            GlobalParameterValue1
+    ]),
+
+    1 = length(global_parameters(Config)),
+
+    GlobalParameterValue1 = rabbit_ct_broker_helpers:rpc(
+        Config, 0,
+        rabbit_runtime_parameters, value_global,
+        [
+            {mqtt_cert_user_vhost, <<"O=client,CN=dummy1">>}
+        ]
+    ),
+
+    GlobalParameterValue2 = <<"vhost2">>,
+    control_action(Config, set_global_parameter,
+        ["{mqtt_cert_user_vhost, <<\"O=client,CN=dummy2\">>}",
+            GlobalParameterValue2
+        ]),
+
+    2 = length(global_parameters(Config)),
+
+    GlobalParameterValue2 = rabbit_ct_broker_helpers:rpc(
+        Config, 0,
+        rabbit_runtime_parameters, value_global,
+        [
+            {mqtt_cert_user_vhost, <<"O=client,CN=dummy2">>}
+        ]
+    ),
+
+    NewGlobalParameterValue = <<"vhost3">>,
+
+    control_action(Config, set_global_parameter,
+        ["{mqtt_cert_user_vhost, <<\"O=client,CN=dummy1\">>}",
+            NewGlobalParameterValue
+        ]),
+
+    2 = length(global_parameters(Config)),
+
+    NewGlobalParameterValue = rabbit_ct_broker_helpers:rpc(
+        Config, 0,
+        rabbit_runtime_parameters, value_global,
+        [
+            {mqtt_cert_user_vhost, <<"O=client,CN=dummy1">>}
+        ]
+    ),
+
+    control_action(Config, clear_global_parameter,
+        ["{mqtt_cert_user_vhost, <<\"O=client,CN=dummy1\">>}"]
+    ),
+
+    1 = length(global_parameters(Config)),
+
+    ok.
+
 %%----------------------------------------------------------------------------
 %% Helpers
 %%----------------------------------------------------------------------------
@@ -144,3 +221,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