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"]))