kubernetes: add opt-in backoff after a failed poll

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## Unreleased
 
+- Add `kubernetes_poll_backoff` to `Cluster.Strategy.Kubernetes`, applying capped
+  exponential backoff with jitter after a failed poll so a degraded API server is
+  not polled at the fixed interval. Disabled by default.
 - Explain mixed-version worker compatibility, staged rollouts and version-scoped discovery tradeoffs (#180).
 - Explain DNS readiness, discovery retries and application-owned Mnesia bootstrap coordination (#111).
 - Document named-port Kubernetes SRV queries, per-Pod identities and bounded distribution/mesh diagnostics (#149).

lib/strategy/kubernetes.ex

@@ -236,6 +236,21 @@ defmodule Cluster.Strategy.Kubernetes do
   The default interval to sync topologies is `#{@default_polling_interval}`
   (#{div(@default_polling_interval, 1000)} seconds). You can configure it with `:polling_interval` option.
 
+  ### Failure backoff
+
+  A poll that fails, for example because the API server is unavailable or rate
+  limiting, keeps the last known membership and retries on the next polling
+  interval. Set `:kubernetes_poll_backoff` to add capped exponential backoff
+  with jitter after each failure, so a degraded API server is not polled again
+  at the fixed interval:
+
+      kubernetes_poll_backoff: [base_ms: 100, max_ms: 30_000, jitter: 0.2]
+
+  `:base_ms` is the delay added after the first failure (default `50`),
+  `:max_ms` caps it (default `60_000`), and `:jitter` spreads retries across
+  clients (default `0.2`, between `0` and `1`). `true` uses the defaults. The
+  option is disabled by default, which keeps the fixed polling interval.
+
   ## Membership during shutdown
 
   Changes to discovered membership take effect on the next successful poll that observes them.
@@ -372,6 +387,7 @@ defmodule Cluster.Strategy.Kubernetes do
   import Cluster.Logger
 
   alias Cluster.Strategy.State
+  alias Cluster.Strategy.Kubernetes.Backoff
   alias Cluster.Strategy.Kubernetes.EndpointSlice
   alias Cluster.Strategy.Kubernetes.TLS
 
@@ -383,7 +399,7 @@ defmodule Cluster.Strategy.Kubernetes do
   end
 
   def init([%State{} = state]) do
-    {:ok, load(state)}
+    {:ok, load(state, 0)}
   end
 
   @impl true
@@ -392,15 +408,35 @@ defmodule Cluster.Strategy.Kubernetes do
   end
 
   def handle_info(:load, %State{} = state) do
-    {:noreply, load(state)}
+    {:noreply, load(state, 0)}
+  end
+
+  def handle_info({:load, attempt}, %State{} = state) do
+    {:noreply, load(state, attempt)}
   end
 
   def handle_info(_, state) do
     {:noreply, state}
   end
 
-  defp load(%State{topology: topology} = state) do
-    new_nodelist = MapSet.new(get_nodes(state))
+  defp load(%State{topology: topology, meta: meta} = state, attempt) do
+    case get_nodes(state) do
+      {:ok, discovered} ->
+        nodes = reconcile(topology, state, MapSet.new(discovered))
+        Process.send_after(self(), {:load, 0}, polling_interval(state))
+        %State{state | meta: nodes}
+
+      {:error, _reason} ->
+        # Keep the last known membership and retry on the next poll, backing off
+        # so a degraded API server is not hit at the fixed polling interval.
+        nodes = reconcile(topology, state, meta)
+        delay = polling_interval(state) + Backoff.delay(backoff_setting(state), attempt)
+        Process.send_after(self(), {:load, attempt + 1}, delay)
+        %State{state | meta: nodes}
+    end
+  end
+
+  defp reconcile(topology, state, new_nodelist) do
     removed = MapSet.difference(state.meta, new_nodelist)
 
     new_nodelist =
@@ -420,28 +456,26 @@ defmodule Cluster.Strategy.Kubernetes do
           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}
+    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
   end
 
+  defp backoff_setting(%State{config: config}),
+    do: Keyword.get(config, :kubernetes_poll_backoff)
+
   defp polling_interval(%State{config: config}) do
     Keyword.get(config, :polling_interval, @default_polling_interval)
   end
@@ -473,8 +507,8 @@ defmodule Cluster.Strategy.Kubernetes do
 
   defp get_namespace(_, namespace), do: namespace
 
-  @spec get_nodes(State.t()) :: [atom()]
-  defp get_nodes(%State{topology: topology, config: config, meta: meta} = state) do
+  @spec get_nodes(State.t()) :: {:ok, [atom()]} | {:error, term()}
+  defp get_nodes(%State{topology: topology, config: config} = state) do
     service_account_path =
       Keyword.get(config, :kubernetes_service_account_path, @service_account_path)
 
@@ -504,12 +538,12 @@ defmodule Cluster.Strategy.Kubernetes do
 
       {:error, reason} ->
         error(topology, "cannot configure kubernetes TLS trust: #{inspect(reason)}")
-        meta
+        {:error, {:tls_options, reason}}
     end
   end
 
   defp get_nodes(
-         %State{topology: topology, config: config, meta: meta},
+         %State{topology: topology, config: config},
          service_account_path,
          base_url,
          http_options,
@@ -555,32 +589,33 @@ defmodule Cluster.Strategy.Kubernetes do
                  namespace
                ) do
             {:ok, nodes} ->
-              format_nodes(nodes, config)
+              {:ok, format_nodes(nodes, config)}
 
             {:error, reason} ->
               error(topology, "cannot list kubernetes EndpointSlices: #{inspect(reason)}")
-              meta
+              {:error, {:endpoint_slices, reason}}
           end
         else
           query = URI.encode_query(query_params, :rfc3986)
 
           case :httpc.request(:get, {~c"#{url}?#{query}", headers}, http_options, request_options) do
             {:ok, {{_version, 200, _status}, _headers, body}} ->
-              parse_response(ip_lookup_mode, Jason.decode!(body), namespace)
-              |> format_nodes(config)
+              {:ok,
+               parse_response(ip_lookup_mode, Jason.decode!(body), namespace)
+               |> format_nodes(config)}
 
             {:ok, {{_version, 403, _status}, _headers, body}} ->
               %{"message" => msg} = Jason.decode!(body)
               warn(topology, "cannot query kubernetes (unauthorized): #{msg}")
-              []
+              {:ok, []}
 
             {:ok, {{_version, code, status}, _headers, body}} ->
               warn(topology, "cannot query kubernetes (#{code} #{status}): #{inspect(body)}")
-              meta
+              {:error, {:http_status, code}}
 
             {:error, reason} ->
               error(topology, "request to kubernetes failed!: #{inspect(reason)}")
-              meta
+              {:error, reason}
           end
         end
 
@@ -590,19 +625,19 @@ defmodule Cluster.Strategy.Kubernetes do
           "kubernetes strategy is selected, but :kubernetes_node_basename is not configured!"
         )
 
-        []
+        {:ok, []}
 
       selector == nil ->
         warn(
           topology,
           "kubernetes strategy is selected, but :kubernetes_selector is not configured!"
         )
 
-        []
+        {:ok, []}
 
       :else ->
         warn(topology, "kubernetes strategy is selected, but is not configured!")
-        []
+        {:ok, []}
     end
   end
 

lib/strategy/kubernetes/backoff.ex

@@ -0,0 +1,39 @@
+defmodule Cluster.Strategy.Kubernetes.Backoff do
+  @moduledoc false
+
+  # The extra delay applied after a failed poll. Keeping the calculation here
+  # makes the growth curve testable without sleeping in the strategy tests.
+  @default_base_ms 50
+  @default_max_ms 60_000
+  @default_jitter 0.2
+
+  # Cap the exponent so a long outage cannot build an enormous integer before
+  # the result is clamped by :max_ms.
+  @max_shift 20
+
+  @doc """
+  Returns the extra delay, in milliseconds, to apply after a failed poll.
+
+  `setting` is the value of the `:kubernetes_poll_backoff` option: `nil` or
+  `false` disables the backoff, `true` uses the defaults, and a keyword list
+  overrides `:base_ms`, `:max_ms` and `:jitter`.
+  """
+  @spec delay(nil | boolean | keyword(), non_neg_integer()) :: non_neg_integer()
+  def delay(nil, _attempt), do: 0
+  def delay(false, _attempt), do: 0
+  def delay(true, attempt), do: delay([], attempt)
+
+  def delay(opts, attempt) when is_list(opts) do
+    base_ms = Keyword.get(opts, :base_ms, @default_base_ms)
+    max_ms = Keyword.get(opts, :max_ms, @default_max_ms)
+    jitter = Keyword.get(opts, :jitter, @default_jitter)
+
+    max_sleep = min(max_ms, base_ms * Bitwise.bsl(1, min(attempt, @max_shift)))
+
+    # Stay in the order of magnitude of the exponential step while spreading
+    # retries across clients, so recovering servers are not hit in lockstep.
+    delay = trunc(max_sleep * (1 - jitter * :rand.uniform()))
+
+    max(delay, 0)
+  end
+end

test/kubernetes_backoff_test.exs

@@ -0,0 +1,38 @@
+defmodule Cluster.Strategy.Kubernetes.BackoffTest do
+  @moduledoc false
+  use ExUnit.Case, async: true
+
+  alias Cluster.Strategy.Kubernetes.Backoff
+
+  describe "delay/2" do
+    test "is disabled when the option is unset or false" do
+      assert Backoff.delay(nil, 0) == 0
+      assert Backoff.delay(false, 3) == 0
+    end
+
+    test "grows exponentially and stops at the configured cap" do
+      opts = [base_ms: 100, max_ms: 1_000, jitter: 0]
+
+      assert Backoff.delay(opts, 0) == 100
+      assert Backoff.delay(opts, 1) == 200
+      assert Backoff.delay(opts, 2) == 400
+      assert Backoff.delay(opts, 3) == 800
+      assert Backoff.delay(opts, 4) == 1_000
+      assert Backoff.delay(opts, 10_000) == 1_000
+    end
+
+    test "stays within the jitter band below the exponential step" do
+      opts = [base_ms: 1_000, max_ms: 1_000, jitter: 0.2]
+
+      for delay <- for(_ <- 1..200, do: Backoff.delay(opts, 0)) do
+        assert delay >= 800 and delay <= 1_000
+      end
+    end
+
+    test "uses the defaults for true" do
+      delay = Backoff.delay(true, 0)
+
+      assert delay >= 40 and delay <= 50
+    end
+  end
+end

test/kubernetes_test.exs

@@ -133,6 +133,37 @@ defmodule Cluster.Strategy.KubernetesTest do
     refute_receive {:disconnect, _}
   end
 
+  test "retries a failed poll and connects once the API recovers", context do
+    body = File.read!(Path.join([__DIR__, "fixtures", "kubernetes", "endpoints.json"]))
+    counter = :counters.new(1, [])
+
+    responder = fn ->
+      :counters.add(counter, 1, 1)
+
+      if :counters.get(counter, 1) == 1 do
+        {503, "temporarily unavailable"}
+      else
+        body
+      end
+    end
+
+    port = start_https(context.valid, responder, 200)
+
+    start_supervised!(
+      {Kubernetes,
+       [
+         state(
+           context,
+           port,
+           [polling_interval: 50, kubernetes_poll_backoff: [base_ms: 10, max_ms: 20, jitter: 0]],
+           MapSet.new()
+         )
+       ]}
+    )
+
+    assert_receive {:connect, :"test_basename@10.48.33.136"}, 1_000
+  end
+
   defp start_strategy(context, config \\ []) do
     resource = Keyword.get(config, :kubernetes_ip_lookup_mode, :endpoints)
     body = File.read!(Path.join([__DIR__, "fixtures", "kubernetes", "#{resource}.json"]))