Implement reconnection [WIP]

Author: chrzaszcz

src/config/mongoose_config_spec.erl

@@ -546,7 +546,8 @@ outgoing_pool_connection(<<"rabbit">>) ->
                  <<"confirms_enabled">> => #option{type = boolean},
                  <<"max_worker_queue_len">> => #option{type = int_or_infinity,
                                                        validate = non_negative},
-                 <<"tls">> => tls([client])
+                 <<"tls">> => tls([client]),
+                 <<"reconnect">> => rabbit_reconnect()
                 },
        include = always,
        defaults = #{<<"host">> => "localhost",
@@ -632,6 +633,15 @@ sql_tls() ->
 sql_tls_extra() ->
     #section{items = #{<<"required">> => #option{type = boolean}}}.
 
+%% path: outgoing_pools.rabbit.*.connection.reconnect
+rabbit_reconnect() ->
+    #section{items = #{<<"attempts">> => #option{type = integer, validate = positive},
+                       <<"delay">> => #option{type = integer, validate = non_negative}},
+             defaults = #{<<"attempts">> => 10,
+                          <<"delay">> => 5000 % milliseconds
+                         }
+            }.
+
 %% TLS options
 
 tls(Entities) when is_list(Entities) ->

src/mongoose_rabbit_worker.erl

@@ -46,7 +46,12 @@
                   password := binary(),
                   virtual_host := binary(),
                   confirms_enabled := boolean(),
-                  max_worker_queue_len := non_neg_integer() | infinity}.
+                  max_worker_queue_len := non_neg_integer() | infinity,
+                  reconnect => reconnect()}.
+
+-type reconnect() :: #{attempts := pos_integer(),
+                       delay := non_neg_integer() % milliseconds
+                      }.
 
 -type publish_result() :: boolean() | timeout | {channel_exception, any(), any()}.
 
@@ -197,7 +202,30 @@ maybe_restart_rabbit_connection(#{connection := Conn} = State) ->
     end.
 
 -spec establish_rabbit_connection(state()) -> state().
+establish_rabbit_connection(State = #{opts := #{reconnect := #{attempts := Attempts}}}) ->
+    establish_rabbit_connection(State, Attempts);
 establish_rabbit_connection(State) ->
+    establish_rabbit_connection(State, 0).
+
+-spec establish_rabbit_connection(state(), non_neg_integer()) -> state().
+establish_rabbit_connection(State, RemainingAttempts) ->
+    case start_amqp_connection(State) of
+        {ok, NewState} ->
+            NewState;
+        {error, Error} when RemainingAttempts > 0 ->
+            ?LOG_WARNING(#{what => rabbit_connection_failed, reason => Error, worker_state => State,
+                           remaining_attempts => RemainingAttempts}),
+            #{opts := #{reconnect := #{delay := Delay}}} = State,
+            timer:sleep(Delay),
+            establish_rabbit_connection(State, RemainingAttempts - 1);
+        {error, Error} when RemainingAttempts =:= 0 ->
+            ErrorInfo = #{what => rabbit_connection_failed, reason => Error, worker_state => State},
+            ?LOG_ERROR(ErrorInfo),
+            exit(ErrorInfo)
+    end.
+
+-spec start_amqp_connection(state()) -> {ok, state()} | {error, term()}.
+start_amqp_connection(State) ->
     #{opts := Opts, host_type := HostType, pool_tag := PoolTag} = State,
     case amqp_connection:start(mongoose_amqp:network_params(Opts)) of
         {ok, Connection} ->
@@ -208,14 +236,12 @@ establish_rabbit_connection(State) ->
             maybe_enable_confirms(Channel, Opts),
             ?LOG_DEBUG(#{what => rabbit_connection_established,
                          host_type => HostType, pool_tag => PoolTag, opts => Opts}),
-            State#{connection => Connection, channel => Channel};
+            {ok, State#{connection => Connection, channel => Channel}};
         {error, Error} ->
             mongoose_instrument:execute(wpool_rabbit_connections,
                                         #{host_type => HostType, pool_tag => PoolTag},
                                         #{failed => 1}),
-            ?LOG_ERROR(#{what => rabbit_connection_failed, reason => Error,
-                         host_type => HostType, pool_tag => PoolTag, opts => Opts}),
-            exit("connection to a Rabbit server failed")
+            {error, Error}
     end.
 
 -spec close_rabbit_connection(Connection :: pid(), Channel :: pid(),

test/common/config_parser_helper.erl

@@ -363,7 +363,8 @@ options("outgoing_pools") ->
          #{type => rabbit, scope => host_type, tag => event_pusher,
            opts => #{workers => 20},
            conn_opts => #{confirms_enabled => true,
-                          max_worker_queue_len => 100}},
+                          max_worker_queue_len => 100,
+                          reconnect => #{attempts => 5, delay => 2000}}},
          #{type => rdbms,
            opts => #{workers => 5},
            conn_opts => #{query_timeout => 5000, keepalive_interval => 30,
@@ -1302,6 +1303,8 @@ default_config([outgoing_pools, Type, _Tag, opts]) ->
     default_pool_wpool_opts(Type);
 default_config([outgoing_pools, Type, _Tag, conn_opts]) ->
     default_pool_conn_opts(Type);
+default_config([outgoing_pools, rabbit, _Tag, conn_opts, reconnect]) ->
+    #{attempts => 10, delay => 5000};
 default_config([outgoing_pools, _Type, _Tag, conn_opts, tls]) ->
     maps:merge(default_tls(), #{server_name_indication => default_sni()});
 default_config([outgoing_pools, _Type, _Tag, conn_opts, tls, server_name_indication]) ->

test/config_parser_SUITE.erl

@@ -145,6 +145,7 @@ groups() ->
                          pool_elastic_connection,
                          pool_rabbit,
                          pool_rabbit_connection,
+                         pool_rabbit_connection_reconnect,
                          pool_rabbit_connection_tls,
                          pool_ldap,
                          pool_ldap_connection,
@@ -1114,6 +1115,15 @@ pool_rabbit_connection(_Config) ->
     ?err(T(#{<<"confirms_enabled">> => <<"yes">>})),
     ?err(T(#{<<"max_worker_queue_len">> => -1})).
 
+pool_rabbit_connection_reconnect(_Config) ->
+    P = [outgoing_pools, 1, conn_opts, reconnect],
+    T = fun(Opts) -> pool_conn_raw(<<"rabbit">>, #{<<"reconnect">> => Opts}) end,
+    ?cfg(P, default_config([outgoing_pools, rabbit, default, conn_opts, reconnect]), T(#{})),
+    ?cfg(P ++ [attempts], 5, T(#{<<"attempts">> => 5})),
+    ?cfg(P ++ [delay], 0, T(#{<<"delay">> => 0})), % not recommended but allowed
+    ?err(T(#{<<"attempts">> => 0})), % disabling is achieved by omitting this subsection instead
+    ?err(T(#{<<"delay">> => <<"infinity">>})).
+
 pool_rabbit_connection_tls(_Config) ->
     P = [outgoing_pools, 1, conn_opts, tls],
     T = fun(Opts) -> pool_conn_raw(<<"rabbit">>, #{<<"tls">> => Opts}) end,

test/config_parser_SUITE_data/outgoing_pools.toml

@@ -55,6 +55,8 @@
     password = "guest"
     confirms_enabled = true
     max_worker_queue_len = 100
+    reconnect.attempts = 5
+    reconnect.delay = 2000
 
 [outgoing_pools.ldap.default]
   scope = "host_type"