Use the TaskContext for passing UIDs of applied resources to the pruner

Author: mortent

pkg/apply/applier.go

@@ -41,7 +41,7 @@ func NewApplier(provider provider.Provider, ioStreams genericclioptions.IOStream
 		ApplyOptions: applyOptions,
 		// VisitedUids keeps track of the unique identifiers for all
 		// currently applied objects. Used to calculate prune set.
-		PruneOptions: prune.NewPruneOptions(applyOptions.VisitedUids),
+		PruneOptions: prune.NewPruneOptions(),
 		provider:     provider,
 		ioStreams:    ioStreams,
 	}

pkg/apply/destroyer.go

@@ -29,9 +29,7 @@ import (
 func NewDestroyer(provider provider.Provider, ioStreams genericclioptions.IOStreams) *Destroyer {
 	return &Destroyer{
 		ApplyOptions: apply.NewApplyOptions(ioStreams),
-		// Create and maintain an empty set of UID's. This empty UID set
-		// is used during prune calculation to prune every object.
-		PruneOptions: prune.NewPruneOptions(sets.NewString()),
+		PruneOptions: prune.NewPruneOptions(),
 		provider:     provider,
 		ioStreams:    ioStreams,
 	}
@@ -98,7 +96,7 @@ func (d *Destroyer) Run(inv *resource.Info) <-chan event.Event {
 		// Events. That we use Prune to implement destroy is an
 		// implementation detail and the events should not be Prune events.
 		tempChannel, completedChannel := runPruneEventTransformer(ch)
-		err := d.PruneOptions.Prune(infos, tempChannel, prune.Options{
+		err := d.PruneOptions.Prune(infos, sets.NewString(), tempChannel, prune.Options{
 			DryRunStrategy:    d.DryRunStrategy,
 			PropagationPolicy: metav1.DeletePropagationBackground,
 		})

pkg/apply/prune/prune.go

@@ -36,11 +36,6 @@ type PruneOptions struct {
 	InvClient inventory.InventoryClient
 	client    dynamic.Interface
 	mapper    meta.RESTMapper
-	// Stores the UID for each of the currently applied objects.
-	// These UID's are written during the apply, and this data
-	// structure is shared. IMPORTANT: the apply task must
-	// always complete before this prune is run.
-	currentUids sets.String
 	// True if we are destroying, which deletes the inventory object
 	// as well (possibly) the inventory namespace.
 	Destroy bool
@@ -49,10 +44,9 @@ type PruneOptions struct {
 // NewPruneOptions returns a struct (PruneOptions) encapsulating the necessary
 // information to run the prune. Returns an error if an error occurs
 // gathering this information.
-func NewPruneOptions(currentUids sets.String) *PruneOptions {
+func NewPruneOptions() *PruneOptions {
 	po := &PruneOptions{
-		currentUids: currentUids,
-		Destroy:     false,
+		Destroy: false,
 	}
 	return po
 }
@@ -86,7 +80,8 @@ type Options struct {
 // (retrieved from previous inventory objects) but omitted in
 // the current apply. Prune also delete all previous inventory
 // objects. Returns an error if there was a problem.
-func (po *PruneOptions) Prune(localInfos []*resource.Info, eventChannel chan<- event.Event, o Options) error {
+func (po *PruneOptions) Prune(localInfos []*resource.Info, currentUIDs sets.String,
+	eventChannel chan<- event.Event, o Options) error {
 	localInv, localInfos, err := inventory.SplitInfos(localInfos)
 	if err != nil {
 		return err
@@ -97,7 +92,7 @@ func (po *PruneOptions) Prune(localInfos []*resource.Info, eventChannel chan<- e
 	if err != nil {
 		return err
 	}
-	klog.V(4).Infof("prune %d currently applied objects", len(po.currentUids))
+	klog.V(4).Infof("prune %d currently applied objects", len(currentUIDs))
 	klog.V(4).Infof("prune %d previously applied objects", len(clusterObjs))
 	// Sort the resources in reverse order using the same rules as is
 	// used for apply.
@@ -125,7 +120,7 @@ func (po *PruneOptions) Prune(localInfos []*resource.Info, eventChannel chan<- e
 		// object is part of the local apply set, skip it.
 		uid := string(metadata.GetUID())
 		klog.V(7).Infof("prune previously applied object UID: %s", uid)
-		if po.currentUids.Has(uid) {
+		if currentUIDs.Has(uid) {
 			klog.V(7).Infof("prune object in current apply; do not prune: %s", uid)
 			continue
 		}

pkg/apply/prune/prune_test.go

@@ -192,7 +192,7 @@ func TestPrune(t *testing.T) {
 		for i := range common.Strategies {
 			drs := common.Strategies[i]
 			t.Run(name, func(t *testing.T) {
-				po := NewPruneOptions(populateObjectIds(tc.currentInfos, t))
+				po := NewPruneOptions()
 				// Set up the previously applied objects.
 				clusterObjs, _ := object.InfosToObjMetas(tc.pastInfos)
 				po.InvClient = inventory.NewFakeInventoryClient(clusterObjs)
@@ -210,7 +210,7 @@ func TestPrune(t *testing.T) {
 				err := func() error {
 					defer close(eventChannel)
 					// Run the prune and validate.
-					return po.Prune(currentInfos, eventChannel, Options{
+					return po.Prune(currentInfos, populateObjectIds(tc.currentInfos, t), eventChannel, Options{
 						DryRunStrategy: drs,
 					})
 				}()

pkg/apply/task/apply_task.go

@@ -119,8 +119,9 @@ func (a *ApplyTask) Start(taskContext *taskrunner.TaskContext) {
 				if err != nil {
 					continue
 				}
+				uid := acc.GetUID()
 				gen := acc.GetGeneration()
-				taskContext.ResourceApplied(id, gen)
+				taskContext.ResourceApplied(id, uid, gen)
 			}
 		}
 		a.sendTaskResult(taskContext, nil)

pkg/apply/task/apply_task_test.go

@@ -10,6 +10,7 @@ import (
 	"gotest.tools/assert"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/cli-runtime/pkg/resource"
 	"sigs.k8s.io/cli-utils/pkg/apply/event"
 	"sigs.k8s.io/cli-utils/pkg/apply/taskrunner"
@@ -24,6 +25,7 @@ type resourceInfo struct {
 	kind       string
 	name       string
 	namespace  string
+	uid        types.UID
 	generation int64
 }
 
@@ -39,6 +41,7 @@ func TestApplyTask_FetchGeneration(t *testing.T) {
 					kind:       "Deployment",
 					name:       "foo",
 					namespace:  "default",
+					uid:        types.UID("my-uid"),
 					generation: int64(42),
 				},
 			},
@@ -50,13 +53,15 @@ func TestApplyTask_FetchGeneration(t *testing.T) {
 					apiVersion: "custom.io/v1beta1",
 					kind:       "Custom",
 					name:       "bar",
+					uid:        types.UID("uid-1"),
 					generation: int64(32),
 				},
 				{
 					group:      "custom2.io",
 					apiVersion: "custom2.io/v1",
 					kind:       "Custom2",
 					name:       "foo",
+					uid:        types.UID("uid-2"),
 					generation: int64(1),
 				},
 			},
@@ -92,7 +97,10 @@ func TestApplyTask_FetchGeneration(t *testing.T) {
 					Name:      info.name,
 					Namespace: info.namespace,
 				}
-				gen := taskContext.ResourceGeneration(id)
+				uid, _ := taskContext.ResourceUID(id)
+				assert.Equal(t, info.uid, uid)
+
+				gen, _ := taskContext.ResourceGeneration(id)
 				assert.Equal(t, info.generation, gen)
 			}
 		})
@@ -294,6 +302,7 @@ func toInfos(rss []resourceInfo) []*resource.Info {
 					"metadata": map[string]interface{}{
 						"name":       rs.name,
 						"namespace":  rs.namespace,
+						"uid":        string(rs.uid),
 						"generation": rs.generation,
 					},
 				},

pkg/apply/task/prune_task.go

@@ -27,7 +27,8 @@ type PruneTask struct {
 // to signal to the taskrunner that the task has completed (or failed).
 func (p *PruneTask) Start(taskContext *taskrunner.TaskContext) {
 	go func() {
-		err := p.PruneOptions.Prune(p.Objects, taskContext.EventChannel(),
+		currentUIDs := taskContext.AllResourceUIDs()
+		err := p.PruneOptions.Prune(p.Objects, currentUIDs, taskContext.EventChannel(),
 			prune.Options{
 				DryRunStrategy:    p.DryRunStrategy,
 				PropagationPolicy: p.PropagationPolicy,

pkg/apply/taskrunner/context.go

@@ -4,6 +4,8 @@
 package taskrunner
 
 import (
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"sigs.k8s.io/cli-utils/pkg/apply/event"
 	"sigs.k8s.io/cli-utils/pkg/object"
 )
@@ -38,20 +40,40 @@ func (tc *TaskContext) EventChannel() chan event.Event {
 // ResourceApplied updates the context with information about the
 // resource identified by the provided id. Currently, we keep information
 // about the generation of the resource after the apply operation completed.
-func (tc *TaskContext) ResourceApplied(id object.ObjMetadata, gen int64) {
+func (tc *TaskContext) ResourceApplied(id object.ObjMetadata, uid types.UID, gen int64) {
 	tc.appliedResources[id] = applyInfo{
 		generation: gen,
+		uid:        uid,
 	}
 }
 
+// ResourceUID looks up the UID of the given resource
+func (tc *TaskContext) ResourceUID(id object.ObjMetadata) (types.UID, bool) {
+	ai, found := tc.appliedResources[id]
+	if !found {
+		return "", false
+	}
+	return ai.uid, true
+}
+
+// AllResourceUIDs returns a set with the UIDs of all the resources in the
+// context.
+func (tc *TaskContext) AllResourceUIDs() sets.String {
+	uids := sets.NewString()
+	for _, ai := range tc.appliedResources {
+		uids.Insert(string(ai.uid))
+	}
+	return uids
+}
+
 // ResourceGeneration looks up the generation of the given resource
 // after it was applied.
-func (tc *TaskContext) ResourceGeneration(id object.ObjMetadata) int64 {
+func (tc *TaskContext) ResourceGeneration(id object.ObjMetadata) (int64, bool) {
 	ai, found := tc.appliedResources[id]
 	if !found {
-		return 0
+		return 0, false
 	}
-	return ai.generation
+	return ai.generation, true
 }
 
 // applyInfo captures information about resources that have been
@@ -63,4 +85,7 @@ type applyInfo struct {
 	// that the APIServer increases every time the desired state of a
 	// resource changes.
 	generation int64
+
+	// uid captures the uid of the resource that has been applied.
+	uid types.UID
 }

pkg/apply/taskrunner/task.go

@@ -112,9 +112,10 @@ func (w *WaitTask) checkCondition(taskContext *TaskContext, coll *resourceStatus
 func (w *WaitTask) computeResourceWaitData(taskContext *TaskContext) []resourceWaitData {
 	var rwd []resourceWaitData
 	for _, id := range w.Identifiers {
+		gen, _ := taskContext.ResourceGeneration(id)
 		rwd = append(rwd, resourceWaitData{
 			identifier: id,
-			generation: taskContext.ResourceGeneration(id),
+			generation: gen,
 		})
 	}
 	return rwd