Merge pull request #4893 from justinsb/fix_resource_watch_terminate_on_delete

fix: resourcewatcher should terminate on delete

Author: google-oss-prow[bot]

pkg/controller/dcl/controller.go

@@ -411,7 +411,7 @@ func (r *Reconciler) handleUnresolvableDeps(ctx context.Context, resource *k8s.R
 		ctx, cancel := context.WithTimeout(ctx, timeoutPeriod)
 		defer cancel()
 		logger.Info("starting wait with timeout on resource's reference", "timeout", timeoutPeriod)
-		if err := watcher.WaitForResourceToBeReady(ctx, refNN, refGVK); err != nil {
+		if err := watcher.WaitForResourceToBeReadyOrDeleted(ctx, refNN, refGVK); err != nil {
 			logger.Error(err, "error while waiting for resource's reference to be ready")
 			return
 		}

pkg/controller/direct/directbase/directbase_controller.go

@@ -550,7 +550,7 @@ func (r *reconcileContext) handleUnresolvableDeps(ctx context.Context, policy *u
 		ctx, cancel := context.WithTimeout(context.TODO(), timeoutPeriod)
 		defer cancel()
 		logger.Info("starting wait with timeout on resource's reference", "timeout", timeoutPeriod)
-		if err := watcher.WaitForResourceToBeReady(ctx, refNN, refGVK); err != nil {
+		if err := watcher.WaitForResourceToBeReadyOrDeleted(ctx, refNN, refGVK); err != nil {
 			logger.Error(err, "error while waiting for resource's reference to be ready")
 			return
 		}

pkg/controller/iam/auditconfig/iamauditconfig_controller.go

@@ -372,7 +372,7 @@ func (r *reconcileContext) handleUnresolvableDeps(auditConfig *iamv1beta1.IAMAud
 		ctx, cancel := context.WithTimeout(context.TODO(), timeoutPeriod)
 		defer cancel()
 		logger.Info("starting wait with timeout on resource's reference", "timeout", timeoutPeriod)
-		if err := watcher.WaitForResourceToBeReady(ctx, refNN, refGVK); err != nil {
+		if err := watcher.WaitForResourceToBeReadyOrDeleted(ctx, refNN, refGVK); err != nil {
 			logger.Error(err, "error while waiting for resource's reference to be ready")
 			return
 		}

pkg/controller/iam/partialpolicy/iampartialpolicy_controller.go

@@ -438,7 +438,7 @@ func (r *reconcileContext) handleUnresolvableDeps(policy *iamv1beta1.IAMPartialP
 		ctx, cancel := context.WithTimeout(context.TODO(), timeoutPeriod)
 		defer cancel()
 		logger.Info("starting wait with timeout on resource's reference", "timeout", timeoutPeriod)
-		if err := watcher.WaitForResourceToBeReady(ctx, refNN, refGVK); err != nil {
+		if err := watcher.WaitForResourceToBeReadyOrDeleted(ctx, refNN, refGVK); err != nil {
 			logger.Error(err, "error while waiting for resource's reference to be ready")
 			return
 		}

pkg/controller/iam/policy/iampolicy_controller.go

@@ -371,7 +371,7 @@ func (r *reconcileContext) handleUnresolvableDeps(policy *iamv1beta1.IAMPolicy,
 		ctx, cancel := context.WithTimeout(context.TODO(), timeoutPeriod)
 		defer cancel()
 		logger.Info("starting wait with timeout on resource's reference", "timeout", timeoutPeriod)
-		if err := watcher.WaitForResourceToBeReady(ctx, refNN, refGVK); err != nil {
+		if err := watcher.WaitForResourceToBeReadyOrDeleted(ctx, refNN, refGVK); err != nil {
 			logger.Error(err, "error while waiting for resource's reference to be ready")
 			return
 		}

pkg/controller/iam/policymember/iampolicymember_controller.go

@@ -383,7 +383,7 @@ func (r *reconcileContext) handleUnresolvableDeps(policyMember *iamv1beta1.IAMPo
 		ctx, cancel := context.WithTimeout(context.TODO(), timeoutPeriod)
 		defer cancel()
 		logger.Info("starting wait with timeout on resource's reference", "timeout", timeoutPeriod)
-		if err := watcher.WaitForResourceToBeReady(ctx, refNN, refGVK); err != nil {
+		if err := watcher.WaitForResourceToBeReadyOrDeleted(ctx, refNN, refGVK); err != nil {
 			logger.Error(err, "error while waiting for resource's reference to be ready")
 			return
 		}

pkg/controller/resourcewatcher/resourcewatcher.go

@@ -53,19 +53,19 @@ func NewWithClient(dynamicClient dynamic.Interface, logger logr.Logger) *Resourc
 	}
 }
 
-// WaitForResourceToBeReady waits for the resource identified by the given GVK
-// and NamespacedName. It blocks until the resource is ready, an error occurs, or a context
+// WaitForResourceToBeReadyOrDeleted waits for the resource identified by the given GVK
+// and NamespacedName. It blocks until the resource is ready or deleted, an error occurs, or a context
 // cancellation occurs. Note that a nil return value signifies that the resource is ready and
 // no errors have occurred.
-func (r *ResourceWatcher) WaitForResourceToBeReady(ctx context.Context, nn types.NamespacedName, gvk schema.GroupVersionKind) error {
+func (r *ResourceWatcher) WaitForResourceToBeReadyOrDeleted(ctx context.Context, nn types.NamespacedName, gvk schema.GroupVersionKind) error {
 	logger := r.logger.WithValues("resource", nn, "resourceGVK", gvk)
 	watch, err := r.WatchResource(ctx, nn, gvk)
 	if err != nil {
 		return err
 	}
 	defer watch.Stop()
 	logger.Info("successfully created watch on resource")
-	return WaitForResourceToBeReadyViaWatch(ctx, watch, logger)
+	return WaitForResourceToBeReadyOrDeletedViaWatch(ctx, watch, logger)
 }
 
 // WatchResource creates a watch on a resource identified by the given GVK and NamespacedName.
@@ -79,25 +79,41 @@ func (r *ResourceWatcher) WatchResource(ctx context.Context, nn types.Namespaced
 	return watch, nil
 }
 
-// WaitForResourceToBeReadyViaWatch monitors a given 'Watch' for any
+// WaitForResourceToBeReadyOrDeletedViaWatch monitors a given 'Watch' for any
 // updates to the resource that the given 'Watch' is targeting. Note that
 // an error is returned to signify a failure during the 'Watch' process,
-// while nil is returned to signify the watched resource is ready.
-func WaitForResourceToBeReadyViaWatch(ctx context.Context, watch watch.Interface, logger logr.Logger) error {
+// while nil is returned to signify the watched resource is ready or deleted.
+func WaitForResourceToBeReadyOrDeletedViaWatch(ctx context.Context, w watch.Interface, logger logr.Logger) error {
 	for {
 		select {
 		case <-ctx.Done():
 			return fmt.Errorf("context was cancelled: %w", ctx.Err())
-		case event, ok := <-watch.ResultChan():
+		case event, ok := <-w.ResultChan():
 			if !ok {
 				return fmt.Errorf("watch channel was closed")
 			}
-			ok, reason, err := isResourceReady(event)
+			if event.Type == watch.Bookmark {
+				continue // ignore
+			}
+			if event.Type == watch.Deleted {
+				logger.Info("resource has been deleted; triggering watch completion")
+				return nil
+			}
+			if event.Type != watch.Modified && event.Type != watch.Added {
+				return fmt.Errorf("unexpected watch event type %v", event.Type)
+			}
+
+			u, ok := event.Object.(*unstructured.Unstructured)
+			if !ok {
+				return fmt.Errorf("error casting event object '%v' of kind '%v' to unstructured", event.Object, event.Object.GetObjectKind())
+			}
+
+			isReady, err := isResourceReady(u)
 			if err != nil {
 				return fmt.Errorf("error checking if resource is ready: %w", err)
 			}
-			if !ok {
-				logger.Info("resource not ready", "reason", reason)
+			if !isReady {
+				logger.Info("resource not ready")
 				continue
 			}
 			logger.Info("resource is ready")
@@ -107,27 +123,19 @@ func WaitForResourceToBeReadyViaWatch(ctx context.Context, watch watch.Interface
 }
 
 // isResourceReady returns whether a resource identified by the given GVK
-// and NamespacedName is ready. Note that a 'reason' for failure is returned only
-// when the resource is not ready and no fatal error has occurred.
-func isResourceReady(event watch.Event) (ok bool, reason string, err error) {
-	if event.Type != watch.Modified && event.Type != watch.Added {
-		return false, fmt.Sprintf("got watch event of type '%v', want event type '%v' or '%v'", event.Type, watch.Modified, watch.Added), nil
-	}
-	u, ok := event.Object.(*unstructured.Unstructured)
-	if !ok {
-		return false, "", fmt.Errorf("error casting event object '%v' of kind '%v' to unstructured", event.Object, event.Object.GetObjectKind())
-	}
+// and NamespacedName is ready.
+func isResourceReady(u *unstructured.Unstructured) (isReady bool, err error) {
 	resource, err := k8s.NewResource(u)
 	if err != nil {
-		return false, "", fmt.Errorf("error converting unstructured to resource: %w", err)
+		return false, fmt.Errorf("error converting unstructured to resource: %w", err)
 	}
 	// Secrets don't have a 'ready' condition. As long as they can be
 	// found on the API server, we consider them ready as resources.
 	if resource.Kind == "Secret" {
-		return true, "", nil
+		return true, nil
 	}
 	if !k8s.IsResourceReady(resource) {
-		return false, "resource not ready", nil
+		return false, nil
 	}
-	return true, "", nil
+	return true, nil
 }

pkg/controller/resourcewatcher/resourcewatcher_test.go

@@ -48,7 +48,7 @@ func TestWatchResourceTimeout(t *testing.T) {
 	}
 	ctx, cancel := context.WithTimeout(context.TODO(), time.Second*10)
 	defer cancel()
-	if err := resourcewatcher.WaitForResourceToBeReadyViaWatch(ctx, watch, logger); !errors.Is(err, context.DeadlineExceeded) {
+	if err := resourcewatcher.WaitForResourceToBeReadyOrDeletedViaWatch(ctx, watch, logger); !errors.Is(err, context.DeadlineExceeded) {
 		t.Fatalf("got error '%v', expected '%v'", err, context.DeadlineExceeded)
 	}
 }
@@ -68,15 +68,67 @@ func TestWatchResourceSuccess(t *testing.T) {
 	if _, err := fake.Resource(k8s.ToGVR(gvk)).
 		Namespace(nn.Namespace).
 		Create(context.TODO(), readyResourceUnstructured, metav1.CreateOptions{}); err != nil {
-		t.Fatal(err)
+		t.Fatalf("creating resource: %v", err)
 	}
 	ctx, cancel := context.WithTimeout(context.TODO(), time.Minute)
 	defer cancel()
-	if err := resourcewatcher.WaitForResourceToBeReadyViaWatch(ctx, watch, logger); err != nil {
+	if err := resourcewatcher.WaitForResourceToBeReadyOrDeletedViaWatch(ctx, watch, logger); err != nil {
 		t.Fatalf("got unexpected error: %v", err)
 	}
 }
 
+func TestWatchResourceDeleted(t *testing.T) {
+	ctx := context.TODO()
+
+	notReadyResourceUnstructured := newResourceUnstructured(newStatus(corev1.ConditionFalse))
+	gvk, nn, err := getResourceInformation(notReadyResourceUnstructured)
+	if err != nil {
+		t.Fatalf("got unexpected error from getResourceInformation: %v", err)
+	}
+	fake := dynamicfake.NewSimpleDynamicClient(runtime.NewScheme())
+	logger := log.Log.WithName("resourcewatcher-test-success")
+	watch, err := resourcewatcher.NewWithClient(fake, logger).WatchResource(ctx, nn, gvk)
+	if err != nil {
+		t.Fatalf("got unexpected error from WatchResource: %v", err)
+	}
+	if _, err := fake.Resource(k8s.ToGVR(gvk)).
+		Namespace(nn.Namespace).
+		Create(ctx, notReadyResourceUnstructured, metav1.CreateOptions{}); err != nil {
+		t.Fatalf("creating resource: %v", err)
+	}
+
+	// We want to ensure that the watch waits for the resource to be deleted.
+	deleted := false
+
+	errChan := make(chan error, 1)
+	go func() {
+		ctx, cancel := context.WithTimeout(ctx, time.Minute)
+		defer cancel()
+		if err := resourcewatcher.WaitForResourceToBeReadyOrDeletedViaWatch(ctx, watch, logger); err != nil {
+			errChan <- fmt.Errorf("got unexpected error: %w", err)
+			return
+		}
+
+		if !deleted {
+			errChan <- errors.New("wait finished before resource was deleted")
+			return
+		}
+
+		errChan <- nil
+	}()
+
+	if err := fake.Resource(k8s.ToGVR(gvk)).
+		Namespace(nn.Namespace).
+		Delete(ctx, nn.Name, metav1.DeleteOptions{}); err != nil {
+		t.Fatalf("deleting resource: %v", err)
+	}
+	deleted = true
+
+	if err := <-errChan; err != nil {
+		t.Fatalf("error from watch: %v", err)
+	}
+}
+
 func newResourceUnstructured(status map[string]interface{}) *unstructured.Unstructured {
 	u := &unstructured.Unstructured{
 		Object: map[string]interface{}{

pkg/controller/tf/controller.go

@@ -460,7 +460,7 @@ func (r *Reconciler) handleUnresolvableDeps(ctx context.Context, resource *k8s.R
 		ctx, cancel := context.WithTimeout(ctx, timeoutPeriod)
 		defer cancel()
 		logger.Info("starting wait with timeout on resource's reference", "timeout", timeoutPeriod)
-		if err := watcher.WaitForResourceToBeReady(ctx, refNN, refGVK); err != nil {
+		if err := watcher.WaitForResourceToBeReadyOrDeleted(ctx, refNN, refGVK); err != nil {
 			logger.Error(err, "error while waiting for resource's reference to be ready")
 			return
 		}