support openapi in apply

pkg/kubectl/cmd/BUILD

@@ -86,6 +86,7 @@ go_library(
         "//pkg/kubectl/cmd/templates:go_default_library",
         "//pkg/kubectl/cmd/util:go_default_library",
         "//pkg/kubectl/cmd/util/editor:go_default_library",
+        "//pkg/kubectl/cmd/util/openapi:go_default_library",
         "//pkg/kubectl/explain:go_default_library",
         "//pkg/kubectl/metricsutil:go_default_library",
         "//pkg/kubectl/plugins:go_default_library",
@@ -147,6 +148,7 @@ go_library(
         "//vendor/k8s.io/client-go/tools/portforward:go_default_library",
         "//vendor/k8s.io/client-go/tools/remotecommand:go_default_library",
         "//vendor/k8s.io/client-go/transport/spdy:go_default_library",
+        "//vendor/k8s.io/kube-openapi/pkg/util/proto:go_default_library",
         "//vendor/k8s.io/utils/exec:go_default_library",
     ],
 )
@@ -214,6 +216,7 @@ go_test(
         "//pkg/kubectl:go_default_library",
         "//pkg/kubectl/cmd/testing:go_default_library",
         "//pkg/kubectl/cmd/util:go_default_library",
+        "//pkg/kubectl/cmd/util/openapi:go_default_library",
         "//pkg/kubectl/plugins:go_default_library",
         "//pkg/kubectl/resource:go_default_library",
         "//pkg/kubectl/scheme:go_default_library",
@@ -241,6 +244,7 @@ go_test(
         "//vendor/k8s.io/apimachinery/pkg/util/intstr:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/strategicpatch:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/strategicpatch/testing:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//vendor/k8s.io/client-go/dynamic:go_default_library",
         "//vendor/k8s.io/client-go/rest:go_default_library",

pkg/kubectl/cmd/apply.go

@@ -37,11 +37,13 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
 	"k8s.io/apimachinery/pkg/util/wait"
+	oapi "k8s.io/kube-openapi/pkg/util/proto"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	"k8s.io/kubernetes/pkg/kubectl"
 	"k8s.io/kubernetes/pkg/kubectl/cmd/templates"
 	cmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
+	"k8s.io/kubernetes/pkg/kubectl/cmd/util/openapi"
 	"k8s.io/kubernetes/pkg/kubectl/resource"
 	"k8s.io/kubernetes/pkg/kubectl/scheme"
 	"k8s.io/kubernetes/pkg/kubectl/util/i18n"
@@ -127,6 +129,7 @@ func NewCmdApply(baseName string, f cmdutil.Factory, out, errOut io.Writer) *cob
 	cmd.Flags().StringVarP(&options.Selector, "selector", "l", "", "Selector (label query) to filter on, supports '=', '==', and '!='.(e.g. -l key1=value1,key2=value2)")
 	cmd.Flags().Bool("all", false, "Select all resources in the namespace of the specified resource types.")
 	cmd.Flags().StringArray("prune-whitelist", []string{}, "Overwrite the default whitelist with <group/version/kind> for --prune")
+	cmd.Flags().Bool("openapi-patch", true, "If true, use openapi to calculate diff when the openapi presents and the resource can be found in the openapi spec. Otherwise, fall back to use baked-in types.")
 	cmdutil.AddDryRunFlag(cmd)
 	cmdutil.AddPrinterFlags(cmd)
 	cmdutil.AddRecordFlag(cmd)
@@ -193,6 +196,14 @@ func RunApply(f cmdutil.Factory, cmd *cobra.Command, out, errOut io.Writer, opti
 		return err
 	}
 
+	var openapiSchema openapi.Resources
+	if cmdutil.GetFlagBool(cmd, "openapi-patch") {
+		openapiSchema, err = f.OpenAPISchema()
+		if err != nil {
+			openapiSchema = nil
+		}
+	}
+
 	cmdNamespace, enforceNamespace, err := f.DefaultNamespace()
 	if err != nil {
 		return err
@@ -312,9 +323,10 @@ func RunApply(f cmdutil.Factory, cmd *cobra.Command, out, errOut io.Writer, opti
 				cascade:       options.Cascade,
 				timeout:       options.Timeout,
 				gracePeriod:   options.GracePeriod,
+				openapiSchema: openapiSchema,
 			}
 
-			patchBytes, patchedObject, err := patcher.patch(info.Object, modified, info.Source, info.Namespace, info.Name)
+			patchBytes, patchedObject, err := patcher.patch(info.Object, modified, info.Source, info.Namespace, info.Name, errOut)
 			if err != nil {
 				return cmdutil.AddSourceToErr(fmt.Sprintf("applying patch:\n%s\nto:\n%v\nfor:", patchBytes, info), info.Source, err)
 			}
@@ -570,9 +582,11 @@ type patcher struct {
 	cascade     bool
 	timeout     time.Duration
 	gracePeriod int
+
+	openapiSchema openapi.Resources
 }
 
-func (p *patcher) patchSimple(obj runtime.Object, modified []byte, source, namespace, name string) ([]byte, runtime.Object, error) {
+func (p *patcher) patchSimple(obj runtime.Object, modified []byte, source, namespace, name string, errOut io.Writer) ([]byte, runtime.Object, error) {
 	// Serialize the current configuration of the object from the server.
 	current, err := runtime.Encode(p.encoder, obj)
 	if err != nil {
@@ -585,12 +599,29 @@ func (p *patcher) patchSimple(obj runtime.Object, modified []byte, source, names
 		return nil, nil, cmdutil.AddSourceToErr(fmt.Sprintf("retrieving original configuration from:\n%v\nfor:", obj), source, err)
 	}
 
+	var patchType types.PatchType
+	var patch []byte
+	var lookupPatchMeta strategicpatch.LookupPatchMeta
+	var schema oapi.Schema
+	createPatchErrFormat := "creating patch with:\noriginal:\n%s\nmodified:\n%s\ncurrent:\n%s\nfor:"
+	// Try to use openapi first if the openapi spec is available and can successfully calculate the patch.
+	// Otherwise, fall back to baked-in types.
+	if p.openapiSchema != nil {
+		if schema = p.openapiSchema.LookupResource(p.mapping.GroupVersionKind); schema != nil {
+			lookupPatchMeta = strategicpatch.PatchMetaFromOpenAPI{Schema: schema}
+			if openapiPatch, err := strategicpatch.CreateThreeWayMergePatch(original, modified, current, lookupPatchMeta, p.overwrite); err != nil {
+				fmt.Fprintf(errOut, "warning: error calculating patch from openapi spec: %v\n", err)
+			} else {
+				patchType = types.StrategicMergePatchType
+				patch = openapiPatch
+			}
+		}
+	}
+
+	if patch == nil {
 		// Create the versioned struct from the type defined in the restmapping
 		// (which is the API version we'll be submitting the patch to)
 		versionedObject, err := scheme.Scheme.New(p.mapping.GroupVersionKind)
-	var patchType types.PatchType
-	var patch []byte
-	createPatchErrFormat := "creating patch with:\noriginal:\n%s\nmodified:\n%s\ncurrent:\n%s\nfor:"
 		switch {
 		case runtime.IsNotRegisteredError(err):
 			// fall back to generic JSON merge patch
@@ -609,10 +640,15 @@ func (p *patcher) patchSimple(obj runtime.Object, modified []byte, source, names
 		case err == nil:
 			// Compute a three way strategic merge patch to send to server.
 			patchType = types.StrategicMergePatchType
-		patch, err = strategicpatch.CreateThreeWayMergePatch(original, modified, current, versionedObject, p.overwrite)
+			lookupPatchMeta, err = strategicpatch.NewPatchMetaFromStruct(versionedObject)
 			if err != nil {
 				return nil, nil, cmdutil.AddSourceToErr(fmt.Sprintf(createPatchErrFormat, original, modified, current), source, err)
 			}
+			patch, err = strategicpatch.CreateThreeWayMergePatch(original, modified, current, lookupPatchMeta, p.overwrite)
+			if err != nil {
+				return nil, nil, cmdutil.AddSourceToErr(fmt.Sprintf(createPatchErrFormat, original, modified, current), source, err)
+			}
+		}
 	}
 
 	if string(patch) == "{}" {
@@ -623,9 +659,9 @@ func (p *patcher) patchSimple(obj runtime.Object, modified []byte, source, names
 	return patch, patchedObj, err
 }
 
-func (p *patcher) patch(current runtime.Object, modified []byte, source, namespace, name string) ([]byte, runtime.Object, error) {
+func (p *patcher) patch(current runtime.Object, modified []byte, source, namespace, name string, errOut io.Writer) ([]byte, runtime.Object, error) {
 	var getErr error
-	patchBytes, patchObject, err := p.patchSimple(current, modified, source, namespace, name)
+	patchBytes, patchObject, err := p.patchSimple(current, modified, source, namespace, name, errOut)
 	for i := 1; i <= maxPatchRetry && errors.IsConflict(err); i++ {
 		if i > triesBeforeBackOff {
 			p.backOff.Sleep(backOffPeriod)
@@ -634,7 +670,7 @@ func (p *patcher) patch(current runtime.Object, modified []byte, source, namespa
 		if getErr != nil {
 			return nil, nil, getErr
 		}
-		patchBytes, patchObject, err = p.patchSimple(current, modified, source, namespace, name)
+		patchBytes, patchObject, err = p.patchSimple(current, modified, source, namespace, name, errOut)
 	}
 	if err != nil && p.force {
 		patchBytes, patchObject, err = p.deleteAndCreate(modified, namespace, name)

pkg/kubectl/cmd/apply_test.go

@@ -19,10 +19,12 @@ package cmd
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"net/http"
 	"os"
+	"path/filepath"
 	"strings"
 	"testing"
 
@@ -33,15 +35,33 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	sptest "k8s.io/apimachinery/pkg/util/strategicpatch/testing"
+	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/rest/fake"
 	"k8s.io/kubernetes/pkg/api/testapi"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/apis/extensions"
 	cmdtesting "k8s.io/kubernetes/pkg/kubectl/cmd/testing"
 	cmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
+	"k8s.io/kubernetes/pkg/kubectl/cmd/util/openapi"
 	"k8s.io/kubernetes/pkg/printers"
 )
 
+var (
+	fakeSchema                 = sptest.Fake{Path: filepath.Join("..", "..", "..", "api", "openapi-spec", "swagger.json")}
+	testingOpenAPISchemaFns    = []func() (openapi.Resources, error){nil, AlwaysErrorOpenAPISchemaFn, openAPISchemaFn}
+	AlwaysErrorOpenAPISchemaFn = func() (openapi.Resources, error) {
+		return nil, errors.New("cannot get openapi spec")
+	}
+	openAPISchemaFn = func() (openapi.Resources, error) {
+		s, err := fakeSchema.OpenAPISchema()
+		if err != nil {
+			return nil, err
+		}
+		return openapi.NewOpenAPIData(s)
+	}
+)
+
 func TestApplyExtraArgsFail(t *testing.T) {
 	buf := bytes.NewBuffer([]byte{})
 	errBuf := bytes.NewBuffer([]byte{})
@@ -410,6 +430,7 @@ func TestApplyObject(t *testing.T) {
 	nameRC, currentRC := readAndAnnotateReplicationController(t, filenameRC)
 	pathRC := "/namespaces/test/replicationcontrollers/" + nameRC
 
+	for _, fn := range testingOpenAPISchemaFns {
 		f, tf, _, _ := cmdtesting.NewAPIFactory()
 		tf.Printer = &testPrinter{}
 		tf.UnstructuredClient = &fake.RESTClient{
@@ -429,6 +450,7 @@ func TestApplyObject(t *testing.T) {
 				}
 			}),
 		}
+		tf.OpenAPISchemaFunc = fn
 		tf.Namespace = "test"
 		buf := bytes.NewBuffer([]byte{})
 		errBuf := bytes.NewBuffer([]byte{})
@@ -443,6 +465,10 @@ func TestApplyObject(t *testing.T) {
 		if buf.String() != expectRC {
 			t.Fatalf("unexpected output: %s\nexpected: %s", buf.String(), expectRC)
 		}
+		if errBuf.String() != "" {
+			t.Fatalf("unexpected error output: %s", errBuf.String())
+		}
+	}
 }
 
 func TestApplyObjectOutput(t *testing.T) {
@@ -466,6 +492,7 @@ func TestApplyObjectOutput(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	for _, fn := range testingOpenAPISchemaFns {
 		f, tf, _, _ := cmdtesting.NewAPIFactory()
 		tf.Printer = &printers.YAMLPrinter{}
 		tf.UnstructuredClient = &fake.RESTClient{
@@ -485,6 +512,7 @@ func TestApplyObjectOutput(t *testing.T) {
 				}
 			}),
 		}
+		tf.OpenAPISchemaFunc = fn
 		tf.Namespace = "test"
 		buf := bytes.NewBuffer([]byte{})
 		errBuf := bytes.NewBuffer([]byte{})
@@ -500,6 +528,10 @@ func TestApplyObjectOutput(t *testing.T) {
 		if !strings.Contains(buf.String(), "post-patch: value") {
 			t.Fatalf("unexpected output: %s\nexpected to contain: %s", buf.String(), "post-patch: value")
 		}
+		if errBuf.String() != "" {
+			t.Fatalf("unexpected error output: %s", errBuf.String())
+		}
+	}
 }
 
 func TestApplyRetry(t *testing.T) {
@@ -507,6 +539,7 @@ func TestApplyRetry(t *testing.T) {
 	nameRC, currentRC := readAndAnnotateReplicationController(t, filenameRC)
 	pathRC := "/namespaces/test/replicationcontrollers/" + nameRC
 
+	for _, fn := range testingOpenAPISchemaFns {
 		firstPatch := true
 		retry := false
 		getCount := 0
@@ -538,6 +571,7 @@ func TestApplyRetry(t *testing.T) {
 				}
 			}),
 		}
+		tf.OpenAPISchemaFunc = fn
 		tf.Namespace = "test"
 		buf := bytes.NewBuffer([]byte{})
 		errBuf := bytes.NewBuffer([]byte{})
@@ -556,6 +590,10 @@ func TestApplyRetry(t *testing.T) {
 		if buf.String() != expectRC {
 			t.Fatalf("unexpected output: %s\nexpected: %s", buf.String(), expectRC)
 		}
+		if errBuf.String() != "" {
+			t.Fatalf("unexpected error output: %s", errBuf.String())
+		}
+	}
 }
 
 func TestApplyNonExistObject(t *testing.T) {
@@ -682,6 +720,7 @@ func testApplyMultipleObjects(t *testing.T, asList bool) {
 	nameSVC, currentSVC := readAndAnnotateService(t, filenameSVC)
 	pathSVC := "/namespaces/test/services/" + nameSVC
 
+	for _, fn := range testingOpenAPISchemaFns {
 		f, tf, _, _ := cmdtesting.NewAPIFactory()
 		tf.Printer = &testPrinter{}
 		tf.UnstructuredClient = &fake.RESTClient{
@@ -708,6 +747,7 @@ func testApplyMultipleObjects(t *testing.T, asList bool) {
 				}
 			}),
 		}
+		tf.OpenAPISchemaFunc = fn
 		tf.Namespace = "test"
 		buf := bytes.NewBuffer([]byte{})
 		errBuf := bytes.NewBuffer([]byte{})
@@ -732,6 +772,10 @@ func testApplyMultipleObjects(t *testing.T, asList bool) {
 		if buf.String() != expectOne && buf.String() != expectTwo {
 			t.Fatalf("unexpected output: %s\nexpected: %s OR %s", buf.String(), expectOne, expectTwo)
 		}
+		if errBuf.String() != "" {
+			t.Fatalf("unexpected error output: %s", errBuf.String())
+		}
+	}
 }
 
 const (
@@ -760,7 +804,8 @@ func TestApplyNULLPreservation(t *testing.T) {
 	verifiedPatch := false
 	deploymentBytes := readDeploymentFromFile(t, filenameDeployObjServerside)
 
-	f, tf, _, _ := cmdtesting.NewTestFactory()
+	for _, fn := range testingOpenAPISchemaFns {
+		f, tf, _, _ := cmdtesting.NewAPIFactory()
 		tf.UnstructuredClient = &fake.RESTClient{
 			NegotiatedSerializer: unstructuredSerializer,
 			Client: fake.CreateHTTPClient(func(req *http.Request) (*http.Response, error) {
@@ -799,8 +844,8 @@ func TestApplyNULLPreservation(t *testing.T) {
 				}
 			}),
 		}
+		tf.OpenAPISchemaFunc = fn
 		tf.Namespace = "test"
-	tf.ClientConfig = defaultClientConfig()
 		buf := bytes.NewBuffer([]byte{})
 		errBuf := bytes.NewBuffer([]byte{})
 
@@ -814,9 +859,13 @@ func TestApplyNULLPreservation(t *testing.T) {
 		if buf.String() != expected {
 			t.Fatalf("unexpected output: %s\nexpected: %s", buf.String(), expected)
 		}
+		if errBuf.String() != "" {
+			t.Fatalf("unexpected error output: %s", errBuf.String())
+		}
 		if !verifiedPatch {
 			t.Fatal("No server-side patch call detected")
 		}
+	}
 }
 
 // TestUnstructuredApply checks apply operations on an unstructured object
@@ -827,6 +876,7 @@ func TestUnstructuredApply(t *testing.T) {
 
 	verifiedPatch := false
 
+	for _, fn := range testingOpenAPISchemaFns {
 		f, tf, _, _ := cmdtesting.NewAPIFactory()
 		tf.Printer = &testPrinter{}
 		tf.UnstructuredClient = &fake.RESTClient{
@@ -858,7 +908,7 @@ func TestUnstructuredApply(t *testing.T) {
 				}
 			}),
 		}
-
+		tf.OpenAPISchemaFunc = fn
 		tf.Namespace = "test"
 		buf := bytes.NewBuffer([]byte{})
 		errBuf := bytes.NewBuffer([]byte{})
@@ -872,9 +922,13 @@ func TestUnstructuredApply(t *testing.T) {
 		if buf.String() != expected {
 			t.Fatalf("unexpected output: %s\nexpected: %s", buf.String(), expected)
 		}
+		if errBuf.String() != "" {
+			t.Fatalf("unexpected error output: %s", errBuf.String())
+		}
 		if !verifiedPatch {
 			t.Fatal("No server-side patch call detected")
 		}
+	}
 }
 
 // TestUnstructuredIdempotentApply checks repeated apply operation on an unstructured object
@@ -890,6 +944,7 @@ func TestUnstructuredIdempotentApply(t *testing.T) {
 
 	verifiedPatch := false
 
+	for _, fn := range testingOpenAPISchemaFns {
 		f, tf, _, _ := cmdtesting.NewAPIFactory()
 		tf.Printer = &testPrinter{}
 		tf.UnstructuredClient = &fake.RESTClient{
@@ -944,7 +999,7 @@ func TestUnstructuredIdempotentApply(t *testing.T) {
 				}
 			}),
 		}
-
+		tf.OpenAPISchemaFunc = fn
 		tf.Namespace = "test"
 		buf := bytes.NewBuffer([]byte{})
 		errBuf := bytes.NewBuffer([]byte{})
@@ -958,9 +1013,13 @@ func TestUnstructuredIdempotentApply(t *testing.T) {
 		if buf.String() != expected {
 			t.Fatalf("unexpected output: %s\nexpected: %s", buf.String(), expected)
 		}
+		if errBuf.String() != "" {
+			t.Fatalf("unexpected error output: %s", errBuf.String())
+		}
 		if !verifiedPatch {
 			t.Fatal("No server-side patch call detected")
 		}
+	}
 }
 
 func TestRunApplySetLastApplied(t *testing.T) {
@@ -1093,8 +1152,6 @@ func TestForceApply(t *testing.T) {
 	nameRC, currentRC := readAndAnnotateReplicationController(t, filenameRC)
 	pathRC := "/namespaces/test/replicationcontrollers/" + nameRC
 	pathRCList := "/namespaces/test/replicationcontrollers"
-	deleted := false
-	counts := map[string]int{}
 	expected := map[string]int{
 		"getOk":       9,
 		"getNotFound": 1,
@@ -1105,6 +1162,9 @@ func TestForceApply(t *testing.T) {
 		"post":        1,
 	}
 
+	for _, fn := range testingOpenAPISchemaFns {
+		deleted := false
+		counts := map[string]int{}
 		f, tf, _, _ := cmdtesting.NewAPIFactory()
 		tf.Printer = &testPrinter{}
 		tf.UnstructuredClient = &fake.RESTClient{
@@ -1164,8 +1224,9 @@ func TestForceApply(t *testing.T) {
 				}
 			}),
 		}
+		tf.OpenAPISchemaFunc = fn
 		tf.Client = tf.UnstructuredClient
-	tf.ClientConfig = defaultClientConfig()
+		tf.ClientConfig = &restclient.Config{}
 		tf.Namespace = "test"
 		buf := bytes.NewBuffer([]byte{})
 		errBuf := bytes.NewBuffer([]byte{})
@@ -1185,4 +1246,8 @@ func TestForceApply(t *testing.T) {
 		if expected := "replicationcontroller/" + nameRC + "\n"; buf.String() != expected {
 			t.Fatalf("unexpected output: %s\nexpected: %s", buf.String(), expected)
 		}
+		if errBuf.String() != "" {
+			t.Fatalf("unexpected error output: %s", errBuf.String())
+		}
+	}
 }