defmodule Cluster.Strategy.DNSPoll do @moduledoc """ Assumes you have nodes that respond to the specified DNS query (A record), and which follow the node name pattern of `@`. If your setup matches those assumptions, this strategy will periodically poll DNS and connect all nodes it finds. ## Options * `poll_interval` - How often to poll in milliseconds (optional; default: 5_000) * `query` - DNS query to use (required; e.g. "my-app.example.com") * `node_basename` - The short name of the nodes you wish to connect to (required; e.g. "my-app") ## Usage config :libcluster, topologies: [ dns_poll_example: [ strategy: #{__MODULE__}, config: [ polling_interval: 5_000, query: "my-app.example.com", node_basename: "my-app"]]] """ use GenServer import Cluster.Logger alias Cluster.Strategy.State alias Cluster.Strategy @default_polling_interval 5_000 def start_link(args), do: GenServer.start_link(__MODULE__, args) @impl true def init([%State{meta: nil} = state]) do init([%State{state | :meta => MapSet.new()}]) end def init([%State{} = state]) do {:ok, do_poll(state)} end @impl true def handle_info(:timeout, state), do: handle_info(:poll, state) def handle_info(:poll, state), do: {:noreply, do_poll(state)} def handle_info(_, state), do: {:noreply, state} defp do_poll( %State{ topology: topology, connect: connect, disconnect: disconnect, list_nodes: list_nodes } = state ) do new_nodelist = state |> get_nodes() |> MapSet.new() removed = MapSet.difference(state.meta, new_nodelist) new_nodelist = case Strategy.disconnect_nodes( topology, disconnect, list_nodes, MapSet.to_list(removed) ) do :ok -> new_nodelist {:error, bad_nodes} -> # Add back the nodes which should have been removed, but which couldn't be for some reason Enum.reduce(bad_nodes, new_nodelist, fn {n, _}, acc -> MapSet.put(acc, n) end) end new_nodelist = case Strategy.connect_nodes( topology, connect, list_nodes, MapSet.to_list(new_nodelist) ) do :ok -> new_nodelist {:error, bad_nodes} -> # Remove the nodes which should have been added, but couldn't be for some reason Enum.reduce(bad_nodes, new_nodelist, fn {n, _}, acc -> MapSet.delete(acc, n) end) end Process.send_after(self(), :poll, polling_interval(state)) %{state | :meta => new_nodelist} end defp polling_interval(%{config: config}) do Keyword.get(config, :polling_interval, @default_polling_interval) end defp get_nodes(%State{config: config} = state) do query = Keyword.fetch(config, :query) node_basename = Keyword.fetch(config, :node_basename) resolver = Keyword.get(config, :resolver, fn query -> query |> String.to_charlist() |> lookup_all_ips end) resolve(query, node_basename, resolver, state) end # query for all ips responding to a given dns query # format ips as node names # filter out me defp resolve({:ok, query}, {:ok, node_basename}, resolver, %State{topology: topology}) when is_binary(query) and is_binary(node_basename) and query != "" and node_basename != "" do debug(topology, "polling dns for '#{query}'") me = node() query |> resolver.() |> Enum.map(&format_node(&1, node_basename)) |> Enum.reject(fn n -> n == me end) end defp resolve({:ok, invalid_query}, {:ok, invalid_basename}, _resolver, %State{ topology: topology }) do warn( topology, "dns polling strategy is selected, but query or basename param is invalid: #{inspect(%{query: invalid_query, node_basename: invalid_basename})}" ) [] end defp resolve(:error, :error, _resolver, %State{topology: topology}) do warn( topology, "dns polling strategy is selected, but query and basename params missed" ) [] end def lookup_all_ips(q) do Enum.flat_map([:a, :aaaa], fn t -> :inet_res.lookup(q, :in, t) end) end # turn an ip into a node name atom, assuming that all other node names looks similar to our own name defp format_node(ip, base_name), do: :"#{base_name}@#{:inet_parse.ntoa(ip)}" end