Merge branch 'stable'

Author: dumbbell

src/inet_proxy_dist.erl

@@ -69,7 +69,11 @@ do_setup(Kernel, Node, Type, MyNode, LongOrShortNames,SetupTime) ->
                         undefined -> []
                     end,
                     ProxyPort = case inet_tcp_proxy:is_enabled() of
-                        true  -> proplists:get_value(TcpPort, PortsMap, TcpPort);
+                        true  -> P = proplists:get_value(TcpPort, PortsMap, TcpPort),
+                                 error_logger:info_msg(
+                                   "Using inet_tcp_proxy to connect to ~s (remote port changed from ~b to ~b)~n",
+                                   [Node, TcpPort, P]),
+                                 P;
                         false -> TcpPort
                     end,
 		    case inet_tcp:connect(Ip, ProxyPort,

src/inet_tcp_proxy.erl

@@ -20,7 +20,8 @@
 
 -export([start/3, reconnect/1, is_enabled/0, allow/1, block/1]).
 
--define(TABLE, ?MODULE).
+-define(NODES_TO_BLOCK, inet_tcp_proxy__nodes_to_block).
+-define(NODES_BLOCKED, inet_tcp_proxy__nodes_blocked).
 
 %% This can't start_link because there's no supervision hierarchy we
 %% can easily fit it into (we need to survive all application
@@ -44,16 +45,16 @@ reconnect(Nodes) ->
     ok.
 
 is_enabled() ->
-    lists:member(?TABLE, ets:all()).
+    lists:member(?NODES_TO_BLOCK, ets:all()).
 
 allow(Node) ->
-    error_logger:info_msg("(~s) Allowing distribution between ~s and ~s~n",
+    rabbit_log:warning("(~s) Allowing distribution between ~s and ~s~n",
       [?MODULE, node(), Node]),
-    ets:delete(?TABLE, Node).
+    true = ets:delete(?NODES_TO_BLOCK, Node).
 block(Node) ->
-    error_logger:info_msg("(~s) BLOCKING distribution between ~s and ~s~n",
+    rabbit_log:warning("(~s) BLOCKING distribution between ~s and ~s~n",
       [?MODULE, node(), Node]),
-    ets:insert(?TABLE, {Node, block}).
+    true = ets:insert(?NODES_TO_BLOCK, {Node, block}).
 
 %%----------------------------------------------------------------------------
 
@@ -67,14 +68,23 @@ error_handler(Thunk) ->
                     %% over; die quietly.
                     ok;
                   _:X ->
-                    io:format(user, "TCP proxy died with ~p~n At ~p~n",
+                    error_logger:error_msg(
+                      "TCP proxy died with ~p~n At ~p~n",
+                      [X, erlang:get_stacktrace()]),
+                    io:format(standard_error,
+                              "TCP proxy died with ~p~n At ~p~n",
                               [X, erlang:get_stacktrace()]),
+                    timer:sleep(1000),
                     erlang:halt(1)
             end
     end.
 
 go(Parent, Port, ProxyPort) ->
-    ets:new(?TABLE, [public, named_table]),
+    ets:new(?NODES_TO_BLOCK, [public, named_table]),
+    ets:new(?NODES_BLOCKED, [public, named_table]),
+    rabbit_log:info(
+      "(~s) Listening on proxy port ~p~n",
+      [?MODULE, ProxyPort]),
     {ok, Sock} = gen_tcp:listen(ProxyPort, [inet,
                                             {reuseaddr, true}]),
     Parent ! ready,
@@ -88,8 +98,12 @@ accept_loop(ListenSock, Port) ->
 
 run_it(SockIn, Port) ->
     case {inet:peername(SockIn), inet:sockname(SockIn)} of
-        {{ok, {_Addr, SrcPort}}, {ok, {Addr, _OtherPort}}} ->
+        {{ok, {_Addr, SrcPort}}, {ok, {Addr, OtherPort}}} ->
             {ok, Remote, This} = inet_tcp_proxy_manager:lookup(SrcPort),
+            rabbit_log:info(
+              "(~s) => Incoming proxied connection from node ~s (port ~b) "
+              "to node ~s (port ~b)~n",
+              [?MODULE, Remote, SrcPort, This, OtherPort]),
             case node() of
                 This  -> ok;
                 _     -> exit({not_me, node(), This})
@@ -101,18 +115,20 @@ run_it(SockIn, Port) ->
     end.
 
 run_loop(Sockets, RemoteNode, Buf0) ->
-    Block = [{RemoteNode, block}] =:= ets:lookup(?TABLE, RemoteNode),
+    Block = [{RemoteNode, block}] =:= ets:lookup(?NODES_TO_BLOCK, RemoteNode),
+    WasBlocked = [{RemoteNode, blocked}] =:= ets:lookup(?NODES_BLOCKED,
+                                                        RemoteNode),
     receive
         {tcp, Sock, Data} ->
             Buf = [Data | Buf0],
-            case {Block, get(dist_was_blocked)} of
-                {true, false} ->
-                    put(dist_was_blocked, Block),
+            case {WasBlocked, Block} of
+                {false, true} ->
+                    true = ets:insert(?NODES_BLOCKED, {RemoteNode, blocked}),
                     rabbit_log:warning(
                       "(~s) Distribution BLOCKED between ~s and ~s~n",
                       [?MODULE, node(), RemoteNode]);
-                {false, S} when S =:= true orelse S =:= undefined ->
-                    put(dist_was_blocked, Block),
+                {true, false} ->
+                    true = ets:delete(?NODES_BLOCKED, RemoteNode),
                     rabbit_log:warning(
                       "(~s) Distribution allowed between ~s and ~s~n",
                       [?MODULE, node(), RemoteNode]);
@@ -125,6 +141,9 @@ run_loop(Sockets, RemoteNode, Buf0) ->
                 true  -> run_loop(Sockets, RemoteNode, Buf)
             end;
         {tcp_closed, Sock} ->
+            rabbit_log:info(
+              "(~s) Distribution closed between ~s and ~s~n",
+              [?MODULE, node(), RemoteNode]),
             gen_tcp:close(other(Sock, Sockets));
         X ->
             exit({weirdness, X})

src/rabbit_ct_broker_helpers.erl

@@ -55,7 +55,8 @@
     set_partition_handling_mode_globally/2,
     enable_dist_proxy_manager/1,
     enable_dist_proxy/1,
-    enable_dist_proxy_on_node/3,
+    start_dist_proxy_on_node/2,
+    disconnect_from_other_nodes/2,
     block_traffic_between/2,
     allow_traffic_between/2,
 
@@ -634,20 +635,38 @@ enable_dist_proxy(Config) ->
     NodeConfigs = rabbit_ct_broker_helpers:get_node_configs(Config),
     Nodes = [?config(nodename, NodeConfig) || NodeConfig <- NodeConfigs],
     ManagerNode = node(),
+    %% We first start the proxy process on all nodes, then we close the
+    %% existing connection.
+    %%
+    %% If we do that in a single loop, i.e. start the proxy on node 1
+    %% and disconnect it, then, start the proxy on node 2 and disconnect
+    %% it, etc., there is a chance that the connection is reopened
+    %% by a node where the proxy is still disabled. Therefore, that
+    %% connection would bypass the proxy process even though we believe
+    %% it to be enabled.
     ok = lists:foreach(
       fun(NodeConfig) ->
           ok = rabbit_ct_broker_helpers:rpc(Config,
             ?config(nodename, NodeConfig),
-            ?MODULE, enable_dist_proxy_on_node,
-            [NodeConfig, ManagerNode, Nodes])
+            ?MODULE, start_dist_proxy_on_node,
+            [NodeConfig, ManagerNode])
+      end, NodeConfigs),
+    ok = lists:foreach(
+      fun(NodeConfig) ->
+          ok = rabbit_ct_broker_helpers:rpc(Config,
+            ?config(nodename, NodeConfig),
+            ?MODULE, disconnect_from_other_nodes,
+            [NodeConfig, Nodes])
       end, NodeConfigs),
     Config.
 
-enable_dist_proxy_on_node(NodeConfig, ManagerNode, Nodes) ->
-    Nodename = ?config(nodename, NodeConfig),
+start_dist_proxy_on_node(NodeConfig, ManagerNode) ->
     DistPort = ?config(tcp_port_erlang_dist, NodeConfig),
     ProxyPort = ?config(tcp_port_erlang_dist_proxy, NodeConfig),
-    ok = inet_tcp_proxy:start(ManagerNode, DistPort, ProxyPort),
+    ok = inet_tcp_proxy:start(ManagerNode, DistPort, ProxyPort).
+
+disconnect_from_other_nodes(NodeConfig, Nodes) ->
+    Nodename = ?config(nodename, NodeConfig),
     ok = inet_tcp_proxy:reconnect(Nodes -- [Nodename]).
 
 block_traffic_between(NodeA, NodeB) ->