Skip to content

Commit e1c442c

Browse files
mkuratczykmkuratczyk
committed
rabbitmq-diagnostics message_size_stats
This command displays cluster-wide message size statistics. It's less detailed than what can be retrieved from the Prometheus endpoint, but it'll be available to all users, regardless of their monitoring setup, or lack thereof.
1 parent 18e4bf6 commit e1c442c

File tree

4 files changed

+315
-41
lines changed

4 files changed

+315
-41
lines changed

‎deps/rabbit/src/rabbit_msg_size_metrics.erl‎

Lines changed: 110 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,10 @@
1111

1212
-export([init/1,
1313
observe/2,
14-
prometheus_format/0]).
14+
prometheus_format/0,
15+
local_summary/0,
16+
cluster_summary/0,
17+
cluster_summary_for_cli/0 ]).
1518

1619
%% Integration tests.
1720
-export([raw_buckets/1,
@@ -52,6 +55,9 @@
5255
-type raw_buckets() :: [{BucketUpperBound :: non_neg_integer(),
5356
NumObservations :: non_neg_integer()}].
5457

58+
-type summary_entry() :: {{non_neg_integer(), non_neg_integer() | infinity}, {non_neg_integer(), float()}}.
59+
-type summary() :: [summary_entry()].
60+
5561
-spec init(atom()) -> ok.
5662
init(Protocol) ->
5763
Size = ?POS_MSG_SIZE_SUM,
@@ -133,6 +139,109 @@ get_labels_counters() ->
133139
[{[{protocol, Protocol}], Counters}
134140
|| {{?MODULE, Protocol}, Counters} <- persistent_term:get()].
135141

142+
get_protocols() ->
143+
[Protocol
144+
|| {{?MODULE, Protocol}, _} <- persistent_term:get()].
145+
146+
%% Aggregates data for all protocols on the local node
147+
-spec local_summary() -> summary().
148+
local_summary() ->
149+
PerProtocolBuckets = lists:map(fun(Protocol) ->
150+
raw_buckets(Protocol)
151+
end, get_protocols()),
152+
153+
%% Sum buckets for all protocols
154+
Buckets0 = [{?BUCKET_1, 0}, {?BUCKET_2, 0}, {?BUCKET_3, 0}, {?BUCKET_4, 0},
155+
{?BUCKET_5, 0}, {?BUCKET_6, 0}, {?BUCKET_7, 0}, {?BUCKET_8, 0}, {?BUCKET_9, 0}],
156+
Buckets = lists:foldl(fun sum_protocol_buckets/2,
157+
Buckets0,
158+
PerProtocolBuckets),
159+
160+
Total = lists:sum([Count || {_UpperBound, Count} <- Buckets]),
161+
162+
Ranges = lists:map(fun({UpperBound, Count}) ->
163+
Percentage = case Total of
164+
0 -> 0.0;
165+
_ -> (Count / Total) * 100
166+
end,
167+
{bucket_range(UpperBound), {Count, Percentage}}
168+
end, Buckets),
169+
170+
Ranges.
171+
172+
sum_protocol_buckets(ProtocolBuckets, Acc) ->
173+
lists:map(fun({UpperBound, AccCount}) ->
174+
ProtocolCount = proplists:get_value(UpperBound, ProtocolBuckets, 0),
175+
{UpperBound, AccCount + ProtocolCount}
176+
end, Acc).
177+
178+
%% Aggregates sumamries from all nodes
179+
-spec cluster_summary() -> summary().
180+
cluster_summary() ->
181+
RemoteNodes = [Node || Node <- rabbit_nodes:list_running(), Node =/= node()],
182+
RemoteSummaries = [ Summary || {ok, Summary} <- erpc:multicall(RemoteNodes,
183+
?MODULE,
184+
local_summary,
185+
[],
186+
5000)],
187+
lists:foldl(fun merge_summaries/2, local_summary(), RemoteSummaries).
188+
189+
bucket_name({_, ?BUCKET_1}) -> <<"below 100B">>;
190+
bucket_name({_, ?BUCKET_2}) -> <<"between 100B and 1KB">>;
191+
bucket_name({_, ?BUCKET_3}) -> <<"between 1KB and 10KB">>;
192+
bucket_name({_, ?BUCKET_4}) -> <<"between 10KB and 100KB">>;
193+
bucket_name({_, ?BUCKET_5}) -> <<"between 100KB and 1MB">>;
194+
bucket_name({_, ?BUCKET_6}) -> <<"between 1MB and 10MB">>;
195+
bucket_name({_, ?BUCKET_7}) -> <<"between 10MB and 50MB">>;
196+
bucket_name({_, ?BUCKET_8}) -> <<"between 50MB and 100MB">>;
197+
bucket_name({_, ?BUCKET_9}) -> <<"above 100MB">>.
198+
199+
cluster_summary_for_cli() ->
200+
[[{<<"Message Size">>, bucket_name(Range)},
201+
{<<"Count">>, Count},
202+
{<<"Percentage">>, iolist_to_binary(io_lib:format("~.2f", [Percentage]))}]
203+
|| {Range, {Count, Percentage}} <- cluster_summary()].
204+
205+
get_count_for_range(Range, SummaryList) ->
206+
case proplists:get_value(Range, SummaryList) of
207+
{Count, _} -> Count;
208+
undefined -> 0
209+
end.
210+
211+
%% Merges two summary lists by adding their counts and recalculating percentages
212+
merge_summaries(Summary1, Summary2) ->
213+
%% Get all bucket ranges
214+
AllRanges = lists:usort([Range || {Range, _} <- Summary1] ++ [Range || {Range, _} <- Summary2]),
215+
216+
MergedRanges = lists:map(fun(Range) ->
217+
Count1 = get_count_for_range(Range, Summary1),
218+
Count2 = get_count_for_range(Range, Summary2),
219+
NewCount = Count1 + Count2,
220+
{Range, NewCount}
221+
end, AllRanges),
222+
223+
%% Calculate total and percentages
224+
NewTotal = lists:sum([Count || {_, Count} <- MergedRanges]),
225+
FinalRanges = lists:map(fun({Range, Count}) ->
226+
NewPercentage = case NewTotal of
227+
0 -> 0.0;
228+
_ -> (Count / NewTotal) * 100
229+
end,
230+
{Range, {Count, NewPercentage}}
231+
end, MergedRanges),
232+
233+
FinalRanges.
234+
235+
bucket_range(?BUCKET_1) -> {0, 100};
236+
bucket_range(?BUCKET_2) -> {101, 1000};
237+
bucket_range(?BUCKET_3) -> {1001, 10000};
238+
bucket_range(?BUCKET_4) -> {10001, 100000};
239+
bucket_range(?BUCKET_5) -> {100001, 1000000};
240+
bucket_range(?BUCKET_6) -> {1000001, 10000000};
241+
bucket_range(?BUCKET_7) -> {10000001, 50000000};
242+
bucket_range(?BUCKET_8) -> {50000001, 100000000};
243+
bucket_range(?BUCKET_9) -> {100000001, infinity}.
244+
136245
-ifdef(TEST).
137246
%% "Counters are not tied to the current process and are automatically
138247
%% garbage collected when they are no longer referenced."

‎deps/rabbit/test/msg_size_metrics_SUITE.erl‎

Lines changed: 94 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -13,18 +13,22 @@
1313
-include_lib("amqp_client/include/amqp_client.hrl").
1414

1515
-import(rabbit_ct_broker_helpers,
16-
[rpc/4]).
16+
[rpc/4, rpc/5]).
1717

1818
all() ->
1919
[
20-
{group, tests}
20+
{group, cluster_size_1},
21+
{group, cluster_size_3}
2122
].
2223

2324
groups() ->
2425
[
25-
{tests, [shuffle],
26+
{cluster_size_1, [],
2627
[message_size,
27-
over_max_message_size]}
28+
over_max_message_size]},
29+
{cluster_size_3, [],
30+
[summary]
31+
}
2832
].
2933

3034
%% -------------------------------------------------------------------
@@ -34,16 +38,31 @@ groups() ->
3438
init_per_suite(Config) ->
3539
{ok, _} = application:ensure_all_started(amqp10_client),
3640
rabbit_ct_helpers:log_environment(),
37-
rabbit_ct_helpers:run_setup_steps(Config).
41+
Config.
3842

3943
end_per_suite(Config) ->
4044
rabbit_ct_helpers:run_teardown_steps(Config).
4145

42-
init_per_group(_Group, Config) ->
43-
rabbit_ct_helpers:run_steps(
44-
Config,
45-
rabbit_ct_broker_helpers:setup_steps() ++
46-
rabbit_ct_client_helpers:setup_steps()).
46+
init_per_group(Group, Config) ->
47+
Nodes = case Group of
48+
cluster_size_1 -> 1;
49+
cluster_size_3 -> 3
50+
end,
51+
Suffix = rabbit_ct_helpers:testcase_absname(Config, "", "-"),
52+
Config1 = rabbit_ct_helpers:set_config(
53+
Config, [{rmq_nodes_count, Nodes},
54+
{rmq_nodename_suffix, Suffix}]),
55+
56+
Config2 = rabbit_ct_helpers:run_setup_steps(
57+
Config1,
58+
rabbit_ct_broker_helpers:setup_steps() ++
59+
rabbit_ct_client_helpers:setup_steps()),
60+
case rabbit_ct_broker_helpers:enable_feature_flag(Config2, 'rabbitmq_4.2.0') of
61+
ok ->
62+
Config2;
63+
{skip, _} = Skip ->
64+
Skip
65+
end.
4766

4867
end_per_group(_Group, Config) ->
4968
rabbit_ct_helpers:run_steps(
@@ -65,32 +84,7 @@ message_size(Config) ->
6584
AmqplBefore = get_msg_size_metrics(amqp091, Config),
6685
AmqpBefore = get_msg_size_metrics(amqp10, Config),
6786

68-
Binary2B = <<"12">>,
69-
Binary200K = binary:copy(<<"x">>, 200_000),
70-
Payloads = [Binary2B, Binary200K, Binary2B],
71-
72-
{AmqplConn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config),
73-
[amqp_channel:call(Ch,
74-
#'basic.publish'{routing_key = <<"nowhere">>},
75-
#amqp_msg{payload = Payload})
76-
|| Payload <- Payloads],
77-
78-
OpnConf = connection_config(Config),
79-
{ok, Connection} = amqp10_client:open_connection(OpnConf),
80-
{ok, Session} = amqp10_client:begin_session_sync(Connection),
81-
Address = rabbitmq_amqp_address:exchange(<<"amq.fanout">>),
82-
{ok, Sender} = amqp10_client:attach_sender_link_sync(Session, <<"sender">>, Address),
83-
receive {amqp10_event, {link, Sender, credited}} -> ok
84-
after 30_000 -> ct:fail(credited_timeout)
85-
end,
86-
87-
ok = amqp10_client:send_msg(Sender, amqp10_msg:new(<<"tag1">>, Binary2B)),
88-
ok = amqp10_client:send_msg(Sender, amqp10_msg:new(<<"tag2">>, Binary200K)),
89-
ok = amqp10_client:send_msg(Sender, amqp10_msg:new(<<"tag3">>, Binary2B)),
90-
91-
ok = wait_for_settlement(released, <<"tag1">>),
92-
ok = wait_for_settlement(released, <<"tag2">>),
93-
ok = wait_for_settlement(released, <<"tag3">>),
87+
publish_messages(Config),
9488

9589
AmqplAfter = get_msg_size_metrics(amqp091, Config),
9690
AmqpAfter = get_msg_size_metrics(amqp10, Config),
@@ -100,10 +94,7 @@ message_size(Config) ->
10094
?assertEqual(ExpectedDiff,
10195
rabbit_msg_size_metrics:diff_raw_buckets(AmqplAfter, AmqplBefore)),
10296
?assertEqual(ExpectedDiff,
103-
rabbit_msg_size_metrics:diff_raw_buckets(AmqpAfter, AmqpBefore)),
104-
105-
ok = amqp10_client:close_connection(Connection),
106-
ok = rabbit_ct_client_helpers:close_connection_and_channel(AmqplConn, Ch).
97+
rabbit_msg_size_metrics:diff_raw_buckets(AmqpAfter, AmqpBefore)).
10798

10899
over_max_message_size(Config) ->
109100
DefaultMaxMessageSize = rpc(Config, persistent_term, get, [max_message_size]),
@@ -134,6 +125,39 @@ over_max_message_size(Config) ->
134125
ok = rabbit_ct_client_helpers:close_connection(Conn),
135126
ok = rpc(Config, persistent_term, put, [max_message_size, DefaultMaxMessageSize]).
136127

128+
summary(Config) ->
129+
ZeroSummary = [{{0, 100}, {0, 0.0}},
130+
{{101, 1000}, {0, 0.0}},
131+
{{1001, 10000}, {0, 0.0}},
132+
{{10001, 100000}, {0, 0.0}},
133+
{{100001, 1000000}, {0, 0.0}},
134+
{{1000001, 10000000}, {0, 0.0}},
135+
{{10000001, 50000000}, {0, 0.0}},
136+
{{50000001, 100000000}, {0, 0.0}},
137+
{{100000001, infinity}, {0, 0.0}}],
138+
139+
?assertEqual(ZeroSummary, rpc(Config, 0, rabbit_msg_size_metrics, local_summary, [])),
140+
?assertEqual(ZeroSummary, rpc(Config, 1, rabbit_msg_size_metrics, cluster_summary, [])),
141+
?assertEqual(ZeroSummary, rpc(Config, 0, rabbit_msg_size_metrics, local_summary, [])),
142+
?assertEqual(ZeroSummary, rpc(Config, 1, rabbit_msg_size_metrics, cluster_summary, [])),
143+
144+
publish_messages(Config),
145+
146+
ExpectedSummary = [{{0, 100}, {4, 66.66666666666666}},
147+
{{101, 1000}, {0, 0.0}},
148+
{{1001, 10000}, {0, 0.0}},
149+
{{10001, 100000}, {0, 0.0}},
150+
{{100001, 1000000}, {2, 33.33333333333333}},
151+
{{1000001, 10000000}, {0, 0.0}},
152+
{{10000001, 50000000}, {0, 0.0}},
153+
{{50000001, 100000000}, {0, 0.0}},
154+
{{100000001, infinity}, {0, 0.0}}],
155+
156+
?assertEqual(ExpectedSummary, rpc(Config, 0, rabbit_msg_size_metrics, local_summary, [])),
157+
?assertEqual(ExpectedSummary, rpc(Config, 0, rabbit_msg_size_metrics, cluster_summary, [])),
158+
?assertEqual(ExpectedSummary, rpc(Config, 1, rabbit_msg_size_metrics, cluster_summary, [])),
159+
?assertEqual(ZeroSummary, rpc(Config, 1, rabbit_msg_size_metrics, local_summary, [])).
160+
137161
get_msg_size_metrics(Protocol, Config) ->
138162
rpc(Config, rabbit_msg_size_metrics, raw_buckets, [Protocol]).
139163

@@ -145,6 +169,36 @@ connection_config(Config) ->
145169
container_id => <<"my container">>,
146170
sasl => anon}.
147171

172+
publish_messages(Config) ->
173+
Binary2B = <<"12">>,
174+
Binary200K = binary:copy(<<"x">>, 200_000),
175+
Payloads = [Binary2B, Binary200K, Binary2B],
176+
177+
{AmqplConn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config),
178+
[amqp_channel:call(Ch,
179+
#'basic.publish'{routing_key = <<"nowhere">>},
180+
#amqp_msg{payload = Payload})
181+
|| Payload <- Payloads],
182+
183+
OpnConf = connection_config(Config),
184+
{ok, Connection} = amqp10_client:open_connection(OpnConf),
185+
{ok, Session} = amqp10_client:begin_session_sync(Connection),
186+
Address = rabbitmq_amqp_address:exchange(<<"amq.fanout">>),
187+
{ok, Sender} = amqp10_client:attach_sender_link_sync(Session, <<"sender">>, Address),
188+
receive {amqp10_event, {link, Sender, credited}} -> ok
189+
after 30_000 -> ct:fail(credited_timeout)
190+
end,
191+
192+
ok = amqp10_client:send_msg(Sender, amqp10_msg:new(<<"tag1">>, Binary2B)),
193+
ok = amqp10_client:send_msg(Sender, amqp10_msg:new(<<"tag2">>, Binary200K)),
194+
ok = amqp10_client:send_msg(Sender, amqp10_msg:new(<<"tag3">>, Binary2B)),
195+
196+
ok = wait_for_settlement(released, <<"tag1">>),
197+
ok = wait_for_settlement(released, <<"tag2">>),
198+
ok = wait_for_settlement(released, <<"tag3">>),
199+
ok = amqp10_client:close_connection(Connection),
200+
ok = rabbit_ct_client_helpers:close_connection_and_channel(AmqplConn, Ch).
201+
148202
wait_for_settlement(State, Tag) ->
149203
receive
150204
{amqp10_disposition, {State, Tag}} ->

‎deps/rabbitmq_cli/lib/rabbitmq/cli/diagnostics/commands/message_size_stats_command.ex‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
## This Source Code Form is subject to the terms of the Mozilla Public
2+
## License, v. 2.0. If a copy of the MPL was not distributed with this
3+
## file, You can obtain one at https://mozilla.org/MPL/2.0/.
4+
##
5+
## Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
6+
7+
defmodule RabbitMQ.CLI.Diagnostics.Commands.MessageSizeStatsCommand do
8+
alias RabbitMQ.CLI.Core.DocGuide
9+
10+
@behaviour RabbitMQ.CLI.CommandBehaviour
11+
12+
@default_timeout 60_000
13+
14+
def scopes(), do: [:ctl, :diagnostics]
15+
16+
def switches(), do: [timeout: :integer]
17+
def aliases(), do: [t: :timeout]
18+
19+
def merge_defaults(args, opts) do
20+
timeout =
21+
case opts[:timeout] do
22+
nil -> @default_timeout
23+
:infinity -> @default_timeout
24+
other -> other
25+
end
26+
27+
{args, Map.merge(%{timeout: timeout}, opts)}
28+
end
29+
30+
use RabbitMQ.CLI.Core.AcceptsNoPositionalArguments
31+
use RabbitMQ.CLI.Core.RequiresRabbitAppRunning
32+
33+
def run([], %{node: node_name, timeout: timeout}) do
34+
:rabbit_misc.rpc_call(node_name, :rabbit_msg_size_metrics, :cluster_summary_for_cli, [], timeout)
35+
end
36+
37+
use RabbitMQ.CLI.DefaultOutput
38+
39+
def formatter(), do: RabbitMQ.CLI.Formatters.PrettyTable
40+
41+
def usage, do: "message_size_stats"
42+
43+
def usage_doc_guides() do
44+
[
45+
DocGuide.monitoring()
46+
]
47+
end
48+
49+
def help_section(), do: :observability_and_health_checks
50+
51+
def description(),
52+
do: "Displays message size distribution statistics aggregated across all cluster nodes"
53+
54+
def banner(_, %{node: node_name}), do: "Gathering message size statistics from cluster via #{node_name} ..."
55+
56+
end

0 commit comments

Comments
 (0)