Merge pull request #33075 from mikedanese/kubectl-prune

Automatic merge from submit-queue

kubectl: implement kubectl apply --prune

Closes https://github.com/kubernetes/kubernetes/issues/19805

Depends on #32599 

@errordeveloper @lukemarsden

hack/make-rules/test-cmd.sh

@@ -1052,6 +1052,29 @@ __EOF__
   kubectl delete pods selector-test-pod
 
 
+  ## kubectl apply --prune
+  # Pre-Condition: no POD exists
+  kube::test::get_object_assert pods "{{range.items}}{{$id_field}}:{{end}}" ''
+
+  # apply a
+  kubectl apply --prune -l prune-group=true -f hack/testdata/prune/a.yaml "${kube_flags[@]}"
+  # check right pod exists
+  kube::test::get_object_assert 'pods a' "{{${id_field}}}" 'a'
+  # check wrong pod doesn't exist
+  output_message=$(! kubectl get pods b 2>&1 "${kube_flags[@]}")
+  kube::test::if_has_string "${output_message}" 'pods "b" not found'
+
+  # apply b
+  kubectl apply --prune -l prune-group=true -f hack/testdata/prune/b.yaml "${kube_flags[@]}"
+  # check right pod exists
+  kube::test::get_object_assert 'pods b' "{{${id_field}}}" 'b'
+  # check wrong pod doesn't exist
+  output_message=$(! kubectl get pods a 2>&1 "${kube_flags[@]}")
+  kube::test::if_has_string "${output_message}" 'pods "a" not found'
+
+  # cleanup
+  kubectl delete pods b
+
   ## kubectl run should create deployments or jobs
   # Pre-Condition: no Job exists
   kube::test::get_object_assert jobs "{{range.items}}{{$id_field}}:{{end}}" ''

hack/testdata/prune/a.yaml

@@ -0,0 +1,10 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: a
+  labels:
+    prune-group: "true"
+spec:
+  containers:
+  - name: kubernetes-pause
+    image: gcr.io/google-containers/pause:2.0

hack/testdata/prune/b.yaml

@@ -0,0 +1,10 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: b
+  labels:
+    prune-group: "true"
+spec:
+  containers:
+  - name: kubernetes-pause
+    image: gcr.io/google-containers/pause:2.0

pkg/kubectl/cmd/apply.go

@@ -26,18 +26,26 @@ import (
 	"github.com/spf13/cobra"
 
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/annotations"
 	"k8s.io/kubernetes/pkg/api/errors"
 	"k8s.io/kubernetes/pkg/api/meta"
+	"k8s.io/kubernetes/pkg/api/unversioned"
+	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	"k8s.io/kubernetes/pkg/kubectl"
 	cmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
 	"k8s.io/kubernetes/pkg/kubectl/resource"
+	"k8s.io/kubernetes/pkg/labels"
 	"k8s.io/kubernetes/pkg/runtime"
+	"k8s.io/kubernetes/pkg/util/sets"
 	"k8s.io/kubernetes/pkg/util/strategicpatch"
 )
 
 type ApplyOptions struct {
 	FilenameOptions resource.FilenameOptions
 	Selector        string
+	Prune           bool
+	Cascade         bool
+	GracePeriod     int
 }
 
 const (
@@ -55,7 +63,9 @@ var (
 		This resource will be created if it doesn't exist yet.
 		To use 'apply', always create the resource initially with either 'apply' or 'create --save-config'.
 
-		JSON and YAML formats are accepted.`)
+		JSON and YAML formats are accepted.
+		
+		Alpha Disclaimer: the --prune functionality is not yet complete. Do not use unless you are aware of what the current state is. See https://issues.k8s.io/34274.`)
 
 	apply_example = dedent.Dedent(`
 		# Apply the configuration in pod.json to a pod.
@@ -76,6 +86,7 @@ func NewCmdApply(f *cmdutil.Factory, out io.Writer) *cobra.Command {
 		Run: func(cmd *cobra.Command, args []string) {
 			cmdutil.CheckErr(validateArgs(cmd, args))
 			cmdutil.CheckErr(cmdutil.ValidateOutputArgs(cmd))
+			cmdutil.CheckErr(validatePruneAll(options.Prune, cmdutil.GetFlagBool(cmd, "all"), options.Selector))
 			cmdutil.CheckErr(RunApply(f, cmd, out, &options))
 		},
 	}
@@ -84,8 +95,12 @@ func NewCmdApply(f *cmdutil.Factory, out io.Writer) *cobra.Command {
 	cmdutil.AddFilenameOptionFlags(cmd, &options.FilenameOptions, usage)
 	cmd.MarkFlagRequired("filename")
 	cmd.Flags().Bool("overwrite", true, "Automatically resolve conflicts between the modified and live configuration by using values from the modified configuration")
+	cmd.Flags().BoolVar(&options.Prune, "prune", false, "Automatically delete resource objects that do not appear in the configs")
+	cmd.Flags().BoolVar(&options.Cascade, "cascade", true, "Only relevant during a prune. If true, cascade the deletion of the resources managed by pruned resources (e.g. Pods created by a ReplicationController).")
+	cmd.Flags().IntVar(&options.GracePeriod, "grace-period", -1, "Period of time in seconds given to pruned resources to terminate gracefully. Ignored if negative.")
 	cmdutil.AddValidateFlags(cmd)
 	cmd.Flags().StringVarP(&options.Selector, "selector", "l", "", "Selector (label query) to filter on")
+	cmd.Flags().Bool("all", false, "[-all] to select all the specified resources.")
 	cmdutil.AddOutputFlagsForMutation(cmd)
 	cmdutil.AddRecordFlag(cmd)
 	cmdutil.AddInclude3rdPartyFlags(cmd)
@@ -100,6 +115,13 @@ func validateArgs(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
+func validatePruneAll(prune, all bool, selector string) error {
+	if prune && !all && selector == "" {
+		return fmt.Errorf("all resources selected for prune without explicitly passing --all. To prune all resources, pass the --all flag. If you did not mean to prune all resources, specify a label selector.")
+	}
+	return nil
+}
+
 func RunApply(f *cmdutil.Factory, cmd *cobra.Command, out io.Writer, options *ApplyOptions) error {
 	shortOutput := cmdutil.GetFlagString(cmd, "output") == "name"
 	schema, err := f.Validator(cmdutil.GetFlagBool(cmd, "validate"), cmdutil.GetFlagString(cmd, "schema-cache-dir"))
@@ -129,6 +151,11 @@ func RunApply(f *cmdutil.Factory, cmd *cobra.Command, out io.Writer, options *Ap
 	encoder := f.JSONEncoder()
 	decoder := f.Decoder(false)
 
+	visitedUids := sets.NewString()
+	visitedNamespaces := sets.NewString()
+	visitedNamespacedRESTMappings := map[unversioned.GroupVersionKind]*meta.RESTMapping{}
+	visitedNonNamespacedRESTMappings := map[unversioned.GroupVersionKind]*meta.RESTMapping{}
+
 	count := 0
 	err = r.Visit(func(info *resource.Info, err error) error {
 		// In this method, info.Object contains the object retrieved from the server
@@ -137,6 +164,13 @@ func RunApply(f *cmdutil.Factory, cmd *cobra.Command, out io.Writer, options *Ap
 			return err
 		}
 
+		if info.Namespaced() {
+			visitedNamespaces.Insert(info.Namespace)
+			visitedNamespacedRESTMappings[info.Mapping.GroupVersionKind] = info.Mapping
+		} else {
+			visitedNonNamespacedRESTMappings[info.Mapping.GroupVersionKind] = info.Mapping
+		}
+
 		// Get the modified configuration of the object. Embed the result
 		// as an annotation in the modified configuration, so that it will appear
 		// in the patch sent to the server.
@@ -165,6 +199,11 @@ func RunApply(f *cmdutil.Factory, cmd *cobra.Command, out io.Writer, options *Ap
 			if err := createAndRefresh(info); err != nil {
 				return cmdutil.AddSourceToErr("creating", info.Source, err)
 			}
+			if uid, err := info.Mapping.UID(info.Object); err != nil {
+				return err
+			} else {
+				visitedUids.Insert(string(uid))
+			}
 			count++
 			cmdutil.PrintSuccess(mapper, shortOutput, out, info.Mapping.Resource, info.Name, false, "created")
 			return nil
@@ -190,6 +229,11 @@ func RunApply(f *cmdutil.Factory, cmd *cobra.Command, out io.Writer, options *Ap
 			}
 		}
 
+		if uid, err := info.Mapping.UID(info.Object); err != nil {
+			return err
+		} else {
+			visitedUids.Insert(string(uid))
+		}
 		count++
 		cmdutil.PrintSuccess(mapper, shortOutput, out, info.Mapping.Resource, info.Name, false, "configured")
 		return nil
@@ -198,11 +242,123 @@ func RunApply(f *cmdutil.Factory, cmd *cobra.Command, out io.Writer, options *Ap
 	if err != nil {
 		return err
 	}
-
 	if count == 0 {
 		return fmt.Errorf("no objects passed to apply")
 	}
 
+	if !options.Prune {
+		return nil
+	}
+
+	selector, err := labels.Parse(options.Selector)
+	if err != nil {
+		return err
+	}
+	p := pruner{
+		mapper:        mapper,
+		clientFunc:    f.ClientForMapping,
+		clientsetFunc: f.ClientSet,
+
+		selector:    selector,
+		visitedUids: visitedUids,
+
+		cascade:     options.Cascade,
+		gracePeriod: options.GracePeriod,
+
+		out: out,
+	}
+	for n := range visitedNamespaces {
+		for _, m := range visitedNamespacedRESTMappings {
+			if err := p.prune(n, m, shortOutput); err != nil {
+				return fmt.Errorf("error pruning objects: %v", err)
+			}
+		}
+	}
+	for _, m := range visitedNonNamespacedRESTMappings {
+		if err := p.prune(api.NamespaceNone, m, shortOutput); err != nil {
+			return fmt.Errorf("error pruning objects: %v", err)
+		}
+	}
+
+	return nil
+}
+
+type pruner struct {
+	mapper        meta.RESTMapper
+	clientFunc    resource.ClientMapperFunc
+	clientsetFunc func() (*internalclientset.Clientset, error)
+
+	visitedUids sets.String
+	selector    labels.Selector
+
+	cascade     bool
+	gracePeriod int
+
+	out io.Writer
+}
+
+func (p *pruner) prune(namespace string, mapping *meta.RESTMapping, shortOutput bool) error {
+	c, err := p.clientFunc(mapping)
+	if err != nil {
+		return err
+	}
+
+	objList, err := resource.NewHelper(c, mapping).List(namespace, mapping.GroupVersionKind.Version, p.selector, false)
+	if err != nil {
+		return err
+	}
+	objs, err := meta.ExtractList(objList)
+	if err != nil {
+		return err
+	}
+
+	for _, obj := range objs {
+		annots, err := mapping.MetadataAccessor.Annotations(obj)
+		if err != nil {
+			return err
+		}
+		if _, ok := annots[annotations.LastAppliedConfigAnnotation]; !ok {
+			// don't prune resources not created with apply
+			continue
+		}
+		uid, err := mapping.UID(obj)
+		if err != nil {
+			return err
+		}
+		if p.visitedUids.Has(string(uid)) {
+			continue
+		}
+
+		name, err := mapping.Name(obj)
+		if err != nil {
+			return err
+		}
+		if err := p.delete(namespace, name, mapping, c); err != nil {
+			return err
+		}
+		cmdutil.PrintSuccess(p.mapper, shortOutput, p.out, mapping.Resource, name, false, "pruned")
+	}
+	return nil
+}
+
+func (p *pruner) delete(namespace, name string, mapping *meta.RESTMapping, c resource.RESTClient) error {
+	if !p.cascade {
+		if err := resource.NewHelper(c, mapping).Delete(namespace, name); err != nil {
+			return err
+		}
+		return nil
+	}
+	cs, err := p.clientsetFunc()
+	if err != nil {
+		return err
+	}
+	r, err := kubectl.ReaperFor(mapping.GroupVersionKind.GroupKind(), cs)
+	if err != nil {
+		return err
+	}
+	if err := r.Stop(namespace, name, 2*time.Minute, api.NewDeleteOptions(int64(p.gracePeriod))); err != nil {
+		return err
+	}
 	return nil
 }