Prevent mero fail when fails a single cluster (#83)

* Added checks for failing clusters

* Added support for failing and recovering clusters

* Renamed Retry/Retries variables and added some formatting to tests

* Renamed Retries variable

* Changed mero config handling to allow addition or removal of cluster configurations

* Removed test case declarations

* Removed retry behavior. Now is not necessary as we are retrying on every heartbeat

* Reverted a small change

Author: ElFantasma

src/mero_conf.erl

@@ -31,9 +31,6 @@
 
 -author('Miriam Pena <[email protected]>').
 
-%% # of times we'll retry to get the configuration from elasticache before raising an error
--define(MAX_RETRIES, 4).
-
 %% It's dynamically invoked using rpc:pmap/3
 -ignore_xref({?MODULE, get_elasticache_cluster_configs, 1}).
 
@@ -46,8 +43,7 @@
          pool_max_connection_delay_time/1, pool_min_connection_interval/1,
          max_connection_delay_time/1, stat_callback/0, stat_callback/1, add_now/1, add_now/2,
          millis_to/1, millis_to/2, process_server_specs/1, elasticache_load_config_delay/0,
-         elasticache_load_config_delay/1, monitor_heartbeat_delay/0, monitor_heartbeat_delay/2,
-         get_elasticache_cluster_configs/1]).
+         elasticache_load_config_delay/1, monitor_heartbeat_delay/0, monitor_heartbeat_delay/2]).
 
 -include_lib("mero/include/mero.hrl").
 
@@ -220,14 +216,32 @@ millis_to(TimeLimit, Then) ->
     end.
 
 -spec process_server_specs(mero:cluster_config()) -> mero:cluster_config().
-process_server_specs(Clusters) ->
-    [{ClusterName, [process_value(Attr) || Attr <- Attrs]}
-     || {ClusterName, Attrs} <- Clusters].
+process_server_specs([]) ->
+    [];
+process_server_specs([Cluster | Clusters]) ->
+    case process_server_spec(Cluster) of
+        {error, _} ->
+            process_server_specs(Clusters);
+        Config ->
+            [Config | process_server_specs(Clusters)]
+    end.
 
 %%%=============================================================================
 %%% Internal functions
 %%%=============================================================================
 
+process_server_spec({ClusterName, Attrs}) ->
+    try
+        {ClusterName, [process_value(Attr) || Attr <- Attrs]}
+    catch
+        Kind:Desc:Stack ->
+            error_logger:error_report([{error, mero_config_failed},
+                                       {kind, Kind},
+                                       {desc, Desc},
+                                       {stack, Stack}]),
+            {error, Desc}
+    end.
+
 get_env(Key) ->
     case application:get_env(mero, Key) of
         {ok, Value} ->
@@ -273,18 +287,9 @@ get_elasticache_cluster_configs({Host, Port}) ->
     [get_elasticache_cluster_config(Host, Port)].
 
 get_elasticache_cluster_config(Host, Port) ->
-    get_elasticache_cluster_config(Host,
-                                   Port,
-                                   0,
-                                   mero_elasticache:get_cluster_config(Host, Port)).
-
-get_elasticache_cluster_config(_Host, _Port, _Retries, {ok, Entries}) ->
-    Entries;
-get_elasticache_cluster_config(Host, Port, ?MAX_RETRIES, {error, Reason}) ->
-    error({Reason, Host, Port});
-get_elasticache_cluster_config(Host, Port, Retries, {error, _Reason}) ->
-    timer:sleep(trunc(math:pow(2, Retries)) * 100),
-    get_elasticache_cluster_config(Host,
-                                   Port,
-                                   Retries + 1,
-                                   mero_elasticache:get_cluster_config(Host, Port)).
+    case mero_elasticache:get_cluster_config(Host, Port) of
+        {ok, Entries} ->
+            Entries;
+        {error, Reason} ->
+            error({Reason, Host, Port})
+    end.

src/mero_conf_monitor.erl

@@ -121,7 +121,7 @@ update_cluster_defs(NewProcessedConfig, State) ->
     ok = mero_cluster:load_clusters(NewProcessedConfig),
     NewClusterVersion = mero_cluster:version(),
 
-    ok = update_clusters(OldProcessedConfig, NewProcessedConfig),
+    ok = update_clusters(lists:sort(OldProcessedConfig), lists:sort(NewProcessedConfig)),
 
     ok = purge_if_version_changed(OldClusterVersion, NewClusterVersion),
 
@@ -132,26 +132,47 @@ purge_if_version_changed(ClusterVersion, ClusterVersion) ->
 purge_if_version_changed(_OldVersion, _NewClusterVersion) ->
     mero_cluster:purge().
 
-%% NOTE: since both cluster definitions are generated through mero_conf:process_server_specs/1
-%%       with the same input, we can be sure that the resulting lists will contain the same number
-%%       of elements, with the same keys in the same order.
+%% NOTE: Old ClusterDefs and New ones are sorted by ClusterName, so it is possible to detect
+%%       when some Def is present or missing in either list
 update_clusters([], []) ->
     ok;
+update_clusters([],
+                [{ClusterName, _} | NewClusterDefs]) -> %% NewCluster is new, start it
+    mero_sup:start_child(ClusterName),
+    update_clusters([], NewClusterDefs);
+update_clusters([{ClusterName, _} | OldClusterDefs],
+                []) -> %% OldCluster not present, terminate it
+    mero_sup:terminate_child(ClusterName),
+    update_clusters(OldClusterDefs, []);
 update_clusters([ClusterDef | OldClusterDefs],
                 [ClusterDef | NewClusterDefs]) -> %% nothing changed
     update_clusters(OldClusterDefs, NewClusterDefs);
 update_clusters([{ClusterName, OldAttrs} | OldClusterDefs],
-                [{ClusterName, NewAttrs} | NewClusterDefs]) ->
-    OldServers =
-        lists:sort(
-            proplists:get_value(servers, OldAttrs)),
+                [{ClusterName, NewAttrs} | NewClusterDefs]) -> %% config changed, check it
+    OldServers = get_servers(OldAttrs),
     ok =
-        case lists:sort(
-                 proplists:get_value(servers, NewAttrs))
-        of
-            OldServers ->
-                ok; %% Nothing of relevance changed
-            _ ->
+        case get_servers(NewAttrs) of
+            OldServers -> %% Nothing of relevance changed
+                ok;
+            _ -> %% Different servers, restart the cluster
                 mero_sup:restart_child(ClusterName)
         end,
+    update_clusters(OldClusterDefs, NewClusterDefs);
+update_clusters([{OldClusterName, _} | OldClusterDefs],
+                [{NewClusterName, _} | _] = NewClusterDefs)
+    when OldClusterName < NewClusterName -> %% OldCluster not present, terminate it
+    mero_sup:terminate_child(OldClusterName),
+    update_clusters(OldClusterDefs, NewClusterDefs);
+update_clusters([{OldClusterName, _} | _] = OldClusterDefs,
+                [{NewClusterName, _} | NewClusterDefs])
+    when OldClusterName > NewClusterName -> %% NewCluster is new, start it
+    mero_sup:start_child(NewClusterName),
     update_clusters(OldClusterDefs, NewClusterDefs).
+
+get_servers(Attrs) ->
+    case proplists:get_value(servers, Attrs) of
+        Error when is_tuple(Error) ->
+            Error;
+        Servers when is_list(Servers) ->
+            lists:sort(Servers)
+    end.

src/mero_sup.erl

@@ -30,7 +30,7 @@
 
 -author('Miriam Pena <[email protected]>').
 
--export([start_link/1, restart_child/1, init/1]).
+-export([start_link/1, start_child/1, restart_child/1, terminate_child/1, init/1]).
 
 -behaviour(supervisor).
 
@@ -49,11 +49,21 @@
 start_link(OrigConfig) ->
     supervisor:start_link({local, ?MODULE}, ?MODULE, #{orig_config => OrigConfig}).
 
+-spec start_child(ClusterName :: atom()) -> ok.
+start_child(ClusterName) ->
+    {ok, _} = supervisor:start_child(?MODULE, cluster_sup_spec(ClusterName)),
+    ok.
+
 -spec restart_child(ClusterName :: atom()) -> ok.
 restart_child(ClusterName) ->
+    terminate_child(ClusterName),
+    start_child(ClusterName),
+    ok.
+
+-spec terminate_child(ClusterName :: atom()) -> ok.
+terminate_child(ClusterName) ->
     _ = supervisor:terminate_child(?MODULE, ClusterName),
     _ = supervisor:delete_child(?MODULE, ClusterName),
-    {ok, _} = supervisor:start_child(?MODULE, cluster_sup_spec(ClusterName)),
     ok.
 
 %%%===================================================================

test/mero_conf_SUITE.erl

@@ -37,7 +37,8 @@
 -export([all/0, init_per_testcase/2, end_per_testcase/2, helper_mfa_config_function/0,
          diff/1, process_server_specs_a_compatible/1, process_server_specs_a/1,
          process_server_specs_a_alternate/1, process_server_specs_a_b/1,
-         process_server_specs_a_b_c/1, process_server_specs_mfa/1, per_pool_config/1]).
+         process_server_specs_a_b_c/1, process_server_specs_mfa/1, per_pool_config/1,
+         process_server_specs_with_failure/1]).
 
 all() ->
     [diff,
@@ -47,7 +48,8 @@ all() ->
      process_server_specs_a_b,
      process_server_specs_a_b_c,
      process_server_specs_mfa,
-     per_pool_config].
+     per_pool_config,
+     process_server_specs_with_failure].
 
 init_per_testcase(diff, Conf) ->
     Conf;
@@ -65,22 +67,24 @@ init_per_testcase(_, Conf) ->
           "c3.com|10.102.00.102|11211 c4.com|10.102.00.102|11211\n">>,
     meck:expect(mero_elasticache,
                 request_response,
-                fun(Type, _, _, _) ->
-                   HostLines =
-                       case Type of
-                           a ->
-                               HostLinea;
-                           b ->
-                               HostLineb;
-                           c ->
-                               HostLinec
-                       end,
-                   {ok,
-                    [{banner, <<"CONFIG cluster ...">>},
-                     {version, <<"version1">>},
-                     {hosts, HostLines},
-                     {crlf, <<"\r\n">>},
-                     {eom, <<"END\r\n">>}]}
+                fun (failing_cluster, _, _, _) ->
+                        {error, econnrefused};
+                    (Type, _, _, _) ->
+                        HostLines =
+                            case Type of
+                                a ->
+                                    HostLinea;
+                                b ->
+                                    HostLineb;
+                                c ->
+                                    HostLinec
+                            end,
+                        {ok,
+                         [{banner, <<"CONFIG cluster ...">>},
+                          {version, <<"version1">>},
+                          {hosts, HostLines},
+                          {crlf, <<"\r\n">>},
+                          {eom, <<"END\r\n">>}]}
                 end),
     Conf.
 
@@ -235,3 +239,47 @@ per_pool_config(_Conf) ->
     ?assertEqual(30, mero_conf:pool_initial_connections(pool_1)),
     ?assertEqual(50, mero_conf:pool_initial_connections(pool_2)),
     ok.
+
+process_server_specs_with_failure(_Conf) ->
+    meck:new(error_logger, [unstick, passthrough]),
+    Spec =
+        [{cluster_a,
+          [{servers, {elasticache, [{a, 11211, 2}]}},
+           {sharding_algorithm, {mero, shard_crc32}},
+           {workers_per_shard, 1},
+           {pool_worker_module, mero_wrk_tcp_binary}]},
+         {cluster_b,
+          [{servers, {elasticache, [{failing_cluster, 11211, 3}]}},
+           {sharding_algorithm, {mero, shard_crc32}},
+           {workers_per_shard, 1},
+           {pool_worker_module, mero_wrk_tcp_binary}]},
+         {cluster_c,
+          [{servers, {elasticache, [{c, 11211, 1}]}},
+           {sharding_algorithm, {mero, shard_crc32}},
+           {workers_per_shard, 1},
+           {pool_worker_module, mero_wrk_tcp_binary}]}],
+    [{cluster_a, ServerSpecsA}, {cluster_c, ServerSpecsC}] =
+        mero_conf:process_server_specs(Spec),
+    ?assertEqual([{"a1.com", 11211},
+                  {"a2.com", 11211},
+                  {"a3.com", 11211},
+                  {"a1.com", 11211},
+                  {"a2.com", 11211},
+                  {"a3.com", 11211}],
+                 proplists:get_value(servers, ServerSpecsA)),
+    ?assertEqual(mero_wrk_tcp_binary, proplists:get_value(pool_worker_module, ServerSpecsA)),
+    ?assertEqual(1, proplists:get_value(workers_per_shard, ServerSpecsA)),
+    ?assertEqual({mero, shard_crc32}, proplists:get_value(sharding_algorithm, ServerSpecsA)),
+
+    ?assertEqual(1,
+                 meck:num_calls(error_logger,
+                                error_report,
+                                [[{error, mero_config_failed},
+                                  {kind, error},
+                                  {desc, {econnrefused, failing_cluster, 11211}},
+                                  {stack, '_'}]])),
+
+    ?assertEqual([{"c1.com", 11211}, {"c2.com", 11211}, {"c3.com", 11211}, {"c4.com", 11211}],
+                 proplists:get_value(servers, ServerSpecsC)),
+    meck:unload([error_logger]),
+    ok.