Fix typo and add tests to verify stats recover after reset

deps/rabbitmq_management_agent/src/rabbit_mgmt_storage.erl

@@ -34,7 +34,7 @@ reset() ->
     rabbit_log:warning("Resetting RabbitMQ management storage~n"),
     [ets:delete_all_objects(IndexTable) || IndexTable <- ?INDEX_TABLES],
     [ets:delete_all_objects(Table) || {Table, _} <- ?TABLES],
-    rabbit_mgmt_metrics_collector:reset(),
+    rabbit_mgmt_metrics_collector:reset_all(),
     ok.
 
 reset_all() ->

deps/rabbitmq_management_agent/test/metrics_SUITE.erl

@@ -17,7 +17,7 @@
 -compile(export_all).
 
 -include_lib("common_test/include/ct.hrl").
-
+-include_lib("amqp_client/include/amqp_client.hrl").
 
 all() ->
     [
@@ -27,7 +27,8 @@ all() ->
 groups() ->
     [
       {non_parallel_tests, [], [
-                                node
+                                node,
+                                storage_reset
                                ]}
     ].
 
@@ -36,11 +37,17 @@ groups() ->
 %% -------------------------------------------------------------------
 
 merge_app_env(Config) ->
-    rabbit_ct_helpers:merge_app_env(Config,
-                                    {rabbit, [
-                                              {collect_statistics, fine},
-                                              {collect_statistics_interval, 500}
-                                             ]}).
+    _Config1 = rabbit_ct_helpers:merge_app_env(Config,
+                                              {rabbit, [
+                                                        {collect_statistics, fine},
+                                                        {collect_statistics_interval, 500}
+                                                       ]}).
+    %% rabbit_ct_helpers:merge_app_env(
+    %%   Config1, {rabbitmq_management_agent, [{sample_retention_policies,
+    %%                                          [{global,   [{605, 500}]},
+    %%                                           {basic,    [{605, 500}]},
+    %%                                           {detailed, [{10, 500}]}] }]}).
+
 init_per_suite(Config) ->
     rabbit_ct_helpers:log_environment(),
     Config1 = rabbit_ct_helpers:set_config(Config, [
@@ -49,10 +56,12 @@ init_per_suite(Config) ->
       ]),
     rabbit_ct_helpers:run_setup_steps(Config1,
       [ fun merge_app_env/1 ] ++
-      rabbit_ct_broker_helpers:setup_steps()).
+      rabbit_ct_broker_helpers:setup_steps() ++
+      rabbit_ct_client_helpers:setup_steps()).
 
 end_per_suite(Config) ->
     rabbit_ct_helpers:run_teardown_steps(Config,
+      rabbit_ct_client_helpers:teardown_steps() ++
       rabbit_ct_broker_helpers:teardown_steps()).
 
 init_per_group(_, Config) ->
@@ -92,3 +101,62 @@ node(Config) ->
     [_, _, _] = read_table_rpc(Config, node_node_metrics). % 3 nodes as ct-helpers adds one
 
 
+storage_reset(Config) ->
+    %% Ensures that core stats are reset, otherwise consume generates negative values
+    %% Doesn't really test if the reset does anything!
+    {Ch, Q} = publish_msg(Config),
+    wait_until(fun() ->
+                       {1, 0, 1} == get_vhost_stats(Config)
+               end),
+    ok = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_mgmt_storage, reset, []),
+    wait_until(fun() ->
+                       {1, 0, 1} == get_vhost_stats(Config)
+               end),
+    consume_msg(Ch, Q),
+    wait_until(fun() ->
+                       {0, 0, 0} == get_vhost_stats(Config)
+               end),
+    rabbit_ct_client_helpers:close_channel(Ch).
+
+%% -------------------------------------------------------------------
+%% Helpers
+%% -------------------------------------------------------------------
+
+publish_msg(Config) ->
+    Ch = rabbit_ct_client_helpers:open_channel(Config, 0),
+    #'queue.declare_ok'{queue = Q} =
+        amqp_channel:call(Ch, #'queue.declare'{exclusive = true}),
+    amqp_channel:cast(Ch, #'basic.publish'{routing_key = Q},
+                      #amqp_msg{props   = #'P_basic'{delivery_mode = 2},
+                                payload = Q}),
+    {Ch, Q}.
+
+consume_msg(Ch, Q) ->
+    amqp_channel:subscribe(Ch, #'basic.consume'{queue    = Q,
+                                                no_ack   = true}, self()),
+    receive #'basic.consume_ok'{} -> ok
+    end,
+    receive {#'basic.deliver'{}, #amqp_msg{payload = Q}} ->
+            ok
+    end.
+
+wait_until(Fun) ->
+    wait_until(Fun, 120).
+
+wait_until(Fun, 0) ->
+    false;
+wait_until(Fun, N) ->
+    case Fun() of
+        true ->
+            true;
+        false ->
+            timer:sleep(1000),
+            wait_until(Fun, N-1)
+    end.
+
+get_vhost_stats(Config) ->
+    Dict = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_mgmt_data, overview_data,
+                                        [none, all ,{no_range, no_range, no_range,no_range},
+                                         [<<"/">>]]),
+    {ok, {VhostMsgStats, _}} = dict:find(vhost_msg_stats, Dict),
+    exometer_slide:last(VhostMsgStats).