ingore the failed resources in wait task

Author: Liujingfang1

pkg/apply/applier_test.go

@@ -265,20 +265,6 @@ func TestApplier(t *testing.T) {
 						Status: status.CurrentStatus,
 					},
 				},
-				{
-					EventType: pollevent.ResourceUpdateEvent,
-					Resource: &pollevent.ResourceStatus{
-						Identifier: toIdentifier(t, resources["deployment"]),
-						Status:     status.InProgressStatus,
-					},
-				},
-				{
-					EventType: pollevent.ResourceUpdateEvent,
-					Resource: &pollevent.ResourceStatus{
-						Identifier: toIdentifier(t, resources["deployment"]),
-						Status:     status.CurrentStatus,
-					},
-				},
 			},
 			expectedEventTypes: []expectedEvent{
 				{
@@ -297,19 +283,12 @@ func TestApplier(t *testing.T) {
 					eventType:       event.StatusType,
 					statusEventType: event.StatusEventResourceUpdate,
 				},
-				{
-					eventType:       event.StatusType,
-					statusEventType: event.StatusEventResourceUpdate,
-				},
-				{
-					eventType:       event.StatusType,
-					statusEventType: event.StatusEventResourceUpdate,
-				},
 				{
 					eventType:       event.StatusType,
 					statusEventType: event.StatusEventCompleted,
 				},
 			},
+			prune:      false,
 			clusterObj: deploymentUnmatched,
 		},
 		"prune with inventory object annotation unmatched": {

pkg/apply/destroyer.go

@@ -11,6 +11,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"sigs.k8s.io/cli-utils/pkg/apply/event"
 	"sigs.k8s.io/cli-utils/pkg/apply/prune"
+	"sigs.k8s.io/cli-utils/pkg/apply/taskrunner"
 	"sigs.k8s.io/cli-utils/pkg/common"
 	"sigs.k8s.io/cli-utils/pkg/inventory"
 	"sigs.k8s.io/cli-utils/pkg/provider"
@@ -70,7 +71,8 @@ func (d *Destroyer) Run(inv inventory.InventoryInfo) <-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(inv, nil, sets.NewString(), tempChannel, prune.Options{
+		taskContext := taskrunner.NewTaskContext(tempChannel)
+		err := d.PruneOptions.Prune(inv, nil, sets.NewString(), taskContext, prune.Options{
 			DryRunStrategy:    d.DryRunStrategy,
 			PropagationPolicy: metav1.DeletePropagationBackground,
 		})

pkg/apply/prune/prune.go

@@ -26,6 +26,7 @@ import (
 	"k8s.io/klog"
 	"k8s.io/kubectl/pkg/cmd/util"
 	"sigs.k8s.io/cli-utils/pkg/apply/event"
+	"sigs.k8s.io/cli-utils/pkg/apply/taskrunner"
 	"sigs.k8s.io/cli-utils/pkg/common"
 	"sigs.k8s.io/cli-utils/pkg/inventory"
 	"sigs.k8s.io/cli-utils/pkg/object"
@@ -86,7 +87,7 @@ type Options struct {
 // the current apply. Prune also delete all previous inventory
 // objects. Returns an error if there was a problem.
 func (po *PruneOptions) Prune(localInv inventory.InventoryInfo, localObjs []*unstructured.Unstructured, currentUIDs sets.String,
-	eventChannel chan<- event.Event, o Options) error {
+	taskContext *taskrunner.TaskContext, o Options) error {
 	if localInv == nil {
 		return fmt.Errorf("the local inventory object can't be nil")
 	}
@@ -114,7 +115,8 @@ func (po *PruneOptions) Prune(localInv inventory.InventoryInfo, localObjs []*uns
 		mapping, err := po.mapper.RESTMapping(clusterObj.GroupKind)
 		if err != nil {
 			localIds = append(localIds, clusterObj)
-			eventChannel <- createPruneFailedEvent(clusterObj, err)
+			taskContext.EventChannel() <- createPruneFailedEvent(clusterObj, err)
+			taskContext.CaptureResourceFailure(clusterObj)
 			continue
 		}
 		namespacedClient := po.client.Resource(mapping.Resource).Namespace(clusterObj.Namespace)
@@ -125,13 +127,15 @@ func (po *PruneOptions) Prune(localInv inventory.InventoryInfo, localObjs []*uns
 				continue
 			}
 			localIds = append(localIds, clusterObj)
-			eventChannel <- createPruneFailedEvent(clusterObj, err)
+			taskContext.EventChannel() <- createPruneFailedEvent(clusterObj, err)
+			taskContext.CaptureResourceFailure(clusterObj)
 			continue
 		}
 		metadata, err := meta.Accessor(obj)
 		if err != nil {
 			localIds = append(localIds, clusterObj)
-			eventChannel <- createPruneFailedEvent(clusterObj, err)
+			taskContext.EventChannel() <- createPruneFailedEvent(clusterObj, err)
+			taskContext.CaptureResourceFailure(clusterObj)
 			continue
 		}
 		// If this cluster object is not also a currently applied
@@ -145,7 +149,7 @@ func (po *PruneOptions) Prune(localInv inventory.InventoryInfo, localObjs []*uns
 		}
 		// Handle lifecycle directive preventing deletion.
 		if !canPrune(localInv, obj, o.InventoryPolicy, uid) {
-			eventChannel <- createPruneEvent(clusterObj, obj, event.PruneSkipped)
+			taskContext.EventChannel() <- createPruneEvent(clusterObj, obj, event.PruneSkipped)
 			localIds = append(localIds, clusterObj)
 			continue
 		}
@@ -155,21 +159,23 @@ func (po *PruneOptions) Prune(localInv inventory.InventoryInfo, localObjs []*uns
 			if clusterObj.GroupKind == object.CoreV1Namespace.GroupKind() &&
 				localNamespaces.Has(clusterObj.Name) {
 				klog.V(4).Infof("skip pruning namespace: %s", clusterObj.Name)
-				eventChannel <- createPruneEvent(clusterObj, obj, event.PruneSkipped)
+				taskContext.EventChannel() <- createPruneEvent(clusterObj, obj, event.PruneSkipped)
 				localIds = append(localIds, clusterObj)
+				taskContext.CaptureResourceFailure(clusterObj)
 				continue
 			}
 		}
 		if !o.DryRunStrategy.ClientOrServerDryRun() {
 			klog.V(4).Infof("prune object delete: %s/%s", clusterObj.Namespace, clusterObj.Name)
 			err = namespacedClient.Delete(context.TODO(), clusterObj.Name, metav1.DeleteOptions{})
 			if err != nil {
-				eventChannel <- createPruneFailedEvent(clusterObj, err)
+				taskContext.EventChannel() <- createPruneFailedEvent(clusterObj, err)
 				localIds = append(localIds, clusterObj)
+				taskContext.CaptureResourceFailure(clusterObj)
 				continue
 			}
 		}
-		eventChannel <- createPruneEvent(clusterObj, obj, event.Pruned)
+		taskContext.EventChannel() <- createPruneEvent(clusterObj, obj, event.Pruned)
 	}
 	return po.InvClient.Replace(localInv, localIds)
 }

pkg/apply/prune/prune_test.go

@@ -21,6 +21,7 @@ import (
 	"k8s.io/client-go/dynamic/fake"
 	"k8s.io/kubectl/pkg/scheme"
 	"sigs.k8s.io/cli-utils/pkg/apply/event"
+	"sigs.k8s.io/cli-utils/pkg/apply/taskrunner"
 	"sigs.k8s.io/cli-utils/pkg/common"
 	"sigs.k8s.io/cli-utils/pkg/inventory"
 	"sigs.k8s.io/cli-utils/pkg/object"
@@ -250,10 +251,11 @@ func TestPrune(t *testing.T) {
 				// The event channel can not block; make sure its bigger than all
 				// the events that can be put on it.
 				eventChannel := make(chan event.Event, len(tc.pastObjs)+1) // Add one for inventory object
+				taskContext := taskrunner.NewTaskContext(eventChannel)
 				err := func() error {
 					defer close(eventChannel)
 					// Run the prune and validate.
-					return po.Prune(currentInventory, tc.currentObjs, populateObjectIds(tc.currentObjs, t), eventChannel, Options{
+					return po.Prune(currentInventory, tc.currentObjs, populateObjectIds(tc.currentObjs, t), taskContext, Options{
 						DryRunStrategy: drs,
 					})
 				}()
@@ -453,10 +455,11 @@ func TestPruneWithError(t *testing.T) {
 			// The event channel can not block; make sure its bigger than all
 			// the events that can be put on it.
 			eventChannel := make(chan event.Event, len(tc.pastObjs)+1) // Add one for inventory object
+			taskContext := taskrunner.NewTaskContext(eventChannel)
 			err := func() error {
 				defer close(eventChannel)
 				// Run the prune and validate.
-				return po.Prune(currentInventory, tc.currentObjs, populateObjectIds(tc.currentObjs, t), eventChannel, Options{
+				return po.Prune(currentInventory, tc.currentObjs, populateObjectIds(tc.currentObjs, t), taskContext, Options{
 					DryRunStrategy: drs,
 				})
 			}()

pkg/apply/task/apply_task.go

@@ -121,28 +121,32 @@ func (a *ApplyTask) Start(taskContext *taskrunner.TaskContext) {
 			// Set the client and mapping fields on the provided
 			// info so they can be applied to the cluster.
 			info, err := a.InfoHelper.BuildInfo(obj)
+			id := object.UnstructuredToObjMeta(obj)
 			if err != nil {
 				taskContext.EventChannel() <- createApplyEvent(
-					object.UnstructuredToObjMeta(obj), event.Failed, applyerror.NewUnknownTypeError(err))
+					id, event.Failed, applyerror.NewUnknownTypeError(err))
+				taskContext.CaptureResourceFailure(id)
 				continue
 			}
 
 			clusterObj, err := getClusterObj(dynamic, info)
 			if err != nil {
 				if !apierrors.IsNotFound(err) {
 					taskContext.EventChannel() <- createApplyEvent(
-						object.UnstructuredToObjMeta(obj),
+						id,
 						event.Unchanged,
 						err)
+					taskContext.CaptureResourceFailure(id)
 					continue
 				}
 			}
 			canApply, err := inventory.CanApply(a.InvInfo, clusterObj, a.InventoryPolicy)
 			if !canApply {
 				taskContext.EventChannel() <- createApplyEvent(
-					object.UnstructuredToObjMeta(obj),
+					id,
 					event.Unchanged,
 					err)
+				taskContext.CaptureResourceFailure(id)
 				continue
 			}
 			// add the inventory annotation to the resource being applied.
@@ -152,7 +156,8 @@ func (a *ApplyTask) Start(taskContext *taskrunner.TaskContext) {
 			err = ao.Run()
 			if err != nil {
 				taskContext.EventChannel() <- createApplyEvent(
-					object.UnstructuredToObjMeta(obj), event.Failed, applyerror.NewApplyRunError(err))
+					id, event.Failed, applyerror.NewApplyRunError(err))
+				taskContext.CaptureResourceFailure(id)
 			}
 		}
 
@@ -331,7 +336,9 @@ func createApplyEvent(id object.ObjMetadata, operation event.ApplyEventOperation
 // a list of resources when failed to initialize the apply process.
 func sendBatchApplyEvents(taskContext *taskrunner.TaskContext, objects []*unstructured.Unstructured, err error) {
 	for _, obj := range objects {
+		id := object.UnstructuredToObjMeta(obj)
 		taskContext.EventChannel() <- createApplyEvent(
-			object.UnstructuredToObjMeta(obj), event.Failed, applyerror.NewInitializeApplyOptionError(err))
+			id, event.Failed, applyerror.NewInitializeApplyOptionError(err))
+		taskContext.CaptureResourceFailure(id)
 	}
 }

pkg/apply/task/prune_task.go

@@ -32,7 +32,7 @@ func (p *PruneTask) Start(taskContext *taskrunner.TaskContext) {
 	go func() {
 		currentUIDs := taskContext.AllResourceUIDs()
 		err := p.PruneOptions.Prune(p.InventoryObject, p.Objects,
-			currentUIDs, taskContext.EventChannel(), prune.Options{
+			currentUIDs, taskContext, prune.Options{
 				DryRunStrategy:    p.DryRunStrategy,
 				PropagationPolicy: p.PropagationPolicy,
 				InventoryPolicy:   p.InventoryPolicy,

pkg/apply/taskrunner/context.go

@@ -16,6 +16,7 @@ func NewTaskContext(eventChannel chan event.Event) *TaskContext {
 		taskChannel:      make(chan TaskResult),
 		eventChannel:     eventChannel,
 		appliedResources: make(map[object.ObjMetadata]applyInfo),
+		failedResources:  make(map[object.ObjMetadata]struct{}),
 	}
 }
 
@@ -27,6 +28,9 @@ type TaskContext struct {
 	eventChannel chan event.Event
 
 	appliedResources map[object.ObjMetadata]applyInfo
+
+	// failedResources records the IDs of resources that are failed during applying and pruning.
+	failedResources map[object.ObjMetadata]struct{}
 }
 
 func (tc *TaskContext) TaskChannel() chan TaskResult {
@@ -76,6 +80,15 @@ func (tc *TaskContext) ResourceGeneration(id object.ObjMetadata) (int64, bool) {
 	return ai.generation, true
 }
 
+func (tc *TaskContext) ResourceFailed(id object.ObjMetadata) bool {
+	_, found := tc.failedResources[id]
+	return found
+}
+
+func (tc *TaskContext) CaptureResourceFailure(id object.ObjMetadata) {
+	tc.failedResources[id] = struct{}{}
+}
+
 // applyInfo captures information about resources that have been
 // applied. This is captured in the TaskContext so other tasks
 // running later might use this information.

pkg/apply/taskrunner/task.go

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