Merge pull request #31220 from mfojtik/pause-patch

Automatic merge from submit-queue

Use PATCH to resume/pause deployment

Alternative version of: https://github.com/kubernetes/kubernetes/pull/30919
Fixes: https://github.com/kubernetes/kubernetes/issues/20437

hack/make-rules/test-cmd.sh

@@ -1580,15 +1580,15 @@ __EOF__
   ## Attempt to pause the replication controllers recursively
   output_message=$(! kubectl rollout pause -f hack/testdata/recursive/rc --recursive 2>&1 "${kube_flags[@]}")
   # Post-condition: busybox0 & busybox1 should error as they are RC's, and since busybox2 is malformed, it should error
-  kube::test::if_has_string "${output_message}" 'error when pausing "hack/testdata/recursive/rc/busybox.yaml'
-  kube::test::if_has_string "${output_message}" 'error when pausing "hack/testdata/recursive/rc/rc/busybox.yaml'
   kube::test::if_has_string "${output_message}" "Object 'Kind' is missing"
+  kube::test::if_has_string "${output_message}" 'replicationcontrollers "busybox0" pausing is not supported'
+  kube::test::if_has_string "${output_message}" 'replicationcontrollers "busybox1" pausing is not supported'
   ## Attempt to resume the replication controllers recursively
   output_message=$(! kubectl rollout resume -f hack/testdata/recursive/rc --recursive 2>&1 "${kube_flags[@]}")
   # Post-condition: busybox0 & busybox1 should error as they are RC's, and since busybox2 is malformed, it should error
-  kube::test::if_has_string "${output_message}" 'error when resuming "hack/testdata/recursive/rc/busybox.yaml'
-  kube::test::if_has_string "${output_message}" 'error when resuming "hack/testdata/recursive/rc/rc/busybox.yaml'
   kube::test::if_has_string "${output_message}" "Object 'Kind' is missing"
+  kube::test::if_has_string "${output_message}" 'replicationcontrollers "busybox0" resuming is not supported'
+  kube::test::if_has_string "${output_message}" 'replicationcontrollers "busybox0" resuming is not supported'
   # Clean up
   ! kubectl delete -f hack/testdata/recursive/rc --recursive "${kube_flags[@]}" --grace-period=0
   sleep 1

pkg/kubectl/cmd/rollout/BUILD

@@ -22,8 +22,10 @@ go_library(
     ],
     tags = ["automanaged"],
     deps = [
+        "//pkg/api:go_default_library",
         "//pkg/api/meta:go_default_library",
         "//pkg/kubectl:go_default_library",
+        "//pkg/kubectl/cmd/set:go_default_library",
         "//pkg/kubectl/cmd/templates:go_default_library",
         "//pkg/kubectl/cmd/util:go_default_library",
         "//pkg/kubectl/resource:go_default_library",

pkg/kubectl/cmd/rollout/rollout_pause.go

@@ -17,12 +17,15 @@ limitations under the License.
 package rollout
 
 import (
+	"fmt"
 	"io"
 
 	"github.com/spf13/cobra"
 
+	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/meta"
 	"k8s.io/kubernetes/pkg/kubectl"
+	"k8s.io/kubernetes/pkg/kubectl/cmd/set"
 	"k8s.io/kubernetes/pkg/kubectl/cmd/templates"
 	cmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
 	"k8s.io/kubernetes/pkg/kubectl/resource"
@@ -35,10 +38,11 @@ import (
 type PauseConfig struct {
 	resource.FilenameOptions
 
-	PauseObject func(object runtime.Object) (bool, error)
+	Pauser  func(info *resource.Info) (bool, error)
-	Mapper      meta.RESTMapper
+	Mapper  meta.RESTMapper
-	Typer       runtime.ObjectTyper
+	Typer   runtime.ObjectTyper
-	Infos       []*resource.Info
+	Encoder runtime.Encoder
+	Infos   []*resource.Info
 
 	Out io.Writer
 }
@@ -96,7 +100,9 @@ func (o *PauseConfig) CompletePause(f cmdutil.Factory, cmd *cobra.Command, out i
 	}
 
 	o.Mapper, o.Typer = f.Object()
-	o.PauseObject = f.PauseObject
+	o.Encoder = f.JSONEncoder()
+
+	o.Pauser = f.Pauser
 	o.Out = out
 
 	cmdNamespace, enforceNamespace, err := f.DefaultNamespace()
@@ -126,17 +132,27 @@ func (o *PauseConfig) CompletePause(f cmdutil.Factory, cmd *cobra.Command, out i
 
 func (o PauseConfig) RunPause() error {
 	allErrs := []error{}
-	for _, info := range o.Infos {
+	for _, patch := range set.CalculatePatches(o.Infos, o.Encoder, o.Pauser) {
-		isAlreadyPaused, err := o.PauseObject(info.Object)
+		info := patch.Info
-		if err != nil {
+		if patch.Err != nil {
-			allErrs = append(allErrs, cmdutil.AddSourceToErr("pausing", info.Source, err))
+			allErrs = append(allErrs, fmt.Errorf("error: %s %q %v", info.Mapping.Resource, info.Name, patch.Err))
 			continue
 		}
-		if isAlreadyPaused {
+
+		if string(patch.Patch) == "{}" || len(patch.Patch) == 0 {
 			cmdutil.PrintSuccess(o.Mapper, false, o.Out, info.Mapping.Resource, info.Name, false, "already paused")
 			continue
 		}
+
+		obj, err := resource.NewHelper(info.Client, info.Mapping).Patch(info.Namespace, info.Name, api.StrategicMergePatchType, patch.Patch)
+		if err != nil {
+			allErrs = append(allErrs, fmt.Errorf("failed to patch: %v", err))
+			continue
+		}
+
+		info.Refresh(obj, true)
 		cmdutil.PrintSuccess(o.Mapper, false, o.Out, info.Mapping.Resource, info.Name, false, "paused")
 	}
+
 	return utilerrors.NewAggregate(allErrs)
 }

pkg/kubectl/cmd/rollout/rollout_resume.go

@@ -17,12 +17,15 @@ limitations under the License.
 package rollout
 
 import (
+	"fmt"
 	"io"
 
 	"github.com/spf13/cobra"
 
+	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/meta"
 	"k8s.io/kubernetes/pkg/kubectl"
+	"k8s.io/kubernetes/pkg/kubectl/cmd/set"
 	"k8s.io/kubernetes/pkg/kubectl/cmd/templates"
 	cmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
 	"k8s.io/kubernetes/pkg/kubectl/resource"
@@ -35,10 +38,11 @@ import (
 type ResumeConfig struct {
 	resource.FilenameOptions
 
-	ResumeObject func(object runtime.Object) (bool, error)
+	Resumer func(object *resource.Info) (bool, error)
-	Mapper       meta.RESTMapper
+	Mapper  meta.RESTMapper
-	Typer        runtime.ObjectTyper
+	Typer   runtime.ObjectTyper
-	Infos        []*resource.Info
+	Encoder runtime.Encoder
+	Infos   []*resource.Info
 
 	Out io.Writer
 }
@@ -94,7 +98,9 @@ func (o *ResumeConfig) CompleteResume(f cmdutil.Factory, cmd *cobra.Command, out
 	}
 
 	o.Mapper, o.Typer = f.Object()
-	o.ResumeObject = f.ResumeObject
+	o.Encoder = f.JSONEncoder()
+
+	o.Resumer = f.Resumer
 	o.Out = out
 
 	cmdNamespace, enforceNamespace, err := f.DefaultNamespace()
@@ -130,17 +136,28 @@ func (o *ResumeConfig) CompleteResume(f cmdutil.Factory, cmd *cobra.Command, out
 
 func (o ResumeConfig) RunResume() error {
 	allErrs := []error{}
-	for _, info := range o.Infos {
+	for _, patch := range set.CalculatePatches(o.Infos, o.Encoder, o.Resumer) {
-		isAlreadyResumed, err := o.ResumeObject(info.Object)
+		info := patch.Info
-		if err != nil {
+
-			allErrs = append(allErrs, cmdutil.AddSourceToErr("resuming", info.Source, err))
+		if patch.Err != nil {
+			allErrs = append(allErrs, fmt.Errorf("error: %s %q %v", info.Mapping.Resource, info.Name, patch.Err))
 			continue
 		}
-		if isAlreadyResumed {
+
+		if string(patch.Patch) == "{}" || len(patch.Patch) == 0 {
 			cmdutil.PrintSuccess(o.Mapper, false, o.Out, info.Mapping.Resource, info.Name, false, "already resumed")
 			continue
 		}
+
+		obj, err := resource.NewHelper(info.Client, info.Mapping).Patch(info.Namespace, info.Name, api.StrategicMergePatchType, patch.Patch)
+		if err != nil {
+			allErrs = append(allErrs, fmt.Errorf("failed to patch: %v", err))
+			continue
+		}
+
+		info.Refresh(obj, true)
 		cmdutil.PrintSuccess(o.Mapper, false, o.Out, info.Mapping.Resource, info.Name, false, "resumed")
 	}
+
 	return utilerrors.NewAggregate(allErrs)
 }

pkg/kubectl/cmd/set/helper.go

@@ -125,13 +125,19 @@ func CalculatePatches(infos []*resource.Info, encoder runtime.Encoder, mutateFn
 	for _, info := range infos {
 		patch := &Patch{Info: info}
 		patch.Before, patch.Err = runtime.Encode(encoder, info.Object)
-
+		if patch.Err != nil {
-		ok, err := mutateFn(info)
+			patches = append(patches, patch)
-		if !ok {
 			continue
 		}
+
+		ok, err := mutateFn(info)
 		if err != nil {
 			patch.Err = err
+			patches = append(patches, patch)
+			continue
+		}
+		if !ok {
+			continue
 		}
 		patches = append(patches, patch)
 		if patch.Err != nil {

pkg/kubectl/cmd/testing/fake.go

@@ -261,11 +261,11 @@ func (f *FakeFactory) LogsForObject(object, options runtime.Object) (*restclient
 	return nil, nil
 }
 
-func (f *FakeFactory) PauseObject(runtime.Object) (bool, error) {
+func (f *FakeFactory) Pauser(info *resource.Info) (bool, error) {
 	return false, nil
 }
 
-func (f *FakeFactory) ResumeObject(runtime.Object) (bool, error) {
+func (f *FakeFactory) Resumer(info *resource.Info) (bool, error) {
 	return false, nil
 }
 

pkg/kubectl/cmd/util/factory.go

@@ -129,10 +129,10 @@ type Factory interface {
 	LabelsForObject(object runtime.Object) (map[string]string, error)
 	// LogsForObject returns a request for the logs associated with the provided object
 	LogsForObject(object, options runtime.Object) (*restclient.Request, error)
-	// PauseObject marks the provided object as paused ie. it will not be reconciled by its controller.
+	// Pauser marks the object in the info as paused ie. it will not be reconciled by its controller.
-	PauseObject(object runtime.Object) (bool, error)
+	Pauser(info *resource.Info) (bool, error)
-	// ResumeObject resumes a paused object ie. it will be reconciled by its controller.
+	// Resumer resumes a paused object inside the info ie. it will be reconciled by its controller.
-	ResumeObject(object runtime.Object) (bool, error)
+	Resumer(info *resource.Info) (bool, error)
 	// Returns a schema that can validate objects stored on disk.
 	Validator(validate bool, cacheDir string) (validation.Schema, error)
 	// SwaggerSchema returns the schema declaration for the provided group version kind.
@@ -644,49 +644,29 @@ func (f *factory) LogsForObject(object, options runtime.Object) (*restclient.Req
 	}
 }
 
-func (f *factory) PauseObject(object runtime.Object) (bool, error) {
+func (f *factory) Pauser(info *resource.Info) (bool, error) {
-	clientset, err := f.clients.ClientSetForVersion(nil)
+	switch obj := info.Object.(type) {
-	if err != nil {
-		return false, err
-	}
-
-	switch t := object.(type) {
 	case *extensions.Deployment:
-		if t.Spec.Paused {
+		if obj.Spec.Paused {
-			return true, nil
+			return true, errors.New("is already paused")
 		}
-		t.Spec.Paused = true
+		obj.Spec.Paused = true
-		_, err := clientset.Extensions().Deployments(t.Namespace).Update(t)
+		return true, nil
-		return false, err
 	default:
-		gvks, _, err := api.Scheme.ObjectKinds(object)
+		return false, fmt.Errorf("pausing is not supported")
-		if err != nil {
-			return false, err
-		}
-		return false, fmt.Errorf("cannot pause %v", gvks[0])
 	}
 }
 
-func (f *factory) ResumeObject(object runtime.Object) (bool, error) {
+func (f *factory) Resumer(info *resource.Info) (bool, error) {
-	clientset, err := f.clients.ClientSetForVersion(nil)
+	switch obj := info.Object.(type) {
-	if err != nil {
-		return false, err
-	}
-
-	switch t := object.(type) {
 	case *extensions.Deployment:
-		if !t.Spec.Paused {
+		if !obj.Spec.Paused {
-			return true, nil
+			return true, errors.New("is not paused")
 		}
-		t.Spec.Paused = false
+		obj.Spec.Paused = false
-		_, err := clientset.Extensions().Deployments(t.Namespace).Update(t)
+		return true, nil
-		return false, err
 	default:
-		gvks, _, err := api.Scheme.ObjectKinds(object)
+		return false, fmt.Errorf("resuming is not supported")
-		if err != nil {
-			return false, err
-		}
-		return false, fmt.Errorf("cannot resume %v", gvks[0])
 	}
 }