From 0b02e3165d50e4934db4109984f32e1540d673d1 Mon Sep 17 00:00:00 2001 From: Aaron Harlap Date: Thu, 2 Jul 2026 11:20:25 -0400 Subject: [PATCH] Retry transient Kubernetes API errors in chi kube driver Get calls A single transient apiserver/network error (connection refused/reset, timeout, 5xx) during a CHI reconcile currently propagates up as-is and aborts the entire reconcile, discarding progress on all remaining hosts. On installations with hundreds of hosts a reconcile makes tens of thousands of API calls over a long wall-clock window, so even a sub-second control-plane blip is likely to hit one of them and abort an otherwise healthy rollout, leaving the CR stuck in Aborted state. Wrap all Get calls in the chi kube drivers with a bounded exponential-backoff retry (5 attempts, 500ms..4s doubling, ~7.5s total budget) that retries only transient network/API errors. Terminal errors (NotFound, AlreadyExists, Conflict, Forbidden, Unauthorized, BadRequest, Invalid, MethodNotSupported) and context cancellation surface immediately, and when retries are exhausted the last error is returned unchanged - so callers behave exactly as before for non-transient errors and sustained outages. Signed-off-by: Aaron Harlap --- pkg/controller/chi/kube/config-map.go | 4 +- pkg/controller/chi/kube/cr.go | 4 +- pkg/controller/chi/kube/pdb.go | 4 +- pkg/controller/chi/kube/pod.go | 4 +- pkg/controller/chi/kube/pvc.go | 4 +- pkg/controller/chi/kube/retry.go | 130 +++++++++++++++++++++ pkg/controller/chi/kube/retry_test.go | 145 ++++++++++++++++++++++++ pkg/controller/chi/kube/secret.go | 4 +- pkg/controller/chi/kube/service.go | 4 +- pkg/controller/chi/kube/statesfulset.go | 4 +- 10 files changed, 299 insertions(+), 8 deletions(-) create mode 100644 pkg/controller/chi/kube/retry.go create mode 100644 pkg/controller/chi/kube/retry_test.go diff --git a/pkg/controller/chi/kube/config-map.go b/pkg/controller/chi/kube/config-map.go index 2e104472c..c28083847 100644 --- a/pkg/controller/chi/kube/config-map.go +++ b/pkg/controller/chi/kube/config-map.go @@ -45,7 +45,9 @@ func (c *ConfigMap) Create(ctx context.Context, cm *core.ConfigMap) (*core.Confi func (c *ConfigMap) Get(ctx context.Context, namespace, name string) (*core.ConfigMap, error) { ctx = k8sCtx(ctx) - return c.kubeClient.CoreV1().ConfigMaps(namespace).Get(ctx, name, controller.NewGetOptions()) + return getWithRetry(ctx, func() (*core.ConfigMap, error) { + return c.kubeClient.CoreV1().ConfigMaps(namespace).Get(ctx, name, controller.NewGetOptions()) + }) } func (c *ConfigMap) Update(ctx context.Context, cm *core.ConfigMap) (*core.ConfigMap, error) { diff --git a/pkg/controller/chi/kube/cr.go b/pkg/controller/chi/kube/cr.go index 44aeace86..c4f405ec7 100644 --- a/pkg/controller/chi/kube/cr.go +++ b/pkg/controller/chi/kube/cr.go @@ -70,7 +70,9 @@ func (c *CR) Get(ctx context.Context, namespace, name string) (api.ICustomResour func (c *CR) getCR(ctx context.Context, namespace, name string) (*api.ClickHouseInstallation, error) { ctx = k8sCtx(ctx) - return c.chopClient.ClickhouseV1().ClickHouseInstallations(namespace).Get(ctx, name, controller.NewGetOptions()) + return getWithRetry(ctx, func() (*api.ClickHouseInstallation, error) { + return c.chopClient.ClickhouseV1().ClickHouseInstallations(namespace).Get(ctx, name, controller.NewGetOptions()) + }) } func (c *CR) getCM(ctx context.Context, chi api.ICustomResource) (*core.ConfigMap, error) { diff --git a/pkg/controller/chi/kube/pdb.go b/pkg/controller/chi/kube/pdb.go index 8acef5a8e..274672802 100644 --- a/pkg/controller/chi/kube/pdb.go +++ b/pkg/controller/chi/kube/pdb.go @@ -45,7 +45,9 @@ func (c *PDB) Create(ctx context.Context, pdb *policy.PodDisruptionBudget) (*pol func (c *PDB) Get(ctx context.Context, namespace, name string) (*policy.PodDisruptionBudget, error) { ctx = k8sCtx(ctx) - return c.kubeClient.PolicyV1().PodDisruptionBudgets(namespace).Get(ctx, name, controller.NewGetOptions()) + return getWithRetry(ctx, func() (*policy.PodDisruptionBudget, error) { + return c.kubeClient.PolicyV1().PodDisruptionBudgets(namespace).Get(ctx, name, controller.NewGetOptions()) + }) } func (c *PDB) Update(ctx context.Context, pdb *policy.PodDisruptionBudget) (*policy.PodDisruptionBudget, error) { diff --git a/pkg/controller/chi/kube/pod.go b/pkg/controller/chi/kube/pod.go index b4bbb40ea..0af32b345 100644 --- a/pkg/controller/chi/kube/pod.go +++ b/pkg/controller/chi/kube/pod.go @@ -66,7 +66,9 @@ func (c *Pod) Get(ctx context.Context, params ...any) (*core.Pod, error) { panic(any("incorrect number or params")) } ctx = k8sCtx(ctx) - return c.kubeClient.CoreV1().Pods(namespace).Get(ctx, name, controller.NewGetOptions()) + return getWithRetry(ctx, func() (*core.Pod, error) { + return c.kubeClient.CoreV1().Pods(namespace).Get(ctx, name, controller.NewGetOptions()) + }) } func (c *Pod) GetRestartCounters(ctx context.Context, params ...any) (map[string]int, error) { diff --git a/pkg/controller/chi/kube/pvc.go b/pkg/controller/chi/kube/pvc.go index ac8d30dcc..e520dd8f1 100644 --- a/pkg/controller/chi/kube/pvc.go +++ b/pkg/controller/chi/kube/pvc.go @@ -44,7 +44,9 @@ func (c *PVC) Create(ctx context.Context, pvc *core.PersistentVolumeClaim) (*cor func (c *PVC) Get(ctx context.Context, namespace, name string) (*core.PersistentVolumeClaim, error) { ctx = k8sCtx(ctx) - return c.kubeClient.CoreV1().PersistentVolumeClaims(namespace).Get(ctx, name, controller.NewGetOptions()) + return getWithRetry(ctx, func() (*core.PersistentVolumeClaim, error) { + return c.kubeClient.CoreV1().PersistentVolumeClaims(namespace).Get(ctx, name, controller.NewGetOptions()) + }) } func (c *PVC) Update(ctx context.Context, pvc *core.PersistentVolumeClaim) (*core.PersistentVolumeClaim, error) { diff --git a/pkg/controller/chi/kube/retry.go b/pkg/controller/chi/kube/retry.go new file mode 100644 index 000000000..4b0796b68 --- /dev/null +++ b/pkg/controller/chi/kube/retry.go @@ -0,0 +1,130 @@ +// Copyright 2019 Altinity Ltd and/or its affiliates. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kube + +import ( + "context" + "errors" + "io" + "net" + "syscall" + "time" + + apiErrors "k8s.io/apimachinery/pkg/api/errors" + + log "github.com/altinity/clickhouse-operator/pkg/announcer" +) + +// transientGetRetryMaxAttempts bounds how many times a transient Get is retried +// (including the first try) before the last error is surfaced to the caller. +const transientGetRetryMaxAttempts = 5 + +// Back-off schedule between retries. These are vars (not consts) so tests can shrink +// them; treat them as read-only at runtime. +var ( + // transientGetRetryBaseDelay is the first back-off pause; it doubles each attempt. + transientGetRetryBaseDelay = 500 * time.Millisecond + // transientGetRetryMaxDelay caps the exponential back-off between attempts. + transientGetRetryMaxDelay = 5 * time.Second +) + +// isTransientAPIError reports whether err is a transient Kubernetes API or network +// error worth retrying, as opposed to a terminal/semantic error (NotFound, Conflict, +// Forbidden, Invalid, ...) that will never succeed on retry. +func isTransientAPIError(err error) bool { + if err == nil { + return false + } + + // Terminal/semantic API errors - never retry, surface immediately. + switch { + case apiErrors.IsNotFound(err), + apiErrors.IsAlreadyExists(err), + apiErrors.IsConflict(err), + apiErrors.IsForbidden(err), + apiErrors.IsUnauthorized(err), + apiErrors.IsBadRequest(err), + apiErrors.IsInvalid(err), + apiErrors.IsMethodNotSupported(err): + return false + } + + // Context cancellation means the reconcile is being torn down - do not retry. + if errors.Is(err, context.Canceled) { + return false + } + + // Transient API-status errors: apiserver overloaded, restarting, or a 5xx. + if apiErrors.IsServerTimeout(err) || + apiErrors.IsTimeout(err) || + apiErrors.IsTooManyRequests(err) || + apiErrors.IsInternalError(err) || + apiErrors.IsServiceUnavailable(err) || + apiErrors.IsUnexpectedServerError(err) { + return true + } + + // Transient transport/network errors. These are not k8s API status errors, so + // apiErrors.Is* does not catch them (this is the connection-refused case above). + if errors.Is(err, io.EOF) || + errors.Is(err, io.ErrUnexpectedEOF) || + errors.Is(err, syscall.ECONNREFUSED) || + errors.Is(err, syscall.ECONNRESET) || + errors.Is(err, syscall.ETIMEDOUT) || + errors.Is(err, syscall.EPIPE) { + return true + } + var netErr net.Error + if errors.As(err, &netErr) { + return true + } + + return false +} + +// getWithRetry runs get and, on a transient API/network error, retries it with a +// bounded exponential back-off. Success or a terminal error returns immediately. +// When retries are exhausted the last error is returned unchanged, so callers behave +// exactly as before for genuine (non-transient or sustained) outages. +// +// No object label is passed in: client-go errors are self-describing (they embed the +// full request URL, e.g. `Get "https://.../namespaces/x/configmaps/y": ...`), so the +// logged error already identifies the verb, kind, namespace and name. +func getWithRetry[T any](ctx context.Context, get func() (T, error)) (T, error) { + var ( + result T + err error + ) + delay := transientGetRetryBaseDelay + for attempt := 1; ; attempt++ { + result, err = get() + if err == nil || !isTransientAPIError(err) { + return result, err + } + if attempt >= transientGetRetryMaxAttempts { + log.V(1).F().Warning("transient API error: giving up after %d attempt(s), err: %v", attempt, err) + return result, err + } + log.V(2).F().Warning("transient API error: retrying (attempt %d/%d) in %s, err: %v", attempt, transientGetRetryMaxAttempts, delay, err) + select { + case <-ctx.Done(): + return result, err + case <-time.After(delay): + } + if delay *= 2; delay > transientGetRetryMaxDelay { + delay = transientGetRetryMaxDelay + } + } +} diff --git a/pkg/controller/chi/kube/retry_test.go b/pkg/controller/chi/kube/retry_test.go new file mode 100644 index 000000000..4e71cc705 --- /dev/null +++ b/pkg/controller/chi/kube/retry_test.go @@ -0,0 +1,145 @@ +// Copyright 2019 Altinity Ltd and/or its affiliates. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kube + +import ( + "context" + "errors" + "io" + "net" + "net/url" + "os" + "syscall" + "testing" + "time" + + apiErrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/runtime/schema" +) + +// connRefused reproduces the error client-go returns on an apiserver blip: +// `Get "https://10.x.x.x:443/...": dial tcp 10.x.x.x:443: connect: connection refused`. +func connRefused() error { + return &url.Error{ + Op: "Get", + URL: "https://10.0.0.1:443/api/v1/namespaces/x/configmaps/y", + Err: &net.OpError{ + Op: "dial", + Net: "tcp", + Err: os.NewSyscallError("connect", syscall.ECONNREFUSED), + }, + } +} + +func TestIsTransientAPIError(t *testing.T) { + gr := schema.GroupResource{Resource: "configmaps"} + tests := []struct { + name string + err error + want bool + }{ + // Not transient - terminal/semantic, or no error. + {"nil", nil, false}, + {"not found", apiErrors.NewNotFound(gr, "y"), false}, + {"conflict", apiErrors.NewConflict(gr, "y", errors.New("x")), false}, + {"forbidden", apiErrors.NewForbidden(gr, "y", errors.New("x")), false}, + {"invalid", apiErrors.NewInvalid(schema.GroupKind{Kind: "ConfigMap"}, "y", nil), false}, + {"context canceled", context.Canceled, false}, + // Transient - network / control-plane blips worth retrying. + {"connection refused", connRefused(), true}, + {"eof", io.EOF, true}, + {"unexpected eof", io.ErrUnexpectedEOF, true}, + {"too many requests", apiErrors.NewTooManyRequests("busy", 1), true}, + {"service unavailable", apiErrors.NewServiceUnavailable("down"), true}, + {"internal error", apiErrors.NewInternalError(errors.New("boom")), true}, + {"server timeout", apiErrors.NewServerTimeout(gr, "get", 1), true}, + {"dns timeout", &net.DNSError{Err: "timeout", IsTimeout: true}, true}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := isTransientAPIError(tc.err); got != tc.want { + t.Fatalf("isTransientAPIError(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} + +func TestGetWithRetry(t *testing.T) { + // Shrink the back-off so the test runs fast. + origBase, origMax := transientGetRetryBaseDelay, transientGetRetryMaxDelay + transientGetRetryBaseDelay, transientGetRetryMaxDelay = time.Millisecond, 4*time.Millisecond + defer func() { transientGetRetryBaseDelay, transientGetRetryMaxDelay = origBase, origMax }() + + t.Run("success on first try", func(t *testing.T) { + calls := 0 + got, err := getWithRetry(context.Background(), func() (int, error) { + calls++ + return 42, nil + }) + if err != nil || got != 42 || calls != 1 { + t.Fatalf("got=%d err=%v calls=%d, want 42/nil/1", got, err, calls) + } + }) + + t.Run("retries transient then succeeds", func(t *testing.T) { + calls := 0 + got, err := getWithRetry(context.Background(), func() (int, error) { + calls++ + if calls < 3 { + return 0, connRefused() + } + return 7, nil + }) + if err != nil || got != 7 || calls != 3 { + t.Fatalf("got=%d err=%v calls=%d, want 7/nil/3", got, err, calls) + } + }) + + t.Run("terminal error returns immediately without retry", func(t *testing.T) { + calls := 0 + _, err := getWithRetry(context.Background(), func() (int, error) { + calls++ + return 0, apiErrors.NewNotFound(schema.GroupResource{Resource: "configmaps"}, "y") + }) + if !apiErrors.IsNotFound(err) || calls != 1 { + t.Fatalf("err=%v calls=%d, want NotFound/1", err, calls) + } + }) + + t.Run("exhausts retries and returns last error", func(t *testing.T) { + calls := 0 + _, err := getWithRetry(context.Background(), func() (int, error) { + calls++ + return 0, connRefused() + }) + if err == nil || calls != transientGetRetryMaxAttempts { + t.Fatalf("err=%v calls=%d, want non-nil/%d", err, calls, transientGetRetryMaxAttempts) + } + }) + + t.Run("stops on context cancellation", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + calls := 0 + _, err := getWithRetry(ctx, func() (int, error) { + calls++ + return 0, connRefused() + }) + // First attempt runs; the transient wait is then short-circuited by ctx.Done(). + if err == nil || calls != 1 { + t.Fatalf("err=%v calls=%d, want non-nil/1", err, calls) + } + }) +} diff --git a/pkg/controller/chi/kube/secret.go b/pkg/controller/chi/kube/secret.go index 9a4c8d4ee..5738b15a0 100644 --- a/pkg/controller/chi/kube/secret.go +++ b/pkg/controller/chi/kube/secret.go @@ -65,7 +65,9 @@ func (c *Secret) Get(ctx context.Context, params ...any) (*core.Secret, error) { } } ctx = k8sCtx(ctx) - return c.kubeClient.CoreV1().Secrets(namespace).Get(ctx, name, controller.NewGetOptions()) + return getWithRetry(ctx, func() (*core.Secret, error) { + return c.kubeClient.CoreV1().Secrets(namespace).Get(ctx, name, controller.NewGetOptions()) + }) } func (c *Secret) Create(ctx context.Context, svc *core.Secret) (*core.Secret, error) { diff --git a/pkg/controller/chi/kube/service.go b/pkg/controller/chi/kube/service.go index 42d9ccb01..b95514128 100644 --- a/pkg/controller/chi/kube/service.go +++ b/pkg/controller/chi/kube/service.go @@ -65,7 +65,9 @@ func (c *Service) Get(ctx context.Context, params ...any) (*core.Service, error) } } ctx = k8sCtx(ctx) - return c.kubeClient.CoreV1().Services(namespace).Get(ctx, name, controller.NewGetOptions()) + return getWithRetry(ctx, func() (*core.Service, error) { + return c.kubeClient.CoreV1().Services(namespace).Get(ctx, name, controller.NewGetOptions()) + }) } func (c *Service) Create(ctx context.Context, svc *core.Service) (*core.Service, error) { diff --git a/pkg/controller/chi/kube/statesfulset.go b/pkg/controller/chi/kube/statesfulset.go index 63d4fde52..68f645ce2 100644 --- a/pkg/controller/chi/kube/statesfulset.go +++ b/pkg/controller/chi/kube/statesfulset.go @@ -70,7 +70,9 @@ func (c *STS) Get(ctx context.Context, params ...any) (*apps.StatefulSet, error) panic(any("unexpected number of args")) } ctx = k8sCtx(ctx) - return c.kubeClient.AppsV1().StatefulSets(namespace).Get(ctx, name, controller.NewGetOptions()) + return getWithRetry(ctx, func() (*apps.StatefulSet, error) { + return c.kubeClient.AppsV1().StatefulSets(namespace).Get(ctx, name, controller.NewGetOptions()) + }) } func (c *STS) Create(ctx context.Context, statefulSet *apps.StatefulSet) (*apps.StatefulSet, error) {