Added version check between patch and live object in server side apply

What is the problem being solved?
https://github.com/kubernetes/kubernetes/pull/75135
Currently version compatibility is not being checked in server side apply between the patch object and the live object.  This is causing a merge that will error to be run and the apiserver returns a 500 error.  The request should fail if the apiVersion provided in the object is different from the apiVersion in the url, but it should fail before trying to merge, and be a 4xx error. Probably a bad request error.

Why is this the best approach?
The approach of serializing the patch byte array and then checking for version equality with the already serialized live object is the simplest and most straightforward solution.

Kubernetes-commit: d5bd17cda0c134e5ef5c03c3eac79a9ce4e18003

pkg/endpoints/handlers/fieldmanager/fieldmanager.go

@@ -20,14 +20,17 @@ import (
 	"fmt"
 	"time"
 
+	"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"
 	"k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager/internal"
 	openapiproto "k8s.io/kube-openapi/pkg/util/proto"
 	"sigs.k8s.io/structured-merge-diff/fieldpath"
 	"sigs.k8s.io/structured-merge-diff/merge"
+	"sigs.k8s.io/yaml"
 )
 
 // FieldManager updates the managed fields and merge applied
@@ -149,9 +152,20 @@ func (f *FieldManager) Apply(liveObj runtime.Object, patch []byte, fieldManager
 	if err != nil {
 		return nil, fmt.Errorf("failed to decode managed fields: %v", err)
 	}
-	// We can assume that patchObj is already on the proper version:
-	// it shouldn't have to be converted so that it's not defaulted.
-	// TODO (jennybuckley): Explicitly checkt that patchObj is in the proper version.
+	// Check that the patch object has the same version as the live object
+	patchObj := &unstructured.Unstructured{Object: map[string]interface{}{}}
+
+	if err := yaml.Unmarshal(patch, &patchObj.Object); err != nil {
+		return nil, fmt.Errorf("error decoding YAML: %v", err)
+	}
+	if patchObj.GetAPIVersion() != f.groupVersion.String() {
+		return nil,
+			errors.NewBadRequest(
+				fmt.Sprintf("Incorrect version specified in apply patch. "+
+					"Specified patch version: %s, expected: %s",
+					patchObj.GetAPIVersion(), f.groupVersion.String()))
+	}
+
 	liveObjVersioned, err := f.toVersioned(liveObj)
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert live object to proper version: %v", err)

pkg/endpoints/handlers/fieldmanager/fieldmanager_test.go

@@ -18,9 +18,11 @@ package fieldmanager_test
 
 import (
 	"errors"
+	"net/http"
 	"testing"
 
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -72,7 +74,7 @@ func TestApplyStripsFields(t *testing.T) {
 	obj := &corev1.Pod{}
 
 	newObj, err := f.Apply(obj, []byte(`{
-		"apiVersion": "v1",
+		"apiVersion": "apps/v1",
 		"kind": "Deployment",
 		"metadata": {
 			"name": "b",
@@ -85,7 +87,7 @@ func TestApplyStripsFields(t *testing.T) {
 			"managedFields": [{
 					"manager": "apply",
 					"operation": "Apply",
-					"apiVersion": "v1",
+					"apiVersion": "apps/v1",
 					"fields": {
 						"f:metadata": {
 							"f:labels": {
@@ -111,13 +113,46 @@ func TestApplyStripsFields(t *testing.T) {
 	}
 }
 
+func TestVersionCheck(t *testing.T) {
+	f := NewTestFieldManager(t)
+
+	obj := &corev1.Pod{}
+
+	// patch has 'apiVersion: apps/v1' and live version is apps/v1 -> no errors
+	_, err := f.Apply(obj, []byte(`{
+		"apiVersion": "apps/v1",
+		"kind": "Deployment",
+	}`), "fieldmanager_test", false)
+	if err != nil {
+		t.Fatalf("failed to apply object: %v", err)
+	}
+
+	// patch has 'apiVersion: apps/v2' but live version is apps/v1 -> error
+	_, err = f.Apply(obj, []byte(`{
+		"apiVersion": "apps/v2",
+		"kind": "Deployment",
+	}`), "fieldmanager_test", false)
+	if err == nil {
+		t.Fatalf("expected an error from mismatched patch and live versions")
+	}
+	switch typ := err.(type) {
+	default:
+		t.Fatalf("expected error to be of type %T was %T", apierrors.StatusError{}, typ)
+	case apierrors.APIStatus:
+		if typ.Status().Code != http.StatusBadRequest {
+			t.Fatalf("expected status code to be %d but was %d",
+				http.StatusBadRequest, typ.Status().Code)
+		}
+	}
+}
+
 func TestApplyDoesNotStripLabels(t *testing.T) {
 	f := NewTestFieldManager(t)
 
 	obj := &corev1.Pod{}
 
 	newObj, err := f.Apply(obj, []byte(`{
-		"apiVersion": "v1",
+		"apiVersion": "apps/v1",
 		"kind": "Pod",
 		"metadata": {
 			"labels": {