Initialise stream metrics to 0

Offsets start at -1, not a valid counter!

2 files changed, 37 insertions, 7 deletions

diff --git a/src/rabbit_osiris_metrics.erl b/src/rabbit_osiris_metrics.erl
index 5ba24ff04c..010855a1c0 100644
--- a/src/rabbit_osiris_metrics.erl
+++ b/src/rabbit_osiris_metrics.erl
@@ -23,8 +23,11 @@
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
          code_change/3]).
 
+-define(TICK_TIMEOUT, 5000).
 -define(SERVER, ?MODULE).
 
+-record(state, {timeout :: non_neg_integer()}).
+
 %%----------------------------------------------------------------------------
 %% Starts the raw metrics storage and owns the ETS tables.
 %%----------------------------------------------------------------------------
@@ -35,8 +38,10 @@ start_link() ->
     gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
 
 init([]) ->
-    erlang:send_after(5000, self(), tick),
-    {ok, none}.
+    Timeout = application:get_env(rabbit, stream2_tick_interval,
+                                  ?TICK_TIMEOUT),
+    erlang:send_after(Timeout, self(), tick),
+    {ok, #state{timeout = Timeout}}.
 
 handle_call(_Request, _From, State) ->
     {noreply, State}.
@@ -44,18 +49,22 @@ handle_call(_Request, _From, State) ->
 handle_cast(_Request, State) ->
     {noreply, State}.
 
-handle_info(tick, State) ->
+handle_info(tick, #state{timeout = Timeout} = State) ->
     Data = osiris_counters:overview(),
     maps:map(
       fun ({osiris_writer, QName}, #{offset := Offs}) ->
-                      rabbit_core_metrics:queue_stats(QName, Offs, 0, Offs, 0),
+                      COffs = case Offs of
+                                  -1 -> 0;
+                                  _ -> Offs
+                              end,
+                      rabbit_core_metrics:queue_stats(QName, COffs, 0, COffs, 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),
+    erlang:send_after(Timeout, self(), tick),
     {noreply, State}.
 
 terminate(_Reason, _State) ->
diff --git a/test/rabbit_stream2_queue_SUITE.erl b/test/rabbit_stream2_queue_SUITE.erl
index eed313aa98..ccd7498a06 100644
--- a/test/rabbit_stream2_queue_SUITE.erl
+++ b/test/rabbit_stream2_queue_SUITE.erl
@@ -52,7 +52,8 @@ all_tests() ->
      idempotent_declare_queue,
      delete_queue,
      zenflix,
-     declare_queue
+     declare_queue,
+     counters_initialised_to_zero
     ].
 
 %% -------------------------------------------------------------------
@@ -62,7 +63,7 @@ all_tests() ->
 init_per_suite(Config0) ->
     rabbit_ct_helpers:log_environment(),
     Config = rabbit_ct_helpers:merge_app_env(
-               Config0, {rabbit, [{quorum_tick_interval, 1000}]}),
+               Config0, {rabbit, [{stream2_tick_interval, 100}]}),
     rabbit_ct_helpers:run_setup_steps(
       Config,
       [fun rabbit_ct_broker_helpers:enable_dist_proxy_manager/1]).
@@ -419,6 +420,26 @@ declare_queue(Config) ->
     flush(100),
     ok.
 
+counters_initialised_to_zero(Config) ->
+    [Server | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    Ch1 = rabbit_ct_client_helpers:open_channel(Config, Server),
+    QName = ?config(queue_name, Config),
+
+    ?assertEqual({'queue.declare_ok', QName, 0, 0},
+                 declare(Ch1, QName, [{<<"x-queue-type">>, longstr,
+                                      ?config(queue_type, Config)}])),
+
+    timer:sleep(500),
+
+    ?assertEqual([[<<"0">>, <<"0">>, <<"0">>]],
+                 rabbit_ct_broker_helpers:rabbitmqctl_list(
+                   Config, 0, ["list_queues", "messages", "messages_ready",
+                               "messages_unacknowledged", "--no-table-headers"])),
+
+    flush(100),
+    ok.
+
 %% HELPERS
 
 assert_declare(Ch, QName, Type) ->