Merge pull request kubernetes#35647 from ymqytw/patch_primitive_list

Automatic merge from submit-queue

Fix strategic patch for list of primitive type with merge sementic

Fix strategic patch for list of primitive type when the patch strategy is `merge`.
Before: we cannot replace or delete an item in a list of primitive, e.g. string, when the patch strategy is `merge`. It will always append new items to the list.
This patch will generate a map to update the list of primitive type.
The server with this patch will accept either a new patch or an old patch.
The client will found out the APIserver version before generate the patch.

Fixes kubernetes#35163, kubernetes#32398

cc: @pwittrock @fabianofranz 

``` release-note
Fix strategic patch for list of primitive type when patch strategy is `merge` to remove deleted objects.
```

pkg/apiserver/resthandler.go

@@ -591,11 +591,11 @@ func patchResource(
 			if err != nil {
 				return nil, err
 			}
-			currentPatch, err := strategicpatch.CreateStrategicMergePatch(originalObjJS, currentObjectJS, versionedObj)
+			currentPatch, err := strategicpatch.CreateStrategicMergePatch(originalObjJS, currentObjectJS, versionedObj, strategicpatch.SMPatchVersionLatest)
 			if err != nil {
 				return nil, err
 			}
-			originalPatch, err := strategicpatch.CreateStrategicMergePatch(originalObjJS, originalPatchedObjJS, versionedObj)
+			originalPatch, err := strategicpatch.CreateStrategicMergePatch(originalObjJS, originalPatchedObjJS, versionedObj, strategicpatch.SMPatchVersionLatest)
 			if err != nil {
 				return nil, err
 			}

pkg/apiserver/resthandler_test.go

@@ -213,7 +213,7 @@ func (tc *patchTestCase) Run(t *testing.T) {
 			continue
 
 		case api.StrategicMergePatchType:
-			patch, err = strategicpatch.CreateStrategicMergePatch(originalObjJS, changedJS, versionedObj)
+			patch, err = strategicpatch.CreateStrategicMergePatch(originalObjJS, changedJS, versionedObj, strategicpatch.SMPatchVersionLatest)
 			if err != nil {
 				t.Errorf("%s: unexpected error: %v", tc.name, err)
 				return

pkg/client/record/events_cache.go

@@ -244,7 +244,9 @@ func (e *eventLogger) eventObserve(newEvent *api.Event) (*api.Event, []byte, err
 
 		newData, _ := json.Marshal(event)
 		oldData, _ := json.Marshal(eventCopy2)
-		patch, err = strategicpatch.CreateStrategicMergePatch(oldData, newData, event)
+		// TODO: need to figure out if we need to let eventObserve() use the new behavior of StrategicMergePatch.
+		// Currently default to old behavior now. Ref: issue #35936
+		patch, err = strategicpatch.CreateStrategicMergePatch(oldData, newData, event, strategicpatch.SMPatchVersion_1_0)
 	}
 
 	// record our new observation

pkg/controller/volume/attachdetach/statusupdater/node_status_updater.go

@@ -59,6 +59,10 @@ type nodeStatusUpdater struct {
 }
 
 func (nsu *nodeStatusUpdater) UpdateNodeStatuses() error {
+	smPatchVersion, err := strategicpatch.GetServerSupportedSMPatchVersion(nsu.kubeClient.Discovery())
+	if err != nil {
+		return err
+	}
 	nodesToUpdate := nsu.actualStateOfWorld.GetVolumesToReportAttached()
 	for nodeName, attachedVolumes := range nodesToUpdate {
 		nodeObj, exists, err := nsu.nodeInformer.GetStore().GetByKey(string(nodeName))
@@ -108,7 +112,7 @@ func (nsu *nodeStatusUpdater) UpdateNodeStatuses() error {
 		}
 
 		patchBytes, err :=
-			strategicpatch.CreateStrategicMergePatch(oldData, newData, node)
+			strategicpatch.CreateStrategicMergePatch(oldData, newData, node, smPatchVersion)
 		if err != nil {
 			return fmt.Errorf(
 				"failed to CreateStrategicMergePatch for node %q. %v",

pkg/kubectl/cmd/BUILD

@@ -193,6 +193,7 @@ go_test(
         "//pkg/util/strings:go_default_library",
         "//pkg/util/term:go_default_library",
         "//pkg/util/wait:go_default_library",
+        "//pkg/version:go_default_library",
         "//pkg/watch:go_default_library",
         "//pkg/watch/versioned:go_default_library",
         "//vendor:github.com/spf13/cobra",

pkg/kubectl/cmd/annotate.go

@@ -223,6 +223,12 @@ func (o AnnotateOptions) RunAnnotate(f cmdutil.Factory, cmd *cobra.Command) erro
 			}
 			outputObj = obj
 		} else {
+			// retrieves server version to determine which SMPatchVersion to use.
+			smPatchVersion, err := cmdutil.GetServerSupportedSMPatchVersionFromFactory(f)
+			if err != nil {
+				return err
+			}
+
 			name, namespace := info.Name, info.Namespace
 			oldData, err := json.Marshal(obj)
 			if err != nil {
@@ -239,7 +245,7 @@ func (o AnnotateOptions) RunAnnotate(f cmdutil.Factory, cmd *cobra.Command) erro
 			if err != nil {
 				return err
 			}
-			patchBytes, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, obj)
+			patchBytes, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, obj, smPatchVersion)
 			createdPatch := err == nil
 			if err != nil {
 				glog.V(2).Infof("couldn't compute patch: %v", err)

pkg/kubectl/cmd/annotate_test.go

@@ -24,8 +24,6 @@ import (
 	"testing"
 
 	"k8s.io/kubernetes/pkg/api"
-	"k8s.io/kubernetes/pkg/apimachinery/registered"
-	"k8s.io/kubernetes/pkg/client/restclient"
 	"k8s.io/kubernetes/pkg/client/restclient/fake"
 	cmdtesting "k8s.io/kubernetes/pkg/kubectl/cmd/testing"
 	"k8s.io/kubernetes/pkg/runtime"
@@ -394,7 +392,7 @@ func TestAnnotateErrors(t *testing.T) {
 		f, tf, _, _ := cmdtesting.NewAPIFactory()
 		tf.Printer = &testPrinter{}
 		tf.Namespace = "test"
-		tf.ClientConfig = &restclient.Config{ContentConfig: restclient.ContentConfig{GroupVersion: &registered.GroupOrDie(api.GroupName).GroupVersion}}
+		tf.ClientConfig = defaultClientConfig()
 
 		buf := bytes.NewBuffer([]byte{})
 		cmd := NewCmdAnnotate(f, buf)
@@ -432,6 +430,12 @@ func TestAnnotateObject(t *testing.T) {
 			switch req.Method {
 			case "GET":
 				switch req.URL.Path {
+				case "/version":
+					resp, err := genResponseWithJsonEncodedBody(serverVersion_1_5_0)
+					if err != nil {
+						t.Fatalf("error: failed to generate server version response: %#v\n", serverVersion_1_5_0)
+					}
+					return resp, nil
 				case "/namespaces/test/pods/foo":
 					return &http.Response{StatusCode: 200, Header: defaultHeader(), Body: objBody(codec, &pods.Items[0])}, nil
 				default:
@@ -453,7 +457,7 @@ func TestAnnotateObject(t *testing.T) {
 		}),
 	}
 	tf.Namespace = "test"
-	tf.ClientConfig = &restclient.Config{ContentConfig: restclient.ContentConfig{GroupVersion: &registered.GroupOrDie(api.GroupName).GroupVersion}}
+	tf.ClientConfig = defaultClientConfig()
 
 	buf := bytes.NewBuffer([]byte{})
 	cmd := NewCmdAnnotate(f, buf)
@@ -482,6 +486,12 @@ func TestAnnotateObjectFromFile(t *testing.T) {
 			switch req.Method {
 			case "GET":
 				switch req.URL.Path {
+				case "/version":
+					resp, err := genResponseWithJsonEncodedBody(serverVersion_1_5_0)
+					if err != nil {
+						t.Fatalf("error: failed to generate server version response: %#v\n", serverVersion_1_5_0)
+					}
+					return resp, nil
 				case "/namespaces/test/replicationcontrollers/cassandra":
 					return &http.Response{StatusCode: 200, Header: defaultHeader(), Body: objBody(codec, &pods.Items[0])}, nil
 				default:
@@ -503,7 +513,7 @@ func TestAnnotateObjectFromFile(t *testing.T) {
 		}),
 	}
 	tf.Namespace = "test"
-	tf.ClientConfig = &restclient.Config{ContentConfig: restclient.ContentConfig{GroupVersion: &registered.GroupOrDie(api.GroupName).GroupVersion}}
+	tf.ClientConfig = defaultClientConfig()
 
 	buf := bytes.NewBuffer([]byte{})
 	cmd := NewCmdAnnotate(f, buf)
@@ -532,7 +542,7 @@ func TestAnnotateLocal(t *testing.T) {
 		}),
 	}
 	tf.Namespace = "test"
-	tf.ClientConfig = &restclient.Config{ContentConfig: restclient.ContentConfig{GroupVersion: &registered.GroupOrDie(api.GroupName).GroupVersion}}
+	tf.ClientConfig = defaultClientConfig()
 
 	buf := bytes.NewBuffer([]byte{})
 	cmd := NewCmdAnnotate(f, buf)
@@ -562,6 +572,12 @@ func TestAnnotateMultipleObjects(t *testing.T) {
 			switch req.Method {
 			case "GET":
 				switch req.URL.Path {
+				case "/version":
+					resp, err := genResponseWithJsonEncodedBody(serverVersion_1_5_0)
+					if err != nil {
+						t.Fatalf("error: failed to generate server version response: %#v\n", serverVersion_1_5_0)
+					}
+					return resp, nil
 				case "/namespaces/test/pods":
 					return &http.Response{StatusCode: 200, Header: defaultHeader(), Body: objBody(codec, pods)}, nil
 				default:
@@ -585,7 +601,7 @@ func TestAnnotateMultipleObjects(t *testing.T) {
 		}),
 	}
 	tf.Namespace = "test"
-	tf.ClientConfig = &restclient.Config{ContentConfig: restclient.ContentConfig{GroupVersion: &registered.GroupOrDie(api.GroupName).GroupVersion}}
+	tf.ClientConfig = defaultClientConfig()
 
 	buf := bytes.NewBuffer([]byte{})
 	cmd := NewCmdAnnotate(f, buf)

pkg/kubectl/cmd/apply.go

@@ -195,6 +195,11 @@ func RunApply(f cmdutil.Factory, cmd *cobra.Command, out io.Writer, options *App
 	visitedUids := sets.NewString()
 	visitedNamespaces := sets.NewString()
 
+	smPatchVersion, err := cmdutil.GetServerSupportedSMPatchVersionFromFactory(f)
+	if err != nil {
+		return err
+	}
+
 	count := 0
 	err = r.Visit(func(info *resource.Info, err error) error {
 		// In this method, info.Object contains the object retrieved from the server
@@ -265,13 +270,13 @@ func RunApply(f cmdutil.Factory, cmd *cobra.Command, out io.Writer, options *App
 				gracePeriod:   options.GracePeriod,
 			}
 
-			patchBytes, err := patcher.patch(info.Object, modified, info.Source, info.Namespace, info.Name)
+			patchBytes, err := patcher.patch(info.Object, modified, info.Source, info.Namespace, info.Name, smPatchVersion)
 			if err != nil {
 				return cmdutil.AddSourceToErr(fmt.Sprintf("applying patch:\n%s\nto:\n%v\nfor:", patchBytes, info), info.Source, err)
 			}
 
 			if cmdutil.ShouldRecord(cmd, info) {
-				patch, err := cmdutil.ChangeResourcePatch(info, f.Command())
+				patch, err := cmdutil.ChangeResourcePatch(info, f.Command(), smPatchVersion)
 				if err != nil {
 					return err
 				}
@@ -507,7 +512,7 @@ type patcher struct {
 	gracePeriod int
 }
 
-func (p *patcher) patchSimple(obj runtime.Object, modified []byte, source, namespace, name string) ([]byte, error) {
+func (p *patcher) patchSimple(obj runtime.Object, modified []byte, source, namespace, name string, smPatchVersion strategicpatch.StrategicMergePatchVersion) ([]byte, error) {
 	// Serialize the current configuration of the object from the server.
 	current, err := runtime.Encode(p.encoder, obj)
 	if err != nil {
@@ -531,7 +536,8 @@ func (p *patcher) patchSimple(obj runtime.Object, modified []byte, source, names
 	}
 
 	// Compute a three way strategic merge patch to send to server.
-	patch, err := strategicpatch.CreateThreeWayMergePatch(original, modified, current, versionedObject, p.overwrite)
+	patch, err := strategicpatch.CreateThreeWayMergePatch(original, modified, current, versionedObject, p.overwrite, smPatchVersion)
+
 	if err != nil {
 		format := "creating patch with:\noriginal:\n%s\nmodified:\n%s\ncurrent:\n%s\nfor:"
 		return nil, cmdutil.AddSourceToErr(fmt.Sprintf(format, original, modified, current), source, err)
@@ -541,9 +547,9 @@ func (p *patcher) patchSimple(obj runtime.Object, modified []byte, source, names
 	return patch, err
 }
 
-func (p *patcher) patch(current runtime.Object, modified []byte, source, namespace, name string) ([]byte, error) {
+func (p *patcher) patch(current runtime.Object, modified []byte, source, namespace, name string, smPatchVersion strategicpatch.StrategicMergePatchVersion) ([]byte, error) {
 	var getErr error
-	patchBytes, err := p.patchSimple(current, modified, source, namespace, name)
+	patchBytes, err := p.patchSimple(current, modified, source, namespace, name, smPatchVersion)
 	for i := 1; i <= maxPatchRetry && errors.IsConflict(err); i++ {
 		if i > triesBeforeBackOff {
 			p.backOff.Sleep(backOffPeriod)
@@ -552,7 +558,7 @@ func (p *patcher) patch(current runtime.Object, modified []byte, source, namespa
 		if getErr != nil {
 			return nil, getErr
 		}
-		patchBytes, err = p.patchSimple(current, modified, source, namespace, name)
+		patchBytes, err = p.patchSimple(current, modified, source, namespace, name, smPatchVersion)
 	}
 	if err != nil && p.force {
 		patchBytes, err = p.deleteAndCreate(modified, namespace, name)

pkg/kubectl/cmd/apply_test.go

@@ -188,6 +188,12 @@ func TestApplyObject(t *testing.T) {
 		NegotiatedSerializer: ns,
 		Client: fake.CreateHTTPClient(func(req *http.Request) (*http.Response, error) {
 			switch p, m := req.URL.Path, req.Method; {
+			case p == "/version" && m == "GET":
+				resp, err := genResponseWithJsonEncodedBody(serverVersion_1_5_0)
+				if err != nil {
+					t.Fatalf("error: failed to generate server version response: %#v\n", serverVersion_1_5_0)
+				}
+				return resp, nil
 			case p == pathRC && m == "GET":
 				bodyRC := ioutil.NopCloser(bytes.NewReader(currentRC))
 				return &http.Response{StatusCode: 200, Header: defaultHeader(), Body: bodyRC}, nil
@@ -202,6 +208,7 @@ func TestApplyObject(t *testing.T) {
 		}),
 	}
 	tf.Namespace = "test"
+	tf.ClientConfig = defaultClientConfig()
 	buf := bytes.NewBuffer([]byte{})
 
 	cmd := NewCmdApply(f, buf)
@@ -230,6 +237,12 @@ func TestApplyRetry(t *testing.T) {
 		NegotiatedSerializer: ns,
 		Client: fake.CreateHTTPClient(func(req *http.Request) (*http.Response, error) {
 			switch p, m := req.URL.Path, req.Method; {
+			case p == "/version" && m == "GET":
+				resp, err := genResponseWithJsonEncodedBody(serverVersion_1_5_0)
+				if err != nil {
+					t.Fatalf("error: failed to generate server version response: %#v\n", serverVersion_1_5_0)
+				}
+				return resp, nil
 			case p == pathRC && m == "GET":
 				getCount++
 				bodyRC := ioutil.NopCloser(bytes.NewReader(currentRC))
@@ -253,6 +266,7 @@ func TestApplyRetry(t *testing.T) {
 		}),
 	}
 	tf.Namespace = "test"
+	tf.ClientConfig = defaultClientConfig()
 	buf := bytes.NewBuffer([]byte{})
 
 	cmd := NewCmdApply(f, buf)
@@ -282,6 +296,12 @@ func TestApplyNonExistObject(t *testing.T) {
 		NegotiatedSerializer: ns,
 		Client: fake.CreateHTTPClient(func(req *http.Request) (*http.Response, error) {
 			switch p, m := req.URL.Path, req.Method; {
+			case p == "/version" && m == "GET":
+				resp, err := genResponseWithJsonEncodedBody(serverVersion_1_5_0)
+				if err != nil {
+					t.Fatalf("error: failed to generate server version response: %#v\n", serverVersion_1_5_0)
+				}
+				return resp, nil
 			case p == "/api/v1/namespaces/test" && m == "GET":
 				return &http.Response{StatusCode: 404, Header: defaultHeader(), Body: ioutil.NopCloser(bytes.NewReader(nil))}, nil
 			case p == pathNameRC && m == "GET":
@@ -296,6 +316,7 @@ func TestApplyNonExistObject(t *testing.T) {
 		}),
 	}
 	tf.Namespace = "test"
+	tf.ClientConfig = defaultClientConfig()
 	buf := bytes.NewBuffer([]byte{})
 
 	cmd := NewCmdApply(f, buf)
@@ -331,6 +352,12 @@ func testApplyMultipleObjects(t *testing.T, asList bool) {
 		NegotiatedSerializer: ns,
 		Client: fake.CreateHTTPClient(func(req *http.Request) (*http.Response, error) {
 			switch p, m := req.URL.Path, req.Method; {
+			case p == "/version" && m == "GET":
+				resp, err := genResponseWithJsonEncodedBody(serverVersion_1_5_0)
+				if err != nil {
+					t.Fatalf("error: failed to generate server version response: %#v\n", serverVersion_1_5_0)
+				}
+				return resp, nil
 			case p == pathRC && m == "GET":
 				bodyRC := ioutil.NopCloser(bytes.NewReader(currentRC))
 				return &http.Response{StatusCode: 200, Header: defaultHeader(), Body: bodyRC}, nil
@@ -352,6 +379,7 @@ func testApplyMultipleObjects(t *testing.T, asList bool) {
 		}),
 	}
 	tf.Namespace = "test"
+	tf.ClientConfig = defaultClientConfig()
 	buf := bytes.NewBuffer([]byte{})
 
 	cmd := NewCmdApply(f, buf)

pkg/kubectl/cmd/cmd_test.go

@@ -39,8 +39,13 @@ import (
 	cmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
 	"k8s.io/kubernetes/pkg/runtime"
 	"k8s.io/kubernetes/pkg/util/strings"
+	"k8s.io/kubernetes/pkg/version"
 )
 
+var serverVersion_1_5_0 = version.Info{
+	GitVersion: "v1.5.0",
+}
+
 func initTestErrorHandler(t *testing.T) {
 	cmdutil.BehaviorOnFatal(func(str string, code int) {
 		t.Errorf("Error running command (exit code %d): %s", code, str)

pkg/kubectl/cmd/edit.go

@@ -281,7 +281,7 @@ func runEdit(f cmdutil.Factory, out, errOut io.Writer, cmd *cobra.Command, args
 
 		switch editMode {
 		case NormalEditMode:
-			err = visitToPatch(originalObj, updates, mapper, resourceMapper, encoder, out, errOut, defaultVersion, &results, file)
+			err = visitToPatch(originalObj, updates, f, mapper, resourceMapper, encoder, out, errOut, defaultVersion, &results, file)
 		case EditBeforeCreateMode:
 			err = visitToCreate(updates, mapper, resourceMapper, out, errOut, defaultVersion, &results, file)
 		default:
@@ -393,9 +393,22 @@ func getMapperAndResult(f cmdutil.Factory, args []string, options *resource.File
 	return mapper, resourceMapper, r, cmdNamespace, err
 }
 
-func visitToPatch(originalObj runtime.Object, updates *resource.Info, mapper meta.RESTMapper, resourceMapper *resource.Mapper, encoder runtime.Encoder, out, errOut io.Writer, defaultVersion unversioned.GroupVersion, results *editResults, file string) error {
+func visitToPatch(originalObj runtime.Object, updates *resource.Info,
+	f cmdutil.Factory,
+	mapper meta.RESTMapper, resourceMapper *resource.Mapper,
+	encoder runtime.Encoder,
+	out, errOut io.Writer,
+	defaultVersion unversioned.GroupVersion,
+	results *editResults,
+	file string) error {
+
+	smPatchVersion, err := cmdutil.GetServerSupportedSMPatchVersionFromFactory(f)
+	if err != nil {
+		return err
+	}
+
 	patchVisitor := resource.NewFlattenListVisitor(updates, resourceMapper)
-	err := patchVisitor.Visit(func(info *resource.Info, incomingErr error) error {
+	err = patchVisitor.Visit(func(info *resource.Info, incomingErr error) error {
 		currOriginalObj := originalObj
 
 		// if we're editing a list, then navigate the list to find the item that we're currently trying to edit
@@ -456,7 +469,7 @@ func visitToPatch(originalObj runtime.Object, updates *resource.Info, mapper met
 
 		preconditions := []strategicpatch.PreconditionFunc{strategicpatch.RequireKeyUnchanged("apiVersion"),
 			strategicpatch.RequireKeyUnchanged("kind"), strategicpatch.RequireMetadataKeyUnchanged("name")}
-		patch, err := strategicpatch.CreateTwoWayMergePatch(originalJS, editedJS, currOriginalObj, preconditions...)
+		patch, err := strategicpatch.CreateTwoWayMergePatch(originalJS, editedJS, currOriginalObj, smPatchVersion, preconditions...)
 		if err != nil {
 			glog.V(4).Infof("Unable to calculate diff, no merge is possible: %v", err)
 			if strategicpatch.IsPreconditionFailed(err) {