Repository: kubernetes/kubernetes (language: go)

Issue:
DeltaFIFO TransformFunc can result in a data race on re-sync
We use a `TransformFunc` to nil out the `ManagedFields`:

```
func(obj interface{}) (interface{}, error) {
    resourceUtil.MustToMeta(obj).SetManagedFields(nil)
    return obj, nil
})
```

However on a re-sync, this results in a data race that was observed when running unit tests:

```
WARNING: DATA RACE
Write at 0x00c0006accc0 by goroutine 81:
  runtime.mapdelete_faststr()
      /usr/lib/golang/src/runtime/map_faststr.go:301 +0x0
  k8s.io/apimachinery/pkg/apis/meta/v1/unstructured.RemoveNestedField()
      /go/pkg/mod/k8s.io/apimachinery@v0.29.3/pkg/apis/meta/v1/unstructured/helpers.go:274 +0x1e6
  k8s.io/apimachinery/pkg/apis/meta/v1/unstructured.(*Unstructured).SetManagedFields()
      /go/pkg/mod/k8s.io/apimachinery@v0.29.3/pkg/apis/meta/v1/unstructured/unstructured.go:476 +0x87
  github.com/submariner-io/admiral/pkg/syncer.NewResourceSyncer.func3()
      /go/pkg/mod/github.com/submariner-io/admiral@v0.18.0-m2.0.20240415195632-6c525b3cb5e9/pkg/syncer/resource_syncer.go:260 +0x4c
  k8s.io/client-go/tools/cache.(*DeltaFIFO).queueActionLocked()
      /go/pkg/mod/k8s.io/client-go@v0.29.3/tools/cache/delta_fifo.go:456 +0x101
  k8s.io/client-go/tools/cache.(*DeltaFIFO).syncKeyLocked()
      /go/pkg/mod/k8s.io/client-go@v0.29.3/tools/cache/delta_fifo.go:746 +0x154
  k8s.io/client-go/tools/cache.(*DeltaFIFO).Resync()
      /go/pkg/mod/k8s.io/client-go@v0.29.3/tools/cache/delta_fifo.go:717 +0x1b6
  k8s.io/client-go/tools/cache.(*Reflector).startResync()
      /go/pkg/mod/k8s.io/client-go@v0.29.3/tools/cache/reflector.go:378 +0x351
  k8s.io/client-go/tools/cache.(*Reflector).ListAndWatch.func2()
      /go/pkg/mod/k8s.io/client-go@v0.29.3/tools/cache/reflector.go:356 +0x5d

Previous read at 0x00c0006accc0 by goroutine 82:
  runtime.mapaccess2_faststr()
      /usr/lib/golang/src/runtime/map_faststr.go:108 +0x0
  k8s.io/apimachinery/pkg/apis/meta/v1/unstructured.NestedFieldNoCopy()
      /go/pkg/mod/k8s.io/apimachinery@v0.29.3/pkg/apis/meta/v1/unstructured/helpers.go:61 +0x10f
  k8s.io/apimachinery/pkg/apis/meta/v1/unstructured.nestedMapNoCopy()
      /go/pkg/mod/k8s.io/apimachinery@v0.29.3/pkg/apis/meta/v1/unstructured/helpers.go:195 +0x66
  k8s.io/apimachinery/pkg/apis/meta/v1/unstructured.NestedStringMap()
      /go/pkg/mod/k8s.io/apimachinery@v0.29.3/pkg/apis/meta/v1/unstructured/helpers.go:167 +0x6e
  k8s.io/apimachinery/pkg/apis/meta/v1/unstructured.(*Unstructured).GetLabels()
      /go/pkg/mod/k8s.io/apimachinery@v0.29.3/pkg/apis/meta/v1/unstructured/unstructured.go:400 +0x94
  github.com/submariner-io/admiral/pkg/syncer.getClusterIDLabel()
      /go/pkg/mod/github.com/submariner-io/admiral@v0.18.0-m2.0.20240415195632-6c525b3cb5e9/pkg/syncer/resource_syncer.go:681 +0x65
  github.com/submariner-io/admiral/pkg/syncer.(*resourceSyncer).shouldSync()
      /go/pkg/mod/github.com/submariner-io/admiral@v0.18.0-m2.0.20240415195632-6c525b3cb5e9/pkg/syncer/resource_syncer.go:647 +0x4d
  github.com/submariner-io/admiral/pkg/syncer.(*resourceSyncer).processNextWorkItem()
      /go/pkg/mod/github.com/submariner-io/admiral@v0.18.0-m2.0.20240415195632-6c525b3cb5e9/pkg/syncer/resource_syncer.go:448 +0x3ad
  github.com/submariner-io/admiral/pkg/syncer.(*resourceSyncer).processNextWorkItem-fm()
      <autogenerated>:1 +0x89
  github.com/submariner-io/admiral/pkg/workqueue.(*queueType).processNextWorkItem.func1()
      /go/pkg/mod/github.com/submariner-io/admiral@v0.18.0-m2.0.20240415195632-6c525b3cb5e9/pkg/workqueue/queue.go:103 +0x59
```

The read had occurred previously during a `ResourceEventHandler` callback.

On re-sync, the `DeltaFIFO` retrieves every object from the `knownObjects` cache,  invokes the transform function, and re-queues them.  However, the object instances are the same as those stored in the cache so mutating them outside the cache without lock protection is unsafe. It is documented that the "_TransformFunc sees the object before any other actor, and it is now safe to mutate the object in place instead of making a copy_" however this is not the case on a re-sync. The `DeltaFIFO` should make a copy of the objects retrieved from the cache store on a re-sync or perhaps elide the `TransformFunc` on a re-sync since the objects would already have been transformed.

As a workaround, I changed our `TransformFunc` to only nil the `ManagedFields` if not already nil but this is a latent bug that others will likely run into.


Discussion / hints:
/assign @linxiulei 

That means that https://github.com/kubernetes/kubernetes/pull/118455 and https://github.com/kubernetes/kubernetes/pull/119556 introduced possible panic regressions into kube-controller-manager and kube-scheduler

/priority critical-urgent
/kind regression
/transfer kubernetes
/sig api-machinery
> We use a TransformFunc to nil out the ManagedFields

kube-controller-manager and kube-scheduler do the same in 1.29+ and likely have the same issue
I think delete has the same problem, and replace may have the same problem:

```go
	// Every object comes through this code path once, so this is a good
	// place to call the transform func.  If obj is a
	// DeletedFinalStateUnknown tombstone, then the containted inner object
	// will already have gone through the transformer, but we document that
	// this can happen. In cases involving Replace(), such an object can
	// come through multiple times.
```

Passing the same object to the transformer after it is already in the cache is unsafe and we should not do that at all.
/assign @linxiulei 
/cc @wojtek-t 
Looking into this. I may take time to come up with a proper fix. But would like to evaluate the risk for the current situation. Not entirely sure how delete and replace would be triggered in normal settings, but resync is fortunately low frequency as the default setting for resync period is between 12h and 24h for `kube-controller-manager` (--min-resync-period duration     Default: 12h0m0s, see [help](https://kubernetes.io/docs/reference/command-line-tools-reference/kube-controller-manager/)) . As for `kube-scheduler` the resync function is [disabled](https://github.com/kubernetes/kubernetes/blob/master/cmd/kube-scheduler/app/options/options.go#L304)

Just think out aloud, I read a thread discussion somewhere that resync was a mechanism in the past to prevent data inconsistency due to bugs but client-go has been mature these days, resync is no-longer needed in theory but for a safe guard. So maybe disabling resync is a reasonable fix if delete/replace can be fixed.



Fix patch:
diff --git a/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go b/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go
index 7160bb1ee72e6..ce74dfb6f104b 100644
--- a/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go
+++ b/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go
@@ -139,20 +139,17 @@ type DeltaFIFO struct {
 }
 
 // TransformFunc allows for transforming an object before it will be processed.
-// TransformFunc (similarly to ResourceEventHandler functions) should be able
-// to correctly handle the tombstone of type cache.DeletedFinalStateUnknown.
-//
-// New in v1.27: In such cases, the contained object will already have gone
-// through the transform object separately (when it was added / updated prior
-// to the delete), so the TransformFunc can likely safely ignore such objects
-// (i.e., just return the input object).
 //
 // The most common usage pattern is to clean-up some parts of the object to
 // reduce component memory usage if a given component doesn't care about them.
 //
-// New in v1.27: unless the object is a DeletedFinalStateUnknown, TransformFunc
-// sees the object before any other actor, and it is now safe to mutate the
-// object in place instead of making a copy.
+// New in v1.27: TransformFunc sees the object before any other actor, and it
+// is now safe to mutate the object in place instead of making a copy.
+//
+// It's recommended for the TransformFunc to be idempotent.
+// It MUST be idempotent if objects already present in the cache are passed to
+// the Replace() to avoid re-mutating them. Default informers do not pass
+// existing objects to Replace though.
 //
 // Note that TransformFunc is called while inserting objects into the
 // notification queue and is therefore extremely performance sensitive; please
@@ -440,22 +437,38 @@ func isDeletionDup(a, b *Delta) *Delta {
 // queueActionLocked appends to the delta list for the object.
 // Caller must lock first.
 func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
+	return f.queueActionInternalLocked(actionType, actionType, obj)
+}
+
+// queueActionInternalLocked appends to the delta list for the object.
+// The actionType is emitted and must honor emitDeltaTypeReplaced.
+// The internalActionType is only used within this function and must
+// ignore emitDeltaTypeReplaced.
+// Caller must lock first.
+func (f *DeltaFIFO) queueActionInternalLocked(actionType, internalActionType DeltaType, obj interface{}) error {
 	id, err := f.KeyOf(obj)
 	if err != nil {
 		return KeyError{obj, err}
 	}
 
 	// Every object comes through this code path once, so this is a good
-	// place to call the transform func.  If obj is a
-	// DeletedFinalStateUnknown tombstone, then the containted inner object
-	// will already have gone through the transformer, but we document that
-	// this can happen. In cases involving Replace(), such an object can
-	// come through multiple times.
+	// place to call the transform func.
+	//
+	// If obj is a DeletedFinalStateUnknown tombstone or the action is a Sync,
+	// then the object have already gone through the transformer.
+	//
+	// If the objects already present in the cache are passed to Replace(),
+	// the transformer must be idempotent to avoid re-mutating them,
+	// or coordinate with all readers from the cache to avoid data races.
+	// Default informers do not pass existing objects to Replace.
 	if f.transformer != nil {
-		var err error
-		obj, err = f.transformer(obj)
-		if err != nil {
-			return err
+		_, isTombstone := obj.(DeletedFinalStateUnknown)
+		if !isTombstone && internalActionType != Sync {
+			var err error
+			obj, err = f.transformer(obj)
+			if err != nil {
+				return err
+			}
 		}
 	}
 
@@ -638,7 +651,7 @@ func (f *DeltaFIFO) Replace(list []interface{}, _ string) error {
 			return KeyError{item, err}
 		}
 		keys.Insert(key)
-		if err := f.queueActionLocked(action, item); err != nil {
+		if err := f.queueActionInternalLocked(action, Replaced, item); err != nil {
 			return fmt.Errorf("couldn't enqueue object: %v", err)
 		}
 	}


Test patch:
diff --git a/staging/src/k8s.io/client-go/tools/cache/controller_test.go b/staging/src/k8s.io/client-go/tools/cache/controller_test.go
index 4058581154fad..992c1f5a4a480 100644
--- a/staging/src/k8s.io/client-go/tools/cache/controller_test.go
+++ b/staging/src/k8s.io/client-go/tools/cache/controller_test.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"math/rand"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -575,6 +576,114 @@ func TestTransformingInformer(t *testing.T) {
 	close(stopCh)
 }
 
+func TestTransformingInformerRace(t *testing.T) {
+	// source simulates an apiserver object endpoint.
+	source := fcache.NewFakeControllerSource()
+
+	label := "to-be-transformed"
+	makePod := func(name string) *v1.Pod {
+		return &v1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      name,
+				Namespace: "namespace",
+				Labels:    map[string]string{label: "true"},
+			},
+			Spec: v1.PodSpec{
+				Hostname: "hostname",
+			},
+		}
+	}
+
+	badTransform := atomic.Bool{}
+	podTransformer := func(obj interface{}) (interface{}, error) {
+		pod, ok := obj.(*v1.Pod)
+		if !ok {
+			return nil, fmt.Errorf("unexpected object type: %T", obj)
+		}
+		if pod.ObjectMeta.Labels[label] != "true" {
+			badTransform.Store(true)
+			return nil, fmt.Errorf("object already transformed: %#v", obj)
+		}
+		pod.ObjectMeta.Labels[label] = "false"
+		return pod, nil
+	}
+
+	numObjs := 5
+	for i := 0; i < numObjs; i++ {
+		source.Add(makePod(fmt.Sprintf("pod-%d", i)))
+	}
+
+	type event struct{}
+	events := make(chan event, numObjs)
+	recordEvent := func(eventType watch.EventType, previous, current interface{}) {
+		events <- event{}
+	}
+	checkEvents := func(count int) {
+		for i := 0; i < count; i++ {
+			<-events
+		}
+	}
+	store, controller := NewTransformingInformer(
+		source,
+		&v1.Pod{},
+		5*time.Millisecond,
+		ResourceEventHandlerDetailedFuncs{
+			AddFunc:    func(obj interface{}, isInInitialList bool) { recordEvent(watch.Added, nil, obj) },
+			UpdateFunc: func(oldObj, newObj interface{}) { recordEvent(watch.Modified, oldObj, newObj) },
+			DeleteFunc: func(obj interface{}) { recordEvent(watch.Deleted, obj, nil) },
+		},
+		podTransformer,
+	)
+
+	stopCh := make(chan struct{})
+	go controller.Run(stopCh)
+
+	checkEvents(numObjs)
+
+	// Periodically fetch objects to ensure no access races.
+	wg := sync.WaitGroup{}
+	errors := make(chan error, numObjs)
+	for i := 0; i < numObjs; i++ {
+		wg.Add(1)
+		go func(index int) {
+			defer wg.Done()
+			key := fmt.Sprintf("namespace/pod-%d", index)
+			for {
+				select {
+				case <-stopCh:
+					return
+				default:
+				}
+
+				obj, ok, err := store.GetByKey(key)
+				if !ok || err != nil {
+					errors <- fmt.Errorf("couldn't get the object for %v", key)
+					return
+				}
+				pod := obj.(*v1.Pod)
+				if pod.ObjectMeta.Labels[label] != "false" {
+					errors <- fmt.Errorf("unexpected object: %#v", pod)
+					return
+				}
+			}
+		}(i)
+	}
+
+	// Let resyncs to happen for some time.
+	time.Sleep(time.Second)
+
+	close(stopCh)
+	wg.Wait()
+	close(errors)
+	for err := range errors {
+		t.Error(err)
+	}
+
+	if badTransform.Load() {
+		t.Errorf("unexpected transformation happened")
+	}
+}
+
 func TestDeletionHandlingObjectToName(t *testing.T) {
 	cm := &v1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{