From 450be0f7304aad3a38c9860c4d4911cfd5c937b3 Mon Sep 17 00:00:00 2001 From: Renato Ceolin Date: Wed, 4 Nov 2026 17:35:00 -0300 Subject: [PATCH] kubernetes: back off after a failed poll Polling a degraded Kubernetes API server at a fixed interval keeps adding load exactly when it can least afford it, which slows recovery. Add an opt-in kubernetes_poll_backoff option that applies capped exponential backoff with jitter after a failed poll, keeping the last known membership in the meantime. Without the option the strategy keeps the fixed polling interval. The backoff curve lives in Cluster.Strategy.Kubernetes.Backoff so it can be tested without sleeping. --- CHANGELOG.md | 3 + lib/strategy/kubernetes.ex | 111 +++++++++++++++++++---------- lib/strategy/kubernetes/backoff.ex | 39 ++++++++++ test/kubernetes_backoff_test.exs | 38 ++++++++++ test/kubernetes_test.exs | 31 ++++++++ 5 files changed, 184 insertions(+), 38 deletions(-) create mode 100644 lib/strategy/kubernetes/backoff.ex create mode 100644 test/kubernetes_backoff_test.exs diff --git a/CHANGELOG.md b/CHANGELOG.md index 1975269..186c82b 100644 --- a/CHANGELOG.md +++ b/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). diff --git a/lib/strategy/kubernetes.ex b/lib/strategy/kubernetes.ex index ac5a287..54274d2 100644 --- a/lib/strategy/kubernetes.ex +++ b/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,7 +625,7 @@ defmodule Cluster.Strategy.Kubernetes do "kubernetes strategy is selected, but :kubernetes_node_basename is not configured!" ) - [] + {:ok, []} selector == nil -> warn( @@ -598,11 +633,11 @@ defmodule Cluster.Strategy.Kubernetes do "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 diff --git a/lib/strategy/kubernetes/backoff.ex b/lib/strategy/kubernetes/backoff.ex new file mode 100644 index 0000000..8c20180 --- /dev/null +++ b/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 diff --git a/test/kubernetes_backoff_test.exs b/test/kubernetes_backoff_test.exs new file mode 100644 index 0000000..b7644b9 --- /dev/null +++ b/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 diff --git a/test/kubernetes_test.exs b/test/kubernetes_test.exs index d48c2f1..ce9d3dc 100644 --- a/test/kubernetes_test.exs +++ b/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"]))