Endpoint to list down links

Makefile

@@ -1,7 +1,7 @@
 PROJECT = rabbitmq_federation_management
 PROJECT_DESCRIPTION = RabbitMQ Federation Management
 
-DEPS = rabbit_common rabbit rabbitmq_management
+DEPS = rabbit_common rabbit rabbitmq_management rabbitmq_federation
 
 DEP_PLUGINS = rabbit_common/mk/rabbitmq-plugin.mk
 

src/rabbit_federation_mgmt.erl

@@ -26,17 +26,20 @@
 
 -include_lib("rabbitmq_management_agent/include/rabbit_mgmt_records.hrl").
 
-dispatcher() -> [{"/federation-links",        ?MODULE, []},
-                 {"/federation-links/:vhost", ?MODULE, []}].
+dispatcher() -> [{"/federation-links", ?MODULE, [all]},
+                 {"/federation-links/:vhost", ?MODULE, [all]},
+                 {"/federation-links/state/down/", ?MODULE, [down]},
+                 {"/federation-links/:vhost/state/down", ?MODULE, [down]}].
+
 web_ui()     -> [{javascript, <<"federation.js">>}].
 
 %%--------------------------------------------------------------------
 
 init(_, _, _) ->
     {upgrade, protocol, cowboy_rest}.
 
-rest_init(Req, _Opts) ->
-    {ok, Req, #context{}}.
+rest_init(Req, [Filter]) ->
+    {ok, Req, {Filter, #context{}}}.
 
 content_types_provided(ReqData, Context) ->
    {[{<<"application/json">>, to_json}], ReqData, Context}.
@@ -47,14 +50,15 @@ resource_exists(ReqData, Context) ->
          _         -> true
      end, ReqData, Context}.
 
-to_json(ReqData, Context) ->
+to_json(ReqData, {Filter, Context}) ->
     Chs = rabbit_mgmt_db:get_all_channels(
             rabbit_mgmt_util:range(ReqData)),
     rabbit_mgmt_util:reply_list(
-      filter_vhost(status(Chs, ReqData, Context), ReqData), ReqData, Context).
+      filter_vhost(status(Chs, ReqData, Context, Filter), ReqData), ReqData, Context).
 
-is_authorized(ReqData, Context) ->
-    rabbit_mgmt_util:is_authorized_monitor(ReqData, Context).
+is_authorized(ReqData, {Filter, Context}) ->
+    {Res, RD, C} = rabbit_mgmt_util:is_authorized_monitor(ReqData, Context),
+    {Res, RD, {Filter, C}}.
 
 %%--------------------------------------------------------------------
 
@@ -63,18 +67,25 @@ filter_vhost(List, ReqData) ->
       ReqData,
       fun(V) -> lists:filter(fun(I) -> pget(vhost, I) =:= V end, List) end).
 
-status(Chs, ReqData, Context) ->
+status(Chs, ReqData, Context, Filter) ->
     rabbit_mgmt_util:filter_vhost(
-      lists:append([status(Node, Chs) || Node <- [node() | nodes()]]),
+      lists:append([status(Node, Chs, Filter) || Node <- [node() | nodes()]]),
       ReqData, Context).
 
-status(Node, Chs) ->
+status(Node, Chs, Filter) ->
     case rpc:call(Node, rabbit_federation_status, status, [], infinity) of
         {badrpc, {'EXIT', {undef, _}}}  -> [];
         {badrpc, {'EXIT', {noproc, _}}} -> [];
-        Status                          -> [format(Node, I, Chs) || I <- Status]
+        Status                          -> [format(Node, I, Chs) || I <- Status,
+                                                                    filter_status(I, Filter)]
     end.
 
+filter_status(_, all) ->
+    true;
+filter_status(Props, down) ->
+    Status = pget(status, Props),
+    not lists:member(Status, [running, starting]).
+
 format(Node, Info, Chs) ->
     LocalCh = case rabbit_mgmt_format:strip_pids(
                      [Ch || Ch <- Chs,

test/federation_mgmt_SUITE.erl

@@ -0,0 +1,197 @@
+%% 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 Federation.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2007-2016 Pivotal Software, Inc.  All rights reserved.
+%%
+
+-module(federation_mgmt_SUITE).
+
+-include_lib("common_test/include/ct.hrl").
+-include_lib("amqp_client/include/amqp_client.hrl").
+-include_lib("rabbitmq_management/include/rabbit_mgmt_test.hrl").
+
+-compile(export_all).
+
+all() ->
+    [
+      {group, non_parallel_tests}
+    ].
+
+groups() ->
+    [
+     {non_parallel_tests, [], [
+                               federation_links,
+                               federation_down_links
+                              ]}
+    ].
+
+%% -------------------------------------------------------------------
+%% Testsuite setup/teardown.
+%% -------------------------------------------------------------------
+init_per_suite(Config) ->
+    rabbit_ct_helpers:log_environment(),
+    inets:start(),
+    Config1 = rabbit_ct_helpers:set_config(Config, [
+                                                    {rmq_nodename_suffix, ?MODULE}
+                                                   ]),
+    rabbit_ct_helpers:run_setup_steps(Config1,
+                                      rabbit_ct_broker_helpers:setup_steps() ++
+                                          rabbit_ct_client_helpers:setup_steps() ++
+                                          [fun setup_federation/1]).
+end_per_suite(Config) ->
+    rabbit_ct_helpers:run_teardown_steps(Config,
+                                         rabbit_ct_client_helpers:teardown_steps() ++
+                                             rabbit_ct_broker_helpers:teardown_steps()).
+
+setup_federation(Config) ->
+    rabbit_ct_broker_helpers:set_policy(
+      Config, 0,
+      <<"fed">>, <<".*">>, <<"all">>, [{<<"federation-upstream-set">>, <<"all">>}]),
+    rabbit_ct_broker_helpers:set_parameter(
+      Config, 0, <<"federation-upstream">>, <<"broken-bunny">>,
+      [{<<"uri">>, <<"amqp://broken-bunny">>},
+       {<<"reconnect-delay">>, 600000}]),
+    rabbit_ct_broker_helpers:set_parameter(
+      Config, 0, <<"federation-upstream">>, <<"bunny">>,
+      [{<<"uri">>, <<"amqp://">>},
+       {<<"reconnect-delay">>, 600000}]),
+    Config.
+
+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.
+%% -------------------------------------------------------------------
+federation_links(Config) ->
+    DefaultExchanges = [<<"amq.direct">>, <<"amq.fanout">>, <<"amq.headers">>,
+                        <<"amq.match">>, <<"amq.topic">>],
+    Running = [{X, <<"bunny">>, <<"running">>} || X <- DefaultExchanges],
+    Down = [{X, <<"broken-bunny">>, <<"error">>} || X <- DefaultExchanges],
+    All = lists:sort(Running ++ Down),
+    Verify = fun(Result) ->
+                     All == lists:sort(Result)
+             end,
+    %% Verify we have 5 running links and 5 down links
+    wait_until(fun() ->
+                       AllLinks = http_get(Config, "/federation-links"),
+                       Result = [{proplists:get_value(exchange, Link),
+                                  proplists:get_value(upstream, Link),
+                                  proplists:get_value(status, Link)} || Link <- AllLinks],
+                       Verify(Result)
+               end).
+
+federation_down_links(Config) ->
+    DefaultExchanges = [<<"amq.direct">>, <<"amq.fanout">>, <<"amq.headers">>,
+                        <<"amq.match">>, <<"amq.topic">>],
+    Down = lists:sort([{X, <<"broken-bunny">>, <<"error">>} || X <- DefaultExchanges]),
+    Verify = fun(Result) ->
+                     %% we might have to wait for all links to get into 'error' status,
+                     %% but no other status is allowed on the meanwhile
+                     lists:all(fun({_, _, <<"error">>}) ->
+                                       true;
+                                  (_) ->
+                                       throw(down_links_returned_wrong_status)
+                               end, Result) andalso (Down == lists:sort(Result))
+             end,
+    wait_until(fun() ->
+                       AllLinks = http_get(Config, "/federation-links/state/down"),
+                       Result = [{proplists:get_value(exchange, Link),
+                                  proplists:get_value(upstream, Link),
+                                  proplists:get_value(status, Link)} || Link <- AllLinks],
+                       Verify(Result)
+               end).
+
+%% -------------------------------------------------------------------
+%% Helpers
+%% -------------------------------------------------------------------
+wait_until(Fun) ->
+    wait_until(Fun, 120).
+
+wait_until(Fun, 0) ->
+    throw(federation_links_timeout);
+wait_until(Fun, N) ->
+    case Fun() of
+        true ->
+            ok;
+        false ->
+            timer:sleep(1000),
+            wait_until(Fun, N-1)
+    end.
+
+%% -------------------------------------------------------------------
+%% Helpers from rabbitmq_management tests
+%% -------------------------------------------------------------------
+http_get(Config, Path) ->
+    http_get(Config, Path, ?OK).
+
+http_get(Config, Path, CodeExp) ->
+    http_get(Config, Path, "guest", "guest", CodeExp).
+
+http_get(Config, Path, User, Pass, CodeExp) ->
+    {ok, {{_HTTP, CodeAct, _}, Headers, ResBody}} =
+        req(Config, 0, get, Path, [auth_header(User, Pass)]),
+    assert_code(CodeExp, CodeAct, "GET", Path, ResBody),
+    decode(CodeExp, Headers, ResBody).
+
+req(Config, Node, Type, Path, Headers) ->
+    httpc:request(Type, {uri_base_from(Config, Node) ++ Path, Headers}, ?HTTPC_OPTS, []).
+
+uri_base_from(Config, Node) ->
+    binary_to_list(
+      rabbit_mgmt_format:print(
+        "http://localhost:~w/api",
+        [mgmt_port(Config, Node)])).
+
+auth_header(Username, Password) ->
+    {"Authorization",
+     "Basic " ++ binary_to_list(base64:encode(Username ++ ":" ++ Password))}.
+
+mgmt_port(Config, Node) ->
+    rabbit_ct_broker_helpers:get_node_config(Config, Node, tcp_port_mgmt).
+
+assert_code(CodesExpected, CodeAct, Type, Path, Body) when is_list(CodesExpected) ->
+    case lists:member(CodeAct, CodesExpected) of
+        true ->
+            ok;
+        false ->
+            throw({expected, CodesExpected, got, CodeAct, type, Type,
+                   path, Path, body, Body})
+    end;
+assert_code(CodeExp, CodeAct, Type, Path, Body) ->
+    case CodeExp of
+        CodeAct -> ok;
+        _       -> throw({expected, CodeExp, got, CodeAct, type, Type,
+                          path, Path, body, Body})
+    end.
+
+decode(?OK, _Headers,  ResBody) -> cleanup(mochijson2:decode(ResBody));
+decode(_,    Headers, _ResBody) -> Headers.
+
+cleanup(L) when is_list(L) ->
+    [cleanup(I) || I <- L];
+cleanup({struct, I}) ->
+    cleanup(I);
+cleanup({K, V}) when is_binary(K) ->
+    {list_to_atom(binary_to_list(K)), cleanup(V)};
+cleanup(I) ->
+    I.