apiserver dedups owner references and adds warning

for CREATE and UPDATE requests, we check duplication before managedFields
update, and after mutating admission; for PATCH requests, we check
duplication after mutating admission

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/create.go

@@ -152,6 +152,8 @@ func createHandler(r rest.NamedCreater, scope *RequestScope, admit admission.Int
 				options,
 			)
 		}
+		// Dedup owner references before updating managed fields
+		dedupOwnerReferencesAndAddWarning(obj, req.Context(), false)
 		result, err := finishRequest(ctx, func() (runtime.Object, error) {
 			if scope.FieldManager != nil {
 				liveObj, err := scope.Creater.New(scope.Kind)
@@ -165,6 +167,8 @@ func createHandler(r rest.NamedCreater, scope *RequestScope, admit admission.Int
 					return nil, err
 				}
 			}
+			// Dedup owner references again after mutating admission happens
+			dedupOwnerReferencesAndAddWarning(obj, req.Context(), true)
 			result, err := requestFunc()
 			// If the object wasn't committed to storage because it's serialized size was too large,
 			// it is safe to remove managedFields (which can be large) and try again.

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/patch.go

@@ -569,9 +569,14 @@ func (p *patcher) patchResource(ctx context.Context, scope *RequestScope) (runti
 	default:
 		return nil, false, fmt.Errorf("%v: unimplemented patch type", p.patchType)
 	}
+	dedupOwnerReferencesTransformer := func(_ context.Context, obj, _ runtime.Object) (runtime.Object, error) {
+		// Dedup owner references after mutating admission happens
+		dedupOwnerReferencesAndAddWarning(obj, ctx, true)
+		return obj, nil
+	}
 
 	wasCreated := false
-	p.updatedObjectInfo = rest.DefaultUpdatedObjectInfo(nil, p.applyPatch, p.applyAdmission)
+	p.updatedObjectInfo = rest.DefaultUpdatedObjectInfo(nil, p.applyPatch, p.applyAdmission, dedupOwnerReferencesTransformer)
 	requestFunc := func() (runtime.Object, error) {
 		// Pass in UpdateOptions to override UpdateStrategy.AllowUpdateOnCreate
 		options := patchToUpdateOptions(p.options)
@@ -585,11 +590,15 @@ func (p *patcher) patchResource(ctx context.Context, scope *RequestScope) (runti
 		// it is safe to remove managedFields (which can be large) and try again.
 		if isTooLargeError(err) && p.patchType != types.ApplyPatchType {
 			if _, accessorErr := meta.Accessor(p.restPatcher.New()); accessorErr == nil {
-				p.updatedObjectInfo = rest.DefaultUpdatedObjectInfo(nil, p.applyPatch, p.applyAdmission, func(_ context.Context, obj, _ runtime.Object) (runtime.Object, error) {
+				p.updatedObjectInfo = rest.DefaultUpdatedObjectInfo(nil,
-					accessor, _ := meta.Accessor(obj)
+					p.applyPatch,
-					accessor.SetManagedFields(nil)
+					p.applyAdmission,
-					return obj, nil
+					dedupOwnerReferencesTransformer,
-				})
+					func(_ context.Context, obj, _ runtime.Object) (runtime.Object, error) {
+						accessor, _ := meta.Accessor(obj)
+						accessor.SetManagedFields(nil)
+						return obj, nil
+					})
 				result, err = requestFunc()
 			}
 		}

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/rest.go

@@ -31,12 +31,14 @@ import (
 	grpccodes "google.golang.org/grpc/codes"
 	grpcstatus "google.golang.org/grpc/status"
 
+	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	metav1beta1 "k8s.io/apimachinery/pkg/apis/meta/v1beta1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apiserver/pkg/admission"
 	"k8s.io/apiserver/pkg/authorization/authorizer"
 	"k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager"
@@ -46,6 +48,7 @@ import (
 	"k8s.io/apiserver/pkg/features"
 	"k8s.io/apiserver/pkg/registry/rest"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	"k8s.io/apiserver/pkg/warning"
 	"k8s.io/klog/v2"
 )
 
@@ -53,6 +56,14 @@ const (
 	// 34 chose as a number close to 30 that is likely to be unique enough to jump out at me the next time I see a timeout.
 	// Everyone chooses 30.
 	requestTimeout = 34 * time.Second
+	// DuplicateOwnerReferencesWarningFormat is the warning that a client receives when a create/update request contains
+	// duplicate owner reference entries.
+	DuplicateOwnerReferencesWarningFormat = ".metadata.ownerReferences contains duplicate entries; API server dedups owner references in 1.20+, and may reject such requests as early as 1.24; please fix your requests; duplicate UID(s) observed: %v"
+	// DuplicateOwnerReferencesAfterMutatingAdmissionWarningFormat indicates the duplication was observed
+	// after mutating admission.
+	// NOTE: For CREATE and UPDATE requests the API server dedups both before and after mutating admission.
+	// For PATCH request the API server only dedups after mutating admission.
+	DuplicateOwnerReferencesAfterMutatingAdmissionWarningFormat = ".metadata.ownerReferences contains duplicate entries after mutating admission happens; API server dedups owner references in 1.20+, and may reject such requests as early as 1.24; please fix your requests; duplicate UID(s) observed: %v"
 )
 
 // RequestScope encapsulates common fields across all RESTful handler methods.
@@ -329,6 +340,68 @@ func checkName(obj runtime.Object, name, namespace string, namer ScopeNamer) err
 	return nil
 }
 
+// dedupOwnerReferences dedups owner references over the entire entry.
+// NOTE: We don't know enough about the existing cases of owner references
+// sharing the same UID but different fields. Nor do we know what might break.
+// In the future we may just dedup/reject owner references with the same UID.
+func dedupOwnerReferences(refs []metav1.OwnerReference) ([]metav1.OwnerReference, []string) {
+	var result []metav1.OwnerReference
+	var duplicates []string
+	seen := make(map[types.UID]struct{})
+	for _, ref := range refs {
+		_, ok := seen[ref.UID]
+		// Short-circuit if we haven't seen the UID before. Otherwise
+		// check the entire list we have so far.
+		if !ok || !hasOwnerReference(result, ref) {
+			seen[ref.UID] = struct{}{}
+			result = append(result, ref)
+		} else {
+			duplicates = append(duplicates, string(ref.UID))
+		}
+	}
+	return result, duplicates
+}
+
+// hasOwnerReference returns true if refs has an item equal to ref. The function
+// focuses on semantic equality instead of memory equality, to catch duplicates
+// with different pointer addresses. The function uses apiequality.Semantic
+// instead of implementing its own comparison, to tolerate API changes to
+// metav1.OwnerReference.
+// NOTE: This is expensive, but we accept it because we've made sure it only
+// happens to owner references containing duplicate UIDs, plus typically the
+// number of items in the list should be small.
+func hasOwnerReference(refs []metav1.OwnerReference, ref metav1.OwnerReference) bool {
+	for _, r := range refs {
+		if apiequality.Semantic.DeepEqual(r, ref) {
+			return true
+		}
+	}
+	return false
+}
+
+// dedupOwnerReferencesAndAddWarning dedups owner references in the object metadata.
+// If duplicates are found, the function records a warning to the provided context.
+func dedupOwnerReferencesAndAddWarning(obj runtime.Object, requestContext context.Context, afterMutatingAdmission bool) {
+	accessor, err := meta.Accessor(obj)
+	if err != nil {
+		// The object doesn't have metadata. Nothing we need to do here.
+		return
+	}
+	refs := accessor.GetOwnerReferences()
+	deduped, duplicates := dedupOwnerReferences(refs)
+	if len(duplicates) > 0 {
+		// NOTE: For CREATE and UPDATE requests the API server dedups both before and after mutating admission.
+		// For PATCH request the API server only dedups after mutating admission.
+		format := DuplicateOwnerReferencesWarningFormat
+		if afterMutatingAdmission {
+			format = DuplicateOwnerReferencesAfterMutatingAdmissionWarningFormat
+		}
+		warning.AddWarning(requestContext, "", fmt.Sprintf(format,
+			strings.Join(duplicates, ", ")))
+		accessor.SetOwnerReferences(deduped)
+	}
+}
+
 // setObjectSelfLink sets the self link of an object as needed.
 // TODO: remove the need for the namer LinkSetters by requiring objects implement either Object or List
 //   interfaces

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/update.go

@@ -150,6 +150,11 @@ func UpdateResource(r rest.Updater, scope *RequestScope, admit admission.Interfa
 				}
 				return newObj, nil
 			})
+			transformers = append(transformers, func(ctx context.Context, newObj, oldObj runtime.Object) (runtime.Object, error) {
+				// Dedup owner references again after mutating admission happens
+				dedupOwnerReferencesAndAddWarning(newObj, req.Context(), true)
+				return newObj, nil
+			})
 		}
 
 		createAuthorizerAttributes := authorizer.AttributesRecord{
@@ -185,6 +190,8 @@ func UpdateResource(r rest.Updater, scope *RequestScope, admit admission.Interfa
 			wasCreated = created
 			return obj, err
 		}
+		// Dedup owner references before updating managed fields
+		dedupOwnerReferencesAndAddWarning(obj, req.Context(), false)
 		result, err := finishRequest(ctx, func() (runtime.Object, error) {
 			result, err := requestFunc()
 			// If the object wasn't committed to storage because it's serialized size was too large,