Merge pull request #78738 from jennybuckley/apply-only

Don't start tracking field management until object has been applied

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager/BUILD

@@ -47,6 +47,7 @@ go_test(
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/meta:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",

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

@@ -104,6 +104,10 @@ func (f *FieldManager) Update(liveObj, newObj runtime.Object, manager string) (r
 			return nil, fmt.Errorf("failed to decode managed fields: %v", err)
 		}
 	}
+	// if managed field is still empty, skip updating managed fields altogether
+	if len(managed.Fields) == 0 {
+		return newObj, nil
+	}
 	newObjVersioned, err := f.toVersioned(newObj)
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert new object to proper version: %v", err)
@@ -165,9 +169,11 @@ func (f *FieldManager) Update(liveObj, newObj runtime.Object, manager string) (r
 // object and update the managed fields.
 func (f *FieldManager) Apply(liveObj runtime.Object, patch []byte, fieldManager string, force bool) (runtime.Object, error) {
 	// If the object doesn't have metadata, apply isn't allowed.
-	if _, err := meta.Accessor(liveObj); err != nil {
+	accessor, err := meta.Accessor(liveObj)
+	if err != nil {
 		return nil, fmt.Errorf("couldn't get accessor: %v", err)
 	}
+	missingManagedFields := (len(accessor.GetManagedFields()) == 0)
 
 	managed, err := internal.DecodeObjectManagedFields(liveObj)
 	if err != nil {
@@ -207,6 +213,23 @@ func (f *FieldManager) Apply(liveObj runtime.Object, patch []byte, fieldManager
 	}
 
 	apiVersion := fieldpath.APIVersion(f.groupVersion.String())
+	// if managed field is missing, create a single entry for all the fields
+	if missingManagedFields {
+		unknownManager, err := internal.BuildManagerIdentifier(&metav1.ManagedFieldsEntry{
+			Manager:    "before-first-apply",
+			Operation:  metav1.ManagedFieldsOperationUpdate,
+			APIVersion: f.groupVersion.String(),
+		})
+		if err != nil {
+			return nil, fmt.Errorf("failed to create manager for existing fields: %v", err)
+		}
+		unknownFieldSet, err := liveObjTyped.ToFieldSet()
+		if err != nil {
+			return nil, fmt.Errorf("failed to create fieldset for existing fields: %v", err)
+		}
+		managed.Fields[unknownManager] = fieldpath.NewVersionedSet(unknownFieldSet, apiVersion, false)
+		f.stripFields(managed.Fields, unknownManager)
+	}
 	newObjTyped, managedFields, err := f.updater.Apply(liveObjTyped, patchObjTyped, apiVersion, managed.Fields, manager, force)
 	if err != nil {
 		if conflicts, ok := err.(merge.Conflicts); ok {

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

@@ -25,6 +25,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"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"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -71,10 +72,34 @@ func TestFieldManagerCreation(t *testing.T) {
 	}
 }
 
+func TestUpdateOnlyDoesNotTrackManagedFields(t *testing.T) {
+	f := NewTestFieldManager()
+
+	liveObj := &corev1.Pod{}
+
+	updatedObj := liveObj.DeepCopy()
+	updatedObj.ObjectMeta.Labels = map[string]string{"k": "v"}
+
+	newObj, err := f.Update(liveObj, updatedObj, "fieldmanager_test")
+	if err != nil {
+		t.Fatalf("failed to update object: %v", err)
+	}
+
+	accessor, err := meta.Accessor(newObj)
+	if err != nil {
+		t.Fatalf("couldn't get accessor: %v", err)
+	}
+
+	if m := accessor.GetManagedFields(); len(m) != 0 {
+		t.Fatalf("managedFields were tracked on update only: %v", m)
+	}
+}
+
 func TestApplyStripsFields(t *testing.T) {
 	f := NewTestFieldManager()
 
 	obj := &corev1.Pod{}
+	obj.ObjectMeta.ManagedFields = []metav1.ManagedFieldsEntry{{}}
 
 	newObj, err := f.Apply(obj, []byte(`{
 		"apiVersion": "apps/v1",
@@ -153,6 +178,7 @@ func TestApplyDoesNotStripLabels(t *testing.T) {
 	f := NewTestFieldManager()
 
 	obj := &corev1.Pod{}
+	obj.ObjectMeta.ManagedFields = []metav1.ManagedFieldsEntry{{}}
 
 	newObj, err := f.Apply(obj, []byte(`{
 		"apiVersion": "apps/v1",

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

@@ -421,7 +421,7 @@ func (p *applyPatcher) applyPatchToCurrentObject(obj runtime.Object) (runtime.Ob
 func (p *applyPatcher) createNewObject() (runtime.Object, error) {
 	obj, err := p.creater.New(p.kind)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create new object: %v", obj)
+		return nil, fmt.Errorf("failed to create new object: %v", err)
 	}
 	return p.applyPatchToCurrentObject(obj)
 }

test/integration/apiserver/apply/apply_test.go

@@ -180,9 +180,11 @@ func TestNoOpUpdateSameResourceVersion(t *testing.T) {
 		}
 	}`)
 
-	_, err := client.CoreV1().RESTClient().Post().
+	o, err := client.CoreV1().RESTClient().Patch(types.ApplyPatchType).
 		Namespace("default").
+		Param("fieldManager", "apply_test").
 		Resource(podResource).
+		Name(podName).
 		Body(podBytes).
 		Do().
 		Get()
@@ -190,6 +192,23 @@ func TestNoOpUpdateSameResourceVersion(t *testing.T) {
 		t.Fatalf("Failed to create object: %v", err)
 	}
 
+	// Need to update once for some reason
+	// TODO (#82042): Remove this update once possible
+	b, err := json.MarshalIndent(o, "\t", "\t")
+	if err != nil {
+		t.Fatalf("Failed to marshal created object: %v", err)
+	}
+	_, err = client.CoreV1().RESTClient().Put().
+		Namespace("default").
+		Resource(podResource).
+		Name(podName).
+		Body(b).
+		Do().
+		Get()
+	if err != nil {
+		t.Fatalf("Failed to apply first no-op update: %v", err)
+	}
+
 	// Sleep for one second to make sure that the times of each update operation is different.
 	time.Sleep(1 * time.Second)
 
@@ -367,6 +386,81 @@ func TestApplyUpdateApplyConflictForced(t *testing.T) {
 	}
 }
 
+// TestUpdateApplyConflict tests that applying to an object, which wasn't created by apply, will give conflicts
+func TestUpdateApplyConflict(t *testing.T) {
+	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, genericfeatures.ServerSideApply, true)()
+
+	_, client, closeFn := setup(t)
+	defer closeFn()
+
+	obj := []byte(`{
+		"apiVersion": "apps/v1",
+		"kind": "Deployment",
+		"metadata": {
+			"name": "deployment",
+			"labels": {"app": "nginx"}
+		},
+		"spec": {
+                        "replicas": 3,
+                        "selector": {
+                                "matchLabels": {
+                                         "app": "nginx"
+                                }
+                        },
+                        "template": {
+                                "metadata": {
+                                        "labels": {
+                                                "app": "nginx"
+                                        }
+                                },
+                                "spec": {
+				        "containers": [{
+					        "name":  "nginx",
+					        "image": "nginx:latest"
+				        }]
+                                }
+                        }
+		}
+	}`)
+
+	_, err := client.CoreV1().RESTClient().Post().
+		AbsPath("/apis/apps/v1").
+		Namespace("default").
+		Resource("deployments").
+		Body(obj).Do().Get()
+	if err != nil {
+		t.Fatalf("Failed to create object using post: %v", err)
+	}
+
+	obj = []byte(`{
+		"apiVersion": "apps/v1",
+		"kind": "Deployment",
+		"metadata": {
+			"name": "deployment",
+		},
+		"spec": {
+			"replicas": 101,
+		}
+	}`)
+	_, err = client.CoreV1().RESTClient().Patch(types.ApplyPatchType).
+		AbsPath("/apis/apps/v1").
+		Namespace("default").
+		Resource("deployments").
+		Name("deployment").
+		Param("fieldManager", "apply_test").
+		Body([]byte(obj)).Do().Get()
+	if err == nil {
+		t.Fatalf("Expecting to get conflicts when applying object")
+	}
+	status, ok := err.(*errors.StatusError)
+	if !ok {
+		t.Fatalf("Expecting to get conflicts as API error")
+	}
+	if len(status.Status().Details.Causes) < 1 {
+		t.Fatalf("Expecting to get at least one conflict when applying object, got: %v", status.Status().Details.Causes)
+	}
+}
+
 // TestApplyManagedFields makes sure that managedFields api does not change
 func TestApplyManagedFields(t *testing.T) {
 	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, genericfeatures.ServerSideApply, true)()
@@ -1178,7 +1272,6 @@ var podBytes = []byte(`
 apiVersion: v1
 kind: Pod
 metadata:
-  creationTimestamp: "2019-07-08T09:31:18Z"
   labels:
     app: some-app
     plugin1: some-value
@@ -1194,8 +1287,6 @@ metadata:
     kind: ReplicaSet
     name: some-name
     uid: 0a9d2b9e-779e-11e7-b422-42010a8001be
-  selfLink: /api/v1/namespaces/pah
-  uid: 23e8f548-a163-11e9-abe4-42010a80026b
 spec:
   containers:
   - args:
@@ -1329,22 +1420,27 @@ func BenchmarkNoServerSideApply(b *testing.B) {
 }
 
 func getPodSizeWhenEnabled(b *testing.B, pod v1.Pod) int {
+	return len(getPodBytesWhenEnabled(b, pod, "application/vnd.kubernetes.protobuf"))
+}
+
+func getPodBytesWhenEnabled(b *testing.B, pod v1.Pod, format string) []byte {
 	defer featuregatetesting.SetFeatureGateDuringTest(b, utilfeature.DefaultFeatureGate, genericfeatures.ServerSideApply, true)()
 	_, client, closeFn := setup(b)
 	defer closeFn()
 	flag.Lookup("v").Value.Set("0")
 
 	pod.Name = "size-pod"
-	podB, err := client.CoreV1().RESTClient().Post().
+	podB, err := client.CoreV1().RESTClient().Patch(types.ApplyPatchType).
+		Name(pod.Name).
 		Namespace("default").
+		Param("fieldManager", "apply_test").
 		Resource("pods").
-		SetHeader("Content-Type", "application/yaml").
+		SetHeader("Accept", format).
-		SetHeader("Accept", "application/vnd.kubernetes.protobuf").
 		Body(encodePod(pod)).DoRaw()
 	if err != nil {
-		b.Fatalf("Failed to create object: %v", err)
+		b.Fatalf("Failed to create object: %#v", err)
 	}
-	return len(podB)
+	return podB
 }
 
 func BenchmarkNoServerSideApplyButSameSize(b *testing.B) {
@@ -1385,16 +1481,24 @@ func BenchmarkNoServerSideApplyButSameSize(b *testing.B) {
 }
 
 func BenchmarkServerSideApply(b *testing.B) {
+	podBytesWhenEnabled := getPodBytesWhenEnabled(b, decodePod(podBytes), "application/yaml")
+
 	defer featuregatetesting.SetFeatureGateDuringTest(b, utilfeature.DefaultFeatureGate, genericfeatures.ServerSideApply, true)()
 
 	_, client, closeFn := setup(b)
 	defer closeFn()
 	flag.Lookup("v").Value.Set("0")
 
-	benchAll(b, client, decodePod(podBytes))
+	benchAll(b, client, decodePod(podBytesWhenEnabled))
 }
 
 func benchAll(b *testing.B, client kubernetes.Interface, pod v1.Pod) {
+	// Make sure pod is ready to post
+	pod.ObjectMeta.CreationTimestamp = metav1.Time{}
+	pod.ObjectMeta.ResourceVersion = ""
+	pod.ObjectMeta.UID = ""
+	pod.ObjectMeta.SelfLink = ""
+
 	// Create pod for repeated-updates
 	pod.Name = "repeated-pod"
 	_, err := client.CoreV1().RESTClient().Post().