Merge branch 'master' into rabbitmq-server-567

3 files changed, 416 insertions, 36 deletions

diff --git a/test/metrics_SUITE.erl b/test/metrics_SUITE.erl
new file mode 100644
index 0000000000..b2b0fe3560
--- /dev/null
+++ b/test/metrics_SUITE.erl
@@ -0,0 +1,376 @@
+%% 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 http://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) 2016 Pivotal Software, Inc.  All rights reserved.
+%%
+-module(metrics_SUITE).
+-compile(export_all).
+
+-include_lib("common_test/include/ct.hrl").
+-include_lib("proper/include/proper.hrl").
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("amqp_client/include/amqp_client.hrl").
+
+
+all() ->
+    [
+      {group, non_parallel_tests}
+    ].
+
+groups() ->
+    [
+      {non_parallel_tests, [], [
+                                connection,
+                                channel,
+                                channel_connection_close,
+                                channel_queue_exchange_consumer_close_connection,
+                                channel_queue_delete_queue,
+                                connection_metric_count_test,
+                                channel_metric_count_test,
+                                queue_metric_count_test,
+                                queue_metric_count_channel_per_queue_test,
+                                connection_metric_idemp_test,
+                                channel_metric_idemp_test,
+                                queue_metric_idemp_test
+                               ]}
+    ].
+
+%% -------------------------------------------------------------------
+%% Testsuite setup/teardown.
+%% -------------------------------------------------------------------
+
+merge_app_env(Config) ->
+    rabbit_ct_helpers:merge_app_env(Config,
+                                    {rabbit, [
+                                              {collect_statistics, fine},
+                                              {collect_statistics_interval, 500}
+                                             ]}).
+init_per_suite(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) ->
+    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).
+
+end_per_testcase(Testcase, Config) ->
+    rabbit_ct_helpers:testcase_finished(Config, Testcase).
+
+
+%% -------------------------------------------------------------------
+%% Testcases.
+%% -------------------------------------------------------------------
+
+% NB: node_stats tests are in the management_agent repo
+
+connection_metric_count_test(Config) ->
+    rabbit_ct_proper_helpers:run_proper(fun prop_connection_metric_count/1, [Config], 25).
+
+channel_metric_count_test(Config) ->
+    rabbit_ct_proper_helpers:run_proper(fun prop_channel_metric_count/1, [Config], 25).
+
+queue_metric_count_test(Config) ->
+    rabbit_ct_proper_helpers:run_proper(fun prop_queue_metric_count/1, [Config], 5).
+
+queue_metric_count_channel_per_queue_test(Config) ->
+    rabbit_ct_proper_helpers:run_proper(fun prop_queue_metric_count_channel_per_queue/1,
+                                     [Config], 5).
+
+connection_metric_idemp_test(Config) ->
+    rabbit_ct_proper_helpers:run_proper(fun prop_connection_metric_idemp/1, [Config], 25).
+
+channel_metric_idemp_test(Config) ->
+    rabbit_ct_proper_helpers:run_proper(fun prop_channel_metric_idemp/1, [Config], 25).
+
+queue_metric_idemp_test(Config) ->
+    rabbit_ct_proper_helpers:run_proper(fun prop_queue_metric_idemp/1, [Config], 25).
+
+prop_connection_metric_idemp(Config) ->
+    ?FORALL(N, {integer(1, 25), integer(1, 25)},
+            connection_metric_idemp(Config, N)).
+
+prop_channel_metric_idemp(Config) ->
+    ?FORALL(N, {integer(1, 25), integer(1, 25)},
+            channel_metric_idemp(Config, N)).
+
+prop_queue_metric_idemp(Config) ->
+    ?FORALL(N, {integer(1, 25), integer(1, 25)},
+            queue_metric_idemp(Config, N)).
+
+prop_connection_metric_count(Config) ->
+    ?FORALL(N, {integer(1, 25), resize(100, list(oneof([add, remove])))},
+            connection_metric_count(Config, N)).
+
+prop_channel_metric_count(Config) ->
+    ?FORALL(N, {integer(1, 25), resize(100, list(oneof([add, remove])))},
+            channel_metric_count(Config, N)).
+
+prop_queue_metric_count(Config) ->
+    ?FORALL(N, {integer(1, 10), resize(10, list(oneof([add, remove])))},
+            queue_metric_count(Config, N)).
+
+prop_queue_metric_count_channel_per_queue(Config) ->
+    ?FORALL(N, {integer(1, 10), resize(10, list(oneof([add, remove])))},
+            queue_metric_count_channel_per_queue(Config, N)).
+
+connection_metric_idemp(Config, {N, R}) ->
+    Conns = [rabbit_ct_client_helpers:open_unmanaged_connection(Config)
+               || _ <- lists:seq(1, N)],
+    Table = [ Pid || {Pid, _} <- read_table_rpc(Config, connection_metrics)],
+    Table2 = [ Pid || {Pid, _} <- read_table_rpc(Config, connection_coarse_metrics)],
+    % referesh stats 'R' times
+    [[Pid ! emit_stats || Pid <- Table] || _ <- lists:seq(1, R)],
+    timer:sleep(100),
+    TableAfter = [ Pid || {Pid, _} <- read_table_rpc(Config, connection_metrics)],
+    TableAfter2 = [ Pid || {Pid, _} <- read_table_rpc(Config, connection_coarse_metrics)],
+    [rabbit_ct_client_helpers:close_connection(Conn) || Conn <- Conns],
+    (Table2 == TableAfter2) and (Table == TableAfter) and
+    (N == length(Table)) and (N == length(TableAfter)).
+
+channel_metric_idemp(Config, {N, R}) ->
+    Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config),
+    [amqp_connection:open_channel(Conn) || _ <- lists:seq(1, N)],
+    Table = [ Pid || {Pid, _} <- read_table_rpc(Config, channel_metrics)],
+    Table2 = [ Pid || {Pid, _} <- read_table_rpc(Config, channel_process_metrics)],
+    % referesh stats 'R' times
+    [[Pid ! emit_stats || Pid <- Table] || _ <- lists:seq(1, R)],
+    timer:sleep(100),
+    TableAfter = [ Pid || {Pid, _} <- read_table_rpc(Config, channel_metrics)],
+    TableAfter2 = [ Pid || {Pid, _} <- read_table_rpc(Config, channel_process_metrics)],
+    rabbit_ct_client_helpers:close_connection(Conn),
+    (Table2 == TableAfter2) and (Table == TableAfter) and
+    (N == length(Table)) and (N == length(TableAfter)).
+
+queue_metric_idemp(Config, {N, R}) ->
+    Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config),
+    {ok, Chan} = amqp_connection:open_channel(Conn),
+    Queues =
+        [begin
+              Queue = declare_queue(Chan),
+              ensure_exchange_metrics_populated(Chan, Queue),
+              ensure_channel_queue_metrics_populated(Chan, Queue),
+              Queue
+          end || _ <- lists:seq(1, N)],
+    Table = [ Pid || {Pid, _} <- read_table_rpc(Config, queue_metrics)],
+    Table2 = [ Pid || {Pid, _} <- read_table_rpc(Config, queue_coarse_metrics)],
+    % referesh stats 'R' times
+    ChanTable = read_table_rpc(Config, channel_created),
+    [[Pid ! emit_stats || {Pid, _} <- ChanTable ] || _ <- lists:seq(1, R)],
+    timer:sleep(100),
+    TableAfter = [ Pid || {Pid, _} <- read_table_rpc(Config,  queue_metrics)],
+    TableAfter2 = [ Pid || {Pid, _} <- read_table_rpc(Config, queue_coarse_metrics)],
+    [ delete_queue(Chan, Q) || Q <- Queues],
+    rabbit_ct_client_helpers:close_connection(Conn),
+    (Table2 == TableAfter2) and (Table == TableAfter) and
+    (N == length(Table)) and (N == length(TableAfter)).
+
+connection_metric_count(Config, Ops) ->
+    add_rem_counter(Config, Ops,
+                    {fun rabbit_ct_client_helpers:open_unmanaged_connection/1,
+                     fun rabbit_ct_client_helpers:close_connection/1},
+                   [ connection_created,
+                     connection_metrics,
+                     connection_coarse_metrics ]).
+
+channel_metric_count(Config, Ops) ->
+    Conn =  rabbit_ct_client_helpers:open_unmanaged_connection(Config),
+    Result = add_rem_counter(Config, Ops,
+                    {fun (_Config) ->
+                             {ok, Chan} = amqp_connection:open_channel(Conn),
+                             Chan
+                     end,
+                     fun amqp_channel:close/1},
+                   [ channel_created,
+                     channel_metrics,
+                     channel_process_metrics ]),
+    ok = rabbit_ct_client_helpers:close_connection(Conn),
+    Result.
+
+queue_metric_count(Config, Ops) ->
+    Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config),
+    {ok, Chan} = amqp_connection:open_channel(Conn),
+    AddFun = fun (_) ->
+                     Queue = declare_queue(Chan),
+                     ensure_exchange_metrics_populated(Chan, Queue),
+                     ensure_channel_queue_metrics_populated(Chan, Queue),
+                     force_channel_stats(Config),
+                     Queue
+             end,
+    Result = add_rem_counter(Config, Ops,
+                    {AddFun,
+                     fun (Q) -> delete_queue(Chan, Q) end},
+                   [ channel_queue_metrics,
+                     channel_queue_exchange_metrics ]),
+    ok = rabbit_ct_client_helpers:close_connection(Conn),
+    Result.
+
+queue_metric_count_channel_per_queue(Config, Ops) ->
+    Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config),
+    AddFun = fun (_) ->
+                     {ok, Chan} = amqp_connection:open_channel(Conn),
+                     Queue = declare_queue(Chan),
+                     ensure_exchange_metrics_populated(Chan, Queue),
+                     ensure_channel_queue_metrics_populated(Chan, Queue),
+                     force_channel_stats(Config),
+                     {Chan, Queue}
+             end,
+    Result = add_rem_counter(Config, Ops,
+                    {AddFun,
+                     fun ({Chan, Q}) -> delete_queue(Chan, Q) end},
+                   [ channel_queue_metrics,
+                     channel_queue_exchange_metrics ]),
+    ok = rabbit_ct_client_helpers:close_connection(Conn),
+    Result.
+
+add_rem_counter(Config, {Initial, Ops}, {AddFun, RemFun}, Tables) ->
+    Things = [ AddFun(Config) || _ <- lists:seq(1, Initial) ],
+    % either add or remove some things
+    {FinalLen, Things1} =
+        lists:foldl(fun(add, {L, Items}) ->
+                                {L+1, [AddFun(Config) | Items]};
+                       (remove, {L, [H|Tail]}) ->
+                                RemFun(H),
+                                {L-1, Tail};
+                       (_, S) -> S end,
+                    {Initial, Things},
+                    Ops),
+    TabLens = lists:map(fun(T) ->
+                                length(read_table_rpc(Config, T)) end, Tables),
+    [RemFun(Thing) || Thing <- Things1],
+    [FinalLen] == lists:usort(TabLens).
+
+
+connection(Config) ->
+    Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config),
+    [_] = read_table_rpc(Config, connection_created),
+    [_] = read_table_rpc(Config, connection_metrics),
+    [_] = read_table_rpc(Config, connection_coarse_metrics),
+    ok = rabbit_ct_client_helpers:close_connection(Conn),
+    [] = read_table_rpc(Config, connection_created),
+    [] = read_table_rpc(Config, connection_metrics),
+    [] = read_table_rpc(Config, connection_coarse_metrics).
+
+channel(Config) ->
+    Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config),
+    {ok, Chan} = amqp_connection:open_channel(Conn),
+    [_] = read_table_rpc(Config, channel_created),
+    [_] = read_table_rpc(Config, channel_metrics),
+    [_] = read_table_rpc(Config, channel_process_metrics),
+    ok = amqp_channel:close(Chan),
+    [] = read_table_rpc(Config, channel_created),
+    [] = read_table_rpc(Config, channel_metrics),
+    [] = read_table_rpc(Config, channel_process_metrics),
+    ok = rabbit_ct_client_helpers:close_connection(Conn).
+
+channel_connection_close(Config) ->
+    Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config),
+    {ok, _} = amqp_connection:open_channel(Conn),
+    [_] = read_table_rpc(Config, channel_created),
+    [_] = read_table_rpc(Config, channel_metrics),
+    [_] = read_table_rpc(Config, channel_process_metrics),
+    ok = rabbit_ct_client_helpers:close_connection(Conn),
+    [] = read_table_rpc(Config, channel_created),
+    [] = read_table_rpc(Config, channel_metrics),
+    [] = read_table_rpc(Config, channel_process_metrics).
+
+channel_queue_delete_queue(Config) ->
+    Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config),
+    {ok, Chan} = amqp_connection:open_channel(Conn),
+    Queue = declare_queue(Chan),
+    ensure_exchange_metrics_populated(Chan, Queue),
+    ensure_channel_queue_metrics_populated(Chan, Queue),
+    force_channel_stats(Config),
+    [_] = read_table_rpc(Config, channel_queue_metrics),
+    [_] = read_table_rpc(Config, channel_queue_exchange_metrics),
+
+    delete_queue(Chan, Queue),
+    % ensure  removal of queue cleans up channel_queue metrics
+    [] = read_table_rpc(Config, channel_queue_exchange_metrics),
+    [] = read_table_rpc(Config, channel_queue_metrics),
+    ok = rabbit_ct_client_helpers:close_connection(Conn).
+
+channel_queue_exchange_consumer_close_connection(Config) ->
+    Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config),
+    {ok, Chan} = amqp_connection:open_channel(Conn),
+    Queue = declare_queue(Chan),
+    ensure_exchange_metrics_populated(Chan, Queue),
+    force_channel_stats(Config),
+
+    [_] = read_table_rpc(Config, channel_exchange_metrics),
+    [_] = read_table_rpc(Config, channel_queue_exchange_metrics),
+
+    ensure_channel_queue_metrics_populated(Chan, Queue),
+    force_channel_stats(Config),
+    [_] = read_table_rpc(Config, channel_queue_metrics),
+
+    Sub = #'basic.consume'{queue = Queue},
+      #'basic.consume_ok'{consumer_tag = _} =
+        amqp_channel:call(Chan, Sub),
+
+    [_] = read_table_rpc(Config, consumer_created),
+
+    ok = rabbit_ct_client_helpers:close_connection(Conn),
+    % ensure cleanup happened
+    [] = read_table_rpc(Config, channel_exchange_metrics),
+    [] = read_table_rpc(Config, channel_queue_exchange_metrics),
+    [] = read_table_rpc(Config, channel_queue_metrics),
+    [] = read_table_rpc(Config, consumer_created).
+
+
+%% -------------------------------------------------------------------
+%% Utilities
+%% -------------------------------------------------------------------
+
+declare_queue(Chan) ->
+    Declare = #'queue.declare'{durable = false, auto_delete = true},
+    #'queue.declare_ok'{queue = Name} = amqp_channel:call(Chan, Declare),
+    Name.
+
+delete_queue(Chan, Name) ->
+    Delete = #'queue.delete'{queue = Name},
+    #'queue.delete_ok'{} = amqp_channel:call(Chan, Delete).
+
+ensure_exchange_metrics_populated(Chan, RoutingKey) ->
+    % need to publish for exchange metrics to be populated
+    Publish = #'basic.publish'{routing_key = RoutingKey},
+    amqp_channel:call(Chan, Publish, #amqp_msg{payload = <<"hello">>}).
+
+ensure_channel_queue_metrics_populated(Chan, Queue) ->
+    % need to get and wait for timer for channel queue metrics to be populated
+    Get = #'basic.get'{queue = Queue, no_ack=true},
+    {#'basic.get_ok'{}, #amqp_msg{}} = amqp_channel:call(Chan, Get).
+
+force_channel_stats(Config) ->
+    [ Pid ! emit_stats || {Pid, _} <- read_table_rpc(Config, channel_created) ],
+    timer:sleep(100).
+
+read_table_rpc(Config, Table) ->
+    rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, read_table, [Table]).
+
+read_table(Table) ->
+    ets:tab2list(Table).
+
diff --git a/test/partitions_SUITE.erl b/test/partitions_SUITE.erl
index e00c015d02..3f0ec419c2 100644
--- a/test/partitions_SUITE.erl
+++ b/test/partitions_SUITE.erl
@@ -33,6 +33,9 @@
 %% It's a lot, but still better than timetrap_timeout
 -define(AWAIT_TIMEOUT, 300000).
 
+suite() ->
+    [{timetrap, {minutes, 60}}].
+
 all() ->
     [
       {group, net_ticktime_1},
diff --git a/test/unit_inbroker_SUITE.erl b/test/unit_inbroker_SUITE.erl
index 569c7a88fa..aee815aee4 100644
--- a/test/unit_inbroker_SUITE.erl
+++ b/test/unit_inbroker_SUITE.erl
@@ -2885,10 +2885,13 @@ channel_statistics1(_Config) ->
     dummy_event_receiver:start(self(), [node()], [channel_stats]),
 
     %% Check stats empty
-    Event = test_ch_statistics_receive_event(Ch, fun (_) -> true end),
-    [] = proplists:get_value(channel_queue_stats, Event),
-    [] = proplists:get_value(channel_exchange_stats, Event),
-    [] = proplists:get_value(channel_queue_exchange_stats, Event),
+    Check1 = fun() ->
+                 [] = ets:match(channel_queue_metrics, {Ch, QRes}),
+                 [] = ets:match(channel_exchange_metrics, {Ch, X}),
+                 [] = ets:match(channel_queue_exchange_metrics,
+                                {Ch, {QRes, X}})
+             end,
+    test_ch_metrics(Check1, ?TIMEOUT),
 
     %% Publish and get a message
     rabbit_channel:do(Ch, #'basic.publish'{exchange = <<"">>,
@@ -2897,46 +2900,44 @@ channel_statistics1(_Config) ->
     rabbit_channel:do(Ch, #'basic.get'{queue = QName}),
 
     %% Check the stats reflect that
-    Event2 = test_ch_statistics_receive_event(
-               Ch,
-               fun (E) ->
-                       length(proplists:get_value(
-                                channel_queue_exchange_stats, E)) > 0
-               end),
-    [{QRes, [{get,1}]}] = proplists:get_value(channel_queue_stats,    Event2),
-    [{X,[{publish,1}]}] = proplists:get_value(channel_exchange_stats, Event2),
-    [{{QRes,X},[{publish,1}]}] =
-        proplists:get_value(channel_queue_exchange_stats, Event2),
+    Check2 = fun() ->
+                 [{{Ch, QRes}, 1, 0, 0, 0, 0, 0}] = ets:lookup(
+                                                      channel_queue_metrics,
+                                                      {Ch, QRes}),
+                 [{{Ch, X}, 1, 0, 0}] = ets:lookup(
+                                          channel_exchange_metrics,
+                                          {Ch, X}),
+                 [{{Ch, {QRes, X}}, 1}] = ets:lookup(
+                                            channel_queue_exchange_metrics,
+                                            {Ch, {QRes, X}})
+             end,
+    test_ch_metrics(Check2, ?TIMEOUT),
 
     %% Check the stats remove stuff on queue deletion
     rabbit_channel:do(Ch, #'queue.delete'{queue = QName}),
-    Event3 = test_ch_statistics_receive_event(
-               Ch,
-               fun (E) ->
-                       length(proplists:get_value(
-                                channel_queue_exchange_stats, E)) == 0
-               end),
-
-    [] = proplists:get_value(channel_queue_stats, Event3),
-    [{X,[{publish,1}]}] = proplists:get_value(channel_exchange_stats, Event3),
-    [] = proplists:get_value(channel_queue_exchange_stats, Event3),
+    Check3 = fun() ->
+                 [] = ets:lookup(channel_queue_metrics, {Ch, QRes}),
+                 [{{Ch, X}, 1, 0, 0}] = ets:lookup(
+                                          channel_exchange_metrics,
+                                          {Ch, X}),
+                 [] = ets:lookup(channel_queue_exchange_metrics,
+                                 {Ch, {QRes, X}})
+             end,
+    test_ch_metrics(Check3, ?TIMEOUT),
 
     rabbit_channel:shutdown(Ch),
     dummy_event_receiver:stop(),
     passed.
 
-test_ch_statistics_receive_event(Ch, Matcher) ->
-    rabbit_channel:flush(Ch),
-    Ch ! emit_stats,
-    test_ch_statistics_receive_event1(Ch, Matcher).
-
-test_ch_statistics_receive_event1(Ch, Matcher) ->
-    receive #event{type = channel_stats, props = Props} ->
-            case Matcher(Props) of
-                true -> Props;
-                _    -> test_ch_statistics_receive_event1(Ch, Matcher)
-            end
-    after ?TIMEOUT -> throw(failed_to_receive_event)
+test_ch_metrics(Fun, Timeout) when Timeout =< 0 ->
+    Fun();
+test_ch_metrics(Fun, Timeout) ->
+    try
+        Fun()
+    catch
+        _:{badmatch, _} ->
+            timer:sleep(1000),
+            test_ch_metrics(Fun, Timeout - 1000)
     end.
 
 head_message_timestamp_statistics(Config) ->