apiserver: clarify watcher fanout deepcopy and self-link code

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

@@ -228,11 +228,13 @@ func PatchResource(r rest.Patcher, scope *RequestScope, admit admission.Interfac
 		}
 		trace.Step("Object stored in database")
 
-		if err := setObjectSelfLink(ctx, result, req, scope.Namer); err != nil {
-			scope.err(err, w, req)
-			return
+		if !utilfeature.DefaultFeatureGate.Enabled(features.RemoveSelfLink) {
+			if err := setObjectSelfLink(ctx, result, req, scope.Namer); err != nil {
+				scope.err(err, w, req)
+				return
+			}
+			trace.Step("Self-link added")
 		}
-		trace.Step("Self-link added")
 
 		status := http.StatusOK
 		if wasCreated {

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

@@ -31,6 +31,8 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apiserver/pkg/endpoints/handlers/negotiation"
 	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
+	"k8s.io/apiserver/pkg/features"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	utiltrace "k8s.io/utils/trace"
 )
 
@@ -59,8 +61,18 @@ func doTransformObject(ctx context.Context, obj runtime.Object, opts interface{}
 	if _, ok := obj.(*metav1.Status); ok {
 		return obj, nil
 	}
-	if err := setObjectSelfLink(ctx, obj, req, scope.Namer); err != nil {
-		return nil, err
+
+	if !utilfeature.DefaultFeatureGate.Enabled(features.RemoveSelfLink) {
+		if err := setObjectSelfLink(ctx, obj, req, scope.Namer); err != nil {
+			return nil, err
+		}
+	}
+
+	// Ensure that for empty lists we don't return <nil> items.
+	if meta.IsListType(obj) && meta.LenList(obj) == 0 {
+		if err := meta.SetList(obj, []runtime.Object{}); err != nil {
+			return nil, err
+		}
 	}
 
 	switch target := mediaType.Convert; {

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

@@ -45,9 +45,7 @@ import (
 	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
 	"k8s.io/apiserver/pkg/endpoints/metrics"
 	"k8s.io/apiserver/pkg/endpoints/request"
-	"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"
 )
@@ -369,16 +367,6 @@ func dedupOwnerReferencesAndAddWarning(obj runtime.Object, requestContext contex
 // TODO: remove the need for the namer LinkSetters by requiring objects implement either Object or List
 //   interfaces
 func setObjectSelfLink(ctx context.Context, obj runtime.Object, req *http.Request, namer ScopeNamer) error {
-	if utilfeature.DefaultFeatureGate.Enabled(features.RemoveSelfLink) {
-		// Ensure that for empty lists we don't return <nil> items.
-		if meta.IsListType(obj) && meta.LenList(obj) == 0 {
-			if err := meta.SetList(obj, []runtime.Object{}); err != nil {
-				return err
-			}
-		}
-		return nil
-	}
-
 	// We only generate list links on objects that implement ListInterface - historically we duck typed this
 	// check via reflection, but as we move away from reflection we require that you not only carry Items but
 	// ListMeta into order to be identified as a list.
@@ -402,18 +390,10 @@ func setObjectSelfLink(ctx context.Context, obj runtime.Object, req *http.Reques
 		return fmt.Errorf("missing requestInfo")
 	}
 
-	count := 0
 	err = meta.EachListItem(obj, func(obj runtime.Object) error {
-		count++
 		return setSelfLink(obj, requestInfo, namer)
 	})
 
-	if count == 0 {
-		if err := meta.SetList(obj, []runtime.Object{}); err != nil {
-			return err
-		}
-	}
-
 	return err
 }
 

staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go

@@ -1308,7 +1308,16 @@ func (c *cacheWatcher) nextBookmarkTime(now time.Time, bookmarkFrequency time.Du
 	return heartbeatTime, true
 }
 
-func getEventObject(object runtime.Object) runtime.Object {
+// getSafeFanoutObject returns an object that can be sent into fan-out
+// and its metadata can be safely mutated in every watcher in parallel.
+// We assume that  every watcher will mutate the object in the same way.
+//
+// Two cases:
+// 1. the CacheableObject object has thread-safe metadata changes. Equal mutations
+//    don't harm performance. Different mutations would harm performance, but we
+//    know they don't happen.
+// 2. a non-CacheableObject is deep-copied and hence every mutation is safe.
+func getSafeFanoutObject(object runtime.Object) runtime.Object {
 	if _, ok := object.(runtime.CacheableObject); ok {
 		// It is safe to return without deep-copy, because the underlying
 		// object was already deep-copied during construction.
@@ -1344,12 +1353,12 @@ func (c *cacheWatcher) convertToWatchEvent(event *watchCacheEvent) *watch.Event
 
 	switch {
 	case curObjPasses && !oldObjPasses:
-		return &watch.Event{Type: watch.Added, Object: getEventObject(event.Object)}
+		return &watch.Event{Type: watch.Added, Object: getSafeFanoutObject(event.Object)}
 	case curObjPasses && oldObjPasses:
-		return &watch.Event{Type: watch.Modified, Object: getEventObject(event.Object)}
+		return &watch.Event{Type: watch.Modified, Object: getSafeFanoutObject(event.Object)}
 	case !curObjPasses && oldObjPasses:
 		// return a delete event with the previous object content, but with the event's resource version
-		oldObj := getEventObject(event.PrevObject)
+		oldObj := getSafeFanoutObject(event.PrevObject)
 		updateResourceVersionIfNeeded(oldObj, c.versioner, event.ResourceVersion)
 		return &watch.Event{Type: watch.Deleted, Object: oldObj}
 	}