Track ownership of deployments scale subresource

pkg/registry/apps/deployment/storage/storage.go

@@ -25,6 +25,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager"
 	"k8s.io/apiserver/pkg/registry/generic"
 	genericregistry "k8s.io/apiserver/pkg/registry/generic/registry"
 	"k8s.io/apiserver/pkg/registry/rest"
@@ -54,6 +55,11 @@ type DeploymentStorage struct {
 	Rollback   *RollbackREST
 }
 
+// maps a group version to the replicas path in a deployment object
+var replicasPathInDeployment = fieldmanager.ResourcePathMappings{
+	schema.GroupVersion{Group: "apps", Version: "v1"}.String(): fieldpath.MakePathOrDie("spec", "replicas"),
+}
+
 // NewStorage returns new instance of DeploymentStorage.
 func NewStorage(optsGetter generic.RESTOptionsGetter) (DeploymentStorage, error) {
 	deploymentRest, deploymentStatusRest, deploymentRollbackRest, err := NewREST(optsGetter)
@@ -337,6 +343,7 @@ func scaleFromDeployment(deployment *apps.Deployment) (*autoscaling.Scale, error
 	if err != nil {
 		return nil, err
 	}
+
 	return &autoscaling.Scale{
 		// TODO: Create a variant of ObjectMeta type that only contains the fields below.
 		ObjectMeta: metav1.ObjectMeta{
@@ -376,11 +383,22 @@ func (i *scaleUpdatedObjectInfo) UpdatedObject(ctx context.Context, oldObj runti
 		return nil, errors.NewNotFound(apps.Resource("deployments/scale"), i.name)
 	}
 
+	managedFieldsHandler := fieldmanager.NewScaleHandler(
+		deployment.ManagedFields,
+		schema.GroupVersion{Group: "apps", Version: "v1"},
+		replicasPathInDeployment,
+	)
+
 	// deployment -> old scale
 	oldScale, err := scaleFromDeployment(deployment)
 	if err != nil {
 		return nil, err
 	}
+	scaleManagedFields, err := managedFieldsHandler.ToSubresource()
+	if err != nil {
+		return nil, err
+	}
+	oldScale.ManagedFields = scaleManagedFields
 
 	// old scale -> new scale
 	newScaleObj, err := i.reqObjInfo.UpdatedObject(ctx, oldScale)
@@ -412,5 +430,12 @@ func (i *scaleUpdatedObjectInfo) UpdatedObject(ctx context.Context, oldObj runti
 	// move replicas/resourceVersion fields to object and return
 	deployment.Spec.Replicas = scale.Spec.Replicas
 	deployment.ResourceVersion = scale.ResourceVersion
+
+	updatedEntries, err := managedFieldsHandler.ToParent(scale.ManagedFields)
+	if err != nil {
+		return nil, err
+	}
+	deployment.ManagedFields = updatedEntries
+
 	return deployment, nil
 }

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager/scalehander.go

@@ -0,0 +1,160 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package fieldmanager
+
+import (
+	"fmt"
+
+	"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/apiserver/pkg/endpoints/handlers/fieldmanager/internal"
+	"sigs.k8s.io/structured-merge-diff/v4/fieldpath"
+)
+
+var (
+	scaleGroupVersion   = schema.GroupVersion{Group: "autoscaling", Version: "v1"}
+	replicasPathInScale = fieldpath.MakePathOrDie("spec", "replicas")
+)
+
+// ResourcePathMappings maps a group/version to its replicas path. The
+// assumption is that all the paths correspond to leaf fields.
+type ResourcePathMappings map[string]fieldpath.Path
+
+// ScaleHandler manages the conversion of managed fields between a main
+// resource and the scale subresource
+type ScaleHandler struct {
+	parentEntries       []metav1.ManagedFieldsEntry
+	defaultGroupVersion schema.GroupVersion
+	mappings            ResourcePathMappings
+}
+
+// NewScaleHandler creates a new ScaleHandler
+func NewScaleHandler(parentEntries []metav1.ManagedFieldsEntry, defaultGroupVersion schema.GroupVersion, mappings ResourcePathMappings) *ScaleHandler {
+	return &ScaleHandler{
+		parentEntries:       parentEntries,
+		defaultGroupVersion: defaultGroupVersion,
+		mappings:            mappings,
+	}
+}
+
+// ToSubresource filter the managed fields of the main resource and convert
+// them so that they can be handled by scale.
+// For the managed fields that have a replicas path it performs two changes:
+// 1. APIVersion is changed to the APIVersion of the scale subresource
+// 2. Replicas path of the main resource is transformed to the replicas path of
+//    the scale subresource
+func (h *ScaleHandler) ToSubresource() ([]metav1.ManagedFieldsEntry, error) {
+	managed, err := DecodeManagedFields(h.parentEntries)
+	if err != nil {
+		return nil, err
+	}
+
+	f := fieldpath.ManagedFields{}
+	t := map[string]*metav1.Time{}
+	for manager, versionedSet := range managed.Fields() {
+		path := h.mappings[string(versionedSet.APIVersion())]
+		if versionedSet.Set().Has(path) {
+			newVersionedSet := fieldpath.NewVersionedSet(
+				fieldpath.NewSet(replicasPathInScale),
+				fieldpath.APIVersion(scaleGroupVersion.String()),
+				versionedSet.Applied(),
+			)
+
+			f[manager] = newVersionedSet
+			t[manager] = managed.Times()[manager]
+		}
+	}
+
+	return managedFieldsEntries(internal.NewManaged(f, t))
+}
+
+// ToParent merges `scaleEntries` with the entries of the main resource and
+// transforms them accordingly
+func (h *ScaleHandler) ToParent(scaleEntries []metav1.ManagedFieldsEntry) ([]metav1.ManagedFieldsEntry, error) {
+	decodedParentEntries, err := DecodeManagedFields(h.parentEntries)
+	if err != nil {
+		return nil, err
+	}
+	parentFields := decodedParentEntries.Fields()
+
+	decodedScaleEntries, err := DecodeManagedFields(scaleEntries)
+	if err != nil {
+		return nil, err
+	}
+	scaleFields := decodedScaleEntries.Fields()
+
+	f := fieldpath.ManagedFields{}
+	t := map[string]*metav1.Time{}
+
+	for manager, versionedSet := range parentFields {
+		// Get the main resource "replicas" path
+		path := h.mappings[string(versionedSet.APIVersion())]
+
+		// If the parent entry does not have the replicas path, just keep it as it is
+		if !versionedSet.Set().Has(path) {
+			f[manager] = versionedSet
+			t[manager] = decodedParentEntries.Times()[manager]
+			continue
+		}
+
+		if _, ok := scaleFields[manager]; !ok {
+			// "Steal" the replicas path from the main resource entry
+			newSet := versionedSet.Set().Difference(fieldpath.NewSet(path))
+
+			if !newSet.Empty() {
+				newVersionedSet := fieldpath.NewVersionedSet(
+					newSet,
+					versionedSet.APIVersion(),
+					versionedSet.Applied(),
+				)
+				f[manager] = newVersionedSet
+				t[manager] = decodedParentEntries.Times()[manager]
+			}
+		} else {
+			// Field wasn't stolen, let's keep the entry as it is.
+			f[manager] = versionedSet
+			t[manager] = decodedParentEntries.Times()[manager]
+			delete(scaleFields, manager)
+		}
+	}
+
+	for manager, versionedSet := range scaleFields {
+		newVersionedSet := fieldpath.NewVersionedSet(
+			fieldpath.NewSet(h.mappings[h.defaultGroupVersion.String()]),
+			fieldpath.APIVersion(h.defaultGroupVersion.String()),
+			versionedSet.Applied(),
+		)
+		f[manager] = newVersionedSet
+		t[manager] = decodedParentEntries.Times()[manager]
+	}
+
+	return managedFieldsEntries(internal.NewManaged(f, t))
+}
+
+func managedFieldsEntries(entries internal.ManagedInterface) ([]metav1.ManagedFieldsEntry, error) {
+	obj := &unstructured.Unstructured{Object: map[string]interface{}{}}
+	if err := internal.EncodeObjectManagedFields(obj, entries); err != nil {
+		return nil, err
+	}
+	accessor, err := meta.Accessor(obj)
+	if err != nil {
+		panic(fmt.Sprintf("couldn't get accessor: %v", err))
+	}
+	return accessor.GetManagedFields(), nil
+}

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager/scalehander_test.go

@@ -0,0 +1,684 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package fieldmanager
+
+import (
+	"reflect"
+	"testing"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/structured-merge-diff/v4/fieldpath"
+)
+
+func TestTransformManagedFieldsToSubresource(t *testing.T) {
+	testTime, _ := time.ParseInLocation("2006-Jan-02", "2013-Feb-03", time.Local)
+	managedFieldTime := v1.NewTime(testTime)
+
+	tests := []struct {
+		desc     string
+		input    []metav1.ManagedFieldsEntry
+		expected []metav1.ManagedFieldsEntry
+	}{
+		{
+			desc: "filter one entry and transform it into a subresource entry",
+			input: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "manager-1",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:another-field":{}}}`)},
+				},
+				{
+					Manager:    "manager-2",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Time:       &managedFieldTime,
+				},
+			},
+			expected: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "manager-2",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "autoscaling/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Time:       &managedFieldTime,
+				},
+			},
+		},
+		{
+			desc: "transform all entries",
+			input: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "manager-1",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+				},
+				{
+					Manager:    "manager-2",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+				},
+				{
+					Manager:     "manager-3",
+					Operation:   metav1.ManagedFieldsOperationApply,
+					APIVersion:  "apps/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+			expected: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "manager-1",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "autoscaling/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+				},
+				{
+					Manager:    "manager-2",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "autoscaling/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+				},
+				{
+					Manager:     "manager-3",
+					Operation:   metav1.ManagedFieldsOperationApply,
+					APIVersion:  "autoscaling/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		handler := NewScaleHandler(
+			test.input,
+			schema.GroupVersion{Group: "apps", Version: "v1"},
+			defaultMappings(),
+		)
+		subresourceEntries, err := handler.ToSubresource()
+		if err != nil {
+			t.Fatalf("test %q - expected no error but got %v", test.desc, err)
+		}
+
+		if !reflect.DeepEqual(subresourceEntries, test.expected) {
+			t.Fatalf("test %q - expected output to be:\n%v\n\nbut got:\n%v", test.desc, test.expected, subresourceEntries)
+		}
+	}
+}
+
+func TestTransformingManagedFieldsToParent(t *testing.T) {
+	tests := []struct {
+		desc        string
+		parent      []metav1.ManagedFieldsEntry
+		subresource []metav1.ManagedFieldsEntry
+		expected    []metav1.ManagedFieldsEntry
+	}{
+		{
+			desc: "different-managers: apply -> update",
+			parent: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{},"f:selector":{}}}`)},
+				},
+			},
+			subresource: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "scale",
+					Operation:   metav1.ManagedFieldsOperationUpdate,
+					APIVersion:  "autoscaling/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+			expected: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:selector":{}}}`)},
+				},
+				{
+					Manager:     "scale",
+					Operation:   metav1.ManagedFieldsOperationUpdate,
+					APIVersion:  "apps/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+		},
+		{
+			desc: "different-managers: apply -> apply",
+			parent: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{},"f:selector":{}}}`)},
+				},
+			},
+			subresource: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "scale",
+					Operation:   metav1.ManagedFieldsOperationApply,
+					APIVersion:  "autoscaling/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+			expected: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "scale",
+					Operation:   metav1.ManagedFieldsOperationApply,
+					APIVersion:  "apps/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:selector":{}}}`)},
+				},
+			},
+		},
+		{
+			desc: "different-managers: update -> update",
+			parent: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationUpdate,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{},"f:selector":{}}}`)},
+				},
+			},
+			subresource: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "scale",
+					Operation:   metav1.ManagedFieldsOperationUpdate,
+					APIVersion:  "autoscaling/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+			expected: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "scale",
+					Operation:   metav1.ManagedFieldsOperationUpdate,
+					APIVersion:  "apps/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationUpdate,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:selector":{}}}`)},
+				},
+			},
+		},
+		{
+			desc: "different-managers: update -> apply",
+			parent: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationUpdate,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{},"f:selector":{}}}`)},
+				},
+			},
+			subresource: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "scale",
+					Operation:   metav1.ManagedFieldsOperationApply,
+					APIVersion:  "autoscaling/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+			expected: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "scale",
+					Operation:   metav1.ManagedFieldsOperationApply,
+					APIVersion:  "apps/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationUpdate,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:selector":{}}}`)},
+				},
+			},
+		},
+		{
+			desc: "same manager: apply -> apply",
+			parent: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{},"f:selector":{}}}`)},
+				},
+			},
+			subresource: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "test",
+					Operation:   metav1.ManagedFieldsOperationApply,
+					APIVersion:  "autoscaling/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+			expected: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:selector":{}}}`)},
+				},
+				{
+					Manager:     "test",
+					Operation:   metav1.ManagedFieldsOperationApply,
+					APIVersion:  "apps/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+		},
+		{
+			desc: "same manager: update -> update",
+			parent: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationUpdate,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{},"f:selector":{}}}`)},
+				},
+			},
+			subresource: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "test",
+					Operation:   metav1.ManagedFieldsOperationUpdate,
+					APIVersion:  "autoscaling/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+			expected: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationUpdate,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:selector":{}}}`)},
+				},
+				{
+					Manager:     "test",
+					Operation:   metav1.ManagedFieldsOperationUpdate,
+					APIVersion:  "apps/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+		},
+		{
+			desc: "same manager: update -> apply",
+			parent: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationUpdate,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{},"f:selector":{}}}`)},
+				},
+			},
+			subresource: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "test",
+					Operation:   metav1.ManagedFieldsOperationApply,
+					APIVersion:  "autoscaling/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+			expected: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "test",
+					Operation:   metav1.ManagedFieldsOperationApply,
+					APIVersion:  "apps/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationUpdate,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:selector":{}}}`)},
+				},
+			},
+		},
+		{
+			desc: "same manager: apply -> update",
+			parent: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{},"f:selector":{}}}`)},
+				},
+			},
+			subresource: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "test",
+					Operation:   metav1.ManagedFieldsOperationUpdate,
+					APIVersion:  "autoscaling/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+			expected: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:selector":{}}}`)},
+				},
+				{
+					Manager:     "test",
+					Operation:   metav1.ManagedFieldsOperationUpdate,
+					APIVersion:  "apps/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+		},
+		{
+			desc: "subresource doesn't own the path anymore",
+			parent: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{},"f:selector":{}}}`)},
+				},
+			},
+			subresource: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "scale",
+					Operation:   metav1.ManagedFieldsOperationUpdate,
+					APIVersion:  "autoscaling/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:another":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+			expected: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:selector":{}}}`)},
+				},
+				{
+					Manager:     "scale",
+					Operation:   metav1.ManagedFieldsOperationUpdate,
+					APIVersion:  "apps/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+		},
+		{
+			desc: "Subresource steals all the fields of the parent resource",
+			parent: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+				},
+			},
+			subresource: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "scale",
+					Operation:   metav1.ManagedFieldsOperationUpdate,
+					APIVersion:  "autoscaling/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+			expected: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "scale",
+					Operation:   metav1.ManagedFieldsOperationUpdate,
+					APIVersion:  "apps/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+		},
+		{
+			desc: "apply without stealing",
+			parent: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{},"f:selector":{}}}`)},
+				},
+			},
+			subresource: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "autoscaling/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+				},
+				{
+					Manager:     "test",
+					Operation:   metav1.ManagedFieldsOperationApply,
+					APIVersion:  "autoscaling/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+			expected: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{},"f:selector":{}}}`)},
+				},
+				{
+					Manager:     "test",
+					Operation:   metav1.ManagedFieldsOperationApply,
+					APIVersion:  "apps/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.desc, func(t *testing.T) {
+			handler := NewScaleHandler(
+				test.parent,
+				schema.GroupVersion{Group: "apps", Version: "v1"},
+				defaultMappings(),
+			)
+			parentEntries, err := handler.ToParent(test.subresource)
+			if err != nil {
+				t.Fatalf("test: %q - expected no error but got %v", test.desc, err)
+			}
+			if !reflect.DeepEqual(parentEntries, test.expected) {
+				t.Fatalf("test: %q - expected output to be:\n%v\n\nbut got:\n%v", test.desc, test.expected, parentEntries)
+			}
+		})
+	}
+}
+
+func TestTransformingManagedFieldsToParentMultiVersion(t *testing.T) {
+	tests := []struct {
+		desc        string
+		mappings    ResourcePathMappings
+		parent      []metav1.ManagedFieldsEntry
+		subresource []metav1.ManagedFieldsEntry
+		expected    []metav1.ManagedFieldsEntry
+	}{
+		{
+			desc: "multi-version",
+			mappings: ResourcePathMappings{
+				"apps/v1": fieldpath.MakePathOrDie("spec", "the-replicas"),
+				"apps/v2": fieldpath.MakePathOrDie("spec", "not-the-replicas"),
+			},
+			parent: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:the-replicas":{},"f:selector":{}}}`)},
+				},
+				{
+					Manager:    "test-other",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v2",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:not-the-replicas":{},"f:selector":{}}}`)},
+				},
+			},
+			subresource: []metav1.ManagedFieldsEntry{
+				{
+					Manager:     "scale",
+					Operation:   metav1.ManagedFieldsOperationUpdate,
+					APIVersion:  "autoscaling/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+			expected: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "test",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v1",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:selector":{}}}`)},
+				},
+				{
+					Manager:    "test-other",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "apps/v2",
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:selector":{}}}`)},
+				},
+				{
+					Manager:     "scale",
+					Operation:   metav1.ManagedFieldsOperationUpdate,
+					APIVersion:  "apps/v1",
+					FieldsType:  "FieldsV1",
+					FieldsV1:    &metav1.FieldsV1{Raw: []byte(`{"f:spec":{"f:the-replicas":{}}}`)},
+					Subresource: "scale",
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.desc, func(t *testing.T) {
+			handler := NewScaleHandler(
+				test.parent,
+				schema.GroupVersion{Group: "apps", Version: "v1"},
+				test.mappings,
+			)
+			parentEntries, err := handler.ToParent(test.subresource)
+			if err != nil {
+				t.Fatalf("test: %q - expected no error but got %v", test.desc, err)
+			}
+			if !reflect.DeepEqual(parentEntries, test.expected) {
+				t.Fatalf("test: %q - expected output to be:\n%v\n\nbut got:\n%v", test.desc, test.expected, parentEntries)
+			}
+		})
+	}
+}
+
+func defaultMappings() ResourcePathMappings {
+	return ResourcePathMappings{
+		"apps/v1": fieldpath.MakePathOrDie("spec", "replicas"),
+	}
+}

test/integration/apiserver/apply/apply_test.go

@@ -3395,9 +3395,9 @@ func TestSubresourceField(t *testing.T) {
 		AbsPath("/apis/apps/v1").
 		Namespace("default").
 		Resource("deployments").
-		SubResource("status").
+		SubResource("scale").
 		Name("deployment").
-		Body([]byte(`{"status":{"unavailableReplicas":32}}`)).
+		Body([]byte(`{"spec":{"replicas":32}}`)).
 		Param("fieldManager", "manager").
 		Do(context.TODO()).
 		Get()
@@ -3405,8 +3405,6 @@ func TestSubresourceField(t *testing.T) {
 		t.Fatalf("Failed to update status: %v", err)
 	}
 
-	// TODO (nodo): add test for "scale" once we start tracking managed fields (#82046)
-
 	deployment, err := client.AppsV1().Deployments("default").Get(context.TODO(), "deployment", metav1.GetOptions{})
 	if err != nil {
 		t.Fatalf("Failed to get object: %v", err)
@@ -3414,15 +3412,312 @@ func TestSubresourceField(t *testing.T) {
 
 	managedFields := deployment.GetManagedFields()
 	if len(managedFields) != 2 {
-		t.Fatalf("Expected object to have 3 managed fields entries, got: %d", len(managedFields))
+		t.Fatalf("Expected object to have 2 managed fields entries, got: %d", len(managedFields))
 	}
 	if managedFields[0].Manager != "manager" || managedFields[0].Operation != "Apply" || managedFields[0].Subresource != "" {
 		t.Fatalf(`Unexpected entry, got: %v`, managedFields[0])
 	}
 	if managedFields[1].Manager != "manager" ||
 		managedFields[1].Operation != "Update" ||
-		managedFields[1].Subresource != "status" ||
-		string(managedFields[1].FieldsV1.Raw) != `{"f:status":{"f:unavailableReplicas":{}}}` {
+		managedFields[1].Subresource != "scale" ||
+		string(managedFields[1].FieldsV1.Raw) != `{"f:spec":{"f:replicas":{}}}` {
 		t.Fatalf(`Unexpected entry, got: %v`, managedFields[1])
 	}
 }
+
+func TestApplyOnScaleDeployment(t *testing.T) {
+	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, genericfeatures.ServerSideApply, true)()
+
+	_, client, closeFn := setup(t)
+	defer closeFn()
+
+	validDeployment := []byte(`{
+		"apiVersion": "apps/v1",
+		"kind": "Deployment",
+		"metadata": {
+			"name": "deployment"
+		},
+		"spec": {
+			"replicas": 1,
+			"selector": {
+				"matchLabels": {
+					 "app": "nginx"
+				}
+			},
+			"template": {
+				"metadata": {
+					"labels": {
+						"app": "nginx"
+					}
+				},
+				"spec": {
+					"containers": [{
+						"name":  "nginx",
+						"image": "nginx:latest"
+					}]
+				}
+			}
+		}
+	}`)
+
+	// Create deployment
+	_, err := client.CoreV1().RESTClient().Patch(types.ApplyPatchType).
+		AbsPath("/apis/apps/v1").
+		Namespace("default").
+		Resource("deployments").
+		Name("deployment").
+		Param("fieldManager", "apply_test").
+		Body(validDeployment).
+		Do(context.TODO()).
+		Get()
+	if err != nil {
+		t.Fatalf("Failed to create object using Apply patch: %v", err)
+	}
+	deployment, err := client.AppsV1().Deployments("default").Get(context.TODO(), "deployment", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Failed to retrieve object: %v", err)
+	}
+	if *deployment.Spec.Replicas != 1 {
+		t.Fatalf("Expected replicas to be 1, but got: %d", *deployment.Spec.Replicas)
+	}
+
+	// Call scale subresource to update replicas
+	_, err = client.CoreV1().RESTClient().
+		Patch(types.MergePatchType).
+		AbsPath("/apis/apps/v1").
+		Name("deployment").
+		Resource("deployments").
+		SubResource("scale").
+		Namespace("default").
+		Param("fieldManager", "scale_test").
+		Body([]byte(`{"spec":{"replicas": 5}}`)).
+		Do(context.TODO()).
+		Get()
+	if err != nil {
+		t.Fatalf("Error updating scale subresource: %v ", err)
+	}
+
+	deployment, err = client.AppsV1().Deployments("default").Get(context.TODO(), "deployment", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Failed to retrieve object: %v", err)
+	}
+
+	if *deployment.Spec.Replicas != 5 {
+		t.Fatalf("Expected replicas to be 5, but got: %d", *deployment.Spec.Replicas)
+	}
+
+	assertReplicasOwnership(t, (*deployment).GetManagedFields(), "scale_test")
+
+	// Re-apply the original object, it should fail with conflict because replicas have changed
+	_, err = client.CoreV1().RESTClient().Patch(types.ApplyPatchType).
+		AbsPath("/apis/apps/v1").
+		Namespace("default").
+		Resource("deployments").
+		Name("deployment").
+		Param("fieldManager", "apply_test").
+		Body(validDeployment).
+		Do(context.TODO()).
+		Get()
+	if !apierrors.IsConflict(err) {
+		t.Fatalf("Expected conflict error but got: %v", err)
+	}
+
+	// Re-apply forcing the changes should succeed
+	_, err = client.CoreV1().RESTClient().Patch(types.ApplyPatchType).
+		AbsPath("/apis/apps/v1").
+		Namespace("default").
+		Resource("deployments").
+		Name("deployment").
+		Param("fieldManager", "apply_test").
+		Param("force", "true").
+		Body(validDeployment).
+		Do(context.TODO()).
+		Get()
+	if err != nil {
+		t.Fatalf("Error updating deployment: %v ", err)
+	}
+
+	deployment, err = client.AppsV1().Deployments("default").Get(context.TODO(), "deployment", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Failed to retrieve object: %v", err)
+	}
+
+	if *deployment.Spec.Replicas != 1 {
+		t.Fatalf("Expected replicas to be 1, but got: %d", *deployment.Spec.Replicas)
+	}
+
+	assertReplicasOwnership(t, (*deployment).GetManagedFields(), "apply_test")
+
+	// Run "Apply" with a scale object with a different number of replicas. It should generate a conflict.
+	_, err = client.CoreV1().RESTClient().
+		Patch(types.ApplyPatchType).
+		AbsPath("/apis/apps/v1").
+		Namespace("default").
+		Resource("deployments").
+		SubResource("scale").
+		Name("deployment").
+		Param("fieldManager", "apply_scale").
+		Body([]byte(`{"kind":"Scale","apiVersion":"autoscaling/v1","metadata":{"name":"deployment","namespace":"default"},"spec":{"replicas":17}}`)).
+		Do(context.TODO()).
+		Get()
+	if !apierrors.IsConflict(err) {
+		t.Fatalf("Expected conflict error but got: %v", err)
+	}
+	if !strings.Contains(err.Error(), "apply_test") {
+		t.Fatalf("Expected conflict with `apply_test` manager but got: %v", err)
+	}
+
+	// Same as before but force. Only the new manager should own .spec.replicas
+	_, err = client.CoreV1().RESTClient().
+		Patch(types.ApplyPatchType).
+		AbsPath("/apis/apps/v1").
+		Namespace("default").
+		Resource("deployments").
+		SubResource("scale").
+		Name("deployment").
+		Param("fieldManager", "apply_scale").
+		Param("force", "true").
+		Body([]byte(`{"kind":"Scale","apiVersion":"autoscaling/v1","metadata":{"name":"deployment","namespace":"default"},"spec":{"replicas":17}}`)).
+		Do(context.TODO()).
+		Get()
+	if err != nil {
+		t.Fatalf("Error updating deployment: %v ", err)
+	}
+
+	deployment, err = client.AppsV1().Deployments("default").Get(context.TODO(), "deployment", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Failed to retrieve object: %v", err)
+	}
+
+	if *deployment.Spec.Replicas != 17 {
+		t.Fatalf("Expected to replicas to be 17, but got: %d", *deployment.Spec.Replicas)
+	}
+
+	assertReplicasOwnership(t, (*deployment).GetManagedFields(), "apply_scale")
+
+	// Replace scale object
+	_, err = client.CoreV1().RESTClient().
+		Put().
+		AbsPath("/apis/apps/v1").
+		Namespace("default").
+		Resource("deployments").
+		SubResource("scale").
+		Name("deployment").
+		Param("fieldManager", "replace_test").
+		Body([]byte(`{"kind":"Scale","apiVersion":"autoscaling/v1","metadata":{"name":"deployment","namespace":"default"},"spec":{"replicas":7}}`)).
+		Do(context.TODO()).
+		Get()
+	if err != nil {
+		t.Fatalf("Error updating deployment: %v ", err)
+	}
+
+	deployment, err = client.AppsV1().Deployments("default").Get(context.TODO(), "deployment", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Failed to retrieve object: %v", err)
+	}
+
+	if *deployment.Spec.Replicas != 7 {
+		t.Fatalf("Expected to replicas to be 7, but got: %d", *deployment.Spec.Replicas)
+	}
+
+	assertReplicasOwnership(t, (*deployment).GetManagedFields(), "replace_test")
+
+	// Apply the same number of replicas, both managers should own the field
+	_, err = client.CoreV1().RESTClient().
+		Patch(types.ApplyPatchType).
+		AbsPath("/apis/apps/v1").
+		Namespace("default").
+		Resource("deployments").
+		SubResource("scale").
+		Name("deployment").
+		Param("fieldManager", "co_owning_test").
+		Param("force", "true").
+		Body([]byte(`{"kind":"Scale","apiVersion":"autoscaling/v1","metadata":{"name":"deployment","namespace":"default"},"spec":{"replicas":7}}`)).
+		Do(context.TODO()).
+		Get()
+	if err != nil {
+		t.Fatalf("Error updating deployment: %v ", err)
+	}
+
+	deployment, err = client.AppsV1().Deployments("default").Get(context.TODO(), "deployment", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Failed to retrieve object: %v", err)
+	}
+
+	if *deployment.Spec.Replicas != 7 {
+		t.Fatalf("Expected to replicas to be 7, but got: %d", *deployment.Spec.Replicas)
+	}
+
+	assertReplicasOwnership(t, (*deployment).GetManagedFields(), "replace_test", "co_owning_test")
+
+	// Scaling again should make this manager the only owner of replicas
+	_, err = client.CoreV1().RESTClient().
+		Patch(types.MergePatchType).
+		AbsPath("/apis/apps/v1").
+		Name("deployment").
+		Resource("deployments").
+		SubResource("scale").
+		Namespace("default").
+		Param("fieldManager", "scale_test").
+		Body([]byte(`{"spec":{"replicas": 5}}`)).
+		Do(context.TODO()).
+		Get()
+	if err != nil {
+		t.Fatalf("Error updating scale subresource: %v ", err)
+	}
+
+	deployment, err = client.AppsV1().Deployments("default").Get(context.TODO(), "deployment", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Failed to retrieve object: %v", err)
+	}
+
+	if *deployment.Spec.Replicas != 5 {
+		t.Fatalf("Expected replicas to be 5, but got: %d", *deployment.Spec.Replicas)
+	}
+
+	assertReplicasOwnership(t, (*deployment).GetManagedFields(), "scale_test")
+}
+
+func assertReplicasOwnership(t *testing.T, managedFields []metav1.ManagedFieldsEntry, fieldManagers ...string) {
+	t.Helper()
+
+	seen := make(map[string]bool)
+	for _, m := range fieldManagers {
+		seen[m] = false
+	}
+
+	for _, managedField := range managedFields {
+		var entryJSON map[string]interface{}
+		if err := json.Unmarshal(managedField.FieldsV1.Raw, &entryJSON); err != nil {
+			t.Fatalf("failed to read into json")
+		}
+
+		spec, ok := entryJSON["f:spec"].(map[string]interface{})
+		if !ok {
+			// continue with the next managedField, as we this field does not hold the spec entry
+			continue
+		}
+
+		if _, ok := spec["f:replicas"]; !ok {
+			// continue with the next managedField, as we this field does not hold the spec.replicas entry
+			continue
+		}
+
+		// check if the manager is one of the ones we expect
+		if _, ok := seen[managedField.Manager]; !ok {
+			t.Fatalf("Unexpected field manager, found %q, expected to be in: %v", managedField.Manager, seen)
+		}
+
+		seen[managedField.Manager] = true
+	}
+
+	var missingManagers []string
+	for manager, managerSeen := range seen {
+		if !managerSeen {
+			missingManagers = append(missingManagers, manager)
+		}
+	}
+	if len(missingManagers) > 0 {
+		t.Fatalf("replicas fields should be owned by %v", missingManagers)
+	}
+}