Added readiness check to avoid errors during topology changes

README.md

@@ -102,6 +102,10 @@ the ring to change.
 
 The whitelist and blacklist only have an effect when `monitor_nodes: true`.
 
+It is possible to have the ring wait until an application starts before it is included in the ring.
+This can be accomplished by setting `wait_for_readiness: true` and listing the app dependencies in
+`readiness_deps: [:app1, :app2]`.
+
 ## Configuration
 
 Below is an example configuration:
@@ -113,7 +117,9 @@ config :libring,
     # but does not allow nodes named "a" or "remsh*" to be added to the ring
     ring_a: [monitor_nodes: true,
              node_type: :visible,
-             node_blacklist: ["a", ~r/^remsh.*$/]],
+             node_blacklist: ["a", ~r/^remsh.*$/],
+             wait_for_readiness: true,
+             readiness_deps: [:myapp]],
     # A ring which is composed of three nodes, of which "c" has a non-default weight of 200
     # The default weight is 128
     ring_b: [nodes: ["a", "b", {"c", 200}]]

lib/managed_ring.ex

@@ -22,12 +22,15 @@ defmodule HashRing.Managed do
   @type weight :: pos_integer
   @type node_list :: [term() | {term(), weight}]
   @type pattern_list :: [String.t() | Regex.t()]
+  @type app_list :: [atom()]
   @type ring_options :: [
           nodes: node_list,
           monitor_nodes: boolean,
           node_blacklist: pattern_list,
           node_whitelist: pattern_list,
-          node_type: :all | :hidden | :visible
+          node_type: :all | :hidden | :visible,
+          wait_for_readiness: boolean,
+          readiness_deps: app_list
         ]
 
   @type child_spec_options :: [
@@ -41,10 +44,22 @@ defmodule HashRing.Managed do
           :nodes => node_list,
           :monitor_nodes => boolean,
           :node_blacklist => pattern_list,
-          :node_whitelist => pattern_list
+          :node_whitelist => pattern_list,
+          :node_type => :all | :hidden | :visible,
+          :wait_for_readiness => boolean,
+          :readiness_deps: app_list,
         ]
 
-  @valid_ring_opts [:name, :nodes, :monitor_nodes, :node_blacklist, :node_whitelist, :node_type]
+  @valid_ring_opts [
+    :name,
+    :nodes,
+    :monitor_nodes,
+    :node_blacklist,
+    :node_whitelist,
+    :node_type,
+    :wait_for_readiness,
+    :readiness_deps
+  ]
 
   @spec child_spec(child_spec_options) :: Supervisor.child_spec
   def child_spec(opts) do
@@ -76,6 +91,8 @@ defmodule HashRing.Managed do
     is provided, the blacklist has no effect.
   * `node_whitelist: [String.t | Regex.t]` - The same as `node_blacklist`, except the opposite; only nodes
     which match a pattern in the whitelist will result in the ring being updated.
+  * `wait_for_readiness: boolean` - Wait for apps listed in `readiness_deps` to start before adding to the ring.
+  * `readiness_deps: [atom]` - List of dependency apps that need to start before the node is considered ready.
   * `node_type: :all | :hidden | :visible`: refers what kind of nodes will be monitored
     when `monitor_nodes` is `true`. For more information, see `:net_kernel.monitor_nodes/2`.
 
@@ -111,6 +128,8 @@ defmodule HashRing.Managed do
             :node_blacklist when is_list(value) -> false
             :node_whitelist when is_list(value) -> false
             :node_type when value in [:all, :hidden, :visible] -> false
+            :wait_for_readiness when is_boolean(value) -> false
+            :readiness_deps when is_list(value) -> false
             _ -> true
           end
       end)

lib/worker.ex

@@ -2,6 +2,19 @@ defmodule HashRing.Worker do
   @moduledoc false
   use GenServer
 
+  @erpc_timeout 500
+  @node_readiness_check_interval :timer.seconds(1)
+
+  defstruct [
+    :table,
+    :node_blacklist,
+    :node_whitelist,
+    :wait_for_readiness,
+    :readiness_deps_set
+  ]
+
+  alias __MODULE__, as: State
+
   def nodes(pid_or_name)
 
   def nodes(pid) when is_pid(pid) do
@@ -14,7 +27,7 @@ defmodule HashRing.Worker do
     |> get_ring()
     |> HashRing.nodes()
   rescue
-    ArgumentError -> 
+    ArgumentError ->
       {:error, :no_such_ring}
   end
 
@@ -69,55 +82,135 @@ defmodule HashRing.Worker do
         nodes = [Node.self() | Node.list(:connected)]
         node_blacklist = Keyword.get(options, :node_blacklist, [~r/^remsh.*$/, ~r/^rem-.*$/])
         node_whitelist = Keyword.get(options, :node_whitelist, [])
+        wait_for_readiness = Keyword.get(options, :wait_for_readiness, false)
+        readiness_deps_set = Keyword.get(options, :readiness_deps, []) |> MapSet.new()
 
         ring =
           Enum.reduce(nodes, ring, fn node, acc ->
-            cond do
-              HashRing.Utils.ignore_node?(node, node_blacklist, node_whitelist) ->
-                acc
-
-              :else ->
+            if HashRing.Utils.ignore_node?(node, node_blacklist, node_whitelist) do
+              acc
+            else
+              if wait_for_readiness do
+                if node_ready?(node, readiness_deps_set) do
+                  HashRing.add_node(acc, node)
+                else
+                  schedule_check_for_node_readiness(node)
+                  acc
+                end
+              else
                 HashRing.add_node(acc, node)
+              end
             end
           end)
 
         node_type = Keyword.get(options, :node_type, :all)
         :ok = :net_kernel.monitor_nodes(true, node_type: node_type)
         true = :ets.insert_new(table, {:ring, ring})
-        {:ok, {table, node_blacklist, node_whitelist}}
+
+        {:ok,
+         %State{
+           table: table,
+           node_blacklist: node_blacklist,
+           node_whitelist: node_whitelist,
+           wait_for_readiness: wait_for_readiness,
+           readiness_deps_set: readiness_deps_set
+         }}
 
       :else ->
         nodes = Keyword.get(options, :nodes, [])
         ring = HashRing.add_nodes(ring, nodes)
         true = :ets.insert_new(table, {:ring, ring})
-        {:ok, {table, [], []}}
+
+        {:ok,
+         %State{
+           table: table,
+           node_blacklist: [],
+           node_whitelist: [],
+           wait_for_readiness: false,
+           readiness_deps_set: MapSet.new()
+         }}
     end
   end
 
-  def handle_call(:list_nodes, _from, {table, _b, _w} = state) do
+  def handle_call(:list_nodes, _from, %State{table: table} = state) do
     {:reply, HashRing.nodes(get_ring(table)), state}
   end
 
-  def handle_call({:key_to_node, key}, _from, {table, _b, _w} = state) do
+  def handle_call({:key_to_node, key}, _from, %State{table: table} = state) do
     {:reply, HashRing.key_to_node(get_ring(table), key), state}
   end
 
-  def handle_call({:add_node, node}, _from, {table, _b, _w} = state) do
-    get_ring(table) |> HashRing.add_node(node) |> update_ring(table)
+  def handle_call(
+        {:add_node, node},
+        _from,
+        %State{
+          table: table,
+          wait_for_readiness: wait_for_readiness,
+          readiness_deps_set: readiness_deps_set
+        } = state
+      ) do
+    if wait_for_readiness and not node_ready?(node, readiness_deps_set) do
+      schedule_check_for_node_readiness(node)
+    else
+      get_ring(table) |> HashRing.add_node(node) |> update_ring(table)
+    end
+
     {:reply, :ok, state}
   end
 
-  def handle_call({:add_node, node, weight}, _from, {table, _b, _w} = state) do
-    get_ring(table) |> HashRing.add_node(node, weight) |> update_ring(table)
+  def handle_call(
+        {:add_node, node, weight},
+        _from,
+        %State{
+          table: table,
+          wait_for_readiness: wait_for_readiness,
+          readiness_deps_set: readiness_deps_set
+        } = state
+      ) do
+    if wait_for_readiness and not node_ready?(node, readiness_deps_set) do
+      schedule_check_for_node_readiness({node, weight})
+    else
+      get_ring(table) |> HashRing.add_node(node, weight) |> update_ring(table)
+    end
+
     {:reply, :ok, state}
   end
 
-  def handle_call({:add_nodes, nodes}, _from, {table, _b, _w} = state) do
-    get_ring(table) |> HashRing.add_nodes(nodes) |> update_ring(table)
+  def handle_call(
+        {:add_nodes, nodes},
+        _from,
+        %State{
+          table: table,
+          wait_for_readiness: wait_for_readiness,
+          readiness_deps_set: readiness_deps_set
+        } = state
+      ) do
+    if wait_for_readiness do
+      %{true: ready_nodes, false: starting_nodes} =
+        Enum.group_by(
+          nodes,
+          fn
+            {node, _weight} ->
+              node_ready?(node, readiness_deps_set)
+
+            node ->
+              node_ready?(node, readiness_deps_set)
+          end
+        )
+
+      get_ring(table) |> HashRing.add_nodes(ready_nodes) |> update_ring(table)
+
+      for starting_node <- starting_nodes do
+        schedule_check_for_node_readiness(starting_node)
+      end
+    else
+      get_ring(table) |> HashRing.add_nodes(nodes) |> update_ring(table)
+    end
+
     {:reply, :ok, state}
   end
 
-  def handle_call({:remove_node, node}, _from, {table, _b, _w} = state) do
+  def handle_call({:remove_node, node}, _from, %State{table: table} = state) do
     get_ring(table) |> HashRing.remove_node(node) |> update_ring(table)
     {:reply, :ok, state}
   end
@@ -127,19 +220,62 @@ defmodule HashRing.Worker do
     {:stop, :shutdown, state}
   end
 
-  def handle_info({:nodeup, node, _info}, {table, b, w} = state) do
+  def handle_info(
+        {:nodeup, node, _info},
+        %State{
+          table: table,
+          node_blacklist: b,
+          node_whitelist: w,
+          wait_for_readiness: wait_for_readiness,
+          readiness_deps_set: readiness_deps_set
+        } = state
+      ) do
     unless HashRing.Utils.ignore_node?(node, b, w) do
-      get_ring(table) |> HashRing.add_node(node) |> update_ring(table)
+      if wait_for_readiness and not node_ready?(node, readiness_deps_set) do
+        schedule_check_for_node_readiness(node)
+      else
+        get_ring(table) |> HashRing.add_node(node) |> update_ring(table)
+      end
     end
 
     {:noreply, state}
   end
 
-  def handle_info({:nodedown, node, _info}, state = {table, _b, _w}) do
+  def handle_info({:nodedown, node, _info}, %State{table: table} = state) do
     get_ring(table) |> HashRing.remove_node(node) |> update_ring(table)
     {:noreply, state}
   end
 
+  def handle_info(
+        {:check_node_readiness, node, weight},
+        %State{table: table, readiness_deps_set: readiness_deps_set} = state
+      ) do
+    if node_ready?(node, readiness_deps_set) do
+      get_ring(table) |> HashRing.add_node(node, weight) |> update_ring(table)
+    else
+      schedule_check_for_node_readiness({node, weight})
+    end
+
+    {:noreply, state}
+  end
+
+  def handle_info(
+        {:check_node_readiness, node},
+        %State{table: table, readiness_deps_set: readiness_deps_set} = state
+      ) do
+    if node_ready?(node, readiness_deps_set) do
+      get_ring(table) |> HashRing.add_node(node) |> update_ring(table)
+    else
+      schedule_check_for_node_readiness(node)
+    end
+
+    {:noreply, state}
+  end
+
+  def handle_info(_msg, state) do
+    {:noreply, state}
+  end
+
   defp get_ets_name(name), do: :"libring_#{name}"
 
   defp do_call(pid_or_name, msg)
@@ -160,6 +296,33 @@ defmodule HashRing.Worker do
 
   defp get_ring(table), do: :ets.lookup_element(table, :ring, 2)
 
-  defp update_ring(ring, table), 
+  defp update_ring(ring, table),
     do: :ets.update_element(table, :ring, {2, ring})
+
+  defp get_started_apps_set(node) do
+    try do
+      :erpc.call(node, Application, :started_applications, [], @erpc_timeout)
+      |> Enum.map(&elem(&1, 0))
+      |> MapSet.new()
+    rescue
+      _e -> MapSet.new()
+    end
+  end
+
+  defp node_ready?(node, readiness_deps_set) do
+    MapSet.difference(readiness_deps_set, get_started_apps_set(node))
+    |> MapSet.equal?(MapSet.new())
+  end
+
+  defp schedule_check_for_node_readiness({node, weight}) do
+    if node in Node.list() do
+      :timer.send_after(@node_readiness_check_interval, {:check_node_readiness, node, weight})
+    end
+  end
+
+  defp schedule_check_for_node_readiness(node) do
+    if node in Node.list() do
+      :timer.send_after(@node_readiness_check_interval, {:check_node_readiness, node})
+    end
+  end
 end

mix.exs

@@ -41,8 +41,9 @@ defmodule HashRing.Mixfile do
 
       {:ex_doc, ">= 0.0.0", only: [:docs]},
       {:benchee, "~> 1.0", only: [:dev]},
-      {:dialyxir, "~> 1.0", only: [:test], runtime: false},
-      {:stream_data, "~> 0.5", only: [:test]}
+      {:dialyxir, "~> 1.0", only: [:dev], runtime: false},
+      {:stream_data, "~> 0.5", only: [:test]},
+      {:local_cluster, "~> 1.2", only: [:test]}
     ]
   end
 

mix.lock

@@ -5,6 +5,8 @@
   "earmark_parser": {:hex, :earmark_parser, "1.4.18", "e1b2be73eb08a49fb032a0208bf647380682374a725dfb5b9e510def8397f6f2", [:mix], [], "hexpm", "114a0e85ec3cf9e04b811009e73c206394ffecfcc313e0b346de0d557774ee97"},
   "erlex": {:hex, :erlex, "0.2.6", "c7987d15e899c7a2f34f5420d2a2ea0d659682c06ac607572df55a43753aa12e", [:mix], [], "hexpm", "2ed2e25711feb44d52b17d2780eabf998452f6efda104877a3881c2f8c0c0c75"},
   "ex_doc": {:hex, :ex_doc, "0.26.0", "1922164bac0b18b02f84d6f69cab1b93bc3e870e2ad18d5dacb50a9e06b542a3", [:mix], [{:earmark_parser, "~> 1.4.0", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1", [hex: :makeup_erlang, repo: "hexpm", optional: false]}], "hexpm", "2775d66e494a9a48355db7867478ffd997864c61c65a47d31c4949459281c78d"},
+  "global_flags": {:hex, :global_flags, "1.0.0", "ee6b864979a1fb38d1fbc67838565644baf632212bce864adca21042df036433", [:rebar3], [], "hexpm", "85d944cecd0f8f96b20ce70b5b16ebccedfcd25e744376b131e89ce61ba93176"},
+  "local_cluster": {:hex, :local_cluster, "1.2.1", "8eab3b8a387680f0872eacfb1a8bd5a91cb1d4d61256eec6a655b07ac7030c73", [:mix], [{:global_flags, "~> 1.0", [hex: :global_flags, repo: "hexpm", optional: false]}], "hexpm", "aae80c9bc92c911cb0be085fdeea2a9f5b88f81b6bec2ff1fec244bb0acc232c"},
   "makeup": {:hex, :makeup, "1.0.5", "d5a830bc42c9800ce07dd97fa94669dfb93d3bf5fcf6ea7a0c67b2e0e4a7f26c", [:mix], [{:nimble_parsec, "~> 0.5 or ~> 1.0", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "cfa158c02d3f5c0c665d0af11512fed3fba0144cf1aadee0f2ce17747fba2ca9"},
   "makeup_elixir": {:hex, :makeup_elixir, "0.15.2", "dc72dfe17eb240552857465cc00cce390960d9a0c055c4ccd38b70629227e97c", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.1", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "fd23ae48d09b32eff49d4ced2b43c9f086d402ee4fd4fcb2d7fad97fa8823e75"},
   "makeup_erlang": {:hex, :makeup_erlang, "0.1.1", "3fcb7f09eb9d98dc4d208f49cc955a34218fc41ff6b84df7c75b3e6e533cc65f", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "174d0809e98a4ef0b3309256cbf97101c6ec01c4ab0b23e926a9e17df2077cbb"},