Merge pull request kubernetes-sigs#325 from yuwenma/applyset

Prune logic for ApplySetApplier

Author: k8s-ci-robot

applylib/applyset/applyset.go

@@ -20,12 +20,19 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"reflect"
 	"sync"
 
 	"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/schema"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/dynamic"
+	"k8s.io/klog/v2"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	kubectlapply "sigs.k8s.io/kubebuilder-declarative-pattern/applylib/third_party/forked/github.com/kubernetes/kubectl/pkg/cmd/apply"
 )
 
 // ApplySet is a set of objects that we want to apply to the cluster.
@@ -37,37 +44,68 @@ import (
 // * We expose a "try once" method to better support running from a controller.
 //
 // TODO: Pluggable health functions.
-// TODO: Pruning
 type ApplySet struct {
 	// client is the dynamic kubernetes client used to apply objects to the k8s cluster.
 	client dynamic.Interface
+	// ParentClient is the controller runtime client used to apply parent.
+	parentClient client.Client
 	// restMapper is used to map object kind to resources, and to know if objects are cluster-scoped.
 	restMapper meta.RESTMapper
 	// patchOptions holds the options used when applying, in particular the fieldManager
 	patchOptions metav1.PatchOptions
 
+	// deleteOptions holds the options used when pruning
+	deleteOptions metav1.DeleteOptions
+
 	// mutex guards trackers
 	mutex sync.Mutex
 	// trackers is a (mutable) pointer to the (immutable) objectTrackerList, containing a list of objects we are applying.
 	trackers *objectTrackerList
+
+	// whether to prune the previously objects that are no longer in the current deployment manifest list.
+	// Finding the objects to prune is done via "apply-set" labels and annotations. See KEP
+	// https://github.com/KnVerey/enhancements/blob/b347756461679f62cf985e7a6b0fd0bc28ea9fd2/keps/sig-cli/3659-kubectl-apply-prune/README.md#optional-hint-annotations
+	prune bool
+	// Parent provides the necessary methods to determine a ApplySet parent object, which can be used to find out all the on-track
+	// deployment manifests.
+	parent Parent
+	// If not given, the tooling value will be the `Parent` Kind.
+	tooling string
 }
 
 // Options holds the parameters for building an ApplySet.
 type Options struct {
 	// Client is the dynamic kubernetes client used to apply objects to the k8s cluster.
 	Client dynamic.Interface
+	// ParentClient is the controller runtime client used to apply parent.
+	ParentClient client.Client
 	// RESTMapper is used to map object kind to resources, and to know if objects are cluster-scoped.
 	RESTMapper meta.RESTMapper
 	// PatchOptions holds the options used when applying, in particular the fieldManager
-	PatchOptions metav1.PatchOptions
+	PatchOptions  metav1.PatchOptions
+	DeleteOptions metav1.DeleteOptions
+	Prune         bool
+	Parent        Parent
+	Tooling       string
 }
 
 // New constructs a new ApplySet
 func New(options Options) (*ApplySet, error) {
+	parent := options.Parent
+	parentRef := &kubectlapply.ApplySetParentRef{Name: parent.Name(), Namespace: parent.Namespace(), RESTMapping: parent.RESTMapping()}
+	kapplyset := kubectlapply.NewApplySet(parentRef, kubectlapply.ApplySetTooling{Name: options.Tooling}, options.RESTMapper)
+	if options.PatchOptions.FieldManager == "" {
+		options.PatchOptions.FieldManager = kapplyset.FieldManager()
+	}
 	a := &ApplySet{
-		client:       options.Client,
-		restMapper:   options.RESTMapper,
-		patchOptions: options.PatchOptions,
+		parentClient:  options.ParentClient,
+		client:        options.Client,
+		restMapper:    options.RESTMapper,
+		patchOptions:  options.PatchOptions,
+		deleteOptions: options.DeleteOptions,
+		prune:         options.Prune,
+		parent:        parent,
+		tooling:       options.Tooling,
 	}
 	a.trackers = &objectTrackerList{}
 	return a, nil
@@ -85,6 +123,11 @@ func (a *ApplySet) SetDesiredObjects(objects []ApplyableObject) error {
 	return nil
 }
 
+type restMappingResult struct {
+	restMapping *meta.RESTMapping
+	err         error
+}
+
 // ApplyOnce will make one attempt to apply all objects and observe their health.
 // It does not wait for the objects to become healthy, but will report their health.
 //
@@ -100,6 +143,43 @@ func (a *ApplySet) ApplyOnce(ctx context.Context) (*ApplyResults, error) {
 	a.mutex.Unlock()
 
 	results := &ApplyResults{total: len(trackers.items)}
+	visitedUids := sets.New[types.UID]()
+
+	// We initialize a new Kubectl ApplySet for each ApplyOnce run. This is because kubectl Applyset is designed for
+	// single actuation and not for reconciliation.
+	// Note: The Kubectl ApplySet will share the RESTMapper with k-d-p/ApplySet, which caches all the manifests in the past.
+	parentRef := &kubectlapply.ApplySetParentRef{Name: a.parent.Name(), Namespace: a.parent.Namespace(), RESTMapping: a.parent.RESTMapping()}
+	kapplyset := kubectlapply.NewApplySet(parentRef, kubectlapply.ApplySetTooling{Name: a.tooling}, a.restMapper)
+
+	// Cache the current RESTMappings to avoid re-fetching the bad ones.
+	restMappings := make(map[schema.GroupVersionKind]restMappingResult)
+	for i := range trackers.items {
+		tracker := &trackers.items[i]
+		obj := tracker.desired
+
+		gvk := obj.GroupVersionKind()
+
+		result, found := restMappings[gvk]
+		if !found {
+			restMapping, err := a.restMapper.RESTMapping(gvk.GroupKind(), gvk.Version)
+			result = restMappingResult{
+				restMapping: restMapping,
+				err:         err,
+			}
+			restMappings[gvk] = result
+		}
+
+		// TODO: Check error is NotFound and not some transient error?
+		restMapping := result.restMapping
+		if restMapping != nil {
+			// cache the GVK in kubectlapply. kubectlapply will use this to calculate
+			// the latest parent "applyset.kubernetes.io/contains-group-resources" annotation after applying.
+			kapplyset.AddResource(restMapping, obj.GetNamespace())
+		}
+	}
+	if err := a.WithParent(ctx, kapplyset); err != nil {
+		return results, fmt.Errorf("unable to update Parent: %w", err)
+	}
 
 	for i := range trackers.items {
 		tracker := &trackers.items[i]
@@ -110,11 +190,23 @@ func (a *ApplySet) ApplyOnce(ctx context.Context) (*ApplyResults, error) {
 		gvk := obj.GroupVersionKind()
 		nn := types.NamespacedName{Namespace: ns, Name: name}
 
-		restMapping, err := a.restMapper.RESTMapping(gvk.GroupKind(), gvk.Version)
-		if err != nil {
-			results.applyError(gvk, nn, fmt.Errorf("error getting rest mapping for %v: %w", gvk, err))
+		restMappingResult := restMappings[gvk]
+		if restMappingResult.err != nil {
+			results.applyError(gvk, nn, fmt.Errorf("error getting rest mapping for %v: %w", gvk, restMappingResult.err))
+			continue
+		}
+
+		restMapping := restMappingResult.restMapping
+		if restMapping == nil {
+			// Should be impossible
+			results.applyError(gvk, nn, fmt.Errorf("rest mapping result not found for %v", gvk))
 			continue
 		}
+
+		if err := a.updateManifestLabel(obj, kapplyset.LabelsForMember()); err != nil {
+			return results, fmt.Errorf("unable to update label for %v/%v %v: %w", obj.GetName(), obj.GetNamespace(), gvk, err)
+		}
+
 		gvr := restMapping.Resource
 
 		var dynamicResource dynamic.ResourceInterface
@@ -140,7 +232,6 @@ func (a *ApplySet) ApplyOnce(ctx context.Context) (*ApplyResults, error) {
 			// Internal error ... this is panic-level
 			return nil, fmt.Errorf("unknown scope for gvk %s: %q", gvk, restMapping.Scope.Name())
 		}
-
 		j, err := json.Marshal(obj)
 		if err != nil {
 			// TODO: Differentiate between server-fixable vs client-fixable errors?
@@ -153,11 +244,144 @@ func (a *ApplySet) ApplyOnce(ctx context.Context) (*ApplyResults, error) {
 			results.applyError(gvk, nn, fmt.Errorf("error from apply: %w", err))
 			continue
 		}
-
+		visitedUids.Insert(lastApplied.GetUID())
 		tracker.lastApplied = lastApplied
 		results.applySuccess(gvk, nn)
 		tracker.isHealthy = isHealthy(lastApplied)
 		results.reportHealth(gvk, nn, tracker.isHealthy)
 	}
+
+	// We want to be more cautions on pruning and only do it if all manifests are applied.
+	if a.prune && results.applyFailCount == 0 {
+		klog.V(4).Infof("Prune is enabled")
+		pruneObjects, err := kapplyset.FindAllObjectsToPrune(ctx, a.client, visitedUids)
+		if err != nil {
+			return results, err
+		}
+		if err = a.deleteObjects(ctx, pruneObjects, results); err != nil {
+			return results, err
+		}
+		// "latest" mode updates the parent "applyset.kubernetes.io/contains-group-resources" annotations
+		// to only contain the current manifest GVRs.
+		if err := a.updateParentLabelsAndAnnotations(ctx, kapplyset, "latest"); err != nil {
+			klog.Errorf("update parent failed %v", err)
+		}
+	}
 	return results, nil
 }
+
+// updateManifestLabel adds the "applyset.kubernetes.io/part-of: Parent-ID" label to the manifest.
+func (a *ApplySet) updateManifestLabel(obj ApplyableObject, applysetLabels map[string]string) error {
+	u, ok := obj.(*unstructured.Unstructured)
+	if !ok {
+		return fmt.Errorf("unable to convert `ApplyableObject` to `unstructured.Unstructured` %v/%v %v",
+			obj.GetName(), obj.GetNamespace(), obj.GroupVersionKind().String())
+	}
+	labels := u.GetLabels()
+	if labels == nil {
+		labels = make(map[string]string)
+	}
+	for k, v := range applysetLabels {
+		labels[k] = v
+	}
+	u.SetLabels(labels)
+	return nil
+}
+
+// updateParentLabelsAndAnnotations updates the parent labels and annotations.
+func (a *ApplySet) updateParentLabelsAndAnnotations(ctx context.Context, kapplyset *kubectlapply.ApplySet, mode kubectlapply.ApplySetUpdateMode) error {
+	parent, err := meta.Accessor(a.parent.GetSubject())
+	if err != nil {
+		return err
+	}
+
+	original, err := meta.Accessor(a.parent.GetSubject().DeepCopyObject())
+	if err != nil {
+		return err
+	}
+	partialParent := kapplyset.BuildParentPatch(mode)
+
+	// update annotation
+	annotations := parent.GetAnnotations()
+	if annotations == nil {
+		annotations = make(map[string]string)
+	}
+	for k, v := range partialParent.Annotations {
+		annotations[k] = v
+	}
+	parent.SetAnnotations(annotations)
+
+	// update labels
+	labels := parent.GetLabels()
+	if labels == nil {
+		labels = make(map[string]string)
+	}
+	for k, v := range partialParent.Labels {
+		labels[k] = v
+	}
+	parent.SetLabels(labels)
+
+	// update parent in the cluster.
+	if !reflect.DeepEqual(original.GetLabels(), parent.GetLabels()) || !reflect.DeepEqual(original.GetAnnotations(), parent.GetAnnotations()) {
+		if err := a.parentClient.Update(ctx, parent.(client.Object)); err != nil {
+			return fmt.Errorf("error updating parent %w", err)
+		}
+	}
+	return nil
+}
+
+func (a *ApplySet) deleteObjects(ctx context.Context, pruneObjects []kubectlapply.PruneObject, results *ApplyResults) error {
+	for i := range pruneObjects {
+		pruneObject := &pruneObjects[i]
+		name := pruneObject.Name
+		namespace := pruneObject.Namespace
+		mapping := pruneObject.Mapping
+		gvk := pruneObject.Object.GetObjectKind().GroupVersionKind()
+		nn := types.NamespacedName{Namespace: namespace, Name: name}
+
+		if err := a.client.Resource(mapping.Resource).Namespace(namespace).Delete(ctx, name, a.deleteOptions); err != nil {
+			results.pruneError(gvk, nn, fmt.Errorf("error from delete: %w", err))
+		} else {
+			klog.Infof("pruned resource %v", pruneObject.String())
+			results.pruneSuccess(gvk, nn)
+		}
+	}
+	return nil
+}
+
+// WithParent guarantees the parent has the right applyset labels.
+// It uses "superset" mode to determine the "applyset.kubernetes.io/contains-group-resources" which contains both
+//
+//	previous manifests GVRs and the current manifests GVRs.
+func (a *ApplySet) WithParent(ctx context.Context, kapplyset *kubectlapply.ApplySet) error {
+	parent := a.parent.GetSubject()
+	object, err := meta.Accessor(parent)
+	if err != nil {
+		return err
+	}
+	//kubectlapply requires the tooling and id to exist.
+	{
+		annotations := object.GetAnnotations()
+		if annotations == nil {
+			annotations = make(map[string]string)
+		}
+		annotations[kubectlapply.ApplySetToolingAnnotation] = a.tooling
+		if _, ok := annotations[kubectlapply.ApplySetGRsAnnotation]; !ok {
+			annotations[kubectlapply.ApplySetGRsAnnotation] = ""
+		}
+		object.SetAnnotations(annotations)
+
+		labels := object.GetLabels()
+		if labels == nil {
+			labels = make(map[string]string)
+		}
+		labels[kubectlapply.ApplySetParentIDLabel] = kapplyset.ID()
+		object.SetLabels(labels)
+	}
+	// This is not a cluster fetch. It builds up the parents labels and annotations information in kapplyset.
+	if err := kapplyset.FetchParent(a.parent.GetSubject()); err != nil {
+		return err
+	}
+
+	return a.updateParentLabelsAndAnnotations(ctx, kapplyset, "superset")
+}

applylib/applyset/applyset_test.go

@@ -31,7 +31,14 @@ import (
 func TestApplySet(t *testing.T) {
 	h := testutils.NewHarness(t)
 
-	existing := ``
+	// The parent should exist in the cluster
+	existing := `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test
+  namespace: default
+`
 
 	apply := `
 apiVersion: v1
@@ -60,11 +67,20 @@ data:
 
 	force := true
 	patchOptions.Force = &force
-
+	parent := h.ParseObjects(existing)[0]
+	parentGVK := parent.GroupVersionKind()
+	restmapping, err := h.RESTMapper().RESTMapping(parentGVK.GroupKind(), parentGVK.Version)
+	if err != nil {
+		h.Fatalf("error building parent restmappaing: %v", err)
+	}
 	s, err := New(Options{
-		RESTMapper:   h.RESTMapper(),
-		Client:       h.DynamicClient(),
-		PatchOptions: patchOptions,
+		Parent:        NewParentRef(parent, "test", "default", restmapping),
+		RESTMapper:    h.RESTMapper(),
+		Client:        h.DynamicClient(),
+		ParentClient:  h.Client(),
+		PatchOptions:  patchOptions,
+		DeleteOptions: metav1.DeleteOptions{},
+		Prune:         true,
 	})
 	if err != nil {
 		h.Fatalf("error building applyset object: %v", err)