osiris metrics

3 files changed, 88 insertions, 42 deletions

diff --git a/src/rabbit.erl b/src/rabbit.erl
index 7e852e89d1..03700bae36 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -111,6 +111,13 @@
                     {requires,    pre_boot},
                     {enables,     external_infrastructure}]}).
 
+-rabbit_boot_step({rabbit_osiris_metrics,
+                   [{description, "osiris metrics scraper"},
+                    {mfa,         {rabbit_sup, start_child,
+                                   [rabbit_osiris_metrics]}},
+                    {requires,    pre_boot},
+                    {enables,     external_infrastructure}]}).
+
 -rabbit_boot_step({rabbit_event,
                    [{description, "statistics event manager"},
                     {mfa,         {rabbit_sup, start_restartable_child,
diff --git a/src/rabbit_osiris_metrics.erl b/src/rabbit_osiris_metrics.erl
new file mode 100644
index 0000000000..df4f609845
--- /dev/null
+++ b/src/rabbit_osiris_metrics.erl
@@ -0,0 +1,66 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License
+%% at https://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and
+%% limitations under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2007-2020 Pivotal Software, Inc.  All rights reserved.
+%%
+
+-module(rabbit_osiris_metrics).
+
+-behaviour(gen_server).
+
+-export([start_link/0]).
+
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
+         code_change/3]).
+
+-define(SERVER, ?MODULE).
+
+%%----------------------------------------------------------------------------
+%% Starts the raw metrics storage and owns the ETS tables.
+%%----------------------------------------------------------------------------
+
+-spec start_link() -> rabbit_types:ok_pid_or_error().
+
+start_link() ->
+    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+
+init([]) ->
+    erlang:send_after(5000, self(), tick),
+    {ok, none}.
+
+handle_call(_Request, _From, State) ->
+    {noreply, State}.
+
+handle_cast(_Request, State) ->
+    {noreply, State}.
+
+handle_info(tick, State) ->
+    rabbit_log:info("osiris metrics tick received", []),
+    Data = osiris_counters:overview(),
+    maps:map(
+      fun ({osiris_writer, QName}, #{offset := Offs}) ->
+                      rabbit_core_metrics:queue_stats(QName, Offs, 0, Offs, 0),
+                      Infos = rabbit_stream2_queue:infos(QName),
+                      rabbit_core_metrics:queue_stats(QName, Infos),
+                      ok;
+          (_, _V) ->
+              ok
+      end, Data),
+    erlang:send_after(5000, self(), tick),
+    {noreply, State}.
+
+terminate(_Reason, _State) ->
+    ok.
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
diff --git a/src/rabbit_stream2_queue.erl b/src/rabbit_stream2_queue.erl
index 433831de1d..954944c413 100644
--- a/src/rabbit_stream2_queue.erl
+++ b/src/rabbit_stream2_queue.erl
@@ -386,8 +386,7 @@ i(durable,     Q) when ?is_amqqueue(Q) -> amqqueue:is_durable(Q);
 i(auto_delete, Q) when ?is_amqqueue(Q) -> amqqueue:is_auto_delete(Q);
 i(arguments,   Q) when ?is_amqqueue(Q) -> amqqueue:get_arguments(Q);
 i(pid, Q) when ?is_amqqueue(Q) ->
-    {Name, _} = amqqueue:get_pid(Q),
-    whereis(Name);
+    amqqueue:get_pid(Q);
 i(messages, Q) when ?is_amqqueue(Q) ->
     QName = amqqueue:get_name(Q),
     case ets:lookup(queue_coarse_metrics, QName) of
@@ -436,9 +435,9 @@ i(consumers, Q) when ?is_amqqueue(Q) ->
             0
     end;
 i(memory, Q) when ?is_amqqueue(Q) ->
-    {Name, _} = amqqueue:get_pid(Q),
+    Pid = amqqueue:get_pid(Q),
     try
-        {memory, M} = process_info(whereis(Name), memory),
+        {memory, M} = process_info(Pid, memory),
         M
     catch
         error:badarg ->
@@ -458,9 +457,9 @@ i(memory, Q) when ?is_amqqueue(Q) ->
 %         _ -> not_member
 %     end;
 i(garbage_collection, Q) when ?is_amqqueue(Q) ->
-    {Name, _} = amqqueue:get_pid(Q),
+    Pid = amqqueue:get_pid(Q),
     try
-        rabbit_misc:get_gc_info(whereis(Name))
+        rabbit_misc:get_gc_info(Pid)
     catch
         error:badarg ->
             []
@@ -470,44 +469,18 @@ i(members, Q) when ?is_amqqueue(Q) ->
 i(online, Q) ->
     get_nodes(Q);
 i(leader, Q) ->
-    {_Name, Leader} = amqqueue:get_pid(Q),
-    Leader;
-i(open_files, Q) when ?is_amqqueue(Q) ->
-    {Name, _} = amqqueue:get_pid(Q),
-    Nodes = get_nodes(Q),
-    {Data, _} = rpc:multicall(Nodes, ?MODULE, open_files, [Name]),
-    lists:flatten(Data);
-i(single_active_consumer_pid, Q) when ?is_amqqueue(Q) ->
-    QPid = amqqueue:get_pid(Q),
-    {ok, {_, SacResult}, _} = ra:local_query(QPid,
-                                             fun rabbit_fifo:query_single_active_consumer/1),
-    case SacResult of
-        {value, {_ConsumerTag, ChPid}} ->
-            ChPid;
-        _ ->
-            ''
-    end;
-i(single_active_consumer_ctag, Q) when ?is_amqqueue(Q) ->
-    QPid = amqqueue:get_pid(Q),
-    {ok, {_, SacResult}, _} = ra:local_query(QPid,
-                                             fun rabbit_fifo:query_single_active_consumer/1),
-    case SacResult of
-        {value, {ConsumerTag, _ChPid}} ->
-            ConsumerTag;
-        _ ->
-            ''
-    end;
-i(type, _) -> quorum;
+    Pid = amqqueue:get_pid(Q),
+    node(Pid);
+% i(open_files, Q) when ?is_amqqueue(Q) ->
+%     Pid = amqqueue:get_pid(Q),
+%     Nodes = get_nodes(Q),
+%     {Data, _} = rpc:multicall(Nodes, ?MODULE, open_files, [Name]),
+%     lists:flatten(Data);
+i(type, _) -> ?MODULE;
 i(messages_ram, Q) when ?is_amqqueue(Q) ->
-    QPid = amqqueue:get_pid(Q),
-    {ok, {_, {Length, _}}, _} = ra:local_query(QPid,
-                                          fun rabbit_fifo:query_in_memory_usage/1),
-    Length;
+    0;
 i(message_bytes_ram, Q) when ?is_amqqueue(Q) ->
-    QPid = amqqueue:get_pid(Q),
-    {ok, {_, {_, Bytes}}, _} = ra:local_query(QPid,
-                                         fun rabbit_fifo:query_in_memory_usage/1),
-    Bytes;
+    0;
 i(_K, _Q) -> ''.
 
 get_nodes(Q) when ?is_amqqueue(Q) ->