diff options
Diffstat (limited to 'deps/rabbitmq_management_agent/src')
16 files changed, 3732 insertions, 0 deletions
diff --git a/deps/rabbitmq_management_agent/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ResetStatsDbCommand.erl b/deps/rabbitmq_management_agent/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ResetStatsDbCommand.erl new file mode 100644 index 0000000000..bc6bdbdc25 --- /dev/null +++ b/deps/rabbitmq_management_agent/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ResetStatsDbCommand.erl @@ -0,0 +1,54 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates. All rights reserved. +%% + +-module('Elixir.RabbitMQ.CLI.Ctl.Commands.ResetStatsDbCommand'). + +-behaviour('Elixir.RabbitMQ.CLI.CommandBehaviour'). + +-export([ + usage/0, + validate/2, + merge_defaults/2, + banner/2, + run/2, + output/2, + switches/0, + description/0 + ]). + + +%%---------------------------------------------------------------------------- +%% Callbacks +%%---------------------------------------------------------------------------- +usage() -> + <<"reset_stats_db [--all]">>. + +validate(_, _) -> + ok. + +merge_defaults(A, Opts) -> + {A, maps:merge(#{all => false}, Opts)}. + +switches() -> + [{all, boolean}]. + +run(_Args, #{node := Node, all := true}) -> + rabbit_misc:rpc_call(Node, rabbit_mgmt_storage, reset_all, []); +run(_Args, #{node := Node, all := false}) -> + rabbit_misc:rpc_call(Node, rabbit_mgmt_storage, reset, []). + +output(Output, _Opts) -> + 'Elixir.RabbitMQ.CLI.DefaultOutput':output(Output). + +banner(_, #{all := true}) -> + <<"Resetting statistics database in all nodes">>; +banner(_, #{node := Node}) -> + erlang:iolist_to_binary([<<"Resetting statistics database on node ">>, + atom_to_binary(Node, utf8)]). + +description() -> + <<"Resets statistics database. This will remove all metrics data!">>. diff --git a/deps/rabbitmq_management_agent/src/exometer_slide.erl b/deps/rabbitmq_management_agent/src/exometer_slide.erl new file mode 100644 index 0000000000..2c4e4c6d35 --- /dev/null +++ b/deps/rabbitmq_management_agent/src/exometer_slide.erl @@ -0,0 +1,551 @@ +%% This file is a copy of exometer_slide.erl from https://github.com/Feuerlabs/exometer_core, +%% with the following modifications: +%% +%% 1) The elements are tuples of numbers +%% +%% 2) Only one element for each expected interval point is added, intermediate values +%% are discarded. Thus, if we have a window of 60s and interval of 5s, at max 12 elements +%% are stored. +%% +%% 3) Additions can be provided as increments to the last value stored +%% +%% 4) sum/1 implements the sum of several slides, generating a new timestamp sequence based +%% on the given intervals. Elements on each window are added to the closest interval point. +%% +%% Original commit: https://github.com/Feuerlabs/exometer_core/commit/2759edc804211b5245867b32c9a20c8fe1d93441 +%% +%% ------------------------------------------------------------------- +%% +%% Copyright (c) 2014 Basho Technologies, Inc. All Rights Reserved. +%% +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% ------------------------------------------------------------------- +%% +%% @author Tony Rogvall <tony@rogvall.se> +%% @author Ulf Wiger <ulf@feuerlabs.com> +%% @author Magnus Feuer <magnus@feuerlabs.com> +%% +%% @doc Efficient sliding-window buffer +%% +%% Initial implementation: 29 Sep 2009 by Tony Rogvall +%% +%% This module implements an efficient sliding window, maintaining +%% two lists - a primary and a secondary. Values are paired with a +%% timestamp (millisecond resolution, see `timestamp/0') +%% and prepended to the primary list. When the time span between the oldest +%% and the newest entry in the primary list exceeds the given window size, +%% the primary list is shifted into the secondary list position, and the +%% new entry is added to a new (empty) primary list. +%% +%% The window can be converted to a list using `to_list/1'. +%% @end +%% +%% +%% All modifications are (C) 2007-2020 VMware, Inc. or its affiliates. All rights reserved. +%% The Initial Developer of the Original Code is Basho Technologies, Inc. + +-module(exometer_slide). + +-export([new/2, new/3, + reset/1, + add_element/3, + to_list/2, + to_list/3, + foldl/5, + map/2, + to_normalized_list/5]). + +-export([timestamp/0, + last_two/1, + last/1]). + +-export([sum/1, + sum/2, + sum/5, + optimize/1]). + +%% For testing +-export([buffer/1]). + +-compile(inline). +-compile(inline_list_funcs). + + +-type value() :: tuple(). +-type internal_value() :: tuple() | drop. +-type timestamp() :: non_neg_integer(). + +-type fold_acc() :: any(). +-type fold_fun() :: fun(({timestamp(), internal_value()}, fold_acc()) -> fold_acc()). + +%% Fixed size event buffer +-record(slide, {size = 0 :: integer(), % ms window + n = 0 :: integer(), % number of elements in buf1 + max_n :: infinity | integer(), % max no of elements + incremental = false :: boolean(), + interval :: integer(), + last = 0 :: integer(), % millisecond timestamp + first = undefined :: undefined | integer(), % millisecond timestamp + buf1 = [] :: [internal_value()], + buf2 = [] :: [internal_value()], + total :: undefined | value()}). + +-opaque slide() :: #slide{}. + +-export_type([slide/0, timestamp/0]). + +-spec timestamp() -> timestamp(). +%% @doc Generate a millisecond-resolution timestamp. +%% +%% This timestamp format is used e.g. by the `exometer_slide' and +%% `exometer_histogram' implementations. +%% @end +timestamp() -> + os:system_time(milli_seconds). + +-spec new(_Size::integer(), _Options::list()) -> slide(). +%% @doc Create a new sliding-window buffer. +%% +%% `Size' determines the size in milliseconds of the sliding window. +%% The implementation prepends values into a primary list until the oldest +%% element in the list is `Size' ms older than the current value. It then +%% swaps the primary list into a secondary list, and starts prepending to +%% a new primary list. This means that more data than fits inside the window +%% will be kept - upwards of twice as much. On the other hand, updating the +%% buffer is very cheap. +%% @end +new(Size, Opts) -> new(timestamp(), Size, Opts). + +-spec new(Timestamp :: timestamp(), Size::integer(), Options::list()) -> slide(). +new(TS, Size, Opts) -> + #slide{size = Size, + max_n = proplists:get_value(max_n, Opts, infinity), + interval = proplists:get_value(interval, Opts, infinity), + last = TS, + first = undefined, + incremental = proplists:get_value(incremental, Opts, false), + buf1 = [], + buf2 = []}. + +-spec reset(slide()) -> slide(). + +%% @doc Empty the buffer +%% +reset(Slide) -> + Slide#slide{n = 0, buf1 = [], buf2 = [], last = 0, first = undefined}. + +%% @doc Add an element to the buffer, tagged with the given timestamp. +%% +%% Apart from the specified timestamp, this function works just like +%% {@link add_element/2}. +%% @end +-spec add_element(timestamp(), value(), slide()) -> slide(). +add_element(_TS, _Evt, Slide) when Slide#slide.size == 0 -> + Slide; +add_element(TS, Evt, #slide{last = Last, interval = Interval, total = Total0, + incremental = true} = Slide) + when (TS - Last) < Interval -> + Total = add_to_total(Evt, Total0), + Slide#slide{total = Total}; +add_element(TS, Evt, #slide{last = Last, interval = Interval} = Slide) + when (TS - Last) < Interval -> + Slide#slide{total = Evt}; +add_element(TS, Evt, #slide{last = Last, size = Sz, incremental = true, + n = N, max_n = MaxN, total = Total0, + buf1 = Buf1} = Slide) -> + N1 = N+1, + Total = add_to_total(Evt, Total0), + %% Total could be the same as the last sample, by adding and substracting + %% the same amout to the totals. That is not strictly a drop, but should + %% generate new samples. + %% I.e. 0, 0, -14, 14 (total = 0, samples = 14, -14, 0, drop) + case {is_zeros(Evt), Buf1} of + {_, []} -> + Slide#slide{n = N1, first = TS, buf1 = [{TS, Total} | Buf1], + last = TS, total = Total}; + _ when TS - Last > Sz; N1 > MaxN -> + %% swap + Slide#slide{last = TS, n = 1, buf1 = [{TS, Total}], + buf2 = Buf1, total = Total}; + {true, [{_, Total}, {_, drop} = Drop | Tail]} -> + %% Memory optimisation + Slide#slide{buf1 = [{TS, Total}, Drop | Tail], + n = N1, last = TS}; + {true, [{DropTS, Total} | Tail]} -> + %% Memory optimisation + Slide#slide{buf1 = [{TS, Total}, {DropTS, drop} | Tail], + n = N1, last = TS}; + _ -> + Slide#slide{n = N1, buf1 = [{TS, Total} | Buf1], + last = TS, total = Total} + end; +add_element(TS, Evt, #slide{last = Last, size = Sz, n = N, max_n = MaxN, + buf1 = Buf1} = Slide) + when TS - Last > Sz; N + 1 > MaxN -> + Slide#slide{last = TS, n = 1, buf1 = [{TS, Evt}], + buf2 = Buf1, total = Evt}; +add_element(TS, Evt, #slide{buf1 = [{_, Evt}, {_, drop} = Drop | Tail], + n = N} = Slide) -> + %% Memory optimisation + Slide#slide{buf1 = [{TS, Evt}, Drop | Tail], n = N + 1, last = TS}; +add_element(TS, Evt, #slide{buf1 = [{DropTS, Evt} | Tail], n = N} = Slide) -> + %% Memory optimisation + Slide#slide{buf1 = [{TS, Evt}, {DropTS, drop} | Tail], + n = N + 1, last = TS}; +add_element(TS, Evt, #slide{n = N, buf1 = Buf1} = Slide) -> + N1 = N+1, + case Buf1 of + [] -> + Slide#slide{n = N1, buf1 = [{TS, Evt} | Buf1], + last = TS, first = TS, total = Evt}; + _ -> + Slide#slide{n = N1, buf1 = [{TS, Evt} | Buf1], + last = TS, total = Evt} + end. + +add_to_total(Evt, undefined) -> + Evt; +add_to_total({A0}, {B0}) -> + {B0 + A0}; +add_to_total({A0, A1}, {B0, B1}) -> + {B0 + A0, B1 + A1}; +add_to_total({A0, A1, A2}, {B0, B1, B2}) -> + {B0 + A0, B1 + A1, B2 + A2}; +add_to_total({A0, A1, A2, A3}, {B0, B1, B2, B3}) -> + {B0 + A0, B1 + A1, B2 + A2, B3 + A3}; +add_to_total({A0, A1, A2, A3, A4}, {B0, B1, B2, B3, B4}) -> + {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4}; +add_to_total({A0, A1, A2, A3, A4, A5}, {B0, B1, B2, B3, B4, B5}) -> + {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4, B5 + A5}; +add_to_total({A0, A1, A2, A3, A4, A5, A6}, {B0, B1, B2, B3, B4, B5, B6}) -> + {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4, B5 + A5, B6 + A6}; +add_to_total({A0, A1, A2, A3, A4, A5, A6, A7}, {B0, B1, B2, B3, B4, B5, B6, B7}) -> + {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4, B5 + A5, B6 + A6, B7 + A7}; +add_to_total({A0, A1, A2, A3, A4, A5, A6, A7, A8}, {B0, B1, B2, B3, B4, B5, B6, B7, B8}) -> + {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4, B5 + A5, B6 + A6, B7 + A7, B8 + A8}; +add_to_total({A0, A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14, + A15, A16, A17, A18, A19}, + {B0, B1, B2, B3, B4, B5, B6, B7, B8, B9, B10, B11, B12, B13, B14, + B15, B16, B17, B18, B19}) -> + {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4, B5 + A5, B6 + A6, B7 + A7, B8 + A8, + B9 + A9, B10 + A10, B11 + A11, B12 + A12, B13 + A13, B14 + A14, B15 + A15, B16 + A16, + B17 + A17, B18 + A18, B19 + A19}. + +is_zeros({0}) -> + true; +is_zeros({0, 0}) -> + true; +is_zeros({0, 0, 0}) -> + true; +is_zeros({0, 0, 0, 0}) -> + true; +is_zeros({0, 0, 0, 0, 0}) -> + true; +is_zeros({0, 0, 0, 0, 0, 0}) -> + true; +is_zeros({0, 0, 0, 0, 0, 0, 0}) -> + true; +is_zeros({0, 0, 0, 0, 0, 0, 0, 0, 0}) -> + true; +is_zeros({0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}) -> + true; +is_zeros(_) -> + false. + +-spec optimize(slide()) -> slide(). +optimize(#slide{buf2 = []} = Slide) -> + Slide; +optimize(#slide{buf1 = Buf1, buf2 = Buf2, max_n = MaxN, n = N} = Slide) + when is_integer(MaxN) andalso length(Buf1) < MaxN -> + Slide#slide{buf1 = Buf1, + buf2 = lists:sublist(Buf2, n_diff(MaxN, N) + 1)}; +optimize(Slide) -> Slide. + +snd(T) when is_tuple(T) -> + element(2, T). + + +-spec to_list(timestamp(), slide()) -> [{timestamp(), value()}]. +%% @doc Convert the sliding window into a list of timestamped values. +%% @end +to_list(_Now, #slide{size = Sz}) when Sz == 0 -> + []; +to_list(Now, #slide{size = Sz} = Slide) -> + snd(to_list_from(Now, Now - Sz, Slide)). + +to_list(Now, Start, Slide) -> + snd(to_list_from(Now, Start, Slide)). + +to_list_from(Now, Start0, #slide{max_n = MaxN, buf2 = Buf2, first = FirstTS, + interval = Interval} = Slide) -> + + {NewN, Buf1} = maybe_add_last_sample(Now, Slide), + Start = first_max(FirstTS, Start0), + {Prev0, Buf1_1} = take_since(Buf1, Now, Start, first_max(MaxN, NewN), [], Interval), + case take_since(Buf2, Now, Start, first_max(MaxN, NewN), Buf1_1, Interval) of + {undefined, Buf1_1} -> + {Prev0, Buf1_1}; + {_Prev, Buf1_1} = Res -> + case Prev0 of + undefined -> + Res; + _ -> + %% If take_since returns the same buffer, that means we don't + %% need Buf2 at all. We might be returning a too old sample + %% in previous, so we must use the one from Buf1 + {Prev0, Buf1_1} + end; + Res -> + Res + end. + +first_max(F, X) when is_integer(F) -> max(F, X); +first_max(_, X) -> X. + +-spec last_two(slide()) -> [{timestamp(), value()}]. +%% @doc Returns the newest 2 elements on the sample +last_two(#slide{buf1 = [{TS, Evt} = H1, {_, drop} | _], interval = Interval}) -> + [H1, {TS - Interval, Evt}]; +last_two(#slide{buf1 = [H1, H2_0 | _], interval = Interval}) -> + H2 = adjust_timestamp(H1, H2_0, Interval), + [H1, H2]; +last_two(#slide{buf1 = [H1], buf2 = [H2_0 | _], + interval = Interval}) -> + H2 = adjust_timestamp(H1, H2_0, Interval), + [H1, H2]; +last_two(#slide{buf1 = [H1], buf2 = []}) -> + [H1]; +last_two(_) -> + []. + +adjust_timestamp({TS1, _}, {TS2, V2}, Interval) -> + case TS1 - TS2 > Interval of + true -> {TS1 - Interval, V2}; + false -> {TS2, V2} + end. + +-spec last(slide()) -> value() | undefined. +last(#slide{total = T}) when T =/= undefined -> + T; +last(#slide{buf1 = [{_TS, T} | _]}) -> + T; +last(#slide{buf2 = [{_TS, T} | _]}) -> + T; +last(_) -> + undefined. + +-spec foldl(timestamp(), timestamp(), fold_fun(), fold_acc(), slide()) -> fold_acc(). +%% @doc Fold over the sliding window, starting from `Timestamp'. +%% Now provides a reference point to evaluate whether to include +%% partial, unrealised sample values in the sequence. Unrealised values will be +%% appended to the sequence when Now >= LastTS + Interval +%% +%% The fun should as `fun({Timestamp, Value}, Acc) -> NewAcc'. +%% The values are processed in order from oldest to newest. +%% @end +foldl(_Now, _Timestamp, _Fun, _Acc, #slide{size = Sz}) when Sz == 0 -> + []; +foldl(Now, Start0, Fun, Acc, #slide{max_n = _MaxN, buf2 = _Buf2, + interval = _Interval} = Slide) -> + lists:foldl(Fun, Acc, element(2, to_list_from(Now, Start0, Slide)) ++ [last]). + +map(Fun, #slide{buf1 = Buf1, buf2 = Buf2, total = Total} = Slide) -> + BufFun = fun({Timestamp, Value}) -> + {Timestamp, Fun(Value)} + end, + MappedBuf1 = lists:map(BufFun, Buf1), + MappedBuf2 = lists:map(BufFun, Buf2), + MappedTotal = Fun(Total), + Slide#slide{buf1 = MappedBuf1, buf2 = MappedBuf2, total = MappedTotal}. + +maybe_add_last_sample(_Now, #slide{total = T, n = N, + buf1 = [{_, T} | _] = Buf1}) -> + {N, Buf1}; +maybe_add_last_sample(Now, #slide{total = T, + n = N, + last = Last, + interval = I, + buf1 = Buf1}) + when T =/= undefined andalso Now >= Last + I -> + {N + 1, [{Last + I, T} | Buf1]}; +maybe_add_last_sample(_Now, #slide{buf1 = Buf1, n = N}) -> + {N, Buf1}. + + +create_normalized_lookup(Start, Interval, RoundFun, Samples) -> + lists:foldl(fun({TS, Value}, Acc) when TS - Start >= 0 -> + NewTS = map_timestamp(TS, Start, Interval, RoundFun), + maps:update_with(NewTS, + fun({T, V}) when T > TS -> + {T, V}; + (_) -> + {TS, Value} + end, {TS, Value}, Acc); + (_, Acc) -> + Acc + end, #{}, Samples). + +-spec to_normalized_list(timestamp(), timestamp(), integer(), slide(), + no_pad | tuple()) -> [tuple()]. +to_normalized_list(Now, Start, Interval, Slide, Empty) -> + to_normalized_list(Now, Start, Interval, Slide, Empty, fun ceil/1). + +to_normalized_list(Now, Start, Interval, #slide{first = FirstTS0, + total = Total} = Slide, + Empty, RoundFun) -> + + RoundTSFun = fun (TS) -> map_timestamp(TS, Start, Interval, RoundFun) end, + + % add interval as we don't want to miss a sample due to rounding + {Prev, Samples} = to_list_from(Now + Interval, Start, Slide), + Lookup = create_normalized_lookup(Start, Interval, RoundFun, Samples), + + NowRound = RoundTSFun(Now), + + Pad = case Samples of + _ when Empty =:= no_pad -> + []; + [{TS, _} | _] when Prev =/= undefined, Start =< TS -> + [{T, snd(Prev)} + || T <- lists:seq(RoundTSFun(TS) - Interval, Start, + -Interval)]; + [{TS, _} | _] when is_number(FirstTS0) andalso Start < FirstTS0 -> + % only if we know there is nothing in the past can we + % generate a 0 pad + [{T, Empty} || T <- lists:seq(RoundTSFun(TS) - Interval, Start, + -Interval)]; + _ when FirstTS0 =:= undefined andalso Total =:= undefined -> + [{T, Empty} || T <- lists:seq(NowRound, Start, -Interval)]; + [] -> % samples have been seen, use the total to pad + [{T, Total} || T <- lists:seq(NowRound, Start, -Interval)]; + _ -> [] + end, + + {_, Res1} = lists:foldl( + fun(T, {Last, Acc}) -> + case maps:find(T, Lookup) of + {ok, {_, V}} -> + {V, [{T, V} | Acc]}; + error when Last =:= undefined -> + {Last, Acc}; + error -> % this pads the last value into the future + {Last, [{T, Last} | Acc]} + end + end, {undefined, []}, + lists:seq(Start, NowRound, Interval)), + Res1 ++ Pad. + + +%% @doc Sums a list of slides +%% +%% Takes the last known timestamp and creates an template version of the +%% sliding window. Timestamps are then truncated and summed with the value +%% in the template slide. +%% @end +-spec sum([slide()]) -> slide(). +sum(Slides) -> sum(Slides, no_pad). + +sum([#slide{size = Size, interval = Interval} | _] = Slides, Pad) -> + % take the freshest timestamp as reference point for summing operation + Now = lists:max([Last || #slide{last = Last} <- Slides]), + Start = Now - Size, + sum(Now, Start, Interval, Slides, Pad). + + +sum(Now, Start, Interval, [Slide | _ ] = All, Pad) -> + Fun = fun({TS, Value}, Acc) -> + maps:update_with(TS, fun(V) -> add_to_total(V, Value) end, + Value, Acc) + end, + {Total, Dict} = + lists:foldl(fun(#slide{total = T} = S, {Tot, Acc}) -> + Samples = to_normalized_list(Now, Start, Interval, S, + Pad, fun ceil/1), + Total = add_to_total(T, Tot), + Folded = lists:foldl(Fun, Acc, Samples), + {Total, Folded} + end, {undefined, #{}}, All), + + {First, Buffer} = case lists:sort(maps:to_list(Dict)) of + [] -> + F = case [TS || #slide{first = TS} <- All, + is_integer(TS)] of + [] -> undefined; + FS -> lists:min(FS) + end, + {F, []}; + [{F, _} | _ ] = B -> + {F, lists:reverse(B)} + end, + Slide#slide{buf1 = Buffer, buf2 = [], total = Total, n = length(Buffer), + first = First, last = Now}. + + +truncated_seq(_First, _Last, _Incr, 0) -> + []; +truncated_seq(TS, TS, _Incr, MaxN) when MaxN > 0 -> + [TS]; +truncated_seq(First, Last, Incr, MaxN) when First =< Last andalso MaxN > 0 -> + End = min(Last, First + (MaxN * Incr) - Incr), + lists:seq(First, End, Incr); +truncated_seq(First, Last, Incr, MaxN) -> + End = max(Last, First + (MaxN * Incr) - Incr), + lists:seq(First, End, Incr). + + +take_since([{DropTS, drop} | T], Now, Start, N, [{TS, Evt} | _] = Acc, + Interval) -> + case T of + [] -> + Fill = [{TS0, Evt} || TS0 <- truncated_seq(TS - Interval, + max(DropTS, Start), + -Interval, N)], + {undefined, lists:reverse(Fill) ++ Acc}; + [{TS0, _} = E | Rest] when TS0 >= Start, N > 0 -> + Fill = [{TS1, Evt} || TS1 <- truncated_seq(TS0 + Interval, + max(TS0 + Interval, TS - Interval), + Interval, N)], + take_since(Rest, Now, Start, decr(N), [E | Fill ++ Acc], Interval); + [Prev | _] -> % next sample is out of range so needs to be filled from Start + Fill = [{TS1, Evt} || TS1 <- truncated_seq(Start, max(Start, TS - Interval), + Interval, N)], + {Prev, Fill ++ Acc} + end; +take_since([{TS, V} = H | T], Now, Start, N, Acc, Interval) when TS >= Start, + N > 0, + TS =< Now, + is_tuple(V) -> + take_since(T, Now, Start, decr(N), [H|Acc], Interval); +take_since([{TS,_} | T], Now, Start, N, Acc, Interval) when TS >= Start, N > 0 -> + take_since(T, Now, Start, decr(N), Acc, Interval); +take_since([Prev | _], _, _, _, Acc, _) -> + {Prev, Acc}; +take_since(_, _, _, _, Acc, _) -> + %% Don't reverse; already the wanted order. + {undefined, Acc}. + +decr(N) when is_integer(N) -> + N-1; +decr(N) -> N. + +n_diff(A, B) when is_integer(A) -> + A - B. + +ceil(X) when X < 0 -> + trunc(X); +ceil(X) -> + T = trunc(X), + case X - T == 0 of + true -> T; + false -> T + 1 + end. + +map_timestamp(TS, Start, Interval, Round) -> + Factor = Round((TS - Start) / Interval), + Start + Interval * Factor. + +buffer(#slide{buf1 = Buf1, buf2 = Buf2}) -> + Buf1 ++ Buf2. diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_agent_app.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_agent_app.erl new file mode 100644 index 0000000000..e889815c2f --- /dev/null +++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_agent_app.erl @@ -0,0 +1,17 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates. All rights reserved. +%% + +-module(rabbit_mgmt_agent_app). + +-behaviour(application). +-export([start/2, stop/1]). + +start(_Type, _StartArgs) -> + rabbit_mgmt_agent_sup_sup:start_link(). + +stop(_State) -> + ok. diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_agent_config.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_agent_config.erl new file mode 100644 index 0000000000..e8d074e891 --- /dev/null +++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_agent_config.erl @@ -0,0 +1,22 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates. All rights reserved. +%% +-module(rabbit_mgmt_agent_config). + +-export([get_env/1, get_env/2]). + +%% some people have reasons to only run with the agent enabled: +%% make it possible for them to configure key management app +%% settings such as rates_mode. +get_env(Key) -> + rabbit_misc:get_env(rabbitmq_management, Key, + rabbit_misc:get_env(rabbitmq_management_agent, Key, + undefined)). + +get_env(Key, Default) -> + rabbit_misc:get_env(rabbitmq_management, Key, + rabbit_misc:get_env(rabbitmq_management_agent, Key, + Default)). diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_agent_sup.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_agent_sup.erl new file mode 100644 index 0000000000..0c4a5465e9 --- /dev/null +++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_agent_sup.erl @@ -0,0 +1,55 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates. All rights reserved. +%% + +-module(rabbit_mgmt_agent_sup). + +%% pg2 is deprecated in OTP 23. +-compile(nowarn_deprecated_function). + +-behaviour(supervisor). + +-include_lib("rabbit_common/include/rabbit.hrl"). +-include_lib("rabbit_common/include/rabbit_core_metrics.hrl"). +-include("rabbit_mgmt_metrics.hrl"). + +-export([init/1]). +-export([start_link/0]). + +init([]) -> + MCs = maybe_enable_metrics_collector(), + ExternalStats = {rabbit_mgmt_external_stats, + {rabbit_mgmt_external_stats, start_link, []}, + permanent, 5000, worker, [rabbit_mgmt_external_stats]}, + {ok, {{one_for_one, 100, 10}, [ExternalStats] ++ MCs}}. + +start_link() -> + supervisor:start_link({local, ?MODULE}, ?MODULE, []). + + +maybe_enable_metrics_collector() -> + case application:get_env(rabbitmq_management_agent, disable_metrics_collector, false) of + false -> + pg2:create(management_db), + ok = pg2:join(management_db, self()), + ST = {rabbit_mgmt_storage, {rabbit_mgmt_storage, start_link, []}, + permanent, ?WORKER_WAIT, worker, [rabbit_mgmt_storage]}, + MD = {delegate_management_sup, {delegate_sup, start_link, [5, ?DELEGATE_PREFIX]}, + permanent, ?SUPERVISOR_WAIT, supervisor, [delegate_sup]}, + MC = [{rabbit_mgmt_metrics_collector:name(Table), + {rabbit_mgmt_metrics_collector, start_link, [Table]}, + permanent, ?WORKER_WAIT, worker, [rabbit_mgmt_metrics_collector]} + || {Table, _} <- ?CORE_TABLES], + MGC = [{rabbit_mgmt_metrics_gc:name(Event), + {rabbit_mgmt_metrics_gc, start_link, [Event]}, + permanent, ?WORKER_WAIT, worker, [rabbit_mgmt_metrics_gc]} + || Event <- ?GC_EVENTS], + GC = {rabbit_mgmt_gc, {rabbit_mgmt_gc, start_link, []}, + permanent, ?WORKER_WAIT, worker, [rabbit_mgmt_gc]}, + [ST, MD, GC | MC ++ MGC]; + true -> + [] + end. diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_agent_sup_sup.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_agent_sup_sup.erl new file mode 100644 index 0000000000..17ffa35307 --- /dev/null +++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_agent_sup_sup.erl @@ -0,0 +1,28 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates. All rights reserved. +%% + +-module(rabbit_mgmt_agent_sup_sup). + +-behaviour(supervisor2). + +-export([init/1]). +-export([start_link/0, start_child/0]). + +-include_lib("rabbit_common/include/rabbit.hrl"). + +start_child() -> + supervisor2:start_child(?MODULE, sup()). + +sup() -> + {rabbit_mgmt_agent_sup, {rabbit_mgmt_agent_sup, start_link, []}, + temporary, ?SUPERVISOR_WAIT, supervisor, [rabbit_mgmt_agent_sup]}. + +init([]) -> + {ok, {{one_for_one, 0, 1}, [sup()]}}. + +start_link() -> + supervisor2:start_link({local, ?MODULE}, ?MODULE, []). diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_data.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_data.erl new file mode 100644 index 0000000000..d73c8a3819 --- /dev/null +++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_data.erl @@ -0,0 +1,572 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2016-2020 VMware, Inc. or its affiliates. All rights reserved. +%% + +-module(rabbit_mgmt_data). + +-include("rabbit_mgmt_records.hrl"). +-include("rabbit_mgmt_metrics.hrl"). +-include_lib("rabbit_common/include/rabbit.hrl"). +-include_lib("rabbit_common/include/rabbit_core_metrics.hrl"). + +-export([empty/2, pick_range/2]). + +% delegate api +-export([overview_data/4, + consumer_data/2, + all_list_queue_data/3, + all_detail_queue_data/3, + all_exchange_data/3, + all_connection_data/3, + all_list_channel_data/3, + all_detail_channel_data/3, + all_vhost_data/3, + all_node_data/3, + augmented_created_stats/2, + augmented_created_stats/3, + augment_channel_pids/2, + augment_details/2, + lookup_element/2, + lookup_element/3 + ]). + +-import(rabbit_misc, [pget/2]). + +-type maybe_slide() :: exometer_slide:slide() | not_found. +-type ranges() :: {maybe_range(), maybe_range(), maybe_range(), maybe_range()}. +-type maybe_range() :: no_range | #range{}. + +%%---------------------------------------------------------------------------- +%% Internal, query-time - node-local operations +%%---------------------------------------------------------------------------- + +created_stats(Name, Type) -> + case ets:select(Type, [{{'_', '$2', '$3'}, [{'==', Name, '$2'}], ['$3']}]) of + [] -> not_found; + [Elem] -> Elem + end. + +created_stats(Type) -> + %% TODO better tab2list? + ets:select(Type, [{{'_', '_', '$3'}, [], ['$3']}]). + +-spec all_detail_queue_data(pid(), [any()], ranges()) -> #{atom() => any()}. +all_detail_queue_data(_Pid, Ids, Ranges) -> + lists:foldl(fun (Id, Acc) -> + Data = detail_queue_data(Ranges, Id), + maps:put(Id, Data, Acc) + end, #{}, Ids). + +all_list_queue_data(_Pid, Ids, Ranges) -> + lists:foldl(fun (Id, Acc) -> + Data = list_queue_data(Ranges, Id), + maps:put(Id, Data, Acc) + end, #{}, Ids). + +all_detail_channel_data(_Pid, Ids, Ranges) -> + lists:foldl(fun (Id, Acc) -> + Data = detail_channel_data(Ranges, Id), + maps:put(Id, Data, Acc) + end, #{}, Ids). + +all_list_channel_data(_Pid, Ids, Ranges) -> + lists:foldl(fun (Id, Acc) -> + Data = list_channel_data(Ranges, Id), + maps:put(Id, Data, Acc) + end, #{}, Ids). + +connection_data(Ranges, Id) -> + maps:from_list([raw_message_data(connection_stats_coarse_conn_stats, + pick_range(coarse_conn_stats, Ranges), Id), + {connection_stats, lookup_element(connection_stats, Id)}]). + +exchange_data(Ranges, Id) -> + maps:from_list( + exchange_raw_detail_stats_data(Ranges, Id) ++ + [raw_message_data(exchange_stats_publish_out, + pick_range(fine_stats, Ranges), Id), + raw_message_data(exchange_stats_publish_in, + pick_range(fine_stats, Ranges), Id)]). + +vhost_data(Ranges, Id) -> + maps:from_list([raw_message_data(vhost_stats_coarse_conn_stats, + pick_range(coarse_conn_stats, Ranges), Id), + raw_message_data(vhost_msg_stats, + pick_range(queue_msg_rates, Ranges), Id), + raw_message_data(vhost_stats_fine_stats, + pick_range(fine_stats, Ranges), Id), + raw_message_data(vhost_stats_deliver_stats, + pick_range(deliver_get, Ranges), Id)]). + +node_data(Ranges, Id) -> + maps:from_list( + [{mgmt_stats, mgmt_queue_length_stats(Id)}] ++ + [{node_node_metrics, node_node_metrics()}] ++ + node_raw_detail_stats_data(Ranges, Id) ++ + [raw_message_data(node_coarse_stats, + pick_range(coarse_node_stats, Ranges), Id), + raw_message_data(node_persister_stats, + pick_range(coarse_node_stats, Ranges), Id), + {node_stats, lookup_element(node_stats, Id)}] ++ + node_connection_churn_rates_data(Ranges, Id)). + +overview_data(_Pid, User, Ranges, VHosts) -> + Raw = [raw_all_message_data(vhost_msg_stats, pick_range(queue_msg_counts, Ranges), VHosts), + raw_all_message_data(vhost_stats_fine_stats, pick_range(fine_stats, Ranges), VHosts), + raw_all_message_data(vhost_msg_rates, pick_range(queue_msg_rates, Ranges), VHosts), + raw_all_message_data(vhost_stats_deliver_stats, pick_range(deliver_get, Ranges), VHosts), + raw_message_data(connection_churn_rates, pick_range(queue_msg_rates, Ranges), node())], + maps:from_list(Raw ++ + [{connections_count, count_created_stats(connection_created_stats, User)}, + {channels_count, count_created_stats(channel_created_stats, User)}, + {consumers_count, ets:info(consumer_stats, size)}]). + +consumer_data(_Pid, VHost) -> + maps:from_list( + [{C, augment_msg_stats(augment_consumer(C))} + || C <- consumers_by_vhost(VHost)]). + +all_connection_data(_Pid, Ids, Ranges) -> + maps:from_list([{Id, connection_data(Ranges, Id)} || Id <- Ids]). + +all_exchange_data(_Pid, Ids, Ranges) -> + maps:from_list([{Id, exchange_data(Ranges, Id)} || Id <- Ids]). + +all_vhost_data(_Pid, Ids, Ranges) -> + maps:from_list([{Id, vhost_data(Ranges, Id)} || Id <- Ids]). + +all_node_data(_Pid, Ids, Ranges) -> + maps:from_list([{Id, node_data(Ranges, Id)} || Id <- Ids]). + +channel_raw_message_data(Ranges, Id) -> + [raw_message_data(channel_stats_fine_stats, pick_range(fine_stats, Ranges), Id), + raw_message_data(channel_stats_deliver_stats, pick_range(deliver_get, Ranges), Id), + raw_message_data(channel_process_stats, pick_range(process_stats, Ranges), Id)]. + +queue_raw_message_data(Ranges, Id) -> + [raw_message_data(queue_stats_publish, pick_range(fine_stats, Ranges), Id), + raw_message_data(queue_stats_deliver_stats, pick_range(deliver_get, Ranges), Id), + raw_message_data(queue_process_stats, pick_range(process_stats, Ranges), Id), + raw_message_data(queue_msg_stats, pick_range(queue_msg_counts, Ranges), Id)]. + +queue_raw_deliver_stats_data(Ranges, Id) -> + [raw_message_data2(channel_queue_stats_deliver_stats, + pick_range(deliver_get, Ranges), Key) + || Key <- get_table_keys(channel_queue_stats_deliver_stats, second(Id))] ++ + [raw_message_data2(queue_exchange_stats_publish, + pick_range(fine_stats, Ranges), Key) + || Key <- get_table_keys(queue_exchange_stats_publish, first(Id))]. + +node_raw_detail_stats_data(Ranges, Id) -> + [raw_message_data2(node_node_coarse_stats, + pick_range(coarse_node_node_stats, Ranges), Key) + || Key <- get_table_keys(node_node_coarse_stats, first(Id))]. + +node_connection_churn_rates_data(Ranges, Id) -> + [raw_message_data(connection_churn_rates, + pick_range(churn_rates, Ranges), Id)]. + +exchange_raw_detail_stats_data(Ranges, Id) -> + [raw_message_data2(channel_exchange_stats_fine_stats, + pick_range(fine_stats, Ranges), Key) + || Key <- get_table_keys(channel_exchange_stats_fine_stats, second(Id))] ++ + [raw_message_data2(queue_exchange_stats_publish, + pick_range(fine_stats, Ranges), Key) + || Key <- get_table_keys(queue_exchange_stats_publish, second(Id))]. + +channel_raw_detail_stats_data(Ranges, Id) -> + [raw_message_data2(channel_exchange_stats_fine_stats, + pick_range(fine_stats, Ranges), Key) + || Key <- get_table_keys(channel_exchange_stats_fine_stats, first(Id))] ++ + [raw_message_data2(channel_queue_stats_deliver_stats, + pick_range(fine_stats, Ranges), Key) + || Key <- get_table_keys(channel_queue_stats_deliver_stats, first(Id))]. + +raw_message_data2(Table, no_range, Id) -> + SmallSample = lookup_smaller_sample(Table, Id), + {{Table, Id}, {SmallSample, not_found}}; +raw_message_data2(Table, Range, Id) -> + SmallSample = lookup_smaller_sample(Table, Id), + Samples = lookup_samples(Table, Id, Range), + {{Table, Id}, {SmallSample, Samples}}. + +detail_queue_data(Ranges, Id) -> + maps:from_list(queue_raw_message_data(Ranges, Id) ++ + queue_raw_deliver_stats_data(Ranges, Id) ++ + [{queue_stats, lookup_element(queue_stats, Id)}, + {consumer_stats, get_queue_consumer_stats(Id)}]). + +list_queue_data(Ranges, Id) -> + maps:from_list(queue_raw_message_data(Ranges, Id) ++ + queue_raw_deliver_stats_data(Ranges, Id) ++ + [{queue_stats, lookup_element(queue_stats, Id)}]). + +detail_channel_data(Ranges, Id) -> + maps:from_list(channel_raw_message_data(Ranges, Id) ++ + channel_raw_detail_stats_data(Ranges, Id) ++ + [{channel_stats, lookup_element(channel_stats, Id)}, + {consumer_stats, get_consumer_stats(Id)}]). + +list_channel_data(Ranges, Id) -> + maps:from_list(channel_raw_message_data(Ranges, Id) ++ + channel_raw_detail_stats_data(Ranges, Id) ++ + [{channel_stats, lookup_element(channel_stats, Id)}]). + +-spec raw_message_data(atom(), maybe_range(), any()) -> + {atom(), {maybe_slide(), maybe_slide()}}. +raw_message_data(Table, no_range, Id) -> + SmallSample = lookup_smaller_sample(Table, Id), + {Table, {SmallSample, not_found}}; +raw_message_data(Table, Range, Id) -> + SmallSample = lookup_smaller_sample(Table, Id), + Samples = lookup_samples(Table, Id, Range), + {Table, {SmallSample, Samples}}. + +raw_all_message_data(Table, Range, VHosts) -> + SmallSample = lookup_all(Table, VHosts, select_smaller_sample(Table)), + RangeSample = case Range of + no_range -> not_found; + _ -> + lookup_all(Table, VHosts, select_range_sample(Table, + Range)) + end, + {Table, {SmallSample, RangeSample}}. + +get_queue_consumer_stats(Id) -> + Consumers = ets:select(consumer_stats, match_queue_consumer_spec(Id)), + [augment_consumer(C) || C <- Consumers]. + +get_consumer_stats(Id) -> + Consumers = ets:select(consumer_stats, match_consumer_spec(Id)), + [augment_consumer(C) || C <- Consumers]. + +count_created_stats(Type, all) -> + ets:info(Type, size); +count_created_stats(Type, User) -> + length(filter_user(created_stats(Type), User)). + +augment_consumer({{Q, Ch, CTag}, Props}) -> + [{queue, format_resource(Q)}, + {channel_details, augment_channel_pid(Ch)}, + {channel_pid, Ch}, + {consumer_tag, CTag} | Props]. + +consumers_by_vhost(VHost) -> + ets:select(consumer_stats, + [{{{#resource{virtual_host = '$1', _ = '_'}, '_', '_'}, '_'}, + [{'orelse', {'==', 'all', VHost}, {'==', VHost, '$1'}}], + ['$_']}]). + +augment_msg_stats(Props) -> + augment_details(Props, []) ++ Props. + +augment_details([{_, none} | T], Acc) -> + augment_details(T, Acc); +augment_details([{_, unknown} | T], Acc) -> + augment_details(T, Acc); +augment_details([{connection, Value} | T], Acc) -> + augment_details(T, [{connection_details, augment_connection_pid(Value)} | Acc]); +augment_details([{channel, Value} | T], Acc) -> + augment_details(T, [{channel_details, augment_channel_pid(Value)} | Acc]); +augment_details([{owner_pid, Value} | T], Acc) -> + augment_details(T, [{owner_pid_details, augment_connection_pid(Value)} | Acc]); +augment_details([_ | T], Acc) -> + augment_details(T, Acc); +augment_details([], Acc) -> + Acc. + +augment_channel_pids(_Pid, ChPids) -> + lists:map(fun (ChPid) -> augment_channel_pid(ChPid) end, ChPids). + +augment_channel_pid(Pid) -> + Ch = lookup_element(channel_created_stats, Pid, 3), + Conn = lookup_element(connection_created_stats, pget(connection, Ch), 3), + case Conn of + [] -> %% If the connection has just been opened, we might not yet have the data + []; + _ -> + [{name, pget(name, Ch)}, + {pid, pget(pid, Ch)}, + {number, pget(number, Ch)}, + {user, pget(user, Ch)}, + {connection_name, pget(name, Conn)}, + {peer_port, pget(peer_port, Conn)}, + {peer_host, pget(peer_host, Conn)}] + end. + +augment_connection_pid(Pid) -> + Conn = lookup_element(connection_created_stats, Pid, 3), + case Conn of + [] -> %% If the connection has just been opened, we might not yet have the data + []; + _ -> + [{name, pget(name, Conn)}, + {peer_port, pget(peer_port, Conn)}, + {peer_host, pget(peer_host, Conn)}] + end. + +augmented_created_stats(_Pid, Key, Type) -> + case created_stats(Key, Type) of + not_found -> not_found; + S -> augment_msg_stats(S) + end. + +augmented_created_stats(_Pid, Type) -> + [ augment_msg_stats(S) || S <- created_stats(Type) ]. + +match_consumer_spec(Id) -> + [{{{'_', '$1', '_'}, '_'}, [{'==', Id, '$1'}], ['$_']}]. + +match_queue_consumer_spec(Id) -> + [{{{'$1', '_', '_'}, '_'}, [{'==', {Id}, '$1'}], ['$_']}]. + +lookup_element(Table, Key) -> lookup_element(Table, Key, 2). + +lookup_element(Table, Key, Pos) -> + try ets:lookup_element(Table, Key, Pos) + catch error:badarg -> [] + end. + +-spec lookup_smaller_sample(atom(), any()) -> maybe_slide(). +lookup_smaller_sample(Table, Id) -> + case ets:lookup(Table, {Id, select_smaller_sample(Table)}) of + [] -> + not_found; + [{_, Slide}] -> + Slide1 = exometer_slide:optimize(Slide), + maybe_convert_for_compatibility(Table, Slide1) + end. + +-spec lookup_samples(atom(), any(), #range{}) -> maybe_slide(). +lookup_samples(Table, Id, Range) -> + case ets:lookup(Table, {Id, select_range_sample(Table, Range)}) of + [] -> + not_found; + [{_, Slide}] -> + Slide1 = exometer_slide:optimize(Slide), + maybe_convert_for_compatibility(Table, Slide1) + end. + +lookup_all(Table, Ids, SecondKey) -> + Slides = lists:foldl(fun(Id, Acc) -> + case ets:lookup(Table, {Id, SecondKey}) of + [] -> + Acc; + [{_, Slide}] -> + [Slide | Acc] + end + end, [], Ids), + case Slides of + [] -> + not_found; + _ -> + Slide = exometer_slide:sum(Slides, empty(Table, 0)), + maybe_convert_for_compatibility(Table, Slide) + end. + +maybe_convert_for_compatibility(Table, Slide) + when Table =:= channel_stats_fine_stats orelse + Table =:= channel_exchange_stats_fine_stats orelse + Table =:= vhost_stats_fine_stats -> + ConversionNeeded = rabbit_feature_flags:is_disabled( + drop_unroutable_metric), + case ConversionNeeded of + false -> + Slide; + true -> + %% drop_drop because the metric is named "drop_unroutable" + rabbit_mgmt_data_compat:drop_drop_unroutable_metric(Slide) + end; +maybe_convert_for_compatibility(Table, Slide) + when Table =:= channel_queue_stats_deliver_stats orelse + Table =:= channel_stats_deliver_stats orelse + Table =:= queue_stats_deliver_stats orelse + Table =:= vhost_stats_deliver_stats -> + ConversionNeeded = rabbit_feature_flags:is_disabled( + empty_basic_get_metric), + case ConversionNeeded of + false -> + Slide; + true -> + rabbit_mgmt_data_compat:drop_get_empty_queue_metric(Slide) + end; +maybe_convert_for_compatibility(_, Slide) -> + Slide. + +get_table_keys(Table, Id0) -> + ets:select(Table, match_spec_keys(Id0)). + +match_spec_keys(Id) -> + MatchCondition = to_match_condition(Id), + MatchHead = {{{'$1', '$2'}, '_'}, '_'}, + [{MatchHead, [MatchCondition], [{{'$1', '$2'}}]}]. + +to_match_condition({'_', Id1}) when is_tuple(Id1) -> + {'==', {Id1}, '$2'}; +to_match_condition({'_', Id1}) -> + {'==', Id1, '$2'}; +to_match_condition({Id0, '_'}) when is_tuple(Id0) -> + {'==', {Id0}, '$1'}; +to_match_condition({Id0, '_'}) -> + {'==', Id0, '$1'}. + +mgmt_queue_length_stats(Id) when Id =:= node() -> + GCsQueueLengths = lists:map(fun (T) -> + case whereis(rabbit_mgmt_metrics_gc:name(T)) of + P when is_pid(P) -> + {message_queue_len, Len} = + erlang:process_info(P, message_queue_len), + {T, Len}; + _ -> {T, 0} + end + end, + ?GC_EVENTS), + [{metrics_gc_queue_length, GCsQueueLengths}]; +mgmt_queue_length_stats(_Id) -> + % if it isn't for the current node just return an empty list + []. + +node_node_metrics() -> + maps:from_list(ets:tab2list(node_node_metrics)). + +select_range_sample(Table, #range{first = First, last = Last}) -> + Range = Last - First, + Policies = rabbit_mgmt_agent_config:get_env(sample_retention_policies), + Policy = retention_policy(Table), + [T | _] = TablePolicies = lists:sort(proplists:get_value(Policy, Policies)), + {_, Sample} = select_smallest_above(T, TablePolicies, Range), + Sample. + +select_smaller_sample(Table) -> + Policies = rabbit_mgmt_agent_config:get_env(sample_retention_policies), + Policy = retention_policy(Table), + TablePolicies = proplists:get_value(Policy, Policies), + [V | _] = lists:sort([I || {_, I} <- TablePolicies]), + V. + +select_smallest_above(V, [], _) -> + V; +select_smallest_above(_, [{H, _} = S | _T], Interval) when (H * 1000) > Interval -> + S; +select_smallest_above(_, [H | T], Interval) -> + select_smallest_above(H, T, Interval). + +pick_range(queue_msg_counts, {RangeL, _RangeM, _RangeD, _RangeN}) -> + RangeL; +pick_range(K, {_RangeL, RangeM, _RangeD, _RangeN}) when K == fine_stats; + K == deliver_get; + K == queue_msg_rates -> + RangeM; +pick_range(K, {_RangeL, _RangeM, RangeD, _RangeN}) when K == coarse_conn_stats; + K == process_stats -> + RangeD; +pick_range(K, {_RangeL, _RangeM, _RangeD, RangeN}) + when K == coarse_node_stats; + K == coarse_node_node_stats; + K == churn_rates -> + RangeN. + +first(Id) -> + {Id, '_'}. + +second(Id) -> + {'_', Id}. + +empty(Type, V) when Type =:= connection_stats_coarse_conn_stats; + Type =:= queue_msg_stats; + Type =:= vhost_msg_stats -> + {V, V, V}; +empty(Type, V) when Type =:= channel_stats_fine_stats; + Type =:= channel_exchange_stats_fine_stats; + Type =:= vhost_stats_fine_stats -> + {V, V, V, V}; +empty(Type, V) when Type =:= channel_queue_stats_deliver_stats; + Type =:= queue_stats_deliver_stats; + Type =:= vhost_stats_deliver_stats; + Type =:= channel_stats_deliver_stats -> + {V, V, V, V, V, V, V, V}; +empty(Type, V) when Type =:= channel_process_stats; + Type =:= queue_process_stats; + Type =:= queue_stats_publish; + Type =:= queue_exchange_stats_publish; + Type =:= exchange_stats_publish_out; + Type =:= exchange_stats_publish_in -> + {V}; +empty(node_coarse_stats, V) -> + {V, V, V, V, V, V, V, V}; +empty(node_persister_stats, V) -> + {V, V, V, V, V, V, V, V, V, V, V, V, V, V, V, V, V, V, V, V}; +empty(Type, V) when Type =:= node_node_coarse_stats; + Type =:= vhost_stats_coarse_conn_stats; + Type =:= queue_msg_rates; + Type =:= vhost_msg_rates -> + {V, V}; +empty(connection_churn_rates, V) -> + {V, V, V, V, V, V, V}. + +retention_policy(connection_stats_coarse_conn_stats) -> + basic; +retention_policy(channel_stats_fine_stats) -> + basic; +retention_policy(channel_queue_stats_deliver_stats) -> + detailed; +retention_policy(channel_exchange_stats_fine_stats) -> + detailed; +retention_policy(channel_process_stats) -> + basic; +retention_policy(vhost_stats_fine_stats) -> + global; +retention_policy(vhost_stats_deliver_stats) -> + global; +retention_policy(vhost_stats_coarse_conn_stats) -> + global; +retention_policy(vhost_msg_rates) -> + global; +retention_policy(channel_stats_deliver_stats) -> + basic; +retention_policy(queue_stats_deliver_stats) -> + basic; +retention_policy(queue_stats_publish) -> + basic; +retention_policy(queue_exchange_stats_publish) -> + basic; +retention_policy(exchange_stats_publish_out) -> + basic; +retention_policy(exchange_stats_publish_in) -> + basic; +retention_policy(queue_process_stats) -> + basic; +retention_policy(queue_msg_stats) -> + basic; +retention_policy(queue_msg_rates) -> + basic; +retention_policy(vhost_msg_stats) -> + global; +retention_policy(node_coarse_stats) -> + global; +retention_policy(node_persister_stats) -> + global; +retention_policy(node_node_coarse_stats) -> + global; +retention_policy(connection_churn_rates) -> + global. + +format_resource(unknown) -> unknown; +format_resource(Res) -> format_resource(name, Res). + +format_resource(_, unknown) -> + unknown; +format_resource(NameAs, #resource{name = Name, virtual_host = VHost}) -> + [{NameAs, Name}, {vhost, VHost}]. + +filter_user(List, #user{username = Username, tags = Tags}) -> + case is_monitor(Tags) of + true -> List; + false -> [I || I <- List, pget(user, I) == Username] + end. + +is_monitor(T) -> intersects(T, [administrator, monitoring]). +intersects(A, B) -> lists:any(fun(I) -> lists:member(I, B) end, A). diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_data_compat.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_data_compat.erl new file mode 100644 index 0000000000..9fd127aff5 --- /dev/null +++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_data_compat.erl @@ -0,0 +1,80 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2018-2020 VMware, Inc. or its affiliates. All rights reserved. +%% + +-module(rabbit_mgmt_data_compat). + +-export([fill_get_empty_queue_metric/1, + drop_get_empty_queue_metric/1, + fill_consumer_active_fields/1, + fill_drop_unroutable_metric/1, + drop_drop_unroutable_metric/1]). + +fill_get_empty_queue_metric(Slide) -> + exometer_slide:map( + fun + (Value) when is_tuple(Value) andalso size(Value) =:= 8 -> + Value; + (Value) when is_tuple(Value) andalso size(Value) =:= 7 -> + %% Inject a 0 for the new metric + list_to_tuple( + tuple_to_list(Value) ++ [0]); + (Value) -> + Value + end, Slide). + +drop_get_empty_queue_metric(Slide) -> + exometer_slide:map( + fun + (Value) when is_tuple(Value) andalso size(Value) =:= 8 -> + %% We want to remove the last element, which is + %% the count of basic.get on empty queues. + list_to_tuple( + lists:sublist( + tuple_to_list(Value), size(Value) - 1)); + (Value) when is_tuple(Value) andalso size(Value) =:= 7 -> + Value; + (Value) -> + Value + end, Slide). + +fill_drop_unroutable_metric(Slide) -> + exometer_slide:map( + fun + (Value) when is_tuple(Value) andalso size(Value) =:= 4 -> + Value; + (Value) when is_tuple(Value) andalso size(Value) =:= 3 -> + %% Inject a 0 + list_to_tuple( + tuple_to_list(Value) ++ [0]); + (Value) -> + Value + end, Slide). + +drop_drop_unroutable_metric(Slide) -> + exometer_slide:map( + fun + (Value) when is_tuple(Value) andalso size(Value) =:= 4 -> + %% Remove the last element. + list_to_tuple( + lists:sublist( + tuple_to_list(Value), size(Value) - 1)); + (Value) when is_tuple(Value) andalso size(Value) =:= 3 -> + Value; + (Value) -> + Value + end, Slide). + +fill_consumer_active_fields(ConsumersStats) -> + [case proplists:get_value(active, ConsumerStats) of + undefined -> + [{active, true}, + {activity_status, up} + | ConsumerStats]; + _ -> + ConsumerStats + end + || ConsumerStats <- ConsumersStats]. diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_db_handler.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_db_handler.erl new file mode 100644 index 0000000000..c1e43223d7 --- /dev/null +++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_db_handler.erl @@ -0,0 +1,99 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates. All rights reserved. +%% + +-module(rabbit_mgmt_db_handler). + +-include_lib("rabbit_common/include/rabbit.hrl"). + +%% Make sure our database is hooked in *before* listening on the network or +%% recovering queues (i.e. so there can't be any events fired before it starts). +-rabbit_boot_step({rabbit_mgmt_db_handler, + [{description, "management agent"}, + {mfa, {?MODULE, add_handler, []}}, + {cleanup, {gen_event, delete_handler, + [rabbit_event, ?MODULE, []]}}, + {requires, rabbit_event}, + {enables, recovery}]}). + +-behaviour(gen_event). + +-export([add_handler/0, gc/0, rates_mode/0]). + +-export([init/1, handle_call/2, handle_event/2, handle_info/2, + terminate/2, code_change/3]). + +%%---------------------------------------------------------------------------- + +add_handler() -> + ok = ensure_statistics_enabled(), + gen_event:add_handler(rabbit_event, ?MODULE, []). + +gc() -> + erlang:garbage_collect(whereis(rabbit_event)). + +rates_mode() -> + case rabbit_mgmt_agent_config:get_env(rates_mode) of + undefined -> basic; + Mode -> Mode + end. + +handle_force_fine_statistics() -> + case rabbit_mgmt_agent_config:get_env(force_fine_statistics) of + undefined -> + ok; + X -> + rabbit_log:warning( + "force_fine_statistics set to ~p; ignored.~n" + "Replaced by {rates_mode, none} in the rabbitmq_management " + "application.~n", [X]) + end. + +%%---------------------------------------------------------------------------- + +ensure_statistics_enabled() -> + ForceStats = rates_mode() =/= none, + handle_force_fine_statistics(), + {ok, StatsLevel} = application:get_env(rabbit, collect_statistics), + rabbit_log:info("Management plugin: using rates mode '~p'~n", [rates_mode()]), + case {ForceStats, StatsLevel} of + {true, fine} -> + ok; + {true, _} -> + application:set_env(rabbit, collect_statistics, fine); + {false, none} -> + application:set_env(rabbit, collect_statistics, coarse); + {_, _} -> + ok + end, + ok = rabbit:force_event_refresh(erlang:make_ref()). + +%%---------------------------------------------------------------------------- + +init([]) -> + {ok, []}. + +handle_call(_Request, State) -> + {ok, not_understood, State}. + +handle_event(#event{type = Type} = Event, State) + when Type == connection_closed; Type == channel_closed; Type == queue_deleted; + Type == exchange_deleted; Type == vhost_deleted; + Type == consumer_deleted; Type == node_node_deleted; + Type == channel_consumer_deleted -> + gen_server:cast(rabbit_mgmt_metrics_gc:name(Type), {event, Event}), + {ok, State}; +handle_event(_, State) -> + {ok, State}. + +handle_info(_Info, State) -> + {ok, State}. + +terminate(_Arg, _State) -> + ok. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_external_stats.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_external_stats.erl new file mode 100644 index 0000000000..5e92d8394c --- /dev/null +++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_external_stats.erl @@ -0,0 +1,501 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates. All rights reserved. +%% + +-module(rabbit_mgmt_external_stats). + +%% Transitional step until we can require Erlang/OTP 21 and +%% use the now recommended try/catch syntax for obtaining the stack trace. +-compile(nowarn_deprecated_function). + +-behaviour(gen_server). + +-export([start_link/0]). +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, + code_change/3]). + +-export([list_registry_plugins/1]). + +-import(rabbit_misc, [pget/2]). + +-include_lib("rabbit_common/include/rabbit.hrl"). + +-define(METRICS_KEYS, [fd_used, sockets_used, mem_used, disk_free, proc_used, gc_num, + gc_bytes_reclaimed, context_switches]). + +-define(PERSISTER_KEYS, [persister_stats]). + +-define(OTHER_KEYS, [name, partitions, os_pid, fd_total, sockets_total, mem_limit, + mem_alarm, disk_free_limit, disk_free_alarm, proc_total, + rates_mode, uptime, run_queue, processors, exchange_types, + auth_mechanisms, applications, contexts, log_files, + db_dir, config_files, net_ticktime, enabled_plugins, + mem_calculation_strategy, ra_open_file_metrics]). + +-define(TEN_MINUTES_AS_SECONDS, 600). + +%%-------------------------------------------------------------------- + +-record(state, { + fd_total, + fhc_stats, + node_owners, + last_ts, + interval, + error_logged_time +}). + +%%-------------------------------------------------------------------- + +start_link() -> + gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). + +%%-------------------------------------------------------------------- + +get_used_fd(State0) -> + try + case get_used_fd(os:type(), State0) of + {State1, UsedFd} when is_number(UsedFd) -> + {State1, UsedFd}; + {State1, _Other} -> + %% Defaults to 0 if data is not available + {State1, 0} + end + catch + _:Error -> + State2 = log_fd_error("Could not infer the number of file handles used: ~p~n", [Error], State0), + {State2, 0} + end. + +get_used_fd({unix, linux}, State0) -> + case file:list_dir("/proc/" ++ os:getpid() ++ "/fd") of + {ok, Files} -> + {State0, length(Files)}; + {error, _} -> + get_used_fd({unix, generic}, State0) + end; + +get_used_fd({unix, BSD}, State0) + when BSD == openbsd; BSD == freebsd; BSD == netbsd -> + IsDigit = fun (D) -> lists:member(D, "0123456789*") end, + Output = os:cmd("fstat -p " ++ os:getpid()), + try + F = fun (Line) -> + lists:all(IsDigit, lists:nth(4, string:tokens(Line, " "))) + end, + UsedFd = length(lists:filter(F, string:tokens(Output, "\n"))), + {State0, UsedFd} + catch _:Error:Stacktrace -> + State1 = log_fd_error("Could not parse fstat output:~n~s~n~p~n", + [Output, {Error, Stacktrace}], State0), + {State1, 0} + end; + +get_used_fd({unix, _}, State0) -> + Cmd = rabbit_misc:format( + "lsof -d \"0-9999999\" -lna -p ~s || echo failed", [os:getpid()]), + Res = os:cmd(Cmd), + case string:right(Res, 7) of + "failed\n" -> + State1 = log_fd_error("Could not obtain lsof output~n", [], State0), + {State1, 0}; + _ -> + UsedFd = string:words(Res, $\n) - 1, + {State0, UsedFd} + end; + +%% handle.exe can be obtained from +%% https://technet.microsoft.com/en-us/sysinternals/bb896655.aspx + +%% Output looks like: + +%% Handle v3.42 +%% Copyright (C) 1997-2008 Mark Russinovich +%% Sysinternals - www.sysinternals.com +%% +%% Handle type summary: +%% ALPC Port : 2 +%% Desktop : 1 +%% Directory : 1 +%% Event : 108 +%% File : 25 +%% IoCompletion : 3 +%% Key : 7 +%% KeyedEvent : 1 +%% Mutant : 1 +%% Process : 3 +%% Process : 38 +%% Thread : 41 +%% Timer : 3 +%% TpWorkerFactory : 2 +%% WindowStation : 2 +%% Total handles: 238 + +%% Nthandle v4.22 - Handle viewer +%% Copyright (C) 1997-2019 Mark Russinovich +%% Sysinternals - www.sysinternals.com +%% +%% Handle type summary: +%% <Unknown type> : 1 +%% <Unknown type> : 166 +%% ALPC Port : 11 +%% Desktop : 1 +%% Directory : 2 +%% Event : 226 +%% File : 122 +%% IoCompletion : 8 +%% IRTimer : 6 +%% Key : 42 +%% Mutant : 7 +%% Process : 3 +%% Section : 2 +%% Semaphore : 43 +%% Thread : 36 +%% TpWorkerFactory : 3 +%% WaitCompletionPacket: 25 +%% WindowStation : 2 +%% Total handles: 706 + +%% Note that the "File" number appears to include network sockets too; I assume +%% that's the number we care about. Note also that if you omit "-s" you will +%% see a list of file handles *without* network sockets. If you then add "-a" +%% you will see a list of handles of various types, including network sockets +%% shown as file handles to \Device\Afd. + +get_used_fd({win32, _}, State0) -> + Handle = rabbit_misc:os_cmd( + "handle.exe /accepteula -s -p " ++ os:getpid() ++ " 2> nul"), + case Handle of + [] -> + State1 = log_fd_error("Could not find handle.exe, please install from sysinternals~n", [], State0), + {State1, 0}; + _ -> + case find_files_line(string:tokens(Handle, "\r\n")) of + unknown -> + State1 = log_fd_error("handle.exe output did not contain " + "a line beginning with ' File ', unable " + "to determine used file descriptor " + "count: ~p~n", [Handle], State0), + {State1, 0}; + UsedFd -> + {State0, UsedFd} + end + end. + +find_files_line([]) -> + unknown; +find_files_line([" File " ++ Rest | _T]) -> + [Files] = string:tokens(Rest, ": "), + list_to_integer(Files); +find_files_line([_H | T]) -> + find_files_line(T). + +-define(SAFE_CALL(Fun, NoProcFailResult), + try + Fun + catch exit:{noproc, _} -> NoProcFailResult + end). + +get_disk_free_limit() -> ?SAFE_CALL(rabbit_disk_monitor:get_disk_free_limit(), + disk_free_monitoring_disabled). + +get_disk_free() -> ?SAFE_CALL(rabbit_disk_monitor:get_disk_free(), + disk_free_monitoring_disabled). + +log_fd_error(Fmt, Args, #state{error_logged_time = undefined}=State) -> + % rabbitmq/rabbitmq-management#90 + % no errors have been logged, so log it and make a note of when + Now = erlang:monotonic_time(second), + ok = rabbit_log:error(Fmt, Args), + State#state{error_logged_time = Now}; +log_fd_error(Fmt, Args, #state{error_logged_time = Time}=State) -> + Now = erlang:monotonic_time(second), + case Now >= Time + ?TEN_MINUTES_AS_SECONDS of + true -> + % rabbitmq/rabbitmq-management#90 + % it has been longer than 10 minutes, + % re-log the error + ok = rabbit_log:error(Fmt, Args), + State#state{error_logged_time = Now}; + _ -> + % 10 minutes have not yet passed + State + end. +%%-------------------------------------------------------------------- + +infos([], Acc, State) -> + {State, lists:reverse(Acc)}; +infos([Item|T], Acc0, State0) -> + {State1, Infos} = i(Item, State0), + Acc1 = [{Item, Infos}|Acc0], + infos(T, Acc1, State1). + +i(name, State) -> + {State, node()}; +i(partitions, State) -> + {State, rabbit_node_monitor:partitions()}; +i(fd_used, State) -> + get_used_fd(State); +i(fd_total, #state{fd_total = FdTotal}=State) -> + {State, FdTotal}; +i(sockets_used, State) -> + {State, proplists:get_value(sockets_used, file_handle_cache:info([sockets_used]))}; +i(sockets_total, State) -> + {State, proplists:get_value(sockets_limit, file_handle_cache:info([sockets_limit]))}; +i(os_pid, State) -> + {State, list_to_binary(os:getpid())}; +i(mem_used, State) -> + {State, vm_memory_monitor:get_process_memory()}; +i(mem_calculation_strategy, State) -> + {State, vm_memory_monitor:get_memory_calculation_strategy()}; +i(mem_limit, State) -> + {State, vm_memory_monitor:get_memory_limit()}; +i(mem_alarm, State) -> + {State, resource_alarm_set(memory)}; +i(proc_used, State) -> + {State, erlang:system_info(process_count)}; +i(proc_total, State) -> + {State, erlang:system_info(process_limit)}; +i(run_queue, State) -> + {State, erlang:statistics(run_queue)}; +i(processors, State) -> + {State, erlang:system_info(logical_processors)}; +i(disk_free_limit, State) -> + {State, get_disk_free_limit()}; +i(disk_free, State) -> + {State, get_disk_free()}; +i(disk_free_alarm, State) -> + {State, resource_alarm_set(disk)}; +i(contexts, State) -> + {State, rabbit_web_dispatch_contexts()}; +i(uptime, State) -> + {Total, _} = erlang:statistics(wall_clock), + {State, Total}; +i(rates_mode, State) -> + {State, rabbit_mgmt_db_handler:rates_mode()}; +i(exchange_types, State) -> + {State, list_registry_plugins(exchange)}; +i(log_files, State) -> + {State, [list_to_binary(F) || F <- rabbit:log_locations()]}; +i(db_dir, State) -> + {State, list_to_binary(rabbit_mnesia:dir())}; +i(config_files, State) -> + {State, [list_to_binary(F) || F <- rabbit:config_files()]}; +i(net_ticktime, State) -> + {State, net_kernel:get_net_ticktime()}; +i(persister_stats, State) -> + {State, persister_stats(State)}; +i(enabled_plugins, State) -> + {ok, Dir} = application:get_env(rabbit, enabled_plugins_file), + {State, rabbit_plugins:read_enabled(Dir)}; +i(auth_mechanisms, State) -> + {ok, Mechanisms} = application:get_env(rabbit, auth_mechanisms), + F = fun (N) -> + lists:member(list_to_atom(binary_to_list(N)), Mechanisms) + end, + {State, list_registry_plugins(auth_mechanism, F)}; +i(applications, State) -> + {State, [format_application(A) || A <- lists:keysort(1, rabbit_misc:which_applications())]}; +i(gc_num, State) -> + {GCs, _, _} = erlang:statistics(garbage_collection), + {State, GCs}; +i(gc_bytes_reclaimed, State) -> + {_, Words, _} = erlang:statistics(garbage_collection), + {State, Words * erlang:system_info(wordsize)}; +i(context_switches, State) -> + {Sw, 0} = erlang:statistics(context_switches), + {State, Sw}; +i(ra_open_file_metrics, State) -> + {State, [{ra_log_wal, ra_metrics(ra_log_wal)}, + {ra_log_segment_writer, ra_metrics(ra_log_segment_writer)}]}. + +ra_metrics(K) -> + try + case ets:lookup(ra_open_file_metrics, whereis(K)) of + [] -> 0; + [{_, C}] -> C + end + catch + error:badarg -> + %% On startup the mgmt might start before ra does + 0 + end. + +resource_alarm_set(Source) -> + lists:member({{resource_limit, Source, node()},[]}, + rabbit_alarm:get_alarms()). + +list_registry_plugins(Type) -> + list_registry_plugins(Type, fun(_) -> true end). + +list_registry_plugins(Type, Fun) -> + [registry_plugin_enabled(set_plugin_name(Name, Module), Fun) || + {Name, Module} <- rabbit_registry:lookup_all(Type)]. + +registry_plugin_enabled(Desc, Fun) -> + Desc ++ [{enabled, Fun(proplists:get_value(name, Desc))}]. + +format_application({Application, Description, Version}) -> + [{name, Application}, + {description, list_to_binary(Description)}, + {version, list_to_binary(Version)}]. + +set_plugin_name(Name, Module) -> + [{name, list_to_binary(atom_to_list(Name))} | + proplists:delete(name, Module:description())]. + +persister_stats(#state{fhc_stats = FHC}) -> + [{flatten_key(K), V} || {{_Op, _Type} = K, V} <- FHC]. + +flatten_key({A, B}) -> + list_to_atom(atom_to_list(A) ++ "_" ++ atom_to_list(B)). + +cluster_links() -> + {ok, Items} = net_kernel:nodes_info(), + [Link || Item <- Items, + Link <- [format_nodes_info(Item)], Link =/= undefined]. + +format_nodes_info({Node, Info}) -> + Owner = proplists:get_value(owner, Info), + case catch process_info(Owner, links) of + {links, Links} -> + case [Link || Link <- Links, is_port(Link)] of + [Port] -> + {Node, Owner, format_nodes_info1(Port)}; + _ -> + undefined + end; + _ -> + undefined + end. + +format_nodes_info1(Port) -> + case {rabbit_net:socket_ends(Port, inbound), + rabbit_net:getstat(Port, [recv_oct, send_oct])} of + {{ok, {PeerAddr, PeerPort, SockAddr, SockPort}}, {ok, Stats}} -> + [{peer_addr, maybe_ntoab(PeerAddr)}, + {peer_port, PeerPort}, + {sock_addr, maybe_ntoab(SockAddr)}, + {sock_port, SockPort}, + {recv_bytes, pget(recv_oct, Stats)}, + {send_bytes, pget(send_oct, Stats)}]; + _ -> + [] + end. + +maybe_ntoab(A) when is_tuple(A) -> list_to_binary(rabbit_misc:ntoab(A)); +maybe_ntoab(H) -> H. + +%%-------------------------------------------------------------------- + +%% This is slightly icky in that we introduce knowledge of +%% rabbit_web_dispatch, which is not a dependency. But the last thing I +%% want to do is create a rabbitmq_mochiweb_management_agent plugin. +rabbit_web_dispatch_contexts() -> + [format_context(C) || C <- rabbit_web_dispatch_registry_list_all()]. + +%% For similar reasons we don't declare a dependency on +%% rabbitmq_mochiweb - so at startup there's no guarantee it will be +%% running. So we have to catch this noproc. +rabbit_web_dispatch_registry_list_all() -> + case code:is_loaded(rabbit_web_dispatch_registry) of + false -> []; + _ -> try + M = rabbit_web_dispatch_registry, %% Fool xref + M:list_all() + catch exit:{noproc, _} -> + [] + end + end. + +format_context({Path, Description, Rest}) -> + [{description, list_to_binary(Description)}, + {path, list_to_binary("/" ++ Path)} | + format_mochiweb_option_list(Rest)]. + +format_mochiweb_option_list(C) -> + [{K, format_mochiweb_option(K, V)} || {K, V} <- C]. + +format_mochiweb_option(ssl_opts, V) -> + format_mochiweb_option_list(V); +format_mochiweb_option(_K, V) -> + case io_lib:printable_list(V) of + true -> list_to_binary(V); + false -> list_to_binary(rabbit_misc:format("~w", [V])) + end. + +%%-------------------------------------------------------------------- + +init([]) -> + {ok, Interval} = application:get_env(rabbit, collect_statistics_interval), + State = #state{fd_total = file_handle_cache:ulimit(), + fhc_stats = get_fhc_stats(), + node_owners = sets:new(), + interval = Interval}, + %% We can update stats straight away as they need to be available + %% when the mgmt plugin starts a collector + {ok, emit_update(State)}. + +handle_call(_Req, _From, State) -> + {reply, unknown_request, State}. + +handle_cast(_C, State) -> + {noreply, State}. + +handle_info(emit_update, State) -> + {noreply, emit_update(State)}; + +handle_info(_I, State) -> + {noreply, State}. + +terminate(_, _) -> ok. + +code_change(_, State, _) -> {ok, State}. + +%%-------------------------------------------------------------------- + +emit_update(State0) -> + State1 = update_state(State0), + {State2, MStats} = infos(?METRICS_KEYS, [], State1), + {State3, PStats} = infos(?PERSISTER_KEYS, [], State2), + {State4, OStats} = infos(?OTHER_KEYS, [], State3), + [{persister_stats, PStats0}] = PStats, + [{name, _Name} | OStats0] = OStats, + rabbit_core_metrics:node_stats(persister_metrics, PStats0), + rabbit_core_metrics:node_stats(coarse_metrics, MStats), + rabbit_core_metrics:node_stats(node_metrics, OStats0), + rabbit_event:notify(node_stats, PStats ++ MStats ++ OStats), + erlang:send_after(State4#state.interval, self(), emit_update), + emit_node_node_stats(State4). + +emit_node_node_stats(State = #state{node_owners = Owners}) -> + Links = cluster_links(), + NewOwners = sets:from_list([{Node, Owner} || {Node, Owner, _} <- Links]), + Dead = sets:to_list(sets:subtract(Owners, NewOwners)), + [rabbit_event:notify( + node_node_deleted, [{route, Route}]) || {Node, _Owner} <- Dead, + Route <- [{node(), Node}, + {Node, node()}]], + [begin + rabbit_core_metrics:node_node_stats({node(), Node}, Stats), + rabbit_event:notify( + node_node_stats, [{route, {node(), Node}} | Stats]) + end || {Node, _Owner, Stats} <- Links], + State#state{node_owners = NewOwners}. + +update_state(State0) -> + %% Store raw data, the average operation time is calculated during querying + %% from the accumulated total + FHC = get_fhc_stats(), + State0#state{fhc_stats = FHC}. + +get_fhc_stats() -> + dict:to_list(dict:merge(fun(_, V1, V2) -> V1 + V2 end, + dict:from_list(file_handle_cache_stats:get()), + dict:from_list(get_ra_io_metrics()))). + +get_ra_io_metrics() -> + lists:sort(ets:tab2list(ra_io_metrics)). diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_ff.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_ff.erl new file mode 100644 index 0000000000..c8173c1244 --- /dev/null +++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_ff.erl @@ -0,0 +1,20 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2018-2020 VMware, Inc. or its affiliates. All rights reserved. +%% + +-module(rabbit_mgmt_ff). + +-rabbit_feature_flag( + {empty_basic_get_metric, + #{desc => "Count AMQP `basic.get` on empty queues in stats", + stability => stable + }}). + +-rabbit_feature_flag( + {drop_unroutable_metric, + #{desc => "Count unroutable publishes to be dropped in stats", + stability => stable + }}). diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_format.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_format.erl new file mode 100644 index 0000000000..4c9e8c189f --- /dev/null +++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_format.erl @@ -0,0 +1,559 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates. All rights reserved. +%% + +-module(rabbit_mgmt_format). + +-export([format/2, ip/1, ipb/1, amqp_table/1, tuple/1]). +-export([parameter/1, now_to_str/1, now_to_str_ms/1, strip_pids/1]). +-export([protocol/1, resource/1, queue/1, queue_state/1, queue_info/1]). +-export([exchange/1, user/1, internal_user/1, binding/1, url/2]). +-export([pack_binding_props/2, tokenise/1]). +-export([to_amqp_table/1, listener/1, web_context/1, properties/1, basic_properties/1]). +-export([record/2, to_basic_properties/1]). +-export([addr/1, port/1]). +-export([format_nulls/1, escape_html_tags/1]). +-export([print/2, print/1]). + +-export([format_queue_stats/1, format_channel_stats/1, + format_consumer_arguments/1, + format_connection_created/1, + format_accept_content/1, format_args/1]). + +-export([strip_queue_pids/1]). + +-export([clean_consumer_details/1, clean_channel_details/1]). + +-export([args_hash/1]). + +-import(rabbit_misc, [pget/2, pget/3, pset/3]). + +-include_lib("rabbit_common/include/rabbit.hrl"). +-include_lib("rabbit_common/include/rabbit_framing.hrl"). +-include_lib("rabbit/include/amqqueue.hrl"). + +%%-------------------------------------------------------------------- + +format(Stats, {[], _}) -> + [Stat || {_Name, Value} = Stat <- Stats, Value =/= unknown]; +format(Stats, {Fs, true}) -> + [Fs(Stat) || {_Name, Value} = Stat <- Stats, Value =/= unknown]; +format(Stats, {Fs, false}) -> + lists:concat([Fs(Stat) || {_Name, Value} = Stat <- Stats, + Value =/= unknown]). + +format_queue_stats({reductions, _}) -> + []; +format_queue_stats({exclusive_consumer_pid, _}) -> + []; +format_queue_stats({single_active_consumer_pid, _}) -> + []; +format_queue_stats({slave_pids, ''}) -> + []; +format_queue_stats({slave_pids, Pids}) -> + [{slave_nodes, [node(Pid) || Pid <- Pids]}]; +format_queue_stats({leader, Leader}) -> + [{node, Leader}]; +format_queue_stats({synchronised_slave_pids, ''}) -> + []; +format_queue_stats({effective_policy_definition, []}) -> + [{effective_policy_definition, #{}}]; +format_queue_stats({synchronised_slave_pids, Pids}) -> + [{synchronised_slave_nodes, [node(Pid) || Pid <- Pids]}]; +format_queue_stats({backing_queue_status, Value}) -> + [{backing_queue_status, properties(Value)}]; +format_queue_stats({idle_since, Value}) -> + [{idle_since, now_to_str(Value)}]; +format_queue_stats({state, Value}) -> + queue_state(Value); +format_queue_stats({disk_reads, _}) -> + []; +format_queue_stats({disk_writes, _}) -> + []; +format_queue_stats(Stat) -> + [Stat]. + +format_channel_stats([{idle_since, Value} | Rest]) -> + [{idle_since, now_to_str(Value)} | Rest]; +format_channel_stats(Stats) -> + Stats. + +%% Conerts an HTTP API request payload value +%% to AMQP 0-9-1 arguments table +format_args({arguments, []}) -> + {arguments, []}; +format_args({arguments, Value}) -> + {arguments, to_amqp_table(Value)}; +format_args(Stat) -> + Stat. + +format_connection_created({host, Value}) -> + {host, addr(Value)}; +format_connection_created({peer_host, Value}) -> + {peer_host, addr(Value)}; +format_connection_created({port, Value}) -> + {port, port(Value)}; +format_connection_created({peer_port, Value}) -> + {peer_port, port(Value)}; +format_connection_created({protocol, Value}) -> + {protocol, protocol(Value)}; +format_connection_created({client_properties, Value}) -> + {client_properties, amqp_table(Value)}; +format_connection_created(Stat) -> + Stat. + +format_exchange_and_queue({policy, Value}) -> + policy(Value); +format_exchange_and_queue({arguments, Value}) -> + [{arguments, amqp_table(Value)}]; +format_exchange_and_queue({name, Value}) -> + resource(Value); +format_exchange_and_queue(Stat) -> + [Stat]. + +format_binding({source, Value}) -> + resource(source, Value); +format_binding({arguments, Value}) -> + [{arguments, amqp_table(Value)}]; +format_binding(Stat) -> + [Stat]. + +format_basic_properties({headers, Value}) -> + {headers, amqp_table(Value)}; +format_basic_properties(Stat) -> + Stat. + +format_accept_content({durable, Value}) -> + {durable, parse_bool(Value)}; +format_accept_content({auto_delete, Value}) -> + {auto_delete, parse_bool(Value)}; +format_accept_content({internal, Value}) -> + {internal, parse_bool(Value)}; +format_accept_content(Stat) -> + Stat. + +print(Fmt, Val) when is_list(Val) -> + list_to_binary(lists:flatten(io_lib:format(Fmt, Val))); +print(Fmt, Val) -> + print(Fmt, [Val]). + +print(Val) when is_list(Val) -> + list_to_binary(lists:flatten(Val)); +print(Val) -> + Val. + +%% TODO - can we remove all these "unknown" cases? Coverage never hits them. + +ip(unknown) -> unknown; +ip(IP) -> list_to_binary(rabbit_misc:ntoa(IP)). + +ipb(unknown) -> unknown; +ipb(IP) -> list_to_binary(rabbit_misc:ntoab(IP)). + +addr(S) when is_list(S); is_atom(S); is_binary(S) -> print("~s", S); +addr(Addr) when is_tuple(Addr) -> ip(Addr). + +port(Port) when is_number(Port) -> Port; +port(Port) -> print("~w", Port). + +properties(unknown) -> unknown; +properties(Table) -> maps:from_list([{Name, tuple(Value)} || + {Name, Value} <- Table]). + +amqp_table(Value) -> rabbit_misc:amqp_table(Value). + +parameter(P) -> pset(value, pget(value, P), P). + +tuple(unknown) -> unknown; +tuple(Tuple) when is_tuple(Tuple) -> [tuple(E) || E <- tuple_to_list(Tuple)]; +tuple(Term) -> Term. + +protocol(unknown) -> + unknown; +protocol(Version = {_Major, _Minor, _Revision}) -> + protocol({'AMQP', Version}); +protocol({Family, Version}) -> + print("~s ~s", [Family, protocol_version(Version)]). + +protocol_version(Arbitrary) + when is_list(Arbitrary) -> Arbitrary; +protocol_version({Major, Minor}) -> io_lib:format("~B-~B", [Major, Minor]); +protocol_version({Major, Minor, 0}) -> protocol_version({Major, Minor}); +protocol_version({Major, Minor, Revision}) -> io_lib:format("~B-~B-~B", + [Major, Minor, Revision]). + +now_to_str(unknown) -> + unknown; +now_to_str(MilliSeconds) -> + BaseDate = calendar:datetime_to_gregorian_seconds({{1970, 1, 1}, + {0, 0, 0}}), + Seconds = BaseDate + (MilliSeconds div 1000), + {{Y, M, D}, {H, Min, S}} = calendar:gregorian_seconds_to_datetime(Seconds), + print("~w-~2.2.0w-~2.2.0w ~w:~2.2.0w:~2.2.0w", [Y, M, D, H, Min, S]). + +now_to_str_ms(unknown) -> + unknown; +now_to_str_ms(MilliSeconds) -> + print("~s:~3.3.0w", [now_to_str(MilliSeconds), MilliSeconds rem 1000]). + +resource(unknown) -> unknown; +resource(Res) -> resource(name, Res). + +resource(_, unknown) -> + unknown; +resource(NameAs, #resource{name = Name, virtual_host = VHost}) -> + [{NameAs, Name}, {vhost, VHost}]. + +policy('') -> []; +policy(Policy) -> [{policy, Policy}]. + +internal_user(User) -> + [{name, internal_user:get_username(User)}, + {password_hash, base64:encode(internal_user:get_password_hash(User))}, + {hashing_algorithm, rabbit_auth_backend_internal:hashing_module_for_user( + User)}, + {tags, tags(internal_user:get_tags(User))}, + {limits, internal_user:get_limits(User)}]. + +user(User) -> + [{name, User#user.username}, + {tags, tags(User#user.tags)}]. + +tags(Tags) -> + list_to_binary(string:join([atom_to_list(T) || T <- Tags], ",")). + +listener(#listener{node = Node, protocol = Protocol, + ip_address = IPAddress, port = Port, opts=Opts}) -> + [{node, Node}, + {protocol, Protocol}, + {ip_address, ip(IPAddress)}, + {port, Port}, + {socket_opts, format_socket_opts(Opts)}]. + +web_context(Props0) -> + SslOpts = pget(ssl_opts, Props0, []), + Props = proplists:delete(ssl_opts, Props0), + [{ssl_opts, format_socket_opts(SslOpts)} | Props]. + +format_socket_opts(Opts) -> + format_socket_opts(Opts, []). + +format_socket_opts([], Acc) -> + lists:reverse(Acc); +%% for HTTP API listeners this will be included into +%% socket_opts +format_socket_opts([{ssl_opts, Value} | Tail], Acc) -> + format_socket_opts(Tail, [{ssl_opts, format_socket_opts(Value)} | Acc]); +%% exclude options that have values that are nested +%% data structures or may include functions. They are fairly +%% obscure and not worth reporting via HTTP API. +format_socket_opts([{verify_fun, _Value} | Tail], Acc) -> + format_socket_opts(Tail, Acc); +format_socket_opts([{crl_cache, _Value} | Tail], Acc) -> + format_socket_opts(Tail, Acc); +format_socket_opts([{partial_chain, _Value} | Tail], Acc) -> + format_socket_opts(Tail, Acc); +format_socket_opts([{user_lookup_fun, _Value} | Tail], Acc) -> + format_socket_opts(Tail, Acc); +format_socket_opts([{sni_fun, _Value} | Tail], Acc) -> + format_socket_opts(Tail, Acc); +%% we do not report SNI host details in the UI, +%% so skip this option and avoid some recursive formatting +%% complexity +format_socket_opts([{sni_hosts, _Value} | Tail], Acc) -> + format_socket_opts(Tail, Acc); +format_socket_opts([{reuse_session, _Value} | Tail], Acc) -> + format_socket_opts(Tail, Acc); +%% we do not want to report configured cipher suites, even +%% though formatting them is straightforward +format_socket_opts([{ciphers, _Value} | Tail], Acc) -> + format_socket_opts(Tail, Acc); +%% single atom options, e.g. `binary` +format_socket_opts([Head | Tail], Acc) when is_atom(Head) -> + format_socket_opts(Tail, [{Head, true} | Acc]); +%% verify_fun value is a tuple that includes a function +format_socket_opts([_Head = {verify_fun, _Value} | Tail], Acc) -> + format_socket_opts(Tail, Acc); +format_socket_opts([Head = {Name, Value} | Tail], Acc) when is_list(Value) -> + case io_lib:printable_unicode_list(Value) of + true -> format_socket_opts(Tail, [{Name, unicode:characters_to_binary(Value)} | Acc]); + false -> format_socket_opts(Tail, [Head | Acc]) + end; +format_socket_opts([{Name, Value} | Tail], Acc) when is_tuple(Value) -> + format_socket_opts(Tail, [{Name, tuple_to_list(Value)} | Acc]); +%% exclude functions from JSON encoding +format_socket_opts([_Head = {_Name, Value} | Tail], Acc) when is_function(Value) -> + format_socket_opts(Tail, Acc); +format_socket_opts([Head | Tail], Acc) -> + format_socket_opts(Tail, [Head | Acc]). + +pack_binding_props(<<"">>, []) -> + <<"~">>; +pack_binding_props(Key, []) -> + list_to_binary(quote_binding(Key)); +pack_binding_props(Key, Args) -> + ArgsEnc = args_hash(Args), + list_to_binary(quote_binding(Key) ++ "~" ++ quote_binding(ArgsEnc)). + +quote_binding(Name) -> + re:replace(rabbit_http_util:quote_plus(Name), "~", "%7E", [global]). + +%% Unfortunately string:tokens("foo~~bar", "~"). -> ["foo","bar"], we lose +%% the fact that there's a double ~. +tokenise("") -> + []; +tokenise(Str) -> + Count = string:cspan(Str, "~"), + case length(Str) of + Count -> [Str]; + _ -> [string:sub_string(Str, 1, Count) | + tokenise(string:sub_string(Str, Count + 2))] + end. + +to_amqp_table(V) -> rabbit_misc:to_amqp_table(V). + +url(Fmt, Vals) -> + print(Fmt, [rabbit_http_util:quote_plus(V) || V <- Vals]). + +exchange(X) -> + format(X, {fun format_exchange_and_queue/1, false}). + +%% We get queues using rabbit_amqqueue:list/1 rather than :info_all/1 since +%% the latter wakes up each queue. Therefore we have a record rather than a +%% proplist to deal with. +queue(Q) when ?is_amqqueue(Q) -> + Name = amqqueue:get_name(Q), + Durable = amqqueue:is_durable(Q), + AutoDelete = amqqueue:is_auto_delete(Q), + ExclusiveOwner = amqqueue:get_exclusive_owner(Q), + Arguments = amqqueue:get_arguments(Q), + Pid = amqqueue:get_pid(Q), + State = amqqueue:get_state(Q), + %% TODO: in the future queue types should be registered with their + %% full and short names and this hard-coded translation should not be + %% necessary + Type = case amqqueue:get_type(Q) of + rabbit_classic_queue -> classic; + rabbit_quorum_queue -> quorum; + T -> T + end, + format( + [{name, Name}, + {durable, Durable}, + {auto_delete, AutoDelete}, + {exclusive, is_pid(ExclusiveOwner)}, + {owner_pid, ExclusiveOwner}, + {arguments, Arguments}, + {pid, Pid}, + {type, Type}, + {state, State}] ++ rabbit_amqqueue:format(Q), + {fun format_exchange_and_queue/1, false}). + +queue_info(List) -> + format(List, {fun format_exchange_and_queue/1, false}). + +queue_state({syncing, Msgs}) -> [{state, syncing}, + {sync_messages, Msgs}]; +queue_state({terminated_by, Name}) -> + [{state, terminated}, + {terminated_by, Name}]; +queue_state(Status) -> [{state, Status}]. + +%% We get bindings using rabbit_binding:list_*/1 rather than :info_all/1 since +%% there are no per-exchange / queue / etc variants for the latter. Therefore +%% we have a record rather than a proplist to deal with. +binding(#binding{source = S, + key = Key, + destination = D, + args = Args}) -> + format( + [{source, S}, + {destination, D#resource.name}, + {destination_type, D#resource.kind}, + {routing_key, Key}, + {arguments, Args}, + {properties_key, pack_binding_props(Key, Args)}], + {fun format_binding/1, false}). + +basic_properties(Props = #'P_basic'{}) -> + Res = record(Props, record_info(fields, 'P_basic')), + format(Res, {fun format_basic_properties/1, true}). + +record(Record, Fields) -> + {Res, _Ix} = lists:foldl(fun (K, {L, Ix}) -> + {case element(Ix, Record) of + undefined -> L; + V -> [{K, V}|L] + end, Ix + 1} + end, {[], 2}, Fields), + Res. + +to_basic_properties(Props) when is_map(Props) -> + E = fun err/2, + Fmt = fun (headers, H) -> to_amqp_table(H); + (delivery_mode, V) when is_integer(V) -> V; + (delivery_mode, _V) -> E(not_int,delivery_mode); + (priority, V) when is_integer(V) -> V; + (priority, _V) -> E(not_int, priority); + (timestamp, V) when is_integer(V) -> V; + (timestamp, _V) -> E(not_int, timestamp); + (_, V) when is_binary(V) -> V; + (K, _V) -> E(not_string, K) + end, + {Res, _Ix} = lists:foldl( + fun (K, {P, Ix}) -> + {case maps:get(a2b(K), Props, undefined) of + undefined -> P; + V -> setelement(Ix, P, Fmt(K, V)) + end, Ix + 1} + end, {#'P_basic'{}, 2}, + record_info(fields, 'P_basic')), + Res. + +-spec err(term(), term()) -> no_return(). +err(A, B) -> + throw({error, {A, B}}). + +a2b(A) -> + list_to_binary(atom_to_list(A)). + +strip_queue_pids(Item) -> + strip_queue_pids(Item, []). + +strip_queue_pids([{_, unknown} | T], Acc) -> + strip_queue_pids(T, Acc); +strip_queue_pids([{pid, Pid} | T], Acc0) when is_pid(Pid) -> + Acc = case proplists:is_defined(node, Acc0) of + false -> [{node, node(Pid)} | Acc0]; + true -> Acc0 + end, + strip_queue_pids(T, Acc); +strip_queue_pids([{pid, _} | T], Acc) -> + strip_queue_pids(T, Acc); +strip_queue_pids([{owner_pid, _} | T], Acc) -> + strip_queue_pids(T, Acc); +strip_queue_pids([Any | T], Acc) -> + strip_queue_pids(T, [Any | Acc]); +strip_queue_pids([], Acc) -> + Acc. + +%% Items can be connections, channels, consumers or queues, hence remove takes +%% various items. +strip_pids(Item = [T | _]) when is_tuple(T) -> + lists:usort(strip_pids(Item, [])); + +strip_pids(Items) -> [lists:usort(strip_pids(I)) || I <- Items]. + +strip_pids([{_, unknown} | T], Acc) -> + strip_pids(T, Acc); +strip_pids([{pid, Pid} | T], Acc) when is_pid(Pid) -> + strip_pids(T, [{node, node(Pid)} | Acc]); +strip_pids([{pid, _} | T], Acc) -> + strip_pids(T, Acc); +strip_pids([{connection, _} | T], Acc) -> + strip_pids(T, Acc); +strip_pids([{owner_pid, _} | T], Acc) -> + strip_pids(T, Acc); +strip_pids([{channel, _} | T], Acc) -> + strip_pids(T, Acc); +strip_pids([{channel_pid, _} | T], Acc) -> + strip_pids(T, Acc); +strip_pids([{exclusive_consumer_pid, _} | T], Acc) -> + strip_pids(T, Acc); +strip_pids([{slave_pids, ''} | T], Acc) -> + strip_pids(T, Acc); +strip_pids([{slave_pids, Pids} | T], Acc) -> + strip_pids(T, [{slave_nodes, [node(Pid) || Pid <- Pids]} | Acc]); +strip_pids([{synchronised_slave_pids, ''} | T], Acc) -> + strip_pids(T, Acc); +strip_pids([{synchronised_slave_pids, Pids} | T], Acc) -> + strip_pids(T, [{synchronised_slave_nodes, [node(Pid) || Pid <- Pids]} | Acc]); +strip_pids([{K, [P|_] = Nested} | T], Acc) when is_tuple(P) -> % recurse + strip_pids(T, [{K, strip_pids(Nested)} | Acc]); +strip_pids([{K, [L|_] = Nested} | T], Acc) when is_list(L) -> % recurse + strip_pids(T, [{K, strip_pids(Nested)} | Acc]); +strip_pids([Any | T], Acc) -> + strip_pids(T, [Any | Acc]); +strip_pids([], Acc) -> + Acc. + +%% Format for JSON replies. Transforms '' into null +format_nulls(Items) when is_list(Items) -> + [format_null_item(Pair) || Pair <- Items]; +format_nulls(Item) -> + format_null_item(Item). + +format_null_item({Key, ''}) -> + {Key, null}; +format_null_item({Key, Value}) when is_list(Value) -> + {Key, format_nulls(Value)}; +format_null_item({Key, Value}) -> + {Key, Value}; +format_null_item([{_K, _V} | _T] = L) -> + format_nulls(L); +format_null_item(Value) -> + Value. + + +-spec escape_html_tags(string()) -> binary(). + +escape_html_tags(S) -> + escape_html_tags(rabbit_data_coercion:to_list(S), []). + + +-spec escape_html_tags(string(), string()) -> binary(). + +escape_html_tags([], Acc) -> + rabbit_data_coercion:to_binary(lists:reverse(Acc)); +escape_html_tags("<" ++ Rest, Acc) -> + escape_html_tags(Rest, lists:reverse("<", Acc)); +escape_html_tags(">" ++ Rest, Acc) -> + escape_html_tags(Rest, lists:reverse(">", Acc)); +escape_html_tags("&" ++ Rest, Acc) -> + escape_html_tags(Rest, lists:reverse("&", Acc)); +escape_html_tags([C | Rest], Acc) -> + escape_html_tags(Rest, [C | Acc]). + + +-spec clean_consumer_details(proplists:proplist()) -> proplists:proplist(). +clean_consumer_details(Obj) -> + case pget(consumer_details, Obj) of + undefined -> Obj; + Cds -> + Cons = [format_consumer_arguments(clean_channel_details(Con)) || Con <- Cds], + pset(consumer_details, Cons, Obj) + end. + +-spec clean_channel_details(proplists:proplist()) -> proplists:proplist(). +clean_channel_details(Obj) -> + Obj0 = lists:keydelete(channel_pid, 1, Obj), + case pget(channel_details, Obj0) of + undefined -> Obj0; + Chd -> + pset(channel_details, + lists:keydelete(pid, 1, Chd), + Obj0) + end. + +-spec format_consumer_arguments(proplists:proplist()) -> proplists:proplist(). +format_consumer_arguments(Obj) -> + case pget(arguments, Obj) of + undefined -> Obj; + #{} -> Obj; + [] -> pset(arguments, #{}, Obj); + Args -> pset(arguments, amqp_table(Args), Obj) + end. + + +parse_bool(<<"true">>) -> true; +parse_bool(<<"false">>) -> false; +parse_bool(true) -> true; +parse_bool(false) -> false; +parse_bool(undefined) -> undefined; +parse_bool(V) -> throw({error, {not_boolean, V}}). + +args_hash(Args) -> + list_to_binary(rabbit_misc:base64url(<<(erlang:phash2(Args, 1 bsl 32)):32>>)). diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_gc.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_gc.erl new file mode 100644 index 0000000000..99ddc89a8e --- /dev/null +++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_gc.erl @@ -0,0 +1,230 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates. All rights reserved. +%% +-module(rabbit_mgmt_gc). + +-include_lib("rabbit_common/include/rabbit.hrl"). + +-record(state, {timer, + interval + }). + +-spec start_link() -> rabbit_types:ok_pid_or_error(). + +-export([start_link/0]). +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, + code_change/3]). + +start_link() -> + gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). + +init(_) -> + Interval = rabbit_misc:get_env(rabbitmq_management_agent, metrics_gc_interval, 120000), + {ok, start_timer(#state{interval = Interval})}. + +handle_call(test, _From, State) -> + {reply, ok, State}. + +handle_cast(_Request, State) -> + {noreply, State}. + +handle_info(start_gc, State) -> + gc_connections(), + gc_vhosts(), + gc_channels(), + gc_queues(), + gc_exchanges(), + gc_nodes(), + {noreply, start_timer(State)}. + +terminate(_Reason, #state{timer = TRef}) -> + _ = erlang:cancel_timer(TRef), + ok. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +start_timer(#state{interval = Interval} = St) -> + TRef = erlang:send_after(Interval, self(), start_gc), + St#state{timer = TRef}. + +gc_connections() -> + gc_process(connection_stats_coarse_conn_stats), + gc_process(connection_created_stats), + gc_process(connection_stats). + +gc_vhosts() -> + VHosts = rabbit_vhost:list(), + GbSet = gb_sets:from_list(VHosts), + gc_entity(vhost_stats_coarse_conn_stats, GbSet), + gc_entity(vhost_stats_fine_stats, GbSet), + gc_entity(vhost_msg_stats, GbSet), + gc_entity(vhost_msg_rates, GbSet), + gc_entity(vhost_stats_deliver_stats, GbSet). + +gc_channels() -> + gc_process(channel_created_stats), + gc_process(channel_stats), + gc_process(channel_stats_fine_stats), + gc_process(channel_process_stats), + gc_process(channel_stats_deliver_stats), + ok. + +gc_queues() -> + Queues = rabbit_amqqueue:list_names(), + GbSet = gb_sets:from_list(Queues), + LocalQueues = rabbit_amqqueue:list_local_names(), + LocalGbSet = gb_sets:from_list(LocalQueues), + gc_entity(queue_stats_publish, GbSet), + gc_entity(queue_stats, LocalGbSet), + gc_entity(queue_msg_stats, LocalGbSet), + gc_entity(queue_process_stats, LocalGbSet), + gc_entity(queue_msg_rates, LocalGbSet), + gc_entity(queue_stats_deliver_stats, GbSet), + gc_process_and_entity(channel_queue_stats_deliver_stats_queue_index, GbSet), + gc_process_and_entity(consumer_stats_queue_index, GbSet), + gc_process_and_entity(consumer_stats_channel_index, GbSet), + gc_process_and_entity(consumer_stats, GbSet), + gc_process_and_entity(channel_exchange_stats_fine_stats_channel_index, GbSet), + gc_process_and_entity(channel_queue_stats_deliver_stats, GbSet), + gc_process_and_entity(channel_queue_stats_deliver_stats_channel_index, GbSet), + ExchangeGbSet = gb_sets:from_list(rabbit_exchange:list_names()), + gc_entities(queue_exchange_stats_publish, GbSet, ExchangeGbSet), + gc_entities(queue_exchange_stats_publish_queue_index, GbSet, ExchangeGbSet), + gc_entities(queue_exchange_stats_publish_exchange_index, GbSet, ExchangeGbSet). + +gc_exchanges() -> + Exchanges = rabbit_exchange:list_names(), + GbSet = gb_sets:from_list(Exchanges), + gc_entity(exchange_stats_publish_in, GbSet), + gc_entity(exchange_stats_publish_out, GbSet), + gc_entity(channel_exchange_stats_fine_stats_exchange_index, GbSet), + gc_process_and_entity(channel_exchange_stats_fine_stats, GbSet). + +gc_nodes() -> + Nodes = rabbit_mnesia:cluster_nodes(all), + GbSet = gb_sets:from_list(Nodes), + gc_entity(node_stats, GbSet), + gc_entity(node_coarse_stats, GbSet), + gc_entity(node_persister_stats, GbSet), + gc_entity(node_node_coarse_stats_node_index, GbSet), + gc_entity(node_node_stats, GbSet), + gc_entity(node_node_coarse_stats, GbSet). + +gc_process(Table) -> + ets:foldl(fun({{Pid, _} = Key, _}, none) -> + gc_process(Pid, Table, Key); + ({Pid = Key, _}, none) -> + gc_process(Pid, Table, Key); + ({Pid = Key, _, _}, none) -> + gc_process(Pid, Table, Key); + ({{Pid, _} = Key, _, _, _, _}, none) -> + gc_process(Pid, Table, Key) + end, none, Table). + +gc_process(Pid, Table, Key) -> + case rabbit_misc:is_process_alive(Pid) of + true -> + none; + false -> + ets:delete(Table, Key), + none + end. + +gc_entity(Table, GbSet) -> + ets:foldl(fun({{_, Id} = Key, _}, none) when Table == node_node_stats -> + gc_entity(Id, Table, Key, GbSet); + ({{{_, Id}, _} = Key, _}, none) when Table == node_node_coarse_stats -> + gc_entity(Id, Table, Key, GbSet); + ({{Id, _} = Key, _}, none) -> + gc_entity(Id, Table, Key, GbSet); + ({Id = Key, _}, none) -> + gc_entity(Id, Table, Key, GbSet); + ({{Id, _} = Key, _}, none) -> + gc_entity(Id, Table, Key, GbSet) + end, none, Table). + +gc_entity(Id, Table, Key, GbSet) -> + case gb_sets:is_member(Id, GbSet) of + true -> + none; + false -> + ets:delete(Table, Key), + none + end. + +gc_process_and_entity(Table, GbSet) -> + ets:foldl(fun({{Id, Pid, _} = Key, _}, none) when Table == consumer_stats -> + gc_process_and_entity(Id, Pid, Table, Key, GbSet); + ({Id = Key, {_, Pid, _}} = Object, none) + when Table == consumer_stats_queue_index -> + gc_object(Pid, Table, Object), + gc_entity(Id, Table, Key, GbSet); + ({Pid = Key, {Id, _, _}} = Object, none) + when Table == consumer_stats_channel_index -> + gc_object(Id, Table, Object, GbSet), + gc_process(Pid, Table, Key); + ({Id = Key, {{Pid, _}, _}} = Object, none) + when Table == channel_exchange_stats_fine_stats_exchange_index; + Table == channel_queue_stats_deliver_stats_queue_index -> + gc_object(Pid, Table, Object), + gc_entity(Id, Table, Key, GbSet); + ({Pid = Key, {{_, Id}, _}} = Object, none) + when Table == channel_exchange_stats_fine_stats_channel_index; + Table == channel_queue_stats_deliver_stats_channel_index -> + gc_object(Id, Table, Object, GbSet), + gc_process(Pid, Table, Key); + ({{{Pid, Id}, _} = Key, _}, none) + when Table == channel_queue_stats_deliver_stats; + Table == channel_exchange_stats_fine_stats -> + gc_process_and_entity(Id, Pid, Table, Key, GbSet); + ({{{Pid, Id}, _} = Key, _, _, _, _, _, _, _, _}, none) -> + gc_process_and_entity(Id, Pid, Table, Key, GbSet); + ({{{Pid, Id}, _} = Key, _, _, _, _}, none) -> + gc_process_and_entity(Id, Pid, Table, Key, GbSet) + end, none, Table). + +gc_process_and_entity(Id, Pid, Table, Key, GbSet) -> + case rabbit_misc:is_process_alive(Pid) andalso gb_sets:is_member(Id, GbSet) of + true -> + none; + false -> + ets:delete(Table, Key), + none + end. + +gc_object(Pid, Table, Object) -> + case rabbit_misc:is_process_alive(Pid) of + true -> + none; + false -> + ets:delete_object(Table, Object), + none + end. + +gc_object(Id, Table, Object, GbSet) -> + case gb_sets:is_member(Id, GbSet) of + true -> + none; + false -> + ets:delete_object(Table, Object), + none + end. + +gc_entities(Table, QueueGbSet, ExchangeGbSet) -> + ets:foldl(fun({{{Q, X}, _} = Key, _}, none) + when Table == queue_exchange_stats_publish -> + gc_entity(Q, Table, Key, QueueGbSet), + gc_entity(X, Table, Key, ExchangeGbSet); + ({Q, {{_, X}, _}} = Object, none) + when Table == queue_exchange_stats_publish_queue_index -> + gc_object(X, Table, Object, ExchangeGbSet), + gc_entity(Q, Table, Q, QueueGbSet); + ({X, {{Q, _}, _}} = Object, none) + when Table == queue_exchange_stats_publish_exchange_index -> + gc_object(Q, Table, Object, QueueGbSet), + gc_entity(X, Table, X, ExchangeGbSet) + end, none, Table). diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_metrics_collector.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_metrics_collector.erl new file mode 100644 index 0000000000..298f17a18d --- /dev/null +++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_metrics_collector.erl @@ -0,0 +1,712 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates. All rights reserved. +%% +-module(rabbit_mgmt_metrics_collector). + +-include_lib("rabbit_common/include/rabbit.hrl"). +-include_lib("rabbit_common/include/rabbit_core_metrics.hrl"). +-include("rabbit_mgmt_metrics.hrl"). + +-behaviour(gen_server). +-compile({no_auto_import, [ceil/1]}). + +-spec start_link(atom()) -> rabbit_types:ok_pid_or_error(). + +-export([name/1]). +-export([start_link/1]). +-export([override_lookups/2, reset_lookups/1]). +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, + code_change/3]). +-export([index_table/2]). +-export([reset_all/0]). + +-import(rabbit_mgmt_data, [lookup_element/3]). + +-record(state, {table, interval, policies, rates_mode, lookup_queue, + lookup_exchange, old_aggr_stats}). + +%% Data is stored in ETS tables: +%% * One ETS table per metric (queue_stats, channel_stats_deliver_stats...) +%% (see ?TABLES in rabbit_mgmt_metrics.hrl) +%% * Stats are stored as key value pairs where the key is a tuple of +%% some value (such as a channel pid) and the retention interval. +%% The value is an instance of an exometer_slide providing a sliding window +%% of samples for some {Object, Interval}. +%% * Each slide can store multiple stats. See stats_per_table in +%% rabbit_mgmt_metrics.hrl for a map of which stats are recorded in which +%% table. + +reset_all() -> + [reset(Table) || {Table, _} <- ?CORE_TABLES]. + +reset(Table) -> + gen_server:cast(name(Table), reset). + +name(Table) -> + list_to_atom((atom_to_list(Table) ++ "_metrics_collector")). + + +start_link(Table) -> + gen_server:start_link({local, name(Table)}, ?MODULE, [Table], []). + +override_lookups(Table, Lookups) -> + gen_server:call(name(Table), {override_lookups, Lookups}, infinity). + +reset_lookups(Table) -> + gen_server:call(name(Table), reset_lookups, infinity). + +init([Table]) -> + {RatesMode, Policies} = load_config(), + Policy = retention_policy(Table), + Interval = take_smaller(proplists:get_value(Policy, Policies, [])) * 1000, + erlang:send_after(Interval, self(), collect_metrics), + {ok, #state{table = Table, interval = Interval, + policies = {proplists:get_value(basic, Policies), + proplists:get_value(detailed, Policies), + proplists:get_value(global, Policies)}, + rates_mode = RatesMode, + old_aggr_stats = #{}, + lookup_queue = fun queue_exists/1, + lookup_exchange = fun exchange_exists/1}}. + +handle_call(reset_lookups, _From, State) -> + {reply, ok, State#state{lookup_queue = fun queue_exists/1, + lookup_exchange = fun exchange_exists/1}}; +handle_call({override_lookups, Lookups}, _From, State) -> + {reply, ok, State#state{lookup_queue = pget(queue, Lookups), + lookup_exchange = pget(exchange, Lookups)}}; +handle_call({submit, Fun}, _From, State) -> + {reply, Fun(), State}; +handle_call(_Request, _From, State) -> + {noreply, State}. + +handle_cast(reset, State) -> + {noreply, State#state{old_aggr_stats = #{}}}; +handle_cast(_Msg, State) -> + {noreply, State}. + + +handle_info(collect_metrics, #state{interval = Interval} = State0) -> + Timestamp = exometer_slide:timestamp(), + State = aggregate_metrics(Timestamp, State0), + erlang:send_after(Interval, self(), collect_metrics), + {noreply, State}; +handle_info(purge_old_stats, State) -> + {noreply, State#state{old_aggr_stats = #{}}}; +handle_info(_Msg, State) -> + {noreply, State}. + +terminate(_Reason, _State) -> + ok. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +retention_policy(connection_created) -> basic; %% really nothing +retention_policy(connection_metrics) -> basic; +retention_policy(connection_coarse_metrics) -> basic; +retention_policy(channel_created) -> basic; +retention_policy(channel_metrics) -> basic; +retention_policy(channel_queue_exchange_metrics) -> detailed; +retention_policy(channel_exchange_metrics) -> detailed; +retention_policy(channel_queue_metrics) -> detailed; +retention_policy(channel_process_metrics) -> basic; +retention_policy(consumer_created) -> basic; +retention_policy(queue_metrics) -> basic; +retention_policy(queue_coarse_metrics) -> basic; +retention_policy(node_persister_metrics) -> global; +retention_policy(node_coarse_metrics) -> global; +retention_policy(node_metrics) -> basic; +retention_policy(node_node_metrics) -> global; +retention_policy(connection_churn_metrics) -> basic. + +take_smaller(Policies) -> + Intervals = [I || {_, I} <- Policies], + case Intervals of + [] -> throw(missing_sample_retention_policies); + _ -> lists:min(Intervals) + end. + +insert_old_aggr_stats(NextStats, Id, Stat) -> + NextStats#{Id => Stat}. + +handle_deleted_queues(queue_coarse_metrics, Remainders, + #state{policies = {BPolicies, _, GPolicies}}) -> + TS = exometer_slide:timestamp(), + lists:foreach(fun ({Queue, {R, U, M}}) -> + NegStats = ?vhost_msg_stats(-R, -U, -M), + [insert_entry(vhost_msg_stats, vhost(Queue), TS, + NegStats, Size, Interval, true) + || {Size, Interval} <- GPolicies], + % zero out msg stats to avoid duplicating msg + % stats when a master queue is migrated + QNegStats = ?queue_msg_stats(0, 0, 0), + [insert_entry(queue_msg_stats, Queue, TS, + QNegStats, Size, Interval, false) + || {Size, Interval} <- BPolicies], + ets:delete(queue_stats, Queue), + ets:delete(queue_process_stats, Queue) + end, maps:to_list(Remainders)); +handle_deleted_queues(_T, _R, _P) -> ok. + +aggregate_metrics(Timestamp, #state{table = Table, + policies = {_, _, _GPolicies}} = State0) -> + Table = State0#state.table, + {Next, Ops, #state{old_aggr_stats = Remainders}} = + ets:foldl(fun(R, {NextStats, O, State}) -> + aggregate_entry(R, NextStats, O, State) + end, {#{}, #{}, State0}, Table), + maps:fold(fun(Tbl, TblOps, Acc) -> + _ = exec_table_ops(Tbl, Timestamp, TblOps), + Acc + end, no_acc, Ops), + + handle_deleted_queues(Table, Remainders, State0), + State0#state{old_aggr_stats = Next}. + +exec_table_ops(Table, Timestamp, TableOps) -> + maps:fold(fun(_Id, {insert, Entry}, A) -> + ets:insert(Table, Entry), + A; + (Id, {insert_with_index, Entry}, A) -> + insert_with_index(Table, Id, Entry), + A; + ({Id, Size, Interval, Incremental}, + {insert_entry, Entry}, A) -> + insert_entry(Table, Id, Timestamp, Entry, Size, + Interval, Incremental), + A + end, no_acc, TableOps). + + +aggregate_entry({Id, Metrics}, NextStats, Ops0, + #state{table = connection_created} = State) -> + case ets:lookup(connection_created_stats, Id) of + [] -> + Ftd = rabbit_mgmt_format:format( + Metrics, + {fun rabbit_mgmt_format:format_connection_created/1, true}), + Entry = ?connection_created_stats(Id, pget(name, Ftd, unknown), Ftd), + Ops = insert_op(connection_created_stats, Id, Entry, Ops0), + {NextStats, Ops, State}; + _ -> + {NextStats, Ops0, State} + end; +aggregate_entry({Id, Metrics}, NextStats, Ops0, + #state{table = connection_metrics} = State) -> + Entry = ?connection_stats(Id, Metrics), + Ops = insert_op(connection_stats, Id, Entry, Ops0), + {NextStats, Ops, State}; +aggregate_entry({Id, RecvOct, SendOct, Reductions, 0}, NextStats, Ops0, + #state{table = connection_coarse_metrics, + policies = {BPolicies, _, GPolicies}} = State) -> + Stats = ?vhost_stats_coarse_conn_stats(RecvOct, SendOct), + Diff = get_difference(Id, Stats, State), + + Ops1 = insert_entry_ops(vhost_stats_coarse_conn_stats, + vhost({connection_created, Id}), true, Diff, Ops0, + GPolicies), + + Entry = ?connection_stats_coarse_conn_stats(RecvOct, SendOct, Reductions), + Ops2 = insert_entry_ops(connection_stats_coarse_conn_stats, Id, false, Entry, + Ops1, BPolicies), + {insert_old_aggr_stats(NextStats, Id, Stats), Ops2, State}; +aggregate_entry({Id, RecvOct, SendOct, _Reductions, 1}, NextStats, Ops0, + #state{table = connection_coarse_metrics, + policies = {_BPolicies, _, GPolicies}} = State) -> + Stats = ?vhost_stats_coarse_conn_stats(RecvOct, SendOct), + Diff = get_difference(Id, Stats, State), + Ops1 = insert_entry_ops(vhost_stats_coarse_conn_stats, + vhost({connection_created, Id}), true, Diff, Ops0, + GPolicies), + rabbit_core_metrics:delete(connection_coarse_metrics, Id), + {NextStats, Ops1, State}; +aggregate_entry({Id, Metrics}, NextStats, Ops0, + #state{table = channel_created} = State) -> + case ets:lookup(channel_created_stats, Id) of + [] -> + Ftd = rabbit_mgmt_format:format(Metrics, {[], false}), + Entry = ?channel_created_stats(Id, pget(name, Ftd, unknown), Ftd), + Ops = insert_op(channel_created_stats, Id, Entry, Ops0), + {NextStats, Ops, State}; + _ -> + {NextStats, Ops0, State} + end; +aggregate_entry({Id, Metrics}, NextStats, Ops0, + #state{table = channel_metrics} = State) -> + %% First metric must be `idle_since` (if available), as expected by + %% `rabbit_mgmt_format:format_channel_stats`. This is a performance + %% optimisation that avoids traversing the whole list when only + %% one element has to be formatted. + Ftd = rabbit_mgmt_format:format_channel_stats(Metrics), + Entry = ?channel_stats(Id, Ftd), + Ops = insert_op(channel_stats, Id, Entry, Ops0), + {NextStats, Ops, State}; +aggregate_entry({{Ch, X} = Id, Publish0, Confirm, ReturnUnroutable, DropUnroutable, 0}, + NextStats, Ops0, + #state{table = channel_exchange_metrics, + policies = {BPolicies, DPolicies, GPolicies}, + rates_mode = RatesMode, + lookup_exchange = ExchangeFun} = State) -> + Stats = ?channel_stats_fine_stats(Publish0, Confirm, ReturnUnroutable, DropUnroutable), + {Publish, _, _, _} = Diff = get_difference(Id, Stats, State), + + Ops1 = insert_entry_ops(channel_stats_fine_stats, Ch, true, Diff, Ops0, + BPolicies), + Ops2 = insert_entry_ops(vhost_stats_fine_stats, vhost(X), true, Diff, Ops1, + GPolicies), + Ops3 = case {ExchangeFun(X), RatesMode} of + {true, basic} -> + Entry = ?exchange_stats_publish_in(Publish), + insert_entry_ops(exchange_stats_publish_in, X, true, Entry, + Ops2, DPolicies); + {true, _} -> + Entry = ?exchange_stats_publish_in(Publish), + O = insert_entry_ops(exchange_stats_publish_in, X, true, + Entry, Ops2, DPolicies), + insert_entry_ops(channel_exchange_stats_fine_stats, Id, + false, Stats, O, DPolicies); + _ -> + Ops2 + end, + {insert_old_aggr_stats(NextStats, Id, Stats), Ops3, State}; +aggregate_entry({{_Ch, X} = Id, Publish0, Confirm, ReturnUnroutable, DropUnroutable, 1}, + NextStats, Ops0, + #state{table = channel_exchange_metrics, + policies = {_BPolicies, DPolicies, GPolicies}, + lookup_exchange = ExchangeFun} = State) -> + Stats = ?channel_stats_fine_stats(Publish0, Confirm, ReturnUnroutable, DropUnroutable), + {Publish, _, _, _} = Diff = get_difference(Id, Stats, State), + Ops1 = insert_entry_ops(vhost_stats_fine_stats, vhost(X), true, Diff, Ops0, + GPolicies), + Ops2 = case ExchangeFun(X) of + true -> + Entry = ?exchange_stats_publish_in(Publish), + insert_entry_ops(exchange_stats_publish_in, X, true, Entry, + Ops1, DPolicies); + _ -> + Ops1 + end, + rabbit_core_metrics:delete(channel_exchange_metrics, Id), + {NextStats, Ops2, State}; +aggregate_entry({{Ch, Q} = Id, Get, GetNoAck, Deliver, DeliverNoAck, + Redeliver, Ack, GetEmpty, 0}, NextStats, Ops0, + #state{table = channel_queue_metrics, + policies = {BPolicies, DPolicies, GPolicies}, + rates_mode = RatesMode, + lookup_queue = QueueFun} = State) -> + Stats = ?vhost_stats_deliver_stats(Get, GetNoAck, Deliver, DeliverNoAck, + Redeliver, Ack, + Deliver + DeliverNoAck + Get + GetNoAck, + GetEmpty), + Diff = get_difference(Id, Stats, State), + + Ops1 = insert_entry_ops(vhost_stats_deliver_stats, vhost(Q), true, Diff, + Ops0, GPolicies), + + Ops2 = insert_entry_ops(channel_stats_deliver_stats, Ch, true, Diff, Ops1, + BPolicies), + + Ops3 = case {QueueFun(Q), RatesMode} of + {true, basic} -> + insert_entry_ops(queue_stats_deliver_stats, Q, true, Diff, + Ops2, BPolicies); + {true, _} -> + O = insert_entry_ops(queue_stats_deliver_stats, Q, true, + Diff, Ops2, BPolicies), + insert_entry_ops(channel_queue_stats_deliver_stats, Id, + false, Stats, O, DPolicies); + _ -> + Ops2 + end, + {insert_old_aggr_stats(NextStats, Id, Stats), Ops3, State}; +aggregate_entry({{_, Q} = Id, Get, GetNoAck, Deliver, DeliverNoAck, + Redeliver, Ack, GetEmpty, 1}, NextStats, Ops0, + #state{table = channel_queue_metrics, + policies = {BPolicies, _, GPolicies}, + lookup_queue = QueueFun} = State) -> + Stats = ?vhost_stats_deliver_stats(Get, GetNoAck, Deliver, DeliverNoAck, + Redeliver, Ack, + Deliver + DeliverNoAck + Get + GetNoAck, + GetEmpty), + Diff = get_difference(Id, Stats, State), + + Ops1 = insert_entry_ops(vhost_stats_deliver_stats, vhost(Q), true, Diff, + Ops0, GPolicies), + Ops2 = case QueueFun(Q) of + true -> + insert_entry_ops(queue_stats_deliver_stats, Q, true, Diff, + Ops1, BPolicies); + _ -> + Ops1 + end, + rabbit_core_metrics:delete(channel_queue_metrics, Id), + {NextStats, Ops2, State}; +aggregate_entry({{_Ch, {Q, X}} = Id, Publish, ToDelete}, NextStats, Ops0, + #state{table = channel_queue_exchange_metrics, + policies = {BPolicies, _, _}, + rates_mode = RatesMode, + lookup_queue = QueueFun, + lookup_exchange = ExchangeFun} = State) -> + Stats = ?queue_stats_publish(Publish), + Diff = get_difference(Id, Stats, State), + Ops1 = case {QueueFun(Q), ExchangeFun(X), RatesMode, ToDelete} of + {true, false, _, _} -> + insert_entry_ops(queue_stats_publish, Q, true, Diff, + Ops0, BPolicies); + {false, true, _, _} -> + insert_entry_ops(exchange_stats_publish_out, X, true, Diff, + Ops0, BPolicies); + {true, true, basic, _} -> + O = insert_entry_ops(queue_stats_publish, Q, true, Diff, + Ops0, BPolicies), + insert_entry_ops(exchange_stats_publish_out, X, true, Diff, + O, BPolicies); + {true, true, _, 0} -> + O1 = insert_entry_ops(queue_stats_publish, Q, true, Diff, + Ops0, BPolicies), + O2 = insert_entry_ops(exchange_stats_publish_out, X, true, + Diff, O1, BPolicies), + insert_entry_ops(queue_exchange_stats_publish, {Q, X}, + true, Diff, O2, BPolicies); + {true, true, _, 1} -> + O = insert_entry_ops(queue_stats_publish, Q, true, Diff, + Ops0, BPolicies), + insert_entry_ops(exchange_stats_publish_out, X, true, + Diff, O, BPolicies); + _ -> + Ops0 + end, + case ToDelete of + 0 -> + {insert_old_aggr_stats(NextStats, Id, Stats), Ops1, State}; + 1 -> + rabbit_core_metrics:delete(channel_queue_exchange_metrics, Id), + {NextStats, Ops1, State} + end; +aggregate_entry({Id, Reductions}, NextStats, Ops0, + #state{table = channel_process_metrics, + policies = {BPolicies, _, _}} = State) -> + Entry = ?channel_process_stats(Reductions), + Ops = insert_entry_ops(channel_process_stats, Id, false, + Entry, Ops0, BPolicies), + {NextStats, Ops, State}; +aggregate_entry({Id, Exclusive, AckRequired, PrefetchCount, + Active, ActivityStatus, Args}, + NextStats, Ops0, + #state{table = consumer_created} = State) -> + case ets:lookup(consumer_stats, Id) of + [] -> + Fmt = rabbit_mgmt_format:format([{exclusive, Exclusive}, + {ack_required, AckRequired}, + {prefetch_count, PrefetchCount}, + {active, Active}, + {activity_status, ActivityStatus}, + {arguments, Args}], {[], false}), + Entry = ?consumer_stats(Id, Fmt), + Ops = insert_with_index_op(consumer_stats, Id, Entry, Ops0), + {NextStats, Ops , State}; + [{_K, V}] -> + CurrentActive = proplists:get_value(active, V, undefined), + case Active =:= CurrentActive of + false -> + Fmt = rabbit_mgmt_format:format([{exclusive, Exclusive}, + {ack_required, AckRequired}, + {prefetch_count, PrefetchCount}, + {active, Active}, + {activity_status, ActivityStatus}, + {arguments, Args}], {[], false}), + Entry = ?consumer_stats(Id, Fmt), + Ops = insert_with_index_op(consumer_stats, Id, Entry, Ops0), + {NextStats, Ops , State}; + _ -> + {NextStats, Ops0, State} + end; + _ -> + {NextStats, Ops0, State} + end; +aggregate_entry({Id, Metrics, 0}, NextStats, Ops0, + #state{table = queue_metrics, + policies = {BPolicies, _, GPolicies}, + lookup_queue = QueueFun} = State) -> + Stats = ?queue_msg_rates(pget(disk_reads, Metrics, 0), + pget(disk_writes, Metrics, 0)), + Diff = get_difference(Id, Stats, State), + Ops1 = insert_entry_ops(vhost_msg_rates, vhost(Id), true, Diff, Ops0, + GPolicies), + Ops2 = case QueueFun(Id) of + true -> + O = insert_entry_ops(queue_msg_rates, Id, false, Stats, Ops1, + BPolicies), + Fmt = rabbit_mgmt_format:format( + Metrics, + {fun rabbit_mgmt_format:format_queue_stats/1, false}), + insert_op(queue_stats, Id, ?queue_stats(Id, Fmt), O); + false -> + Ops1 + end, + {insert_old_aggr_stats(NextStats, Id, Stats), Ops2, State}; +aggregate_entry({Id, Metrics, 1}, NextStats, Ops0, + #state{table = queue_metrics, + policies = {_, _, GPolicies}} = State) -> + Stats = ?queue_msg_rates(pget(disk_reads, Metrics, 0), + pget(disk_writes, Metrics, 0)), + Diff = get_difference(Id, Stats, State), + Ops = insert_entry_ops(vhost_msg_rates, vhost(Id), true, Diff, Ops0, + GPolicies), + rabbit_core_metrics:delete(queue_metrics, Id), + {NextStats, Ops, State}; +aggregate_entry({Name, Ready, Unack, Msgs, Red}, NextStats, Ops0, + #state{table = queue_coarse_metrics, + old_aggr_stats = Old, + policies = {BPolicies, _, GPolicies}, + lookup_queue = QueueFun} = State) -> + Stats = ?vhost_msg_stats(Ready, Unack, Msgs), + Diff = get_difference(Name, Stats, State), + Ops1 = insert_entry_ops(vhost_msg_stats, vhost(Name), true, Diff, Ops0, + GPolicies), + Ops2 = case QueueFun(Name) of + true -> + QPS =?queue_process_stats(Red), + O1 = insert_entry_ops(queue_process_stats, Name, false, QPS, + Ops1, BPolicies), + QMS = ?queue_msg_stats(Ready, Unack, Msgs), + insert_entry_ops(queue_msg_stats, Name, false, QMS, + O1, BPolicies); + _ -> + Ops1 + end, + State1 = State#state{old_aggr_stats = maps:remove(Name, Old)}, + {insert_old_aggr_stats(NextStats, Name, Stats), Ops2, State1}; +aggregate_entry({Id, Metrics}, NextStats, Ops0, + #state{table = node_metrics} = State) -> + Ops = insert_op(node_stats, Id, {Id, Metrics}, Ops0), + {NextStats, Ops, State}; +aggregate_entry({Id, Metrics}, NextStats, Ops0, + #state{table = node_coarse_metrics, + policies = {_, _, GPolicies}} = State) -> + Stats = ?node_coarse_stats( + pget(fd_used, Metrics, 0), pget(sockets_used, Metrics, 0), + pget(mem_used, Metrics, 0), pget(disk_free, Metrics, 0), + pget(proc_used, Metrics, 0), pget(gc_num, Metrics, 0), + pget(gc_bytes_reclaimed, Metrics, 0), + pget(context_switches, Metrics, 0)), + Ops = insert_entry_ops(node_coarse_stats, Id, false, Stats, Ops0, + GPolicies), + {NextStats, Ops, State}; +aggregate_entry({Id, Metrics}, NextStats, Ops0, + #state{table = node_persister_metrics, + policies = {_, _, GPolicies}} = State) -> + Stats = ?node_persister_stats( + pget(io_read_count, Metrics, 0), pget(io_read_bytes, Metrics, 0), + pget(io_read_time, Metrics, 0), pget(io_write_count, Metrics, 0), + pget(io_write_bytes, Metrics, 0), pget(io_write_time, Metrics, 0), + pget(io_sync_count, Metrics, 0), pget(io_sync_time, Metrics, 0), + pget(io_seek_count, Metrics, 0), pget(io_seek_time, Metrics, 0), + pget(io_reopen_count, Metrics, 0), pget(mnesia_ram_tx_count, Metrics, 0), + pget(mnesia_disk_tx_count, Metrics, 0), pget(msg_store_read_count, Metrics, 0), + pget(msg_store_write_count, Metrics, 0), + pget(queue_index_journal_write_count, Metrics, 0), + pget(queue_index_write_count, Metrics, 0), pget(queue_index_read_count, Metrics, 0), + pget(io_file_handle_open_attempt_count, Metrics, 0), + pget(io_file_handle_open_attempt_time, Metrics, 0)), + Ops = insert_entry_ops(node_persister_stats, Id, false, Stats, Ops0, + GPolicies), + {NextStats, Ops, State}; +aggregate_entry({Id, Metrics}, NextStats, Ops0, + #state{table = node_node_metrics, + policies = {_, _, GPolicies}} = State) -> + Stats = ?node_node_coarse_stats(pget(send_bytes, Metrics, 0), + pget(recv_bytes, Metrics, 0)), + CleanMetrics = lists:keydelete(recv_bytes, 1, + lists:keydelete(send_bytes, 1, Metrics)), + Ops1 = insert_op(node_node_stats, Id, ?node_node_stats(Id, CleanMetrics), + Ops0), + Ops = insert_entry_ops(node_node_coarse_stats, Id, false, Stats, Ops1, + GPolicies), + {NextStats, Ops, State}; +aggregate_entry({Id, ConnCreated, ConnClosed, ChCreated, ChClosed, + QueueDeclared, QueueCreated, QueueDeleted}, NextStats, Ops0, + #state{table = connection_churn_metrics, + policies = {_, _, GPolicies}} = State) -> + %% Id is the local node. There is only one entry on every ETS table. + Stats = ?connection_churn_rates(ConnCreated, ConnClosed, ChCreated, ChClosed, + QueueDeclared, QueueCreated, QueueDeleted), + Diff = get_difference(Id, Stats, State), + Ops = insert_entry_ops(connection_churn_rates, Id, true, Diff, Ops0, + GPolicies), + {insert_old_aggr_stats(NextStats, Id, Stats), Ops, State}. + +insert_entry(Table, Id, TS, Entry, Size, Interval0, Incremental) -> + Key = {Id, Interval0}, + Slide = + case ets:lookup(Table, Key) of + [{Key, S}] -> + S; + [] -> + IntervalMs = Interval0 * 1000, + % add some margin to Size and max_n to reduce chances of off-by-one errors + exometer_slide:new(TS - IntervalMs, (Size + Interval0) * 1000, + [{interval, IntervalMs}, + {max_n, ceil(Size / Interval0) + 1}, + {incremental, Incremental}]) + end, + insert_with_index(Table, Key, {Key, exometer_slide:add_element(TS, Entry, + Slide)}). + +update_op(Table, Key, Op, Ops) -> + TableOps = case maps:find(Table, Ops) of + {ok, Inner} -> + maps:put(Key, Op, Inner); + error -> + Inner = #{}, + maps:put(Key, Op, Inner) + end, + maps:put(Table, TableOps, Ops). + +insert_with_index_op(Table, Key, Entry, Ops) -> + update_op(Table, Key, {insert_with_index, Entry}, Ops). + +insert_op(Table, Key, Entry, Ops) -> + update_op(Table, Key, {insert, Entry}, Ops). + +insert_entry_op(Table, Key, Entry, Ops) -> + TableOps0 = case maps:find(Table, Ops) of + {ok, Inner} -> Inner; + error -> #{} + end, + TableOps = maps:update_with(Key, fun({insert_entry, Entry0}) -> + {insert_entry, sum_entry(Entry0, Entry)} + end, {insert_entry, Entry}, TableOps0), + maps:put(Table, TableOps, Ops). + +insert_entry_ops(Table, Id, Incr, Entry, Ops, Policies) -> + lists:foldl(fun({Size, Interval}, Acc) -> + Key = {Id, Size, Interval, Incr}, + insert_entry_op(Table, Key, Entry, Acc) + end, Ops, Policies). + +get_difference(Id, Stats, #state{old_aggr_stats = OldStats}) -> + case maps:find(Id, OldStats) of + error -> + Stats; + {ok, OldStat} -> + difference(OldStat, Stats) + end. + +sum_entry({A0}, {B0}) -> + {B0 + A0}; +sum_entry({A0, A1}, {B0, B1}) -> + {B0 + A0, B1 + A1}; +sum_entry({A0, A1, A2}, {B0, B1, B2}) -> + {B0 + A0, B1 + A1, B2 + A2}; +sum_entry({A0, A1, A2, A3}, {B0, B1, B2, B3}) -> + {B0 + A0, B1 + A1, B2 + A2, B3 + A3}; +sum_entry({A0, A1, A2, A3, A4}, {B0, B1, B2, B3, B4}) -> + {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4}; +sum_entry({A0, A1, A2, A3, A4, A5}, {B0, B1, B2, B3, B4, B5}) -> + {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4, B5 + A5}; +sum_entry({A0, A1, A2, A3, A4, A5, A6}, {B0, B1, B2, B3, B4, B5, B6}) -> + {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4, B5 + A5, B6 + A6}; +sum_entry({A0, A1, A2, A3, A4, A5, A6, A7}, {B0, B1, B2, B3, B4, B5, B6, B7}) -> + {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4, B5 + A5, B6 + A6, B7 + A7}. + +difference({A0}, {B0}) -> + {B0 - A0}; +difference({A0, A1}, {B0, B1}) -> + {B0 - A0, B1 - A1}; +difference({A0, A1, A2}, {B0, B1, B2}) -> + {B0 - A0, B1 - A1, B2 - A2}; +difference({A0, A1, A2, A3}, {B0, B1, B2, B3}) -> + {B0 - A0, B1 - A1, B2 - A2, B3 - A3}; +difference({A0, A1, A2, A3, A4}, {B0, B1, B2, B3, B4}) -> + {B0 - A0, B1 - A1, B2 - A2, B3 - A3, B4 - A4}; +difference({A0, A1, A2, A3, A4, A5}, {B0, B1, B2, B3, B4, B5}) -> + {B0 - A0, B1 - A1, B2 - A2, B3 - A3, B4 - A4, B5 - A5}; +difference({A0, A1, A2, A3, A4, A5, A6}, {B0, B1, B2, B3, B4, B5, B6}) -> + {B0 - A0, B1 - A1, B2 - A2, B3 - A3, B4 - A4, B5 - A5, B6 - A6}; +difference({A0, A1, A2, A3, A4, A5, A6, A7}, {B0, B1, B2, B3, B4, B5, B6, B7}) -> + {B0 - A0, B1 - A1, B2 - A2, B3 - A3, B4 - A4, B5 - A5, B6 - A6, B7 - A7}. + +vhost(#resource{virtual_host = VHost}) -> + VHost; +vhost({queue_stats, #resource{virtual_host = VHost}}) -> + VHost; +vhost({TName, Pid}) -> + pget(vhost, lookup_element(TName, Pid, 2)). + +exchange_exists(Name) -> + case rabbit_exchange:lookup(Name) of + {ok, _} -> + true; + _ -> + false + end. + +queue_exists(Name) -> + case rabbit_amqqueue:lookup(Name) of + {ok, _} -> + true; + _ -> + false + end. + +insert_with_index(Table, Key, Tuple) -> + Insert = ets:insert(Table, Tuple), + insert_index(Table, Key), + Insert. + +insert_index(consumer_stats, {Q, Ch, _} = Key) -> + ets:insert(index_table(consumer_stats, queue), {Q, Key}), + ets:insert(index_table(consumer_stats, channel), {Ch, Key}); +insert_index(channel_exchange_stats_fine_stats, {{Ch, Ex}, _} = Key) -> + ets:insert(index_table(channel_exchange_stats_fine_stats, exchange), {Ex, Key}), + ets:insert(index_table(channel_exchange_stats_fine_stats, channel), {Ch, Key}); +insert_index(channel_queue_stats_deliver_stats, {{Ch, Q}, _} = Key) -> + ets:insert(index_table(channel_queue_stats_deliver_stats, queue), {Q, Key}), + ets:insert(index_table(channel_queue_stats_deliver_stats, channel), {Ch, Key}); +insert_index(queue_exchange_stats_publish, {{Q, Ex}, _} = Key) -> + ets:insert(index_table(queue_exchange_stats_publish, queue), {Q, Key}), + ets:insert(index_table(queue_exchange_stats_publish, exchange), {Ex, Key}); +insert_index(node_node_coarse_stats, {{_, Node}, _} = Key) -> + ets:insert(index_table(node_node_coarse_stats, node), {Node, Key}); +insert_index(_, _) -> ok. + +index_table(consumer_stats, queue) -> consumer_stats_queue_index; +index_table(consumer_stats, channel) -> consumer_stats_channel_index; +index_table(channel_exchange_stats_fine_stats, exchange) -> channel_exchange_stats_fine_stats_exchange_index; +index_table(channel_exchange_stats_fine_stats, channel) -> channel_exchange_stats_fine_stats_channel_index; +index_table(channel_queue_stats_deliver_stats, queue) -> channel_queue_stats_deliver_stats_queue_index; +index_table(channel_queue_stats_deliver_stats, channel) -> channel_queue_stats_deliver_stats_channel_index; +index_table(queue_exchange_stats_publish, queue) -> queue_exchange_stats_publish_queue_index; +index_table(queue_exchange_stats_publish, exchange) -> queue_exchange_stats_publish_exchange_index; +index_table(node_node_coarse_stats, node) -> node_node_coarse_stats_node_index. + +load_config() -> + RatesMode = rabbit_mgmt_agent_config:get_env(rates_mode), + Policies = rabbit_mgmt_agent_config:get_env(sample_retention_policies, []), + {RatesMode, Policies}. + +ceil(X) when X < 0 -> + trunc(X); +ceil(X) -> + T = trunc(X), + case X - T == 0 of + true -> T; + false -> T + 1 + end. + +pget(Key, List) -> pget(Key, List, unknown). + +pget(Key, List, Default) when is_number(Default) -> + case rabbit_misc:pget(Key, List) of + Number when is_number(Number) -> + Number; + _Other -> + Default + end; +pget(Key, List, Default) -> + rabbit_misc:pget(Key, List, Default). diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_metrics_gc.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_metrics_gc.erl new file mode 100644 index 0000000000..f1ae48e0e4 --- /dev/null +++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_metrics_gc.erl @@ -0,0 +1,175 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates. All rights reserved. +%% +-module(rabbit_mgmt_metrics_gc). + +-record(state, {basic_i, + detailed_i, + global_i}). + +-include_lib("rabbit_common/include/rabbit.hrl"). + +-spec start_link(atom()) -> rabbit_types:ok_pid_or_error(). + +-export([name/1]). +-export([start_link/1]). +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, + code_change/3]). + +name(EventType) -> + list_to_atom((atom_to_list(EventType) ++ "_metrics_gc")). + +start_link(EventType) -> + gen_server:start_link({local, name(EventType)}, ?MODULE, [], []). + +init(_) -> + Policies = rabbit_mgmt_agent_config:get_env(sample_retention_policies), + {ok, #state{basic_i = intervals(basic, Policies), + global_i = intervals(global, Policies), + detailed_i = intervals(detailed, Policies)}}. + +handle_call(_Request, _From, State) -> + {noreply, State}. + +handle_cast({event, #event{type = connection_closed, props = Props}}, + State = #state{basic_i = BIntervals}) -> + Pid = pget(pid, Props), + remove_connection(Pid, BIntervals), + {noreply, State}; +handle_cast({event, #event{type = channel_closed, props = Props}}, + State = #state{basic_i = BIntervals}) -> + Pid = pget(pid, Props), + remove_channel(Pid, BIntervals), + {noreply, State}; +handle_cast({event, #event{type = consumer_deleted, props = Props}}, State) -> + remove_consumer(Props), + {noreply, State}; +handle_cast({event, #event{type = exchange_deleted, props = Props}}, + State = #state{basic_i = BIntervals}) -> + Name = pget(name, Props), + remove_exchange(Name, BIntervals), + {noreply, State}; +handle_cast({event, #event{type = queue_deleted, props = Props}}, + State = #state{basic_i = BIntervals}) -> + Name = pget(name, Props), + remove_queue(Name, BIntervals), + {noreply, State}; +handle_cast({event, #event{type = vhost_deleted, props = Props}}, + State = #state{global_i = GIntervals}) -> + Name = pget(name, Props), + remove_vhost(Name, GIntervals), + {noreply, State}; +handle_cast({event, #event{type = node_node_deleted, props = Props}}, State) -> + Name = pget(route, Props), + remove_node_node(Name), + {noreply, State}. + +handle_info(_Msg, State) -> + {noreply, State}. + +terminate(_Reason, _State) -> + ok. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +remove_connection(Id, BIntervals) -> + ets:delete(connection_created_stats, Id), + ets:delete(connection_stats, Id), + delete_samples(connection_stats_coarse_conn_stats, Id, BIntervals), + ok. + +remove_channel(Id, BIntervals) -> + ets:delete(channel_created_stats, Id), + ets:delete(channel_stats, Id), + delete_samples(channel_process_stats, Id, BIntervals), + delete_samples(channel_stats_fine_stats, Id, BIntervals), + delete_samples(channel_stats_deliver_stats, Id, BIntervals), + index_delete(consumer_stats, channel, Id), + index_delete(channel_exchange_stats_fine_stats, channel, Id), + index_delete(channel_queue_stats_deliver_stats, channel, Id), + ok. + +remove_consumer(Props) -> + Id = {pget(queue, Props), pget(channel, Props), pget(consumer_tag, Props)}, + ets:delete(consumer_stats, Id), + cleanup_index(consumer_stats, Id), + ok. + +remove_exchange(Name, BIntervals) -> + delete_samples(exchange_stats_publish_out, Name, BIntervals), + delete_samples(exchange_stats_publish_in, Name, BIntervals), + index_delete(queue_exchange_stats_publish, exchange, Name), + index_delete(channel_exchange_stats_fine_stats, exchange, Name), + ok. + +remove_queue(Name, BIntervals) -> + ets:delete(queue_stats, Name), + delete_samples(queue_stats_publish, Name, BIntervals), + delete_samples(queue_stats_deliver_stats, Name, BIntervals), + delete_samples(queue_process_stats, Name, BIntervals), + delete_samples(queue_msg_stats, Name, BIntervals), + delete_samples(queue_msg_rates, Name, BIntervals), + index_delete(channel_queue_stats_deliver_stats, queue, Name), + index_delete(queue_exchange_stats_publish, queue, Name), + index_delete(consumer_stats, queue, Name), + + ok. + +remove_vhost(Name, GIntervals) -> + delete_samples(vhost_stats_coarse_conn_stats, Name, GIntervals), + delete_samples(vhost_stats_fine_stats, Name, GIntervals), + delete_samples(vhost_stats_deliver_stats, Name, GIntervals), + ok. + +remove_node_node(Name) -> + index_delete(node_node_coarse_stats, node, Name), + ok. + +intervals(Type, Policies) -> + [I || {_, I} <- proplists:get_value(Type, Policies)]. + +delete_samples(Table, Id, Intervals) -> + [ets:delete(Table, {Id, I}) || I <- Intervals], + ok. + +index_delete(Table, Type, Id) -> + IndexTable = rabbit_mgmt_metrics_collector:index_table(Table, Type), + Keys = ets:lookup(IndexTable, Id), + [ begin + ets:delete(Table, Key), + cleanup_index(Table, Key) + end + || {_Index, Key} <- Keys ], + ets:delete(IndexTable, Id), + ok. + +cleanup_index(consumer_stats, {Q, Ch, _} = Key) -> + delete_index(consumer_stats, queue, {Q, Key}), + delete_index(consumer_stats, channel, {Ch, Key}), + ok; +cleanup_index(channel_exchange_stats_fine_stats, {{Ch, Ex}, _} = Key) -> + delete_index(channel_exchange_stats_fine_stats, exchange, {Ex, Key}), + delete_index(channel_exchange_stats_fine_stats, channel, {Ch, Key}), + ok; +cleanup_index(channel_queue_stats_deliver_stats, {{Ch, Q}, _} = Key) -> + delete_index(channel_queue_stats_deliver_stats, queue, {Q, Key}), + delete_index(channel_queue_stats_deliver_stats, channel, {Ch, Key}), + ok; +cleanup_index(queue_exchange_stats_publish, {{Q, Ex}, _} = Key) -> + delete_index(queue_exchange_stats_publish, queue, {Q, Key}), + delete_index(queue_exchange_stats_publish, exchange, {Ex, Key}), + ok; +cleanup_index(node_node_coarse_stats, {{_, Node}, _} = Key) -> + delete_index(node_node_coarse_stats, node, {Node, Key}), + ok; +cleanup_index(_, _) -> ok. + +delete_index(Table, Index, Obj) -> + ets:delete_object(rabbit_mgmt_metrics_collector:index_table(Table, Index), + Obj). + +pget(Key, List) -> rabbit_misc:pget(Key, List, unknown). diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_storage.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_storage.erl new file mode 100644 index 0000000000..4c5c8c18ef --- /dev/null +++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_storage.erl @@ -0,0 +1,57 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates. All rights reserved. +%% +-module(rabbit_mgmt_storage). +-behaviour(gen_server2). +-record(state, {}). + +-spec start_link() -> rabbit_types:ok_pid_or_error(). + +-export([start_link/0]). +-export([reset/0, reset_all/0]). +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, + code_change/3]). + +-include("rabbit_mgmt_metrics.hrl"). + +%% ETS owner +start_link() -> + gen_server2:start_link({local, ?MODULE}, ?MODULE, [], []). + +reset() -> + rabbit_log:warning("Resetting RabbitMQ management storage"), + [ets:delete_all_objects(IndexTable) || IndexTable <- ?INDEX_TABLES], + [ets:delete_all_objects(Table) || {Table, _} <- ?TABLES], + _ = rabbit_mgmt_metrics_collector:reset_all(), + ok. + +reset_all() -> + _ = [rpc:call(Node, rabbit_mgmt_storage, reset, []) + || Node <- rabbit_nodes:all_running()], + ok. + +init(_) -> + _ = [ets:new(IndexTable, [public, bag, named_table]) + || IndexTable <- ?INDEX_TABLES], + _ = [ets:new(Table, [public, Type, named_table]) + || {Table, Type} <- ?TABLES], + _ = ets:new(rabbit_mgmt_db_cache, [public, set, named_table]), + {ok, #state{}}. + +handle_call(_Request, _From, State) -> + {noreply, State}. + +handle_cast(_Msg, State) -> + {noreply, State}. + +handle_info(_Msg, State) -> + {noreply, State}. + +terminate(_Reason, _State) -> + ok. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. |