From eca880b81d0948f26cfda664c94bcf0dc8d2fb08 Mon Sep 17 00:00:00 2001 From: Scot Wells Date: Thu, 29 Oct 2026 23:49:00 -0500 Subject: [PATCH 1/2] fix: retry Instance DNS and projection status conflicts --- internal/controller/instance_projector.go | 24 ++++++---- .../controller/instance_projector_test.go | 24 ++++++++++ internal/controller/internaldns_publisher.go | 2 +- internal/controller/internaldns_status.go | 28 ++++++++--- .../controller/internaldns_status_test.go | 48 ++++++++++++++++++- 5 files changed, 109 insertions(+), 17 deletions(-) diff --git a/internal/controller/instance_projector.go b/internal/controller/instance_projector.go index b4979607..6b45c17d 100644 --- a/internal/controller/instance_projector.go +++ b/internal/controller/instance_projector.go @@ -12,6 +12,7 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/util/retry" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/cluster" @@ -230,14 +231,21 @@ func (r *InstanceProjector) reconcile(ctx context.Context, req ctrl.Request) err logger.Info("reconciled Instance projection", "operation", operationResult, "namespace", targetNamespace, "cluster", clusterName) - // Status is a separate subresource. - dnsStatus := projection.Status.DNS - projection.Status = downstreamInstance.Status - // Private name publication belongs to the project-side DNS publisher. - if r.InternalDNSEnabled { - projection.Status.DNS = dnsStatus - } - if err := projectClient.Status().Update(ctx, projection); err != nil && !apierrors.IsNotFound(err) { + if err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + latest := &computev1alpha.Instance{} + if err := projectCluster.GetAPIReader().Get(ctx, client.ObjectKeyFromObject(projection), latest); err != nil { + return err + } + if latest.UID != projection.UID || latest.Generation != projection.Generation { + return fmt.Errorf("instance projection changed during status publication") + } + dnsStatus := latest.Status.DNS + latest.Status = downstreamInstance.Status + if r.InternalDNSEnabled { + latest.Status.DNS = dnsStatus + } + return projectClient.Status().Update(ctx, latest) + }); err != nil && !apierrors.IsNotFound(err) { return fmt.Errorf("failed updating Instance projection status: %w", err) } diff --git a/internal/controller/instance_projector_test.go b/internal/controller/instance_projector_test.go index c1e2f223..da93422d 100644 --- a/internal/controller/instance_projector_test.go +++ b/internal/controller/instance_projector_test.go @@ -926,3 +926,27 @@ func TestReconcileDNSProjectionLifetimeLeavesProjectionUntouched(t *testing.T) { require.Empty(t, projection.ResourceVersion) require.Empty(t, projection.Labels) } + +func TestInstanceProjectorRetriesConcurrentDNSPublication(t *testing.T) { + upstream := projTestKarmadaInstance(map[string]string{computev1alpha.InstanceSourceUIDLabel: internalDNSTestEdgeUID}) + projection := &computev1alpha.Instance{ObjectMeta: metav1.ObjectMeta{Name: projTestInstanceName, Namespace: projTestProjNS, Labels: map[string]string{computev1alpha.InstanceSourceUIDLabel: internalDNSTestEdgeUID}}} + base := fake.NewClientBuilder().WithScheme(newProjectScheme()).WithObjects(projTestWorkloadDeployment(), projection).WithStatusSubresource(projection).Build() + writes := 0 + cl := interceptor.NewClient(base, interceptor.Funcs{SubResourceUpdate: func(ctx context.Context, cl client.Client, sub string, obj client.Object, opts ...client.SubResourceUpdateOption) error { + writes++ + if writes == 1 { + current := &computev1alpha.Instance{} + require.NoError(t, base.Get(ctx, client.ObjectKeyFromObject(projection), current)) + current.Status.DNS = []computev1alpha.InstanceDNSStatus{{Network: networkingv1alpha.NetworkRef{Name: "concurrent-publication"}}} + require.NoError(t, base.Status().Update(ctx, current)) + } + return cl.SubResource(sub).Update(ctx, obj, opts...) + }}) + r := newTestProjector(newKarmadaFakeClient(upstream), cl) + r.InternalDNSEnabled = true + _, err := r.Reconcile(context.Background(), projectorRequest()) + require.NoError(t, err) + require.Equal(t, 2, writes) + require.NoError(t, base.Get(context.Background(), client.ObjectKeyFromObject(projection), projection)) + require.Equal(t, "concurrent-publication", projection.Status.DNS[0].Network.Name) +} diff --git a/internal/controller/internaldns_publisher.go b/internal/controller/internaldns_publisher.go index 27356540..28decd4a 100644 --- a/internal/controller/internaldns_publisher.go +++ b/internal/controller/internaldns_publisher.go @@ -190,7 +190,7 @@ func (r *InternalDNSPublisherReconciler) Reconcile(ctx context.Context, req mcre publicationErr = errors.Join(publicationErr, err) } } - if err := r.reportPublication(ctx, access.Writer, &projectInstance, publicationStatus); err != nil { + if err := r.reportPublication(ctx, access.Writer, access.Reader, &projectInstance, publicationStatus); err != nil { return ctrl.Result{}, err } if err := r.deletePublishedResources(ctx, access, projectNamespace, projectInstance.UID, desired); err != nil { diff --git a/internal/controller/internaldns_status.go b/internal/controller/internaldns_status.go index da050a01..d3560874 100644 --- a/internal/controller/internaldns_status.go +++ b/internal/controller/internaldns_status.go @@ -14,6 +14,8 @@ import ( apimeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/util/retry" "sigs.k8s.io/controller-runtime/pkg/client" ) @@ -106,14 +108,26 @@ func publicationMatchesContext(ctx context.Context, reader client.Reader, identi return canonical == instanceDNSScope+"."+suffix && referenceName(registeredZone) == referenceName(zone), nil } -func (r *InternalDNSPublisherReconciler) reportPublication(ctx context.Context, writer client.Client, instance *computev1alpha.Instance, status []computev1alpha.InstanceDNSStatus) error { +func (r *InternalDNSPublisherReconciler) reportPublication(ctx context.Context, writer client.Client, reader client.Reader, instance *computev1alpha.Instance, status []computev1alpha.InstanceDNSStatus) error { if reflect.DeepEqual(instance.Status.DNS, status) { return nil } - base := instance.DeepCopy() - instance.Status.DNS = status - if err := writer.Status().Patch(ctx, instance, client.MergeFromWithOptions(base, client.MergeFromWithOptimisticLock{})); err != nil { - return fmt.Errorf("report Instance DNS publication: %w", err) - } - return nil + return retry.RetryOnConflict(retry.DefaultRetry, func() error { + latest := &computev1alpha.Instance{} + if err := reader.Get(ctx, client.ObjectKeyFromObject(instance), latest); err != nil { + return client.IgnoreNotFound(err) + } + if latest.UID != instance.UID || latest.Generation != instance.Generation { + return apierrors.NewConflict(schema.GroupResource{Group: "compute.datumapis.com", Resource: "instances"}, instance.Name, fmt.Errorf("instance changed during DNS publication")) + } + if reflect.DeepEqual(latest.Status.DNS, status) { + return nil + } + base := latest.DeepCopy() + latest.Status.DNS = status + if err := writer.Status().Patch(ctx, latest, client.MergeFromWithOptions(base, client.MergeFromWithOptimisticLock{})); err != nil { + return fmt.Errorf("report Instance DNS publication: %w", err) + } + return nil + }) } diff --git a/internal/controller/internaldns_status_test.go b/internal/controller/internaldns_status_test.go index 92ea0e70..40151675 100644 --- a/internal/controller/internaldns_status_test.go +++ b/internal/controller/internaldns_status_test.go @@ -8,6 +8,8 @@ import ( "testing" "time" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + "github.com/stretchr/testify/require" computev1alpha "go.datum.net/compute/api/v1alpha" networkingv1alpha "go.datum.net/network-services-operator/api/v1alpha" @@ -120,7 +122,7 @@ func TestReportPublicationPreservesRuntimeStatus(t *testing.T) { require.NoError(t, cl.Get(context.Background(), client.ObjectKeyFromObject(instance), instance)) r := &InternalDNSPublisherReconciler{} network := networkingv1alpha.NetworkRef{Name: internalDNSTestNetworkName, Namespace: internalDNSTestNamespace} - require.NoError(t, r.reportPublication(context.Background(), cl, instance, []computev1alpha.InstanceDNSStatus{{Network: network}})) + require.NoError(t, r.reportPublication(context.Background(), cl, cl, instance, []computev1alpha.InstanceDNSStatus{{Network: network}})) require.NoError(t, cl.Get(context.Background(), client.ObjectKeyFromObject(instance), instance)) require.Equal(t, metav1.ConditionTrue, instance.Status.Conditions[0].Status) require.Len(t, instance.Status.DNS, 1) @@ -156,3 +158,47 @@ func TestPublicationStatusMatchesNamespacedNetworkReference(t *testing.T) { }) } } + +func TestReportPublicationRetriesConcurrentRuntimeStatus(t *testing.T) { + s := runtime.NewScheme() + require.NoError(t, computev1alpha.AddToScheme(s)) + instance := &computev1alpha.Instance{ObjectMeta: metav1.ObjectMeta{Name: "publication-race", Namespace: "default", UID: "original", Generation: 1}} + base := fake.NewClientBuilder().WithScheme(s).WithObjects(instance).WithStatusSubresource(instance).Build() + require.NoError(t, base.Get(context.Background(), client.ObjectKeyFromObject(instance), instance)) + writes := 0 + writer := interceptor.NewClient(base, interceptor.Funcs{SubResourcePatch: func(ctx context.Context, cl client.Client, sub string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + writes++ + if writes == 1 { + current := &computev1alpha.Instance{} + require.NoError(t, base.Get(ctx, client.ObjectKeyFromObject(instance), current)) + current.Status.Conditions = []metav1.Condition{{Type: computev1alpha.InstanceReady, Status: metav1.ConditionTrue}} + require.NoError(t, base.Status().Update(ctx, current)) + } + return cl.SubResource(sub).Patch(ctx, obj, patch, opts...) + }}) + status := []computev1alpha.InstanceDNSStatus{{Network: networkingv1alpha.NetworkRef{Name: "private"}}} + require.NoError(t, (&InternalDNSPublisherReconciler{}).reportPublication(context.Background(), writer, base, instance, status)) + require.Equal(t, 2, writes) + require.NoError(t, base.Get(context.Background(), client.ObjectKeyFromObject(instance), instance)) + require.Equal(t, status, instance.Status.DNS) + require.Equal(t, metav1.ConditionTrue, instance.Status.Conditions[0].Status) +} + +func TestReportPublicationRejectsChangedInstance(t *testing.T) { + for _, changedUID := range []bool{false, true} { + s := runtime.NewScheme() + require.NoError(t, computev1alpha.AddToScheme(s)) + original := &computev1alpha.Instance{ObjectMeta: metav1.ObjectMeta{Name: "changed", Namespace: "default", UID: "original", Generation: 1}} + latest := original.DeepCopy() + if changedUID { + latest.UID = "replacement" + } else { + latest.Generation++ + } + cl := fake.NewClientBuilder().WithScheme(s).WithObjects(latest).WithStatusSubresource(latest).Build() + err := (&InternalDNSPublisherReconciler{}).reportPublication(context.Background(), cl, cl, original, []computev1alpha.InstanceDNSStatus{{Network: networkingv1alpha.NetworkRef{Name: "private"}}}) + require.Error(t, err) + require.NoError(t, cl.Get(context.Background(), client.ObjectKeyFromObject(latest), latest)) + require.Empty(t, latest.Status.DNS) + } +} From 161afb3509ef35f7f6fcd0807bffd4c71aa94333 Mon Sep 17 00:00:00 2001 From: Scot Wells Date: Thu, 29 Oct 2026 23:51:53 -0500 Subject: [PATCH 2/2] fix: reuse API group and test namespace in conflict tests --- internal/controller/internaldns_status.go | 2 +- internal/controller/internaldns_status_test.go | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/internal/controller/internaldns_status.go b/internal/controller/internaldns_status.go index d3560874..d9fe84d5 100644 --- a/internal/controller/internaldns_status.go +++ b/internal/controller/internaldns_status.go @@ -118,7 +118,7 @@ func (r *InternalDNSPublisherReconciler) reportPublication(ctx context.Context, return client.IgnoreNotFound(err) } if latest.UID != instance.UID || latest.Generation != instance.Generation { - return apierrors.NewConflict(schema.GroupResource{Group: "compute.datumapis.com", Resource: "instances"}, instance.Name, fmt.Errorf("instance changed during DNS publication")) + return apierrors.NewConflict(schema.GroupResource{Group: computev1alpha.GroupVersion.Group, Resource: "instances"}, instance.Name, fmt.Errorf("instance changed during DNS publication")) } if reflect.DeepEqual(latest.Status.DNS, status) { return nil diff --git a/internal/controller/internaldns_status_test.go b/internal/controller/internaldns_status_test.go index 40151675..da1c94f3 100644 --- a/internal/controller/internaldns_status_test.go +++ b/internal/controller/internaldns_status_test.go @@ -162,7 +162,7 @@ func TestPublicationStatusMatchesNamespacedNetworkReference(t *testing.T) { func TestReportPublicationRetriesConcurrentRuntimeStatus(t *testing.T) { s := runtime.NewScheme() require.NoError(t, computev1alpha.AddToScheme(s)) - instance := &computev1alpha.Instance{ObjectMeta: metav1.ObjectMeta{Name: "publication-race", Namespace: "default", UID: "original", Generation: 1}} + instance := &computev1alpha.Instance{ObjectMeta: metav1.ObjectMeta{Name: "publication-race", Namespace: internalDNSTestNamespace, UID: "original", Generation: 1}} base := fake.NewClientBuilder().WithScheme(s).WithObjects(instance).WithStatusSubresource(instance).Build() require.NoError(t, base.Get(context.Background(), client.ObjectKeyFromObject(instance), instance)) writes := 0 @@ -188,7 +188,7 @@ func TestReportPublicationRejectsChangedInstance(t *testing.T) { for _, changedUID := range []bool{false, true} { s := runtime.NewScheme() require.NoError(t, computev1alpha.AddToScheme(s)) - original := &computev1alpha.Instance{ObjectMeta: metav1.ObjectMeta{Name: "changed", Namespace: "default", UID: "original", Generation: 1}} + original := &computev1alpha.Instance{ObjectMeta: metav1.ObjectMeta{Name: "changed", Namespace: internalDNSTestNamespace, UID: "original", Generation: 1}} latest := original.DeepCopy() if changedUID { latest.UID = "replacement"