diff --git a/internal/composition/composition.go b/internal/composition/composition.go index 0199141..cc93f51 100644 --- a/internal/composition/composition.go +++ b/internal/composition/composition.go @@ -266,7 +266,7 @@ func (h *handler) Observe(ctx context.Context, mg *unstructured.Unstructured) (c "waiting for an in-flight helm operation to settle (status %s, pending for %s)", string(rel.Status), pendingFor.Truncate(time.Second)) unstructuredtools.SetConditions(mg, pendingCond) - if _, uerr := tools.UpdateStatus(ctx, mg, updateOpts); uerr != nil { + if _, uerr := updateStatusWithRetry(ctx, mg, updateOpts); uerr != nil { // Non-fatal: the wait itself is not an error, and failing the reconcile here would // burn retry budget on a purely cosmetic write. log.Debug("Could not record the pending-operation condition", "error", uerr) @@ -310,7 +310,7 @@ func (h *handler) Observe(ctx context.Context, mg *unstructured.Unstructured) (c condition := condition.Unavailable() condition.Message = retErr.Error() unstructuredtools.SetConditions(mg, condition) - _, err = tools.UpdateStatus(ctx, mg, updateOpts) + _, err = updateStatusWithRetry(ctx, mg, updateOpts) if err != nil { return controller.ExternalObservation{}, fmt.Errorf("updating status after failure: %w", err) } @@ -326,7 +326,7 @@ func (h *handler) Observe(ctx context.Context, mg *unstructured.Unstructured) (c condition := condition.Unavailable() condition.Message = retErr.Error() unstructuredtools.SetConditions(mg, condition) - _, err = tools.UpdateStatus(ctx, mg, updateOpts) + _, err = updateStatusWithRetry(ctx, mg, updateOpts) if err != nil { return controller.ExternalObservation{}, fmt.Errorf("updating status after failure: %w", err) } @@ -398,7 +398,7 @@ func (h *handler) Observe(ctx context.Context, mg *unstructured.Unstructured) (c condition := condition.Unavailable() condition.Message = retErr.Error() unstructuredtools.SetConditions(mg, condition) - _, err = tools.UpdateStatus(ctx, mg, updateOpts) + _, err = updateStatusWithRetry(ctx, mg, updateOpts) if err != nil { return controller.ExternalObservation{}, fmt.Errorf("updating status after failure: %w", err) } @@ -445,7 +445,7 @@ func (h *handler) Observe(ctx context.Context, mg *unstructured.Unstructured) (c return controller.ExternalObservation{}, err } - _, err = tools.UpdateStatus(ctx, mg, updateOpts) + _, err = updateStatusWithRetry(ctx, mg, updateOpts) if err != nil { return controller.ExternalObservation{}, err } @@ -621,7 +621,7 @@ func (h *handler) Create(ctx context.Context, mg *unstructured.Unstructured) err log.Debug("Composition created.", "package", pkg.URL) h.eventRecorder.Event(mg, event.Normal(reasonCreated, "Create", fmt.Sprintf("Composition created: %s", mg.GetName()))) - mg, err = tools.UpdateStatus(ctx, mg, updateOpts) + mg, err = updateStatusWithRetry(ctx, mg, updateOpts) if err != nil { return fmt.Errorf("updating cr with values: %w", err) } @@ -765,7 +765,7 @@ func (h *handler) Update(ctx context.Context, mg *unstructured.Unstructured) err return fmt.Errorf("setting status: %w", err) } - mg, err = tools.UpdateStatus(ctx, mg, tools.UpdateOptions{ + mg, err = updateStatusWithRetry(ctx, mg, tools.UpdateOptions{ Pluralizer: h.pluralizer, DynamicClient: dyn, }) diff --git a/internal/composition/status_update.go b/internal/composition/status_update.go new file mode 100644 index 0000000..750f26a --- /dev/null +++ b/internal/composition/status_update.go @@ -0,0 +1,57 @@ +package composition + +import ( + "context" + + "github.com/krateoplatformops/unstructured-runtime/pkg/tools" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/client-go/util/retry" +) + +// updateStatusWithRetry writes mg's status subresource, retrying on optimistic-concurrency +// conflicts. On a 409 it re-reads the object solely to pick up the current resourceVersion, then +// re-submits the status this reconcile already computed. It is a drop-in replacement for +// tools.UpdateStatus. +// +// Why this exists: tools.UpdateStatus submits mg with whatever resourceVersion it was read at and +// performs no retry, so a benign "the object has been modified" 409 aborts the whole reconcile. +// During a GVK-migration handover the retiring per-version controller and this one briefly contend +// on the same composition CR's status, making that 409 routine; without a retry the healthy +// Ready/Synced condition never lands and the composition stays wedged Ready=False until someone +// manually restarts the controller (braghettos/krateo-core-provider#57). Re-submitting our status +// is safe because this controller owns the status subresource (last-writer-wins). +// +// Note: we carry mg's status as-is and refresh only its resourceVersion — we deliberately do NOT +// deep-copy the status subtree. mg.Object["status"] can hold typed values that +// runtime.DeepCopyJSONValue (used by unstructured.NestedFieldCopy) panics on. +func updateStatusWithRetry(ctx context.Context, mg *unstructured.Unstructured, opts tools.UpdateOptions) (*unstructured.Unstructured, error) { + var result *unstructured.Unstructured + err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + updated, uErr := tools.UpdateStatus(ctx, mg, opts) + if uErr == nil { + result = updated + return nil + } + if !apierrors.IsConflict(uErr) { + return uErr + } + // Conflict: re-read for a fresh resourceVersion and let RetryOnConflict re-invoke. + // Pluralizer/DynamicClient are non-nil here — tools.UpdateStatus validates them and would + // have returned a non-conflict error otherwise. + gvr, gErr := opts.Pluralizer.GVKtoGVR(mg.GroupVersionKind()) + if gErr != nil { + return gErr + } + latest, gErr := opts.DynamicClient.Resource(gvr). + Namespace(mg.GetNamespace()). + Get(ctx, mg.GetName(), metav1.GetOptions{}) + if gErr != nil { + return gErr + } + mg.SetResourceVersion(latest.GetResourceVersion()) + return uErr + }) + return result, err +} diff --git a/internal/composition/status_update_test.go b/internal/composition/status_update_test.go new file mode 100644 index 0000000..bc7aac3 --- /dev/null +++ b/internal/composition/status_update_test.go @@ -0,0 +1,133 @@ +package composition + +import ( + "context" + "fmt" + "strings" + "testing" + + "github.com/gobuffalo/flect" + "github.com/krateoplatformops/unstructured-runtime/pkg/pluralizer" + "github.com/krateoplatformops/unstructured-runtime/pkg/tools" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + dynamicfake "k8s.io/client-go/dynamic/fake" + clienttesting "k8s.io/client-go/testing" +) + +type statusRetryPluralizer struct{} + +var _ pluralizer.PluralizerInterface = statusRetryPluralizer{} + +func (statusRetryPluralizer) GVKtoGVR(gvk schema.GroupVersionKind) (schema.GroupVersionResource, error) { + return schema.GroupVersionResource{ + Group: gvk.Group, + Version: gvk.Version, + Resource: flect.Pluralize(strings.ToLower(gvk.Kind)), + }, nil +} + +var srGVR = schema.GroupVersionResource{Group: "composition.krateo.io", Version: "v0-1-0", Resource: "portals"} + +func srPortal(rv, statusMarker string) *unstructured.Unstructured { + u := &unstructured.Unstructured{} + u.SetGroupVersionKind(schema.GroupVersionKind{Group: "composition.krateo.io", Version: "v0-1-0", Kind: "Portal"}) + u.SetName("portal") + u.SetNamespace("krateo-system") + u.SetResourceVersion(rv) + _ = unstructured.SetNestedField(u.Object, statusMarker, "status", "marker") + return u +} + +func srFakeDyn(objs ...runtime.Object) *dynamicfake.FakeDynamicClient { + scheme := runtime.NewScheme() + gvrToListKind := map[schema.GroupVersionResource]string{srGVR: "PortalList"} + return dynamicfake.NewSimpleDynamicClientWithCustomListKinds(scheme, gvrToListKind, objs...) +} + +func srOpts(dyn *dynamicfake.FakeDynamicClient) tools.UpdateOptions { + return tools.UpdateOptions{Pluralizer: statusRetryPluralizer{}, DynamicClient: dyn} +} + +// A 409 on the first status write (as happens during a GVK-migration handover) must be retried +// after a re-fetch, and the status this reconcile computed must ultimately land. Regression guard +// for braghettos/krateo-core-provider#57. +func TestUpdateStatusWithRetry_RecoversFromConflict(t *testing.T) { + // The stored object was concurrently modified (rv=2) after we read it (rv=1). + dyn := srFakeDyn(srPortal("2", "concurrent")) + + var statusUpdates int + dyn.PrependReactor("update", "portals", func(action clienttesting.Action) (bool, runtime.Object, error) { + ua, ok := action.(clienttesting.UpdateAction) + if !ok || ua.GetSubresource() != "status" { + return false, nil, nil + } + statusUpdates++ + if statusUpdates == 1 { + return true, nil, apierrors.NewConflict(srGVR.GroupResource(), "portal", fmt.Errorf("the object has been modified")) + } + return false, nil, nil // fall through to the tracker (success) + }) + + // Our reconcile computed a "healthy" status against the now-stale rv=1. + mg := srPortal("1", "healthy") + got, err := updateStatusWithRetry(context.Background(), mg, srOpts(dyn)) + if err != nil { + t.Fatalf("expected recovery from the conflict, got error: %v", err) + } + if statusUpdates < 2 { + t.Fatalf("expected a retry (>=2 status writes), got %d", statusUpdates) + } + if m, _, _ := unstructured.NestedString(got.Object, "status", "marker"); m != "healthy" { + t.Fatalf("computed status should win after re-fetch; got marker=%q", m) + } +} + +// A non-conflict error must propagate immediately and never be retried. +func TestUpdateStatusWithRetry_NonConflictNotRetried(t *testing.T) { + dyn := srFakeDyn(srPortal("1", "x")) + + var statusUpdates int + dyn.PrependReactor("update", "portals", func(action clienttesting.Action) (bool, runtime.Object, error) { + ua, ok := action.(clienttesting.UpdateAction) + if !ok || ua.GetSubresource() != "status" { + return false, nil, nil + } + statusUpdates++ + return true, nil, apierrors.NewInternalError(fmt.Errorf("boom")) + }) + + _, err := updateStatusWithRetry(context.Background(), srPortal("1", "healthy"), srOpts(dyn)) + if err == nil { + t.Fatal("expected the non-conflict error to propagate") + } + if statusUpdates != 1 { + t.Fatalf("a non-conflict error must not be retried; got %d attempts", statusUpdates) + } +} + +// The happy path: no conflict -> a single write, computed status returned. +func TestUpdateStatusWithRetry_SuccessFirstTry(t *testing.T) { + dyn := srFakeDyn(srPortal("1", "x")) + + var statusUpdates int + dyn.PrependReactor("update", "portals", func(action clienttesting.Action) (bool, runtime.Object, error) { + if ua, ok := action.(clienttesting.UpdateAction); ok && ua.GetSubresource() == "status" { + statusUpdates++ + } + return false, nil, nil + }) + + got, err := updateStatusWithRetry(context.Background(), srPortal("1", "healthy"), srOpts(dyn)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if statusUpdates != 1 { + t.Fatalf("expected exactly 1 status write, got %d", statusUpdates) + } + if m, _, _ := unstructured.NestedString(got.Object, "status", "marker"); m != "healthy" { + t.Fatalf("expected healthy marker, got %q", m) + } +}