Add cluster tag config that is exposed via HTTP /api/overview and CTL cluster_status

This commit is contained in:
Simon Unge 2024-10-23 20:57:08 +00:00
parent 8b554474a6
commit f5ef64ad06
4 changed files with 70 additions and 5 deletions

View File

@ -2688,6 +2688,20 @@ fun(Conf) ->
end
end}.
{mapping, "cluster_tags.$tag", "rabbit.cluster_tags", [
{datatype, [binary]}
]}.
{translation, "rabbit.cluster_tags",
fun(Conf) ->
case cuttlefish:conf_get("cluster_tags", Conf, undefined) of
none -> [];
_ ->
Settings = cuttlefish_variable:filter_by_prefix("cluster_tags", Conf),
[ {list_to_binary(K), V} || {[_, K], V} <- Settings]
end
end}.
% ===============================
% Validators
% ===============================

View File

@ -33,7 +33,7 @@ defmodule RabbitMQ.CLI.Ctl.Commands.ClusterStatusCommand do
use RabbitMQ.CLI.Core.RequiresRabbitAppRunning
def run([], %{node: node_name, timeout: timeout} = opts) do
status =
status0 =
case :rabbit_misc.rpc_call(node_name, :rabbit_db_cluster, :cli_cluster_status, []) do
{:badrpc, {:EXIT, {:undef, _}}} ->
:rabbit_misc.rpc_call(node_name, :rabbit_mnesia, :status, [])
@ -45,11 +45,13 @@ defmodule RabbitMQ.CLI.Ctl.Commands.ClusterStatusCommand do
status
end
case status do
case status0 do
{:badrpc, _} = err ->
err
status ->
status0 ->
tags = cluster_tags(node_name, timeout)
status = status0 ++ [{:cluster_tags, tags}]
case :rabbit_misc.rpc_call(node_name, :rabbit_nodes, :list_running, []) do
{:badrpc, _} = err ->
err
@ -122,7 +124,6 @@ defmodule RabbitMQ.CLI.Ctl.Commands.ClusterStatusCommand do
def output(result, %{node: node_name}) when is_list(result) do
m = result_map(result)
total_cores = Enum.reduce(m[:cpu_cores], 0, fn {_, val}, acc -> acc + val end)
cluster_name_section = [
@ -131,6 +132,15 @@ defmodule RabbitMQ.CLI.Ctl.Commands.ClusterStatusCommand do
"Total CPU cores available cluster-wide: #{total_cores}"
]
cluster_tag_section =
[
"\n#{bright("Cluster Tags")}\n"
] ++
case m[:cluster_tags] do
[] -> ["(none)"]
tags -> cluster_tag_lines(tags)
end
disk_nodes_section =
[
"\n#{bright("Disk Nodes")}\n"
@ -210,6 +220,7 @@ defmodule RabbitMQ.CLI.Ctl.Commands.ClusterStatusCommand do
lines =
cluster_name_section ++
cluster_tag_section ++
disk_nodes_section ++
ram_nodes_section ++
running_nodes_section ++
@ -260,6 +271,7 @@ defmodule RabbitMQ.CLI.Ctl.Commands.ClusterStatusCommand do
# {rabbit@warp10,[{resource_limit,memory,rabbit@warp10}]}]}]
%{
cluster_name: Keyword.get(result, :cluster_name),
cluster_tags: result |> Keyword.get(:cluster_tags, []),
disk_nodes: result |> Keyword.get(:nodes, []) |> Keyword.get(:disc, []),
ram_nodes: result |> Keyword.get(:nodes, []) |> Keyword.get(:ram, []),
running_nodes: result |> Keyword.get(:running_nodes, []) |> Enum.map(&to_string/1),
@ -383,6 +395,18 @@ defmodule RabbitMQ.CLI.Ctl.Commands.ClusterStatusCommand do
{node, result}
end
defp cluster_tags(node, timeout) do
case :rabbit_misc.rpc_call(
node,
:application,
:get_env,
[:rabbit, :cluster_tags],
timeout) do
{:ok, tags} -> tags
_ -> []
end
end
defp node_lines(nodes) do
Enum.map(nodes, &to_string/1) |> Enum.sort()
end
@ -413,4 +437,10 @@ defmodule RabbitMQ.CLI.Ctl.Commands.ClusterStatusCommand do
defp maintenance_lines(mapping) do
Enum.map(mapping, fn {node, status} -> "Node: #{node}, status: #{status}" end)
end
defp cluster_tag_lines(mapping) do
Enum.map(mapping, fn {key, value} ->
"#{key}: #{value}"
end)
end
end

View File

@ -47,6 +47,7 @@ to_json(ReqData, Context = #context{user = User = #user{tags = Tags}}) ->
{product_name, list_to_binary(rabbit:product_name())},
{rabbitmq_version, list_to_binary(rabbit:base_product_version())},
{cluster_name, rabbit_nodes:cluster_name()},
{cluster_tags, cluster_tags()},
{erlang_version, erlang_version()},
{erlang_full_version, erlang_full_version()},
{release_series_support_status, rabbit_release_series:readable_support_status()},
@ -182,3 +183,6 @@ transform_retention_intervals([{MaxAgeInSeconds, _}|Rest], Acc) ->
0
end,
transform_retention_intervals(Rest, [AccVal|Acc]).
cluster_tags() ->
application:get_env(rabbit, cluster_tags, []).

View File

@ -204,7 +204,8 @@ all_tests() -> [
connections_amqp,
amqp_sessions,
amqpl_sessions,
enable_plugin_amqp
enable_plugin_amqp,
cluster_tags_test
].
%% -------------------------------------------------------------------
@ -285,6 +286,11 @@ init_per_testcase(Testcase = disabled_qq_replica_opers_test, Config) ->
rabbit_ct_broker_helpers:rpc_all(Config,
application, set_env, [rabbitmq_management, restrictions, Restrictions]),
rabbit_ct_helpers:testcase_started(Config, Testcase);
init_per_testcase(Testcase = cluster_tags_test, Config) ->
Tags = [{<<"az">>, <<"us-east-3">>}, {<<"region">>,<<"us-east">>}, {<<"environment">>,<<"production">>}],
rabbit_ct_broker_helpers:rpc_all(Config,
application, set_env, [rabbit, cluster_tags, Tags]),
rabbit_ct_helpers:testcase_started(Config, Testcase);
init_per_testcase(queues_detailed_test, Config) ->
IsEnabled = rabbit_ct_broker_helpers:is_feature_flag_enabled(
Config, detailed_queues_endpoint),
@ -351,6 +357,9 @@ end_per_testcase0(disabled_operator_policy_test, Config) ->
end_per_testcase0(disabled_qq_replica_opers_test, Config) ->
rpc(Config, application, unset_env, [rabbitmq_management, restrictions]),
Config;
end_per_testcase0(cluster_tags_test, Config) ->
rpc(Config, application, unset_env, [rabbit, cluster_tags]),
Config;
end_per_testcase0(Testcase, Config)
when Testcase == list_deprecated_features_test;
Testcase == list_used_deprecated_features_test ->
@ -4083,6 +4092,14 @@ list_used_deprecated_features_test(Config) ->
?assertEqual(list_to_binary(Desc), maps:get(desc, Feature)),
?assertEqual(list_to_binary(DocUrl), maps:get(doc_url, Feature)).
cluster_tags_test(Config) ->
Overview = http_get(Config, "/overview"),
Tags = maps:get(cluster_tags, Overview),
ExpectedTags = #{az => <<"us-east-3">>,environment => <<"production">>,
region => <<"us-east">>},
?assertEqual(ExpectedTags, Tags),
passed.
%% -------------------------------------------------------------------
%% Helpers.
%% -------------------------------------------------------------------