Merge pull request #1278 from rabbitmq/rabbitmq-management-427

Only preserve stats for local queues

1 files changed, 80 insertions, 25 deletions

diff --git a/test/rabbit_core_metrics_gc_SUITE.erl b/test/rabbit_core_metrics_gc_SUITE.erl
index 0ef86717ac..d63c9ab2fb 100644
--- a/test/rabbit_core_metrics_gc_SUITE.erl
+++ b/test/rabbit_core_metrics_gc_SUITE.erl
@@ -24,7 +24,8 @@
 
 all() ->
     [
-      {group, non_parallel_tests}
+      {group, non_parallel_tests},
+      {group, cluster_tests}
     ].
 
 groups() ->
@@ -37,7 +38,8 @@ groups() ->
         gen_server2_metrics,
         consumer_metrics
       ]
-     }
+     },
+     {cluster_tests, [], [cluster_queue_metrics]}
     ].
 
 %% -------------------------------------------------------------------
@@ -45,33 +47,27 @@ groups() ->
 %% -------------------------------------------------------------------
 
 merge_app_env(Config) ->
-    rabbit_ct_helpers:merge_app_env(Config,
-                                    {rabbit, [
-                                              {core_metrics_gc_interval, 6000000},
-                                              {collect_statistics_interval, 100},
-                                              {collect_statistics, fine}
-                                             ]}).
-
-init_per_suite(Config) ->
+    AppEnv = {rabbit, [{core_metrics_gc_interval, 6000000},
+                       {collect_statistics_interval, 100},
+                       {collect_statistics, fine}]},
+    rabbit_ct_helpers:merge_app_env(Config, AppEnv).
+
+init_per_group(cluster_tests, Config) ->
+    rabbit_ct_helpers:log_environment(),
+    Conf = [{rmq_nodename_suffix, cluster_tests}, {rmq_nodes_count, 2}],
+    Config1 = rabbit_ct_helpers:set_config(Config, Conf),
+    rabbit_ct_helpers:run_setup_steps(Config1, setup_steps());
+init_per_group(non_parallel_tests, Config) ->
     rabbit_ct_helpers:log_environment(),
-    Config1 = rabbit_ct_helpers:set_config(Config, [
-                                                    {rmq_nodename_suffix, ?MODULE}
-                                                   ]),
-    rabbit_ct_helpers:run_setup_steps(
-      Config1,
-      [ fun merge_app_env/1 ] ++ rabbit_ct_broker_helpers:setup_steps()).
-
-end_per_suite(Config) ->
+    Conf = [{rmq_nodename_suffix, non_parallel_tests}],
+    Config1 = rabbit_ct_helpers:set_config(Config, Conf),
+    rabbit_ct_helpers:run_setup_steps(Config1, setup_steps()).
+
+end_per_group(_, Config) ->
     rabbit_ct_helpers:run_teardown_steps(
       Config,
       rabbit_ct_broker_helpers:teardown_steps()).
 
-init_per_group(_, Config) ->
-    Config.
-
-end_per_group(_, Config) ->
-    Config.
-
 init_per_testcase(Testcase, Config) ->
     rabbit_ct_helpers:testcase_started(Config, Testcase),
     rabbit_ct_helpers:run_steps(Config,
@@ -83,8 +79,11 @@ end_per_testcase(Testcase, Config) ->
       Config,
       rabbit_ct_client_helpers:teardown_steps()).
 
+setup_steps() ->
+    [ fun merge_app_env/1 ] ++ rabbit_ct_broker_helpers:setup_steps().
+
 %% -------------------------------------------------------------------
-%% Testcases.
+%% Single-node Testcases.
 %% -------------------------------------------------------------------
 
 queue_metrics(Config) ->
@@ -329,3 +328,59 @@ x(Name) ->
     #resource{ virtual_host = <<"/">>,
                kind = exchange,
                name = Name }.
+
+%% -------------------------------------------------------------------
+%% Cluster Testcases.
+%% -------------------------------------------------------------------
+
+cluster_queue_metrics(Config) ->
+    VHost = <<"/">>,
+    QueueName = <<"cluster_queue_metrics">>,
+    PolicyName = <<"ha-policy-1">>,
+    PolicyPattern = <<".*">>,
+    PolicyAppliesTo = <<"queues">>,
+
+    Node0 = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    Node1 = rabbit_ct_broker_helpers:get_node_config(Config, 1, nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Node0),
+
+    Node0Name = rabbit_data_coercion:to_binary(Node0),
+    Definition0 = [{<<"ha-mode">>, <<"nodes">>}, {<<"ha-params">>, [Node0Name]}],
+    ok = rabbit_ct_broker_helpers:set_policy(Config, 0,
+                                             PolicyName, PolicyPattern,
+                                             PolicyAppliesTo, Definition0),
+
+    amqp_channel:call(Ch, #'queue.declare'{queue = QueueName}),
+    amqp_channel:call(Ch, #'basic.publish'{routing_key = QueueName},
+                          #amqp_msg{payload = <<"hello">>}),
+
+    % Update policy to point to other node
+    Node1Name = rabbit_data_coercion:to_binary(Node1),
+    Definition1 = [{<<"ha-mode">>, <<"nodes">>}, {<<"ha-params">>, [Node1Name]}],
+    ok = rabbit_ct_broker_helpers:set_policy(Config, 0,
+                                             PolicyName, PolicyPattern,
+                                             PolicyAppliesTo, Definition1),
+
+    % Synchronize
+    Name = rabbit_misc:r(VHost, queue, QueueName),
+    [#amqqueue{pid = QPid}] = rabbit_ct_broker_helpers:rpc(Config, Node0,
+                                                           ets, lookup,
+                                                           [rabbit_queue, Name]),
+    ok = rabbit_ct_broker_helpers:rpc(Config, Node0, rabbit_amqqueue,
+                                      sync_mirrors, [QPid]),
+
+    timer:sleep(1500),
+
+    % Check ETS table for data
+    % rabbit_core_metrics:queue_stats
+    [] = rabbit_ct_broker_helpers:rpc(Config, Node0, ets, tab2list,
+                                      [queue_coarse_metrics]),
+
+    [{Name, 1, 0, 1, _}] = rabbit_ct_broker_helpers:rpc(Config, Node1, ets,
+                                                        tab2list,
+                                                        [queue_coarse_metrics]),
+
+    amqp_channel:call(Ch, #'queue.delete'{queue=QueueName}),
+    rabbit_ct_client_helpers:close_channel(Ch),
+    Config.