defmodule Cluster.Strategy.Kubernetes.DNSSRV do
@default_polling_interval 5_000
@moduledoc """
This clustering strategy works by issuing a SRV query for the headless service where the StatefulSet
containing your nodes is running.
> This strategy requires deploying pods as a StatefulSet which is exposed by a headless service.
> If you want to avoid that, you could use `Cluster.Strategy.Kubernetes.DNS`.
It assumes that all Erlang nodes are using longnames - `<basename>@<domain>`:
+ all nodes are using the same `<basename>`
+ all nodes are using unique `<domain>`
In `<basename>@<domain>`:
+ `<basename>` would be the value configured by `:application_name` option.
+ `<domain>` would be the value which is controlled by following options:
- `:service`
- `:namespace`
- `:resolver`
## Getting `<basename>`
As said above, the basename is configured by `:application_name` option.
Just one thing to keep in mind - when building an OTP release, make sure that the name of the OTP
release matches the name configured by `:application_name`.
## Getting `<domain>`
> For more information, see the kubernetes stateful-application [documentation](https://kubernetes.io/docs/tutorials/stateful-application/basic-stateful-set/#using-stable-network-identities)
## Setup
Getting this strategy to work requires:
1. deploying pods as a StatefulSet (otherwise, hostname won't set for pods)
2. exposing above StatefulSet by a headless service (otherwise, the SRV query won't work as expected)
3. setting the name of Erlang node according to hostname of pods
First, deploying pods as a StatefulSet which is exposed by a headless service. And here is an
example of a corresponding Kubernetes definition:
```yaml
apiVersion: v1
kind: Service
metadata:
name: "myapp-headless"
labels:
app: myapp
spec:
ports:
- port: 4000
name: web
clusterIP: None
selector:
app: myapp
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: myapp
spec:
serviceName: "myapp-headless"
replicas: 2
selector:
matchLabels:
app: myapp
template:
metadata:
labels:
app: myapp
spec:
containers:
- name: myapp
image: myapp:v1.0.0
imagePullPolicy: Always
ports:
- containerPort: 4000
name: http
protocol: TCP
```
Then, set the name of Erlang node by using the hostname of pod. If you use mix releases, you
can configure the required options in `rel/env.sh.eex`:
# rel/env.sh.eex
export RELEASE_DISTRIBUTION=name
export RELEASE_NODE=<%= @release.name %>@$(hostname -f)
## Polling Interval
The default interval to sync topologies is `#{@default_polling_interval}`
(#{div(@default_polling_interval, 1000)} seconds). You can configure it with `:polling_interval` option.
## An example configuration
config :libcluster,
topologies: [
erlang_nodes_in_k8s: [
strategy: #{__MODULE__},
config: [
service: "myapp-headless",
application_name: "myapp",
namespace: "default",
polling_interval: 10_000
]
]
]
## An example of how this strategy extracts topology information from DNS
```sh
$ hostname -f
myapp-1.myapp-headless.default.svc.cluster.local
# An SRV query for a headless service returns multiple entries
$ dig SRV myapp-headless.default.svc.cluster.local
; <<>> DiG 9.14.3 <<>> SRV myapp-headless.default.svc.cluster.local
;; global options: +cmd
;; Got answer:
;; WARNING: .local is reserved for Multicast DNS
;; You are currently testing what happens when an mDNS query is leaked to DNS
;; ->>HEADER<<- opcode: QUERY, status: NOERROR, id: 7169
;; flags: qr aa rd ra; QUERY: 1, ANSWER: 2, AUTHORITY: 0, ADDITIONAL: 2
;; QUESTION SECTION:
;myapp-headless.default.svc.cluster.local. IN SRV
;; ANSWER SECTION:
myapp-headless.default.svc.cluster.local. 30 IN SRV 10 50 0 myapp-0.myapp-headless.default.svc.cluster.local.
myapp-headless.default.svc.cluster.local. 30 IN SRV 10 50 0 myapp-1.myapp-headless.default.svc.cluster.local.
;; ADDITIONAL SECTION:
myapp-0.myapp-headless.default.svc.cluster.local. 30 IN A 10.1.0.95
myapp--1.myapp-headless.default.svc.cluster.local. 30 IN A 10.1.0.96
;; Query time: 0 msec
;; SERVER: 10.96.0.10#53(10.96.0.10)
;; WHEN: Wed Jul 03 11:55:27 UTC 2019
;; MSG SIZE rcvd: 167
```
"""
use GenServer
use Cluster.Strategy
import Cluster.Logger
alias Cluster.Strategy.State
@impl true
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, load(state), 0}
end
@impl true
def handle_info(:timeout, state) do
handle_info(:load, state)
end
def handle_info(:load, state) do
{:noreply, load(state)}
end
def handle_info(_, state) do
{:noreply, state}
end
defp load(%State{topology: topology, meta: meta} = state) do
new_nodelist = MapSet.new(get_nodes(state))
removed = MapSet.difference(meta, new_nodelist)
new_nodelist =
case Cluster.Strategy.disconnect_nodes(
topology,
state.disconnect,
state.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 Cluster.Strategy.connect_nodes(
topology,
state.connect,
state.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(),
:load,
polling_interval(state)
)
%State{state | :meta => new_nodelist}
end
@spec get_nodes(State.t()) :: [atom()]
defp get_nodes(%State{topology: topology, config: config}) do
app_name = Keyword.fetch!(config, :application_name)
service = Keyword.fetch!(config, :service)
namespace = Keyword.fetch!(config, :namespace)
service_k8s_path =
"#{service}.#{namespace}.svc.#{System.get_env("CLUSTER_DOMAIN", "cluster.local.")}"
resolver = Keyword.get(config, :resolver, &:inet_res.getbyname(&1, :srv))
cond do
app_name != nil and service != nil ->
headless_service = to_charlist(service_k8s_path)
case resolver.(headless_service) do
{:ok, {:hostent, _, _, :srv, _count, addresses}} ->
parse_response(addresses, app_name)
{:error, reason} ->
error(
topology,
"#{inspect(headless_service)} : lookup against #{service} failed: #{inspect(reason)}"
)
[]
end
app_name == nil ->
warn(
topology,
"kubernetes.DNS strategy is selected, but :application_name is not configured!"
)
[]
service == nil ->
warn(topology, "kubernetes strategy is selected, but :service is not configured!")
[]
:else ->
warn(topology, "kubernetes strategy is selected, but is not configured!")
[]
end
end
defp polling_interval(%State{config: config}) do
Keyword.get(config, :polling_interval, @default_polling_interval)
end
defp parse_response(addresses, app_name) do
addresses
|> Enum.map(&:erlang.list_to_binary(elem(&1, 3)))
|> Enum.map(&"#{app_name}@#{&1}")
|> Enum.map(&String.to_atom(&1))
end
end
