Skip to content

Commit 503792d

Browse files
committed
chore: add retry to main kubernetes operations
retry on failed apply and diff operations Signed-off-by: Orzelius <33936483+Orzelius@users.noreply.github.com>
1 parent 6a00c4f commit 503792d

5 files changed

Lines changed: 223 additions & 7 deletions

File tree

kubernetes/ssa/apply.go

Lines changed: 26 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -8,15 +8,18 @@ import (
88
"context"
99
"errors"
1010
"fmt"
11+
"slices"
1112
"time"
1213

1314
"github.com/fluxcd/pkg/ssa"
1415
"github.com/fluxcd/pkg/ssa/utils"
1516
"github.com/go-logr/logr"
17+
"github.com/siderolabs/go-retry/retry"
1618
apierrors "k8s.io/apimachinery/pkg/api/errors"
1719
v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
1820
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
1921

22+
"github.com/siderolabs/go-kubernetes/kubernetes"
2023
"github.com/siderolabs/go-kubernetes/kubernetes/ssa/object"
2124
)
2225

@@ -119,10 +122,29 @@ func (m *Manager) Apply(ctx context.Context, objects []*unstructured.Unstructure
119122
m.mapper.Reset()
120123
}
121124

122-
changeSet, applyErr := m.resourceManager.ApplyAllStaged(ctx, objects, ssa.ApplyOptions{
123-
Force: ops.Force,
124-
WaitInterval: ops.WaitInterval,
125-
WaitTimeout: ops.WaitTimeout,
125+
changeSet := ssa.NewChangeSet()
126+
127+
applyErr := retry.Constant(3*time.Minute, retry.WithUnits(10*time.Second), retry.WithErrorLogging(true)).RetryWithContext(ctx, func(ctx context.Context) error {
128+
var result *ssa.ChangeSet
129+
130+
result, err = m.resourceManager.ApplyAllStaged(ctx, objects, ssa.ApplyOptions{
131+
Force: ops.Force,
132+
WaitInterval: ops.WaitInterval,
133+
WaitTimeout: ops.WaitTimeout,
134+
})
135+
136+
// only push new results to avoid "unchanged" results for objects that were already applied
137+
for _, entry := range result.Entries {
138+
if !slices.ContainsFunc(changeSet.Entries, func(e ssa.ChangeSetEntry) bool { return e.Subject == entry.Subject }) {
139+
changeSet.Add(entry)
140+
}
141+
}
142+
143+
if kubernetes.IsRetryableError(err) {
144+
return retry.ExpectedError(err)
145+
}
146+
147+
return err
126148
})
127149
if applyErr != nil && changeSet == nil {
128150
return nil, fmt.Errorf("apply failed: %w", applyErr)

kubernetes/ssa/apply_test.go

Lines changed: 56 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,11 +9,13 @@ import (
99
"context"
1010
_ "embed"
1111
"errors"
12+
"sync/atomic"
1213
"testing"
1314

1415
fluxssa "github.com/fluxcd/pkg/ssa"
1516
"github.com/stretchr/testify/assert"
1617
"github.com/stretchr/testify/require"
18+
apierrors "k8s.io/apimachinery/pkg/api/errors"
1719
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
1820
sigsyaml "sigs.k8s.io/yaml"
1921

@@ -79,7 +81,7 @@ func TestApplyError(t *testing.T) {
7981
obj2 := getConfigmapManifest("configmap2")
8082

8183
results, err := manager.Apply(t.Context(), []*unstructured.Unstructured{obj1, obj2}, ssa.ApplyOptions{})
82-
require.EqualError(t, err, "apply failed", "the manager should return the error from the resourceManager apply")
84+
require.ErrorContains(t, err, "apply failed", "the manager should return the error from the resourceManager apply")
8385

8486
require.Len(t, results, 1, "results for applied objects should exist")
8587

@@ -426,6 +428,59 @@ func TestApplyEdgeCases(t *testing.T) {
426428
})
427429
}
428430

431+
// transientFailResourceManager fails ApplyAllStaged with a retryable internal error
432+
// for the first N calls, then delegates to the embedded Mock.
433+
type transientFailResourceManager struct {
434+
resourcemanager.Mock
435+
remaining atomic.Int32
436+
}
437+
438+
func (m *transientFailResourceManager) ApplyAllStaged(ctx context.Context, objects []*unstructured.Unstructured, opts fluxssa.ApplyOptions) (*fluxssa.ChangeSet, error) {
439+
if m.remaining.Add(-1) >= 0 {
440+
// Apply the first object successfully, then return a retryable error.
441+
cs := fluxssa.NewChangeSet()
442+
443+
entry, err := m.Apply(ctx, objects[0], opts)
444+
if err != nil {
445+
return cs, err
446+
}
447+
448+
cs.Add(*entry)
449+
450+
return cs, apierrors.NewInternalError(errors.New("transient API server error"))
451+
}
452+
453+
return m.Mock.ApplyAllStaged(ctx, objects, opts)
454+
}
455+
456+
func TestApplyRetry(t *testing.T) {
457+
// First attempt: cm-1 applied successfully, cm-2 fails with conflict.
458+
// Second attempt: both succeed. Verify results are correct and not duplicated.
459+
rm := &transientFailResourceManager{}
460+
rm.remaining.Store(1)
461+
462+
inv := memory.NewInventory("test-inventory")
463+
manager := ssa.NewCustomManager(rm, testInventoryClosure(t.Context(), inv), nil, &mapperMock{})
464+
465+
obj1 := getConfigmapManifest("cm-1")
466+
obj2 := getConfigmapManifest("cm-2")
467+
468+
results, err := manager.Apply(t.Context(), []*unstructured.Unstructured{obj1, obj2}, ssa.ApplyOptions{})
469+
require.NoError(t, err)
470+
require.Len(t, results, 2, "each object should appear exactly once in results")
471+
472+
resultsByName := map[string]ssa.Change{}
473+
for _, r := range results {
474+
resultsByName[r.ObjMetadata.Name] = r
475+
}
476+
477+
assert.Equal(t, ssa.CreatedAction, resultsByName["cm-1"].Action)
478+
assert.Equal(t, ssa.CreatedAction, resultsByName["cm-2"].Action)
479+
480+
invContents := inv.Get()
481+
require.Len(t, invContents, 2, "both objects should be in inventory after successful retry")
482+
}
483+
429484
func getConfigmapManifest(name string) *unstructured.Unstructured {
430485
obj := &unstructured.Unstructured{
431486
Object: map[string]any{

kubernetes/ssa/diff.go

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,15 +8,19 @@ import (
88
"context"
99
"fmt"
1010
"strings"
11+
"time"
1112

1213
"github.com/fluxcd/cli-utils/pkg/object"
1314
"github.com/fluxcd/pkg/ssa"
1415
"github.com/go-logr/logr"
16+
"github.com/siderolabs/go-retry/retry"
1517
"github.com/siderolabs/talos/pkg/machinery/textdiff"
1618
apierrors "k8s.io/apimachinery/pkg/api/errors"
1719
"k8s.io/apimachinery/pkg/api/meta"
1820
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
1921
k8syaml "sigs.k8s.io/yaml"
22+
23+
"github.com/siderolabs/go-kubernetes/kubernetes"
2024
)
2125

2226
// DiffOptions defines the options for the Diff method.
@@ -148,7 +152,24 @@ func (m *Manager) diff(
148152
invPolicy InventoryPolicy,
149153
invID string,
150154
) (*ssa.ChangeSetEntry, string, error) {
151-
changeSet, inClusterObj, dryRunResult, err := m.resourceManager.Diff(ctx, inputObj, ssa.DiffOptions{Force: force})
155+
var (
156+
changeSet *ssa.ChangeSetEntry
157+
inClusterObj *unstructured.Unstructured
158+
dryRunResult *unstructured.Unstructured
159+
)
160+
161+
err := retry.Constant(30*time.Second, retry.WithUnits(5*time.Second), retry.WithErrorLogging(true)).RetryWithContext(ctx, func(ctx context.Context) error {
162+
var err error
163+
164+
changeSet, inClusterObj, dryRunResult, err = m.resourceManager.Diff(ctx, inputObj, ssa.DiffOptions{Force: force})
165+
166+
if kubernetes.IsRetryableError(err) {
167+
return retry.ExpectedError(err)
168+
}
169+
170+
return err
171+
})
172+
152173
if err != nil && (apierrors.IsNotFound(err) || meta.IsNoMatchError(err) || strings.Contains(err.Error(), "not found")) {
153174
if changeSet == nil {
154175
changeSet = &ssa.ChangeSetEntry{

kubernetes/ssa/wait.go

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,12 +10,18 @@ import (
1010
"github.com/fluxcd/cli-utils/pkg/object"
1111
"github.com/fluxcd/pkg/ssa"
1212
"github.com/go-logr/logr"
13+
"github.com/siderolabs/go-retry/retry"
14+
15+
"github.com/siderolabs/go-kubernetes/kubernetes"
1316
)
1417

1518
// WaitOptions contains options for wait requests.
1619
type WaitOptions = ssa.WaitOptions
1720

1821
// Wait checks if the given set of objects has been fully reconciled.
22+
//
23+
// The total wait time is bound by ops.Timeout. Transient network errors
24+
// (connection resets, API server timeouts) are retried within that budget.
1925
func (m *Manager) Wait(ctx context.Context, set object.ObjMetadataSet, ops WaitOptions) error {
2026
ctx = logr.NewContext(ctx, logr.FromContextOrDiscard(ctx))
2127

@@ -27,5 +33,13 @@ func (m *Manager) Wait(ctx context.Context, set object.ObjMetadataSet, ops WaitO
2733
ops.Timeout = ssa.DefaultApplyOptions().WaitTimeout
2834
}
2935

30-
return m.resourceManager.WaitForSetWithContext(ctx, set, ops)
36+
return retry.Constant(ops.Timeout, retry.WithUnits(ops.Interval), retry.WithErrorLogging(true)).RetryWithContext(ctx, func(ctx context.Context) error {
37+
err := m.resourceManager.WaitForSetWithContext(ctx, set, ops)
38+
39+
if kubernetes.IsRetryableError(err) {
40+
return retry.ExpectedError(err)
41+
}
42+
43+
return err
44+
})
3145
}

kubernetes/ssa/wait_test.go

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
// This Source Code Form is subject to the terms of the Mozilla Public
2+
// License, v. 2.0. If a copy of the MPL was not distributed with this
3+
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
4+
5+
// nolint: contextcheck,godoclint
6+
package ssa_test
7+
8+
import (
9+
"context"
10+
"errors"
11+
"sync/atomic"
12+
"testing"
13+
"time"
14+
15+
"github.com/fluxcd/cli-utils/pkg/object"
16+
fluxssa "github.com/fluxcd/pkg/ssa"
17+
"github.com/stretchr/testify/assert"
18+
"github.com/stretchr/testify/require"
19+
apierrors "k8s.io/apimachinery/pkg/api/errors"
20+
21+
"github.com/siderolabs/go-kubernetes/kubernetes/ssa"
22+
"github.com/siderolabs/go-kubernetes/kubernetes/ssa/internal/resourcemanager"
23+
)
24+
25+
// retryableWaitResourceManager returns a retryable error for the first N calls
26+
// to WaitForSetWithContext, then succeeds.
27+
type retryableWaitResourceManager struct {
28+
resourcemanager.Mock
29+
remaining atomic.Int32
30+
}
31+
32+
func (m *retryableWaitResourceManager) WaitForSetWithContext(_ context.Context, _ object.ObjMetadataSet, _ fluxssa.WaitOptions) error {
33+
if m.remaining.Add(-1) >= 0 {
34+
return apierrors.NewInternalError(errors.New("transient API server error"))
35+
}
36+
37+
return nil
38+
}
39+
40+
func TestWait(t *testing.T) {
41+
t.Run("retries_on_transient_error_then_succeeds", func(t *testing.T) {
42+
rm := &retryableWaitResourceManager{}
43+
rm.remaining.Store(2) // fail twice, succeed on third
44+
45+
manager := ssa.NewCustomManager(rm, testInventoryFactory, nil, &mapperMock{})
46+
47+
err := manager.Wait(t.Context(), object.ObjMetadataSet{}, ssa.WaitOptions{
48+
Timeout: 30 * time.Second,
49+
Interval: 1 * time.Second,
50+
})
51+
require.NoError(t, err)
52+
53+
// remaining should be -1
54+
assert.Equal(t, rm.remaining.Load(), int32(-1))
55+
})
56+
57+
t.Run("respects_timeout_budget", func(t *testing.T) {
58+
// Always returns a retryable error — Wait must still return within the timeout.
59+
rm := &retryableWaitResourceManager{}
60+
rm.remaining.Store(1000) // never succeeds
61+
62+
manager := ssa.NewCustomManager(rm, testInventoryFactory, nil, &mapperMock{})
63+
64+
timeout := 3 * time.Second
65+
66+
start := time.Now()
67+
68+
err := manager.Wait(t.Context(), object.ObjMetadataSet{}, ssa.WaitOptions{
69+
Timeout: timeout,
70+
Interval: 500 * time.Millisecond,
71+
})
72+
elapsed := time.Since(start)
73+
74+
require.Error(t, err)
75+
assert.Less(t, elapsed, timeout+100*time.Millisecond, "Wait should not exceed timeout by more than a small margin")
76+
})
77+
78+
t.Run("non_retryable_error_returns_immediately", func(t *testing.T) {
79+
rm := &permanentWaitFailResourceManager{err: errors.New("resources failed")}
80+
81+
manager := ssa.NewCustomManager(rm, testInventoryFactory, nil, &mapperMock{})
82+
83+
start := time.Now()
84+
85+
err := manager.Wait(t.Context(), object.ObjMetadataSet{}, ssa.WaitOptions{
86+
Timeout: 30 * time.Second,
87+
Interval: 1 * time.Second,
88+
})
89+
elapsed := time.Since(start)
90+
91+
require.ErrorContains(t, err, "resources failed")
92+
assert.Less(t, elapsed, 2*time.Second, "non-retryable error should return immediately")
93+
})
94+
}
95+
96+
// permanentWaitFailResourceManager always returns a non-retryable error.
97+
type permanentWaitFailResourceManager struct {
98+
resourcemanager.Mock
99+
err error
100+
}
101+
102+
func (m *permanentWaitFailResourceManager) WaitForSetWithContext(_ context.Context, _ object.ObjMetadataSet, _ fluxssa.WaitOptions) error {
103+
return m.err
104+
}

0 commit comments

Comments
 (0)