Retry transient Kubernetes API errors in chi kube driver Get calls

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) {

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) {

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) {

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) {

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) {

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
+		}
+	}
+}

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)
+		}
+	})
+}

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) {

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) {

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) {