From 6ec81006cfa140a6bcff1be4ca3a4070921d4997 Mon Sep 17 00:00:00 2001 From: OpenClaw Agent Date: Mon, 15 Jun 2026 14:09:34 +0000 Subject: [PATCH 1/3] fix: harden topology sync against transient API server errors at startup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wrap SynchronizeTopology in a bounded exponential-backoff retry loop (SynchronizeTopologyWithRetry) so that transient Kubernetes API server errors — 5xx, service-unavailable, request-timeout, rate-limiting — no longer cause the Grove operator process to exit at startup. Permanent errors (Forbidden, Unauthorized) are returned immediately without retrying. The default retry budget is six steps with doubling delay (approx. 63 s total), with 10% jitter to reduce thundering-herd load on a recovering control plane. main.go is updated to call SynchronizeTopologyWithRetry with the default backoff. Three unit tests are added to cover: - success after N transient failures - permanent errors not retried (single call only) - transient errors exhaust the retry budget and return the last error Closes #654 Signed-off-by: OpenClaw Agent --- operator/cmd/main.go | 3 +- .../clustertopology/clustertopology.go | 38 +++++++++ .../clustertopology/clustertopology_test.go | 79 +++++++++++++++++++ 3 files changed, 119 insertions(+), 1 deletion(-) diff --git a/operator/cmd/main.go b/operator/cmd/main.go index 620654a3f..1a5968f6d 100644 --- a/operator/cmd/main.go +++ b/operator/cmd/main.go @@ -102,7 +102,8 @@ func main() { // Synchronize backend topologies for all existing ClusterTopologyBinding resources. // This must be done before starting the controllers that may depend on the ClusterTopologyBinding resource. - if err = clustertopology.SynchronizeTopology(ctx, cl, logger, schedRegistry.AllTopologyAware()); err != nil { + // Uses a bounded exponential backoff so that transient API-server errors do not crash the operator. + if err = clustertopology.SynchronizeTopologyWithRetry(ctx, cl, logger, schedRegistry.AllTopologyAware(), clustertopology.DefaultSyncRetryBackoff); err != nil { logger.Error(err, "failed to synchronize cluster topology") handleErrorAndExit(err, cli.ExitErrSynchronizeTopology) } diff --git a/operator/internal/clustertopology/clustertopology.go b/operator/internal/clustertopology/clustertopology.go index c8d513a21..69933dece 100644 --- a/operator/internal/clustertopology/clustertopology.go +++ b/operator/internal/clustertopology/clustertopology.go @@ -19,14 +19,52 @@ package clustertopology import ( "context" "fmt" + "time" grovecorev1alpha1 "github.com/ai-dynamo/grove/operator/api/core/v1alpha1" "github.com/ai-dynamo/grove/operator/internal/scheduler" "github.com/go-logr/logr" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/util/retry" "sigs.k8s.io/controller-runtime/pkg/client" ) +// DefaultSyncRetryBackoff is the retry backoff used by SynchronizeTopologyWithRetry. +// Six steps with doubling delay gives roughly 60 seconds of retry budget (1+2+4+8+16+32). +var DefaultSyncRetryBackoff = wait.Backoff{ + Duration: time.Second, + Factor: 2.0, + Jitter: 0.1, + Steps: 6, +} + +// SynchronizeTopologyWithRetry wraps SynchronizeTopology with a bounded exponential +// backoff so that transient API-server errors (5xx, timeouts, service-unavailable) +// at operator startup do not crash the operator process. +// Permanent errors (Forbidden, Unauthorized) are returned immediately without retrying. +func SynchronizeTopologyWithRetry(ctx context.Context, cl client.Client, logger logr.Logger, backends map[string]scheduler.TopologyAwareBackend, backoff wait.Backoff) error { + attempt := 0 + return retry.OnError(backoff, isTransientAPIError, func() error { + if attempt > 0 { + logger.Info("Retrying topology synchronization after transient error", "attempt", attempt+1) + } + attempt++ + return SynchronizeTopology(ctx, cl, logger, backends) + }) +} + +// isTransientAPIError reports whether err is a transient Kubernetes API server error +// that is safe to retry. Permanent errors (Forbidden, Unauthorized) return false. +func isTransientAPIError(err error) bool { + return apierrors.IsServerTimeout(err) || + apierrors.IsServiceUnavailable(err) || + apierrors.IsInternalError(err) || + apierrors.IsTimeout(err) || + apierrors.IsTooManyRequests(err) +} + // SynchronizeTopology synchronizes scheduler-specific topology resources at operator startup. // Lists all existing ClusterTopologyBinding resources and ensures backend topologies exist for each. // Called before controllers start to avoid races with PCS reconciliation. diff --git a/operator/internal/clustertopology/clustertopology_test.go b/operator/internal/clustertopology/clustertopology_test.go index 52a063e3f..92a3229f2 100644 --- a/operator/internal/clustertopology/clustertopology_test.go +++ b/operator/internal/clustertopology/clustertopology_test.go @@ -19,6 +19,7 @@ package clustertopology import ( "context" "testing" + "time" apicommonconstants "github.com/ai-dynamo/grove/operator/api/common/constants" configv1alpha1 "github.com/ai-dynamo/grove/operator/api/config/v1alpha1" @@ -34,6 +35,7 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/util/uuid" "sigs.k8s.io/controller-runtime/pkg/client" ) @@ -303,6 +305,83 @@ func createTestClusterTopology(name string, levels []grovecorev1alpha1.TopologyL } } +// transientListClient wraps a client.Client and returns failErr for the first failCount List calls, +// then delegates all subsequent calls to the underlying client. Used to simulate transient API errors. +type transientListClient struct { + client.Client + failCount int + callCount int + failErr error +} + +func (c *transientListClient) List(ctx context.Context, objList client.ObjectList, opts ...client.ListOption) error { + c.callCount++ + if c.callCount <= c.failCount { + return c.failErr + } + return c.Client.List(ctx, objList, opts...) +} + +// fastBackoff is a zero-duration backoff suitable for unit tests to avoid slowing them down. +var fastBackoff = wait.Backoff{Duration: time.Millisecond, Factor: 1.0, Steps: 10} + +func TestSynchronizeTopologyWithRetry_SucceedsAfterTransientError(t *testing.T) { + ctx := context.Background() + ct := createTestClusterTopology(topologyName, []grovecorev1alpha1.TopologyLevel{ + {Domain: grovecorev1alpha1.TopologyDomainHost, Key: "kubernetes.io/hostname"}, + }) + base := testutils.CreateDefaultFakeClient([]client.Object{ct}) + // Inject a transient 500 error for the first 2 List calls; the 3rd call succeeds. + cl := &transientListClient{ + Client: base, + failCount: 2, + failErr: apierrors.NewInternalError(assert.AnError), + } + + err := SynchronizeTopologyWithRetry(ctx, cl, logr.Discard(), newKaiBackends(base), fastBackoff) + require.NoError(t, err) + assert.Equal(t, 3, cl.callCount, "expected 2 transient failures then 1 success") + + // Verify topology was still created correctly. + kaiTopology := &kaitopologyv1alpha1.Topology{} + require.NoError(t, base.Get(ctx, client.ObjectKey{Name: topologyName}, kaiTopology)) + assert.Len(t, kaiTopology.Spec.Levels, 1) +} + +func TestSynchronizeTopologyWithRetry_PermanentErrorNotRetried(t *testing.T) { + ctx := context.Background() + permErr := apierrors.NewForbidden(schema.GroupResource{Resource: "clustertopologybindings"}, "", assert.AnError) + ctListGVK := grovecorev1alpha1.SchemeGroupVersion.WithKind("ClusterTopologyBindingList") + cl := testutils.NewTestClientBuilder(). + RecordErrorForObjectsMatchingLabels(testutils.ClientMethodList, client.ObjectKey{}, ctListGVK, nil, permErr). + Build() + + callCount := 0 + // Wrap the permanent-error client in a counter to verify it is only called once. + countingCl := &transientListClient{Client: cl, failCount: 100, failErr: permErr} + + err := SynchronizeTopologyWithRetry(ctx, countingCl, logr.Discard(), newKaiBackends(cl), fastBackoff) + require.Error(t, err) + assert.True(t, apierrors.IsForbidden(err), "expected Forbidden error to propagate") + // The permanent error must not be retried: fn is called exactly once. + _ = callCount + assert.Equal(t, 1, countingCl.callCount, "permanent error should not be retried") +} + +func TestSynchronizeTopologyWithRetry_ExhaustsRetriesOnPersistentTransientError(t *testing.T) { + ctx := context.Background() + transientErr := apierrors.NewServiceUnavailable("apiserver temporarily unavailable") + ctListGVK := grovecorev1alpha1.SchemeGroupVersion.WithKind("ClusterTopologyBindingList") + cl := testutils.NewTestClientBuilder(). + RecordErrorForObjectsMatchingLabels(testutils.ClientMethodList, client.ObjectKey{}, ctListGVK, nil, transientErr). + Build() + + shortBackoff := wait.Backoff{Duration: time.Millisecond, Factor: 1.0, Steps: 3} + err := SynchronizeTopologyWithRetry(ctx, cl, logr.Discard(), newKaiBackends(cl), shortBackoff) + require.Error(t, err, "should fail after exhausting retry budget") + assert.True(t, apierrors.IsServiceUnavailable(err)) +} + func TestBuildSchedulerReferenceMap(t *testing.T) { refs := []grovecorev1alpha1.SchedulerTopologyBinding{ {SchedulerName: "kai-scheduler", TopologyReference: "kai-topo"}, From 52fa34440ad3c6c60846450c936c2ba7cf83a810 Mon Sep 17 00:00:00 2001 From: OpenClaw Agent Date: Mon, 15 Jun 2026 18:01:41 +0000 Subject: [PATCH 2/3] fix: sort imports alphabetically in clustertopology_test.go uuid must precede wait to satisfy goimports ordering enforced by CI check. Co-Authored-By: Claude Sonnet 4.6 Signed-off-by: OpenClaw Agent --- operator/internal/clustertopology/clustertopology_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/operator/internal/clustertopology/clustertopology_test.go b/operator/internal/clustertopology/clustertopology_test.go index 92a3229f2..a5e2a296d 100644 --- a/operator/internal/clustertopology/clustertopology_test.go +++ b/operator/internal/clustertopology/clustertopology_test.go @@ -35,8 +35,8 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime/schema" - "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/util/uuid" + "k8s.io/apimachinery/pkg/util/wait" "sigs.k8s.io/controller-runtime/pkg/client" ) From fc5b7be944cf175afaef46b4713b473fcd198474 Mon Sep 17 00:00:00 2001 From: OpenClaw Agent Date: Tue, 16 Jun 2026 18:08:59 +0000 Subject: [PATCH 3/3] fix: extend isTransientAPIError to cover transport-level errors Several startup failures from issue #654 (API-server rolling restart, network partition, connection timeout) surface as transport-level errors rather than Kubernetes StatusErrors. The Go client returns these as net.Error timeouts, io.EOF, or syscall.ECONNREFUSED before wrapping them in a StatusError, so all apierrors.IsXxx() calls return false and the operator would not retry them. Add transport-level checks to isTransientAPIError: - net.Error with Timeout() == true (dial/read timeouts) - io.EOF (connection dropped mid-request) - syscall.ECONNREFUSED (API server not yet accepting connections) Also adds TestIsTransientAPIError (table-driven) and TestSynchronizeTopologyWithRetry_TransportErrorRetried to cover the new paths. Co-Authored-By: Claude Sonnet 4.6 Signed-off-by: OpenClaw Agent --- .../clustertopology/clustertopology.go | 23 ++++- .../clustertopology/clustertopology_test.go | 89 +++++++++++++++++++ 2 files changed, 108 insertions(+), 4 deletions(-) diff --git a/operator/internal/clustertopology/clustertopology.go b/operator/internal/clustertopology/clustertopology.go index 69933dece..61db5b69c 100644 --- a/operator/internal/clustertopology/clustertopology.go +++ b/operator/internal/clustertopology/clustertopology.go @@ -18,7 +18,11 @@ package clustertopology import ( "context" + "errors" "fmt" + "io" + "net" + "syscall" "time" grovecorev1alpha1 "github.com/ai-dynamo/grove/operator/api/core/v1alpha1" @@ -55,14 +59,25 @@ func SynchronizeTopologyWithRetry(ctx context.Context, cl client.Client, logger }) } -// isTransientAPIError reports whether err is a transient Kubernetes API server error -// that is safe to retry. Permanent errors (Forbidden, Unauthorized) return false. +// isTransientAPIError reports whether err is a transient error that is safe to retry. +// It covers both Kubernetes API-level errors (5xx, timeouts) and transport-level +// errors (connection refused, EOF, net timeout) that surface before a StatusError +// is returned — e.g. during an API-server rolling restart or network partition. +// Permanent errors (Forbidden, Unauthorized) return false. func isTransientAPIError(err error) bool { - return apierrors.IsServerTimeout(err) || + if apierrors.IsServerTimeout(err) || apierrors.IsServiceUnavailable(err) || apierrors.IsInternalError(err) || apierrors.IsTimeout(err) || - apierrors.IsTooManyRequests(err) + apierrors.IsTooManyRequests(err) { + return true + } + // Transport-level transient errors. + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return true + } + return errors.Is(err, io.EOF) || errors.Is(err, syscall.ECONNREFUSED) } // SynchronizeTopology synchronizes scheduler-specific topology resources at operator startup. diff --git a/operator/internal/clustertopology/clustertopology_test.go b/operator/internal/clustertopology/clustertopology_test.go index a5e2a296d..848402764 100644 --- a/operator/internal/clustertopology/clustertopology_test.go +++ b/operator/internal/clustertopology/clustertopology_test.go @@ -18,6 +18,10 @@ package clustertopology import ( "context" + "fmt" + "io" + "net" + "syscall" "testing" "time" @@ -382,6 +386,91 @@ func TestSynchronizeTopologyWithRetry_ExhaustsRetriesOnPersistentTransientError( assert.True(t, apierrors.IsServiceUnavailable(err)) } +// mockTimeoutError is a net.Error that reports Timeout() == true, mimicking a +// transport-level dial timeout (e.g. API server rolling restart, network partition). +type mockTimeoutError struct{} + +func (mockTimeoutError) Error() string { return "i/o timeout" } +func (mockTimeoutError) Timeout() bool { return true } +func (mockTimeoutError) Temporary() bool { return true } + +func TestIsTransientAPIError(t *testing.T) { + tests := []struct { + name string + err error + wantRetry bool + }{ + { + name: "apierrors.InternalError is transient", + err: apierrors.NewInternalError(fmt.Errorf("internal")), + wantRetry: true, + }, + { + name: "apierrors.ServiceUnavailable is transient", + err: apierrors.NewServiceUnavailable("unavailable"), + wantRetry: true, + }, + { + name: "apierrors.Forbidden is permanent", + err: apierrors.NewForbidden(schema.GroupResource{}, "", fmt.Errorf("denied")), + wantRetry: false, + }, + { + name: "net.Error timeout is transient", + err: mockTimeoutError{}, + wantRetry: true, + }, + { + name: "io.EOF is transient", + err: io.EOF, + wantRetry: true, + }, + { + name: "ECONNREFUSED is transient", + err: syscall.ECONNREFUSED, + wantRetry: true, + }, + { + name: "wrapped net.Error timeout is transient", + err: fmt.Errorf("wrapped: %w", mockTimeoutError{}), + wantRetry: true, + }, + { + name: "wrapped ECONNREFUSED is transient", + err: fmt.Errorf("wrapped: %w", syscall.ECONNREFUSED), + wantRetry: true, + }, + { + name: "net.Error non-timeout is not transient", + err: &net.OpError{Err: fmt.Errorf("some non-timeout net error")}, + wantRetry: false, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.wantRetry, isTransientAPIError(tc.err)) + }) + } +} + +func TestSynchronizeTopologyWithRetry_TransportErrorRetried(t *testing.T) { + ctx := context.Background() + ct := createTestClusterTopology(topologyName, []grovecorev1alpha1.TopologyLevel{ + {Domain: grovecorev1alpha1.TopologyDomainHost, Key: "kubernetes.io/hostname"}, + }) + base := testutils.CreateDefaultFakeClient([]client.Object{ct}) + // Inject a transport-level timeout for the first 2 List calls; the 3rd succeeds. + cl := &transientListClient{ + Client: base, + failCount: 2, + failErr: mockTimeoutError{}, + } + + err := SynchronizeTopologyWithRetry(ctx, cl, logr.Discard(), newKaiBackends(base), fastBackoff) + require.NoError(t, err) + assert.Equal(t, 3, cl.callCount, "expected 2 transient transport failures then 1 success") +} + func TestBuildSchedulerReferenceMap(t *testing.T) { refs := []grovecorev1alpha1.SchedulerTopologyBinding{ {SchedulerName: "kai-scheduler", TopologyReference: "kai-topo"},