Repository: numaproj/numaflow
Pull request #1393 (2023-11-28 01:07:02)

feat: improve numaflow k8s events
Explain what this PR does.

Currently Numaflow is emitting events for the pipeline and vertex controllers. This PR adds events for creating updating and deleting resources for the pipeline, vertex and now isbsvcs. 

<!--

Before you push your changes:

* Run `make pre-push -B` to fix codegen and lint problems.

Then, you MUST:

* Sign-off your commit (otherwise the DCO check will fail).
* Use [a conventional commit message](https://www.conventionalcommits.org/en/v1.0.0/) (otherwise the commit message check will fail).

If you did not do this, reset all your commit and replace them with a single commit:

```
git reset HEAD~1 ;# change 1 to how many commits you made
git commit --signoff -m 'feat: my feat. Fixes #1234'
```

When creating your PR: 

* Make sure that "Fixes #" or "Closes #" is in both the PR title (for release notes) and description (to automatically link and close the issue).
* Say how you tested your changes. If you changed the UI, attach screenshots.
* Set your PR as a draft initially.
* Your PR needs to pass the required checks before it can be approved. 
* Once required tests have passed, mark your PR "Ready for review".

If changes were requested, once you've made them, you MUST dismiss the review to get it reviewed again.

-->



Patch:
diff --git a/pkg/reconciler/cmd/start.go b/pkg/reconciler/cmd/start.go
index fe1c9d78..ae1d7ab2 100644
--- a/pkg/reconciler/cmd/start.go
+++ b/pkg/reconciler/cmd/start.go
@@ -94,7 +94,7 @@ func Start(namespaced bool, managedNamespace string) {
 	}
 
 	isbSvcController, err := controller.New(dfv1.ControllerISBSvc, mgr, controller.Options{
-		Reconciler: isbsvcctrl.NewReconciler(mgr.GetClient(), kubeClient, mgr.GetScheme(), config, logger),
+		Reconciler: isbsvcctrl.NewReconciler(mgr.GetClient(), kubeClient, mgr.GetScheme(), config, logger, mgr.GetEventRecorderFor(dfv1.ControllerISBSvc)),
 	})
 	if err != nil {
 		logger.Fatalw("Unable to set up ISB controller", zap.Error(err))
diff --git a/pkg/reconciler/isbsvc/controller.go b/pkg/reconciler/isbsvc/controller.go
index 89e83250..d55efe56 100644
--- a/pkg/reconciler/isbsvc/controller.go
+++ b/pkg/reconciler/isbsvc/controller.go
@@ -24,6 +24,7 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/record"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
@@ -44,12 +45,13 @@ type interStepBufferServiceReconciler struct {
 	kubeClient kubernetes.Interface
 	scheme     *runtime.Scheme
 
-	config *reconciler.GlobalConfig
-	logger *zap.SugaredLogger
+	config   *reconciler.GlobalConfig
+	logger   *zap.SugaredLogger
+	recorder record.EventRecorder
 }
 
-func NewReconciler(client client.Client, kubeClient kubernetes.Interface, scheme *runtime.Scheme, config *reconciler.GlobalConfig, logger *zap.SugaredLogger) reconcile.Reconciler {
-	return &interStepBufferServiceReconciler{client: client, kubeClient: kubeClient, scheme: scheme, config: config, logger: logger}
+func NewReconciler(client client.Client, kubeClient kubernetes.Interface, scheme *runtime.Scheme, config *reconciler.GlobalConfig, logger *zap.SugaredLogger, recorder record.EventRecorder) reconcile.Reconciler {
+	return &interStepBufferServiceReconciler{client: client, kubeClient: kubeClient, scheme: scheme, config: config, logger: logger, recorder: recorder}
 }
 
 func (r *interStepBufferServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
@@ -86,7 +88,7 @@ func (r *interStepBufferServiceReconciler) reconcile(ctx context.Context, isbSvc
 		log.Info("Deleting isbsvc")
 		if controllerutil.ContainsFinalizer(isbSvc, finalizerName) {
 			// Finalizer logic should be added here.
-			if err := installer.Uninstall(ctx, isbSvc, r.client, r.kubeClient, r.config, log); err != nil {
+			if err := installer.Uninstall(ctx, isbSvc, r.client, r.kubeClient, r.config, log, r.recorder); err != nil {
 				log.Errorw("Failed to uninstall", zap.Error(err))
 				return err
 			}
@@ -106,7 +108,7 @@ func (r *interStepBufferServiceReconciler) reconcile(ctx context.Context, isbSvc
 	} else {
 		isbSvc.Status.MarkConfigured()
 	}
-	return installer.Install(ctx, isbSvc, r.client, r.kubeClient, r.config, log)
+	return installer.Install(ctx, isbSvc, r.client, r.kubeClient, r.config, log, r.recorder)
 }
 
 func (r *interStepBufferServiceReconciler) needsUpdate(old, new *dfv1.InterStepBufferService) bool {
diff --git a/pkg/reconciler/isbsvc/installer/installer.go b/pkg/reconciler/isbsvc/installer/installer.go
index 3a1e135d..f92b21f5 100644
--- a/pkg/reconciler/isbsvc/installer/installer.go
+++ b/pkg/reconciler/isbsvc/installer/installer.go
@@ -24,6 +24,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/reconciler"
 	"go.uber.org/zap"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -36,8 +37,8 @@ type Installer interface {
 }
 
 // Install function installs the ISBS
-func Install(ctx context.Context, isbSvc *dfv1.InterStepBufferService, client client.Client, kubeClient kubernetes.Interface, config *reconciler.GlobalConfig, logger *zap.SugaredLogger) error {
-	installer, err := getInstaller(isbSvc, client, kubeClient, config, logger)
+func Install(ctx context.Context, isbSvc *dfv1.InterStepBufferService, client client.Client, kubeClient kubernetes.Interface, config *reconciler.GlobalConfig, logger *zap.SugaredLogger, recorder record.EventRecorder) error {
+	installer, err := getInstaller(isbSvc, client, kubeClient, config, logger, recorder)
 	if err != nil {
 		logger.Errorw("failed to get an installer", zap.Error(err))
 		return err
@@ -52,7 +53,7 @@ func Install(ctx context.Context, isbSvc *dfv1.InterStepBufferService, client cl
 }
 
 // GetInstaller returns Installer implementation
-func getInstaller(isbSvc *dfv1.InterStepBufferService, client client.Client, kubeClient kubernetes.Interface, config *reconciler.GlobalConfig, logger *zap.SugaredLogger) (Installer, error) {
+func getInstaller(isbSvc *dfv1.InterStepBufferService, client client.Client, kubeClient kubernetes.Interface, config *reconciler.GlobalConfig, logger *zap.SugaredLogger, recorder record.EventRecorder) (Installer, error) {
 	labels := map[string]string{
 		dfv1.KeyPartOf:     dfv1.Project,
 		dfv1.KeyManagedBy:  dfv1.ControllerISBSvc,
@@ -64,11 +65,11 @@ func getInstaller(isbSvc *dfv1.InterStepBufferService, client client.Client, kub
 		if redis.External != nil {
 			return NewExternalRedisInstaller(isbSvc, logger), nil
 		} else if redis.Native != nil {
-			return NewNativeRedisInstaller(client, kubeClient, isbSvc, config, labels, logger), nil
+			return NewNativeRedisInstaller(client, kubeClient, isbSvc, config, labels, logger, recorder), nil
 		}
 	} else if js := isbSvc.Spec.JetStream; js != nil {
 		labels[dfv1.KeyISBSvcType] = string(dfv1.ISBSvcTypeJetStream)
-		return NewJetStreamInstaller(client, kubeClient, isbSvc, config, labels, logger), nil
+		return NewJetStreamInstaller(client, kubeClient, isbSvc, config, labels, logger, recorder), nil
 	}
 	return nil, fmt.Errorf("invalid isb service spec")
 }
@@ -81,8 +82,8 @@ func getInstaller(isbSvc *dfv1.InterStepBufferService, client client.Client, kub
 // separately.
 //
 // It could also be used to check if the ISB Service object can be safely deleted.
-func Uninstall(ctx context.Context, isbSvc *dfv1.InterStepBufferService, client client.Client, kubeClient kubernetes.Interface, config *reconciler.GlobalConfig, logger *zap.SugaredLogger) error {
-	installer, err := getInstaller(isbSvc, client, kubeClient, config, logger)
+func Uninstall(ctx context.Context, isbSvc *dfv1.InterStepBufferService, client client.Client, kubeClient kubernetes.Interface, config *reconciler.GlobalConfig, logger *zap.SugaredLogger, recorder record.EventRecorder) error {
+	installer, err := getInstaller(isbSvc, client, kubeClient, config, logger, recorder)
 	if err != nil {
 		logger.Errorw("Failed to get an installer", zap.Error(err))
 		return err
diff --git a/pkg/reconciler/isbsvc/installer/jetstream.go b/pkg/reconciler/isbsvc/installer/jetstream.go
index 5ebee712..03caf1e3 100644
--- a/pkg/reconciler/isbsvc/installer/jetstream.go
+++ b/pkg/reconciler/isbsvc/installer/jetstream.go
@@ -38,6 +38,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/yaml"
 )
@@ -61,9 +62,10 @@ type jetStreamInstaller struct {
 	config     *reconciler.GlobalConfig
 	labels     map[string]string
 	logger     *zap.SugaredLogger
+	recorder   record.EventRecorder
 }
 
-func NewJetStreamInstaller(client client.Client, kubeClient kubernetes.Interface, isbSvc *dfv1.InterStepBufferService, config *reconciler.GlobalConfig, labels map[string]string, logger *zap.SugaredLogger) Installer {
+func NewJetStreamInstaller(client client.Client, kubeClient kubernetes.Interface, isbSvc *dfv1.InterStepBufferService, config *reconciler.GlobalConfig, labels map[string]string, logger *zap.SugaredLogger, recorder record.EventRecorder) Installer {
 	return &jetStreamInstaller{
 		client:     client,
 		kubeClient: kubeClient,
@@ -71,6 +73,7 @@ func NewJetStreamInstaller(client client.Client, kubeClient kubernetes.Interface
 		config:     config,
 		labels:     labels,
 		logger:     logger.With("isbsvc", isbSvc.Name),
+		recorder:   recorder,
 	}
 }
 
@@ -104,21 +107,25 @@ func (r *jetStreamInstaller) Install(ctx context.Context) (*dfv1.BufferServiceCo
 	if err := r.createSecrets(ctx); err != nil {
 		r.logger.Errorw("Failed to create jetstream auth secrets", zap.Error(err))
 		r.isbSvc.Status.MarkDeployFailed("JetStreamAuthSecretsFailed", err.Error())
+		r.recorder.Eventf(r.isbSvc, corev1.EventTypeWarning, "JetStreamAuthSecretsFailed", "Failed to create jetstream auth secrets: %v", err.Error())
 		return nil, err
 	}
 	if err := r.createConfigMap(ctx); err != nil {
 		r.logger.Errorw("Failed to create jetstream ConfigMap", zap.Error(err))
 		r.isbSvc.Status.MarkDeployFailed("JetStreamConfigMapFailed", err.Error())
+		r.recorder.Eventf(r.isbSvc, corev1.EventTypeWarning, "JetStreamConfigMapFailed", "Failed to create jetstream ConfigMap: %v", err.Error())
 		return nil, err
 	}
 	if err := r.createService(ctx); err != nil {
 		r.logger.Errorw("Failed to create jetstream Service", zap.Error(err))
 		r.isbSvc.Status.MarkDeployFailed("JetStreamServiceFailed", err.Error())
+		r.recorder.Eventf(r.isbSvc, corev1.EventTypeWarning, "JetStreamServiceFailed", "Failed to create jetstream Service: %v", err.Error())
 		return nil, err
 	}
 	if err := r.createStatefulSet(ctx); err != nil {
 		r.logger.Errorw("Failed to create jetstream StatefulSet", zap.Error(err))
 		r.isbSvc.Status.MarkDeployFailed("JetStreamStatefulSetFailed", err.Error())
+		r.recorder.Eventf(r.isbSvc, corev1.EventTypeWarning, "JetStreamStatefulSetFailed", "Failed to create jetstream StatefulSet: %v", err.Error())
 		return nil, err
 	}
 	r.isbSvc.Status.MarkDeployed()
@@ -177,6 +184,7 @@ func (r *jetStreamInstaller) createService(ctx context.Context) error {
 				return fmt.Errorf("failed to create jetstream service, err: %w", err)
 			}
 			r.logger.Info("Created jetstream service successfully")
+			r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "JetStreamServiceSuccess", "Created jetstream service successfully")
 			return nil
 		} else {
 			return fmt.Errorf("failed to check if jetstream service is existing, err: %w", err)
@@ -189,6 +197,7 @@ func (r *jetStreamInstaller) createService(ctx context.Context) error {
 			return fmt.Errorf("failed to update jetstream service, err: %w", err)
 		}
 		r.logger.Info("Updated jetstream service successfully")
+		r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "JetStreamServiceSuccess", "Updated jetstream service successfully")
 	}
 	return nil
 }
@@ -236,6 +245,7 @@ func (r *jetStreamInstaller) createStatefulSet(ctx context.Context) error {
 				return fmt.Errorf("failed to create jetstream statefulset, err: %w", err)
 			}
 			r.logger.Info("Created jetstream statefulset successfully")
+			r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "JetStreamStatefulSetSuccess", "Created jetstream stateful successfully")
 			return nil
 		} else {
 			return fmt.Errorf("failed to check if jetstream statefulset is existing, err: %w", err)
@@ -248,6 +258,7 @@ func (r *jetStreamInstaller) createStatefulSet(ctx context.Context) error {
 			return fmt.Errorf("failed to update jetstream statefulset, err: %w", err)
 		}
 		r.logger.Info("Updated jetstream statefulset successfully")
+		r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "JetStreamStatefulSet", "Updated jetstream statefulset successfully")
 	}
 	return nil
 }
@@ -482,6 +493,7 @@ func (r *jetStreamInstaller) createConfigMap(ctx context.Context) error {
 				return fmt.Errorf("failed to create jetstream configmap, err: %w", err)
 			}
 			r.logger.Info("Created jetstream configmap successfully")
+			r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "JetStreamConfigMap", "Created jetstream configmap successfully")
 			return nil
 		} else {
 			return fmt.Errorf("failed to check if jetstream configmap is existing, err: %w", err)
@@ -494,6 +506,7 @@ func (r *jetStreamInstaller) createConfigMap(ctx context.Context) error {
 			return fmt.Errorf("failed to update jetstream configmap, err: %w", err)
 		}
 		r.logger.Info("Updated jetstream configmap successfully")
+		r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "JetStreamConfigMap", "Updated jetstream configmap successfully")
 	}
 	return nil
 }
diff --git a/pkg/reconciler/isbsvc/installer/native_redis.go b/pkg/reconciler/isbsvc/installer/native_redis.go
index aef4dd38..f1651dd1 100644
--- a/pkg/reconciler/isbsvc/installer/native_redis.go
+++ b/pkg/reconciler/isbsvc/installer/native_redis.go
@@ -33,6 +33,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -59,9 +60,10 @@ type redisInstaller struct {
 	config     *reconciler.GlobalConfig
 	labels     map[string]string
 	logger     *zap.SugaredLogger
+	recorder   record.EventRecorder
 }
 
-func NewNativeRedisInstaller(client client.Client, kubeClient kubernetes.Interface, isbSvc *dfv1.InterStepBufferService, config *reconciler.GlobalConfig, labels map[string]string, logger *zap.SugaredLogger) Installer {
+func NewNativeRedisInstaller(client client.Client, kubeClient kubernetes.Interface, isbSvc *dfv1.InterStepBufferService, config *reconciler.GlobalConfig, labels map[string]string, logger *zap.SugaredLogger, recorder record.EventRecorder) Installer {
 	return &redisInstaller{
 		client:     client,
 		kubeClient: kubeClient,
@@ -69,6 +71,7 @@ func NewNativeRedisInstaller(client client.Client, kubeClient kubernetes.Interfa
 		config:     config,
 		labels:     labels,
 		logger:     logger.With("isbsvc", isbSvc.Name),
+		recorder:   recorder,
 	}
 }
 
@@ -79,16 +82,19 @@ func (r *redisInstaller) Install(ctx context.Context) (*dfv1.BufferServiceConfig
 	r.isbSvc.Status.SetType(dfv1.ISBSvcTypeRedis)
 	if err := r.createScriptsConfigMap(ctx); err != nil {
 		r.logger.Errorw("Failed to create Redis scripts ConfigMap", zap.Error(err))
+		r.recorder.Eventf(r.isbSvc, corev1.EventTypeWarning, "RedisScriptsConfigMapFailed", "Failed to create Redis scripts ConfigMap: %v", err.Error())
 		r.isbSvc.Status.MarkDeployFailed("RedisScriptsConfigMapFailed", err.Error())
 		return nil, err
 	}
 	if err := r.createHealthConfigMap(ctx); err != nil {
 		r.logger.Errorw("Failed to create Redis health ConfigMap", zap.Error(err))
+		r.recorder.Eventf(r.isbSvc, corev1.EventTypeWarning, "RedisHealthConfigMapFailed", "Failed to create Redis health ConfigMap: %v", err.Error())
 		r.isbSvc.Status.MarkDeployFailed("RedisHealthConfigMapFailed", err.Error())
 		return nil, err
 	}
 	if err := r.createConfConfigMap(ctx); err != nil {
 		r.logger.Errorw("Failed to create Redis config ConfigMap", zap.Error(err))
+		r.recorder.Eventf(r.isbSvc, corev1.EventTypeWarning, "RedisConfConfigMapFailed", "Failed to create Redis config ConfigMap: %v", err.Error())
 		r.isbSvc.Status.MarkDeployFailed("RedisConfConfigMapFailed", err.Error())
 		return nil, err
 	}
@@ -99,16 +105,19 @@ func (r *redisInstaller) Install(ctx context.Context) (*dfv1.BufferServiceConfig
 	}
 	if err := r.createRedisService(ctx); err != nil {
 		r.logger.Errorw("Failed to create Redis Service", zap.Error(err))
+		r.recorder.Eventf(r.isbSvc, corev1.EventTypeWarning, "RedisServiceFailed", "Failed to create Redis Serivce: %v", err.Error())
 		r.isbSvc.Status.MarkDeployFailed("RedisServiceFailed", err.Error())
 		return nil, err
 	}
 	if err := r.createRedisHeadlessService(ctx); err != nil {
 		r.logger.Errorw("Failed to create Redis Headless Service", zap.Error(err))
+		r.recorder.Eventf(r.isbSvc, corev1.EventTypeWarning, "RedisHeadlessServiceFailed", "Failed to create Redis Headless Service: %v", err.Error())
 		r.isbSvc.Status.MarkDeployFailed("RedisHeadlessServiceFailed", err.Error())
 		return nil, err
 	}
 	if err := r.createStatefulSet(ctx); err != nil {
 		r.logger.Errorw("Failed to create Redis StatefulSet", zap.Error(err))
+		r.recorder.Eventf(r.isbSvc, corev1.EventTypeWarning, "RedisStatefulSetFailed", "Failed to create Redis StatefulSet: %v", err.Error())
 		r.isbSvc.Status.MarkDeployFailed("RedisStatefulSetFailed", err.Error())
 		return nil, err
 	}
@@ -219,6 +228,7 @@ func (r *redisInstaller) createScriptsConfigMap(ctx context.Context) error {
 				return fmt.Errorf("failed to create redis scripts configmap, err: %w", err)
 			}
 			r.logger.Info("Created redis scripts configmap successfully")
+			r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "RedisScriptsConfigMapSuccess", "Created redis scripts configmap successfully")
 			return nil
 		} else {
 			return fmt.Errorf("failed to check if redis scripts configmap is existing, err: %w", err)
@@ -231,6 +241,7 @@ func (r *redisInstaller) createScriptsConfigMap(ctx context.Context) error {
 			return fmt.Errorf("failed to update redis scripts configmap, err: %w", err)
 		}
 		r.logger.Info("Updated redis scripts configmap successfully")
+		r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "RedisScriptsConfigMapSuccess", "Updated redis scripts configmap successfully")
 	}
 	return nil
 }
@@ -268,6 +279,7 @@ func (r *redisInstaller) createHealthConfigMap(ctx context.Context) error {
 				return fmt.Errorf("failed to create redis health configmap, err: %w", err)
 			}
 			r.logger.Info("Created redis health configmap successfully")
+			r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "RedisHealthConfigMapSuccess", "Created redis health configmap successfully")
 			return nil
 		} else {
 			return fmt.Errorf("failed to check if redis health configmap is existing, err: %w", err)
@@ -280,6 +292,7 @@ func (r *redisInstaller) createHealthConfigMap(ctx context.Context) error {
 			return fmt.Errorf("failed to update redis health configmap, err: %w", err)
 		}
 		r.logger.Info("Updated redis health configmap successfully")
+		r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "RedisHealthConfigMapSuccess", "Updated redis health configmap successfully")
 	}
 	return nil
 }
@@ -394,6 +407,7 @@ func (r *redisInstaller) createConfConfigMap(ctx context.Context) error {
 				return fmt.Errorf("failed to create redis config configmap, err: %w", err)
 			}
 			r.logger.Info("Created redis config configmap successfully")
+			r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "RedisConfConfigMapSuccess", "Created redis config configmap successfully")
 			return nil
 		} else {
 			return fmt.Errorf("failed to check if redis config configmap is existing, err: %w", err)
@@ -406,6 +420,7 @@ func (r *redisInstaller) createConfConfigMap(ctx context.Context) error {
 			return fmt.Errorf("failed to update redis config configmap, err: %w", err)
 		}
 		r.logger.Info("Updated redis config configmap successfully")
+		r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "RedisConfConfigMapSuccess", "Updated redis config configmap successfully")
 	}
 	return nil
 }
@@ -438,6 +453,7 @@ func (r *redisInstaller) createRedisService(ctx context.Context) error {
 				return fmt.Errorf("failed to create redis service, err: %w", err)
 			}
 			r.logger.Info("Created redis service successfully")
+			r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "RedisServiceSuccess", "Created redis service successfully")
 			return nil
 		} else {
 			return fmt.Errorf("failed to check if redis service is existing, err: %w", err)
@@ -450,6 +466,7 @@ func (r *redisInstaller) createRedisService(ctx context.Context) error {
 			return fmt.Errorf("failed to update redis service, err: %w", err)
 		}
 		r.logger.Info("Updated redis service successfully")
+		r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "RedisServiceSuccess", "Updated redis service successfully")
 	}
 	return nil
 }
@@ -482,6 +499,7 @@ func (r *redisInstaller) createRedisHeadlessService(ctx context.Context) error {
 				return fmt.Errorf("failed to create redis headless service, err: %w", err)
 			}
 			r.logger.Info("Created redis headless service successfully")
+			r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "RedisHeadlessServiceSuccess", "Created redis headless service successfully")
 			return nil
 		} else {
 			return fmt.Errorf("failed to check if redis headless service is existing, err: %w", err)
@@ -494,6 +512,7 @@ func (r *redisInstaller) createRedisHeadlessService(ctx context.Context) error {
 			return fmt.Errorf("failed to update redis headless service, err: %w", err)
 		}
 		r.logger.Info("Updated redis headless service successfully")
+		r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "RedisHeadlessServiceSuccess", "Updated redis headless service successfully")
 	}
 	return nil
 }
@@ -542,6 +561,7 @@ func (r *redisInstaller) createStatefulSet(ctx context.Context) error {
 				return fmt.Errorf("failed to create redis statefulset, err: %w", err)
 			}
 			r.logger.Info("Created redis statefulset successfully")
+			r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "RedisStatefulSetSuccess", "Created redis statefulset successfully")
 			return nil
 		} else {
 			return fmt.Errorf("failed to check if redis statefulset is existing, err: %w", err)
@@ -554,6 +574,7 @@ func (r *redisInstaller) createStatefulSet(ctx context.Context) error {
 			return fmt.Errorf("failed to update redis statefulset, err: %w", err)
 		}
 		r.logger.Info("Updated redis statefulset successfully")
+		r.recorder.Event(r.isbSvc, corev1.EventTypeNormal, "RedisStatefulSetSuccess", "Updated redis statefulset successfully")
 	}
 	return nil
 }
diff --git a/pkg/reconciler/pipeline/controller.go b/pkg/reconciler/pipeline/controller.go
index 56418674..7444198d 100644
--- a/pkg/reconciler/pipeline/controller.go
+++ b/pkg/reconciler/pipeline/controller.go
@@ -104,9 +104,8 @@ func (r *pipelineReconciler) reconcile(ctx context.Context, pl *dfv1.Pipeline) (
 			if time.Now().Before(pl.DeletionTimestamp.Add(time.Duration(pl.Spec.Lifecycle.GetDeleteGracePeriodSeconds()) * time.Second)) {
 				safeToDelete, err := r.safeToDelete(ctx, pl)
 				if err != nil {
-					logMsg := fmt.Sprintf("Failed to check if it's safe to delete the pipeline: %v", err.Error())
+					logMsg := fmt.Sprintf("Failed to check if it's safe to delete pipeline %s: %v", pl.Name, err.Error())
 					log.Error(logMsg)
-					r.recorder.Event(pl, corev1.EventTypeWarning, "ReconcilePipelineFailed", logMsg)
 					return ctrl.Result{}, err
 				}
 
@@ -205,6 +204,7 @@ func (r *pipelineReconciler) reconcileNonLifecycleChanges(ctx context.Context, p
 	if err := r.createOrUpdateSIMDeployments(ctx, pl, isbSvc.Status.Config); err != nil {
 		log.Errorw("Failed to create or update Side Inputs Manager deployments", zap.Error(err))
 		pl.Status.MarkDeployFailed("CreateOrUpdateSIMDeploymentsFailed", err.Error())
+		r.recorder.Eventf(pl, corev1.EventTypeWarning, "CreateOrUpdateSIMDeploymentsFailed", "Failed to create or update Side Inputs Manager deployments: %w", err.Error())
 		return ctrl.Result{}, err
 	}
 
@@ -251,19 +251,23 @@ func (r *pipelineReconciler) reconcileNonLifecycleChanges(ctx context.Context, p
 					continue
 				} else {
 					pl.Status.MarkDeployFailed("CreateVertexFailed", err.Error())
+					r.recorder.Eventf(pl, corev1.EventTypeWarning, "CreateVertexFailed", "Failed to create vertex: %w", err.Error())
 					return ctrl.Result{}, fmt.Errorf("failed to create vertex, err: %w", err)
 				}
 			}
 			log.Infow("Created vertex successfully", zap.String("vertex", vertexName))
+			r.recorder.Eventf(pl, corev1.EventTypeNormal, "CreateVertexSuccess", "Created vertex %s successfully", vertexName)
 		} else {
 			if oldObj.GetAnnotations()[dfv1.KeyHash] != newObj.GetAnnotations()[dfv1.KeyHash] { // need to update
 				oldObj.Spec = newObj.Spec
 				oldObj.Annotations[dfv1.KeyHash] = newObj.GetAnnotations()[dfv1.KeyHash]
 				if err := r.client.Update(ctx, &oldObj); err != nil {
 					pl.Status.MarkDeployFailed("UpdateVertexFailed", err.Error())
+					r.recorder.Eventf(pl, corev1.EventTypeWarning, "UpdateVertexFailed", "Failed to update vertex: %w", err.Error())
 					return ctrl.Result{}, fmt.Errorf("failed to update vertex, err: %w", err)
 				}
 				log.Infow("Updated vertex successfully", zap.String("vertex", vertexName))
+				r.recorder.Eventf(pl, corev1.EventTypeNormal, "UpdateVertexSuccess", "Updated vertex %s successfully", vertexName)
 			}
 			delete(existingObjs, vertexName)
 		}
@@ -271,9 +275,11 @@ func (r *pipelineReconciler) reconcileNonLifecycleChanges(ctx context.Context, p
 	for _, v := range existingObjs {
 		if err := r.client.Delete(ctx, &v); err != nil {
 			pl.Status.MarkDeployFailed("DeleteStaleVertexFailed", err.Error())
+			r.recorder.Eventf(pl, corev1.EventTypeWarning, "DeleteStaleVertexFailed", "Failed to delete vertex: %w", err.Error())
 			return ctrl.Result{}, fmt.Errorf("failed to delete vertex, err: %w", err)
 		}
 		log.Infow("Deleted stale vertex successfully", zap.String("vertex", v.Name))
+		r.recorder.Eventf(pl, corev1.EventTypeNormal, "DeleteStaleVertexSuccess", "Deleted stale vertex %s successfully", v.Name)
 	}
 
 	// create batch job
@@ -347,6 +353,7 @@ func (r *pipelineReconciler) createOrUpdateDaemonService(ctx context.Context, pl
 		if err := r.client.Delete(ctx, existingSvc); err != nil && !apierrors.IsNotFound(err) {
 			log.Errorw("Failed to delete existing daemon service", zap.String("service", existingSvc.Name), zap.Error(err))
 			pl.Status.MarkDeployFailed("DelDaemonSvcFailed", err.Error())
+			r.recorder.Eventf(pl, corev1.EventTypeWarning, "DelDaemonSvcFailed", "Failed to delete existing daemon service: %w", err.Error())
 			return fmt.Errorf("failed to delete existing daemon service, %w", err)
 		}
 		needToCreatDaemonSvc = true
@@ -355,9 +362,11 @@ func (r *pipelineReconciler) createOrUpdateDaemonService(ctx context.Context, pl
 		if err := r.client.Create(ctx, svc); err != nil {
 			log.Errorw("Failed to create daemon service", zap.String("service", svc.Name), zap.Error(err))
 			pl.Status.MarkDeployFailed("CreateDaemonSvcFailed", err.Error())
+			r.recorder.Eventf(pl, corev1.EventTypeWarning, "CreateDaemonSvcFailed", "Failed to create daemon service: %w", err.Error())
 			return fmt.Errorf("failed to create daemon service, %w", err)
 		}
 		log.Infow("Succeeded to create a daemon service", zap.String("service", svc.Name))
+		r.recorder.Eventf(pl, corev1.EventTypeNormal, "CreateDaemonSvcSuccess", "Succeeded to create daemon service %s", svc.Name)
 	}
 	return nil
 }
@@ -395,6 +404,7 @@ func (r *pipelineReconciler) createOrUpdateDaemonDeployment(ctx context.Context,
 			if err := r.client.Delete(ctx, existingDeploy); err != nil {
 				log.Errorw("Failed to delete the outdated daemon deployment", zap.String("deployment", existingDeploy.Name), zap.Error(err))
 				pl.Status.MarkDeployFailed("DeleteOldDaemonDeployFailed", err.Error())
+				r.recorder.Eventf(pl, corev1.EventTypeWarning, "DeleteOldDaemonDeployFailed", "Failed to delete the outdated daemon deployment: %w", err.Error())
 				return fmt.Errorf("failed to delete an outdated daemon deployment, %w", err)
 			}
 			needToCreate = true
@@ -404,9 +414,11 @@ func (r *pipelineReconciler) createOrUpdateDaemonDeployment(ctx context.Context,
 		if err := r.client.Create(ctx, deploy); err != nil && !apierrors.IsAlreadyExists(err) {
 			log.Errorw("Failed to create a daemon deployment", zap.String("deployment", deploy.Name), zap.Error(err))
 			pl.Status.MarkDeployFailed("CreateDaemonDeployFailed", err.Error())
+			r.recorder.Eventf(pl, corev1.EventTypeWarning, "CreateDaemonDeployFailed", "Failed to create a daemon deployment: %w", err.Error())
 			return fmt.Errorf("failed to create a daemon deployment, %w", err)
 		}
-		log.Infow("Succeeded to created/recreatd a daemon deployment", zap.String("deployment", deploy.Name))
+		log.Infow("Succeeded to create/recreate a daemon deployment", zap.String("deployment", deploy.Name))
+		r.recorder.Eventf(pl, corev1.EventTypeNormal, "CreateDaemonDeploySuccess", "Succeeded to create/recreate daemon deployment %s", deploy.Name)
 	}
 	return nil
 }
@@ -476,6 +488,7 @@ func (r *pipelineReconciler) createOrUpdateSIMDeployments(ctx context.Context, p
 				}
 			}
 			log.Infow("Succeeded to create/recreate Side Inputs Manager Deployment", zap.String("deployment", newObj.Name))
+			r.recorder.Event(pl, corev1.EventTypeNormal, "CreateSIMDeployment", "Succeeded to create/recreate Side Inputs Manager Deployment")
 		}
 	}
 	for _, v := range existingObjs {
diff --git a/pkg/reconciler/vertex/controller.go b/pkg/reconciler/vertex/controller.go
index a6ccbfce..d542eb7e 100644
--- a/pkg/reconciler/vertex/controller.go
+++ b/pkg/reconciler/vertex/controller.go
@@ -105,14 +105,15 @@ func (r *vertexReconciler) reconcile(ctx context.Context, vertex *dfv1.Vertex) (
 		if apierrors.IsNotFound(err) {
 			e := fmt.Errorf("isbsvc %s not found", isbSvcName)
 			vertex.Status.MarkPhaseFailed("ISBSvcNotFound", e.Error())
-			r.recorder.Event(vertex, corev1.EventTypeWarning, "ISBSvcNotFound", e.Error())
 			return ctrl.Result{}, e
 		}
-		r.markPhaseLogEvent(vertex, log, "FindISBSvcFailed", err.Error(), "Failed to get ISB Service", zap.String("isbsvc", isbSvcName), zap.Error(err))
+		log.Errorw("Failed to get ISB Service", zap.String("isbsvc", isbSvcName), zap.Error(err))
+		vertex.Status.MarkPhaseFailed("FindISBSvcFailed", err.Error())
 		return ctrl.Result{}, err
 	}
 	if !isbSvc.Status.IsReady() {
-		r.markPhaseLogEvent(vertex, log, "ISBSvcNotReady", "isbsvc not ready", "isbsvc not ready", zap.String("isbsvc", isbSvcName))
+		log.Errorw("ISB Service is not in ready status", zap.String("isbsvc", isbSvcName), zap.Error(err))
+		vertex.Status.MarkPhaseFailed("ISBSvcNotReady", "isbsvc not ready")
 		return ctrl.Result{}, fmt.Errorf("isbsvc not ready")
 	}
 
@@ -127,7 +128,8 @@ func (r *vertexReconciler) reconcile(ctx context.Context, vertex *dfv1.Vertex) (
 			for i := 0; i < desiredReplicas; i++ {
 				newPvc, err := r.buildReduceVertexPVCSpec(vertex, i)
 				if err != nil {
-					r.markPhaseLogEvent(vertex, log, "BuildPVCSpecFailed", err.Error(), "Error building a PVC spec", zap.Error(err))
+					log.Errorw("Error building a PVC spec", zap.Error(err))
+					vertex.Status.MarkPhaseFailed("BuildPVCSpecFailed", err.Error())
 					return ctrl.Result{}, err
 				}
 				hash := sharedutil.MustHash(newPvc.Spec)
@@ -135,13 +137,15 @@ func (r *vertexReconciler) reconcile(ctx context.Context, vertex *dfv1.Vertex) (
 				existingPvc := &corev1.PersistentVolumeClaim{}
 				if err := r.client.Get(ctx, types.NamespacedName{Namespace: vertex.Namespace, Name: newPvc.Name}, existingPvc); err != nil {
 					if !apierrors.IsNotFound(err) {
-						r.markPhaseLogEvent(vertex, log, "FindExistingPVCFailed", err.Error(), "Error finding existing PVC", zap.Error(err))
+						log.Errorw("Error finding existing PVC", zap.Error(err))
+						vertex.Status.MarkPhaseFailed("FindExistingPVCFailed", err.Error())
 						return ctrl.Result{}, err
 					}
 					if err := r.client.Create(ctx, newPvc); err != nil && !apierrors.IsAlreadyExists(err) {
 						r.markPhaseLogEvent(vertex, log, "CreatePVCFailed", err.Error(), "Error creating a PVC", zap.Error(err))
 						return ctrl.Result{}, err
 					}
+					r.recorder.Eventf(vertex, corev1.EventTypeNormal, "CreatePVCSuccess", "Successfully created PVC %s", newPvc.Name)
 				} else {
 					if existingPvc.GetAnnotations()[dfv1.KeyHash] != hash {
 						// TODO: deal with spec difference
@@ -162,7 +166,8 @@ func (r *vertexReconciler) reconcile(ctx context.Context, vertex *dfv1.Vertex) (
 	// E.g., the vertex processing rate calculation relies on the headless service to determine the number of active pods.
 	existingSvcs, err := r.findExistingServices(ctx, vertex)
 	if err != nil {
-		r.markPhaseLogEvent(vertex, log, "FindExistingSvcsFailed", err.Error(), "Failed to find existing services", zap.Error(err))
+		log.Errorw("Failed to find existing services", zap.Error(err))
+		vertex.Status.MarkPhaseFailed("FindExistingSvcsFailed", err.Error())
 		return ctrl.Result{}, err
 	}
 	for _, s := range vertex.GetServiceObjs() {
@@ -178,6 +183,7 @@ func (r *vertexReconciler) reconcile(ctx context.Context, vertex *dfv1.Vertex) (
 					}
 				} else {
 					log.Infow("Deleted a stale service to recreate", zap.String("service", existingSvc.Name))
+					r.recorder.Eventf(vertex, corev1.EventTypeNormal, "DelSvcSuccess", "Deleted stale service %s to recreate", existingSvc.Name)
 				}
 				needToCreate = true
 			}
@@ -194,6 +200,7 @@ func (r *vertexReconciler) reconcile(ctx context.Context, vertex *dfv1.Vertex) (
 				return ctrl.Result{}, err
 			} else {
 				log.Infow("Succeeded to create a service", zap.String("service", s.Name))
+				r.recorder.Eventf(vertex, corev1.EventTypeNormal, "CreateSvcSuccess", "Succeeded to create service %s", s.Name)
 			}
 		}
 	}
@@ -205,24 +212,28 @@ func (r *vertexReconciler) reconcile(ctx context.Context, vertex *dfv1.Vertex) (
 			}
 		} else {
 			log.Infow("Deleted a stale service", zap.String("service", v.Name))
+			r.recorder.Eventf(vertex, corev1.EventTypeNormal, "DelSvcSuccess", "Deleted stale service %s", v.Name)
 		}
 	}
 
 	pipeline := &dfv1.Pipeline{}
 	if err := r.client.Get(ctx, types.NamespacedName{Namespace: vertex.Namespace, Name: vertex.Spec.PipelineName}, pipeline); err != nil {
-		r.markPhaseLogEvent(vertex, log, "GetPipelineFailed", err.Error(), "Failed to get pipeline object", zap.Error(err))
+		log.Errorw("Failed to get pipeline object", zap.Error(err))
+		vertex.Status.MarkPhaseFailed("GetPipelineFailed", err.Error())
 		return ctrl.Result{}, err
 	}
 	// Create pods
 	existingPods, err := r.findExistingPods(ctx, vertex)
 	if err != nil {
-		r.markPhaseLogEvent(vertex, log, "FindExistingPodFailed", err.Error(), "Failed to find existing pods", zap.Error(err))
+		log.Errorw("Failed to find existing pods", zap.Error(err))
+		vertex.Status.MarkPhaseFailed("FindExistingPodFailed", err.Error())
 		return ctrl.Result{}, err
 	}
 	for replica := 0; replica < desiredReplicas; replica++ {
 		podSpec, err := r.buildPodSpec(vertex, pipeline, isbSvc.Status.Config, replica)
 		if err != nil {
-			r.markPhaseLogEvent(vertex, log, "PodSpecGenFailed", err.Error(), "Failed to generate pod spec", zap.Error(err))
+			log.Errorw("Failed to generate pod spec", zap.Error(err))
+			vertex.Status.MarkPhaseFailed("PodSpecGenFailed", err.Error())
 			return ctrl.Result{}, err
 		}
 		hash := sharedutil.MustHash(podSpec)
@@ -295,6 +306,7 @@ func (r *vertexReconciler) reconcile(ctx context.Context, vertex *dfv1.Vertex) (
 				return ctrl.Result{}, err
 			}
 			log.Infow("Succeeded to create a pod", zap.String("pod", pod.Name))
+			r.recorder.Eventf(vertex, corev1.EventTypeNormal, "CreatePodSuccess", "Succeeded to create pod %s", pod.Name)
 		}
 	}
 	for _, v := range existingPods {
@@ -307,6 +319,7 @@ func (r *vertexReconciler) reconcile(ctx context.Context, vertex *dfv1.Vertex) (
 	currentReplicas := int(vertex.Status.Replicas)
 	if currentReplicas != desiredReplicas || vertex.Status.Selector == "" {
 		log.Infow("Replicas changed", "currentReplicas", currentReplicas, "desiredReplicas", desiredReplicas)
+		r.recorder.Eventf(vertex, corev1.EventTypeNormal, "ReplicasScaled", "Replicas changed from %d to %d", currentReplicas, desiredReplicas)
 		vertex.Status.Replicas = uint32(desiredReplicas)
 		vertex.Status.LastScaledAt = metav1.Time{Time: time.Now()}
 	}
@@ -432,6 +445,7 @@ func (r *vertexReconciler) findExistingServices(ctx context.Context, vertex *dfv
 	return result, nil
 }
 
+// helper function for warning event types
 func (r *vertexReconciler) markPhaseLogEvent(vertex *dfv1.Vertex, log *zap.SugaredLogger, reason, message, logMsg string, logWith ...interface{}) {
 	log.Errorw(logMsg, logWith)
 	vertex.Status.MarkPhaseFailed(reason, message)


Test patch:
diff --git a/pkg/reconciler/isbsvc/controller_test.go b/pkg/reconciler/isbsvc/controller_test.go
index 1192db8a..c2ab161e 100644
--- a/pkg/reconciler/isbsvc/controller_test.go
+++ b/pkg/reconciler/isbsvc/controller_test.go
@@ -28,6 +28,7 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	k8sfake "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
@@ -110,7 +111,7 @@ func init() {
 func Test_NewReconciler(t *testing.T) {
 	cl := fake.NewClientBuilder().Build()
 	kubeClient := k8sfake.NewSimpleClientset()
-	r := NewReconciler(cl, kubeClient, scheme.Scheme, fakeConfig, zaptest.NewLogger(t).Sugar())
+	r := NewReconciler(cl, kubeClient, scheme.Scheme, fakeConfig, zaptest.NewLogger(t).Sugar(), record.NewFakeRecorder(64))
 	_, ok := r.(*interStepBufferServiceReconciler)
 	assert.True(t, ok)
 }
@@ -126,6 +127,7 @@ func TestReconcileNativeRedis(t *testing.T) {
 			scheme:     scheme.Scheme,
 			config:     fakeConfig,
 			logger:     zaptest.NewLogger(t).Sugar(),
+			recorder:   record.NewFakeRecorder(64),
 		}
 		err := r.reconcile(ctx, testIsb)
 		assert.NoError(t, err)
@@ -166,6 +168,7 @@ func TestReconcileJetStream(t *testing.T) {
 			scheme:     scheme.Scheme,
 			config:     fakeConfig,
 			logger:     zaptest.NewLogger(t).Sugar(),
+			recorder:   record.NewFakeRecorder(64),
 		}
 		err := r.reconcile(ctx, testIsb)
 		assert.NoError(t, err)
diff --git a/pkg/reconciler/isbsvc/installer/installer_test.go b/pkg/reconciler/isbsvc/installer/installer_test.go
index 60edb52c..796cd139 100644
--- a/pkg/reconciler/isbsvc/installer/installer_test.go
+++ b/pkg/reconciler/isbsvc/installer/installer_test.go
@@ -29,6 +29,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	k8sfake "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
@@ -122,7 +123,7 @@ func init() {
 
 func TestGetInstaller(t *testing.T) {
 	t.Run("get native redis installer", func(t *testing.T) {
-		installer, err := getInstaller(testNativeRedisIsbSvc, nil, nil, fakeConfig, zaptest.NewLogger(t).Sugar())
+		installer, err := getInstaller(testNativeRedisIsbSvc, nil, nil, fakeConfig, zaptest.NewLogger(t).Sugar(), nil)
 		assert.NoError(t, err)
 		assert.NotNil(t, installer)
 		_, ok := installer.(*redisInstaller)
@@ -130,7 +131,7 @@ func TestGetInstaller(t *testing.T) {
 	})
 
 	t.Run("get jetstream installer", func(t *testing.T) {
-		installer, err := getInstaller(testJetStreamIsbSvc, nil, nil, fakeConfig, zaptest.NewLogger(t).Sugar())
+		installer, err := getInstaller(testJetStreamIsbSvc, nil, nil, fakeConfig, zaptest.NewLogger(t).Sugar(), nil)
 		assert.NoError(t, err)
 		assert.NotNil(t, installer)
 		_, ok := installer.(*jetStreamInstaller)
@@ -138,7 +139,7 @@ func TestGetInstaller(t *testing.T) {
 	})
 
 	t.Run("get external redis installer", func(t *testing.T) {
-		installer, err := getInstaller(testExternalRedisIsbSvc, nil, nil, fakeConfig, zaptest.NewLogger(t).Sugar())
+		installer, err := getInstaller(testExternalRedisIsbSvc, nil, nil, fakeConfig, zaptest.NewLogger(t).Sugar(), nil)
 		assert.NoError(t, err)
 		assert.NotNil(t, installer)
 		_, ok := installer.(*externalRedisInstaller)
@@ -148,7 +149,7 @@ func TestGetInstaller(t *testing.T) {
 	t.Run("test error", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
 		testObj.Spec.Redis = nil
-		_, err := getInstaller(testObj, nil, nil, fakeConfig, zaptest.NewLogger(t).Sugar())
+		_, err := getInstaller(testObj, nil, nil, fakeConfig, zaptest.NewLogger(t).Sugar(), nil)
 		assert.Error(t, err)
 	})
 }
@@ -160,14 +161,14 @@ func TestInstall(t *testing.T) {
 	t.Run("test redis error", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
 		testObj.Spec.Redis = nil
-		err := Install(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Install(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar(), record.NewFakeRecorder(64))
 		assert.Error(t, err)
 		assert.Equal(t, "invalid isb service spec", err.Error())
 	})
 
 	t.Run("test redis install ok", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
-		err := Install(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Install(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar(), record.NewFakeRecorder(64))
 		assert.NoError(t, err)
 		assert.True(t, testObj.Status.IsReady())
 		assert.NotNil(t, testObj.Status.Config.Redis)
@@ -180,14 +181,14 @@ func TestInstall(t *testing.T) {
 	t.Run("test jetstream error", func(t *testing.T) {
 		testObj := testJetStreamIsbSvc.DeepCopy()
 		testObj.Spec.JetStream = nil
-		err := Install(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Install(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar(), record.NewFakeRecorder(64))
 		assert.Error(t, err)
 		assert.Equal(t, "invalid isb service spec", err.Error())
 	})
 
 	t.Run("test jetstream install ok", func(t *testing.T) {
 		testObj := testJetStreamIsbSvc.DeepCopy()
-		err := Install(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Install(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar(), record.NewFakeRecorder(64))
 		assert.NoError(t, err)
 		assert.True(t, testObj.Status.IsReady())
 		assert.NotNil(t, testObj.Status.Config.JetStream)
@@ -207,28 +208,28 @@ func TestUnInstall(t *testing.T) {
 	t.Run("test redis error", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
 		testObj.Spec.Redis = nil
-		err := Uninstall(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Uninstall(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar(), record.NewFakeRecorder(64))
 		assert.Error(t, err)
 		assert.Equal(t, "invalid isb service spec", err.Error())
 	})
 
 	t.Run("test redis uninstall ok", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
-		err := Uninstall(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Uninstall(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar(), record.NewFakeRecorder(64))
 		assert.NoError(t, err)
 	})
 
 	t.Run("test jetstream error", func(t *testing.T) {
 		testObj := testJetStreamIsbSvc.DeepCopy()
 		testObj.Spec.JetStream = nil
-		err := Uninstall(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Uninstall(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar(), record.NewFakeRecorder(64))
 		assert.Error(t, err)
 		assert.Equal(t, "invalid isb service spec", err.Error())
 	})
 
 	t.Run("test jetstream uninstall ok", func(t *testing.T) {
 		testObj := testJetStreamIsbSvc.DeepCopy()
-		err := Uninstall(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Uninstall(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar(), record.NewFakeRecorder(64))
 		assert.NoError(t, err)
 	})
 }
diff --git a/pkg/reconciler/isbsvc/installer/jetstream_test.go b/pkg/reconciler/isbsvc/installer/jetstream_test.go
index bdfdbb09..81f0749f 100644
--- a/pkg/reconciler/isbsvc/installer/jetstream_test.go
+++ b/pkg/reconciler/isbsvc/installer/jetstream_test.go
@@ -28,6 +28,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	k8sfake "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
@@ -42,11 +43,12 @@ func TestJetStreamBadInstallation(t *testing.T) {
 		badIsbs := testJetStreamIsbSvc.DeepCopy()
 		badIsbs.Spec.JetStream = nil
 		installer := &jetStreamInstaller{
-			client: fake.NewClientBuilder().Build(),
-			isbSvc: badIsbs,
-			config: fakeConfig,
-			labels: testLabels,
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:   fake.NewClientBuilder().Build(),
+			isbSvc:   badIsbs,
+			config:   fakeConfig,
+			labels:   testLabels,
+			logger:   zaptest.NewLogger(t).Sugar(),
+			recorder: record.NewFakeRecorder(64),
 		}
 		_, err := installer.Install(context.TODO())
 		assert.Error(t, err)
@@ -78,6 +80,7 @@ func TestJetStreamCreateObjects(t *testing.T) {
 		config:     fakeConfig,
 		labels:     testLabels,
 		logger:     zaptest.NewLogger(t).Sugar(),
+		recorder:   record.NewFakeRecorder(64),
 	}
 
 	t.Run("test create sts", func(t *testing.T) {
@@ -156,6 +159,7 @@ func Test_JetStreamInstall_Uninstall(t *testing.T) {
 		config:     fakeConfig,
 		labels:     testLabels,
 		logger:     zaptest.NewLogger(t).Sugar(),
+		recorder:   record.NewFakeRecorder(64),
 	}
 	t.Run("test install", func(t *testing.T) {
 		c, err := i.Install(ctx)
@@ -169,6 +173,8 @@ func Test_JetStreamInstall_Uninstall(t *testing.T) {
 		assert.NotNil(t, c.JetStream.Auth.Basic.Password)
 		assert.True(t, testJetStreamIsbSvc.Status.IsReady())
 		assert.False(t, c.JetStream.TLSEnabled)
+		events := getEvents(i.recorder)
+		assert.Contains(t, events, "Normal JetStreamConfigMap Created jetstream configmap successfully", "Normal JetStreamServiceSuccess Created jetstream service successfully", "Normal JetStreamStatefulSetSuccess Created jetstream stateful successfully")
 		svc := &corev1.Service{}
 		err = cl.Get(ctx, types.NamespacedName{Namespace: testJetStreamIsbSvc.Namespace, Name: generateJetStreamServiceName(testJetStreamIsbSvc)}, svc)
 		assert.NoError(t, err)
@@ -184,3 +190,13 @@ func Test_JetStreamInstall_Uninstall(t *testing.T) {
 		assert.NoError(t, err)
 	})
 }
+
+func getEvents(recorder record.EventRecorder) []string {
+	c := recorder.(*record.FakeRecorder).Events
+	close(c)
+	events := make([]string, len(c))
+	for msg := range c {
+		events = append(events, msg)
+	}
+	return events
+}
diff --git a/pkg/reconciler/isbsvc/installer/native_redis_test.go b/pkg/reconciler/isbsvc/installer/native_redis_test.go
index 7f72e640..3a336853 100644
--- a/pkg/reconciler/isbsvc/installer/native_redis_test.go
+++ b/pkg/reconciler/isbsvc/installer/native_redis_test.go
@@ -28,6 +28,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	k8sfake "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
@@ -42,11 +43,12 @@ func TestNativeRedisBadInstallation(t *testing.T) {
 		badIsbs := testNativeRedisIsbSvc.DeepCopy()
 		badIsbs.Spec.Redis = nil
 		installer := &redisInstaller{
-			client: fake.NewClientBuilder().Build(),
-			isbSvc: badIsbs,
-			config: fakeConfig,
-			labels: testLabels,
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:   fake.NewClientBuilder().Build(),
+			isbSvc:   badIsbs,
+			config:   fakeConfig,
+			labels:   testLabels,
+			logger:   zaptest.NewLogger(t).Sugar(),
+			recorder: record.NewFakeRecorder(64),
 		}
 		_, err := installer.Install(context.TODO())
 		assert.Error(t, err)
@@ -82,6 +84,7 @@ func TestNativeRedisCreateObjects(t *testing.T) {
 		config:     fakeConfig,
 		labels:     testLabels,
 		logger:     zaptest.NewLogger(t).Sugar(),
+		recorder:   record.NewFakeRecorder(64),
 	}
 
 	t.Run("test create sts", func(t *testing.T) {
@@ -183,6 +186,7 @@ func Test_NativeRedisInstall_Uninstall(t *testing.T) {
 		config:     fakeConfig,
 		labels:     testLabels,
 		logger:     zaptest.NewLogger(t).Sugar(),
+		recorder:   record.NewFakeRecorder(64),
 	}
 	t.Run("test install", func(t *testing.T) {
 		c, err := i.Install(ctx)
@@ -195,6 +199,8 @@ func Test_NativeRedisInstall_Uninstall(t *testing.T) {
 		assert.NotNil(t, c.Redis.Password)
 		assert.NotNil(t, c.Redis.SentinelPassword)
 		assert.True(t, testNativeRedisIsbSvc.Status.IsReady())
+		events := getEvents(i.recorder)
+		assert.Contains(t, events, "Normal RedisScriptsConfigMapSuccess Created redis scripts configmap successfully", "Normal RedisHealthConfigMapSuccess Created redis health configmap successfully", "Normal RedisConfConfigMapSuccess Created redis config configmap successfully", "Normal RedisServiceSuccess Created redis service successfully", "Normal RedisHeadlessServiceSuccess Created redis headless service successfully", "Normal RedisStatefulSetSuccess Created redis statefulset successfully")
 		svc := &corev1.Service{}
 		err = cl.Get(ctx, types.NamespacedName{Namespace: testNativeRedisIsbSvc.Namespace, Name: generateRedisHeadlessServiceName(testNativeRedisIsbSvc)}, svc)
 		assert.NoError(t, err)
diff --git a/pkg/reconciler/pipeline/controller_test.go b/pkg/reconciler/pipeline/controller_test.go
index 906d56e1..70826ad0 100644
--- a/pkg/reconciler/pipeline/controller_test.go
+++ b/pkg/reconciler/pipeline/controller_test.go
@@ -171,9 +171,9 @@ func Test_reconcileEvents(t *testing.T) {
 		testObj.Name = "very-very-very-loooooooooooooooooooooooooooooooooooong"
 		_, err = r.reconcile(ctx, testObj)
 		assert.Error(t, err)
-		events := getEvents(r, 2)
-		assert.Equal(t, "Normal UpdatePipelinePhase Updated pipeline phase from Paused to Running", events[0])
-		assert.Equal(t, "Warning ReconcilePipelineFailed Failed to reconcile pipeline: the length of the pipeline name plus the vertex name is over the max limit. (very-very-very-loooooooooooooooooooooooooooooooooooong-input), [must be no more than 63 characters]", events[1])
+		events := getEvents(r)
+		assert.Contains(t, events, "Normal UpdatePipelinePhase Updated pipeline phase from Paused to Running")
+		assert.Contains(t, events, "Warning ReconcilePipelineFailed Failed to reconcile pipeline: the length of the pipeline name plus the vertex name is over the max limit. (very-very-very-loooooooooooooooooooooooooooooooooooong-input), [must be no more than 63 characters]")
 	})
 
 	t.Run("test reconcile - duplicate vertex", func(t *testing.T) {
@@ -198,8 +198,8 @@ func Test_reconcileEvents(t *testing.T) {
 		testObj.Spec.Vertices = append(testObj.Spec.Vertices, dfv1.AbstractVertex{Name: "input", Source: &dfv1.Source{}})
 		_, err = r.reconcile(ctx, testObj)
 		assert.Error(t, err)
-		events := getEvents(r, 1)
-		assert.Equal(t, "Warning ReconcilePipelineFailed Failed to reconcile pipeline: duplicate vertex name \"input\"", events[0])
+		events := getEvents(r)
+		assert.Contains(t, events, "Warning ReconcilePipelineFailed Failed to reconcile pipeline: duplicate vertex name \"input\"")
 	})
 }
 
@@ -483,11 +483,12 @@ func TestCreateOrUpdateDaemon(t *testing.T) {
 	cl := fake.NewClientBuilder().Build()
 	ctx := context.TODO()
 	r := &pipelineReconciler{
-		client: cl,
-		scheme: scheme.Scheme,
-		config: fakeConfig,
-		image:  testFlowImage,
-		logger: zaptest.NewLogger(t).Sugar(),
+		client:   cl,
+		scheme:   scheme.Scheme,
+		config:   fakeConfig,
+		image:    testFlowImage,
+		logger:   zaptest.NewLogger(t).Sugar(),
+		recorder: record.NewFakeRecorder(64),
 	}
 
 	t.Run("test create or update service", func(t *testing.T) {
@@ -517,11 +518,12 @@ func Test_createOrUpdateSIMDeployments(t *testing.T) {
 	cl := fake.NewClientBuilder().Build()
 	ctx := context.TODO()
 	r := &pipelineReconciler{
-		client: cl,
-		scheme: scheme.Scheme,
-		config: fakeConfig,
-		image:  testFlowImage,
-		logger: zaptest.NewLogger(t).Sugar(),
+		client:   cl,
+		scheme:   scheme.Scheme,
+		config:   fakeConfig,
+		image:    testFlowImage,
+		logger:   zaptest.NewLogger(t).Sugar(),
+		recorder: record.NewFakeRecorder(64),
 	}
 
 	t.Run("no side inputs", func(t *testing.T) {
@@ -613,11 +615,12 @@ func Test_createOrUpdateSIMDeployments(t *testing.T) {
 	})
 }
 
-func getEvents(reconciler *pipelineReconciler, num int) []string {
+func getEvents(reconciler *pipelineReconciler) []string {
 	c := reconciler.recorder.(*record.FakeRecorder).Events
-	events := make([]string, num)
-	for i := 0; i < num; i++ {
-		events[i] = <-c
+	close(c)
+	events := make([]string, len(c))
+	for msg := range c {
+		events = append(events, msg)
 	}
 	return events
 }
diff --git a/pkg/reconciler/vertex/controller_test.go b/pkg/reconciler/vertex/controller_test.go
index e80f63db..f74c4625 100644
--- a/pkg/reconciler/vertex/controller_test.go
+++ b/pkg/reconciler/vertex/controller_test.go
@@ -179,11 +179,12 @@ func Test_BuildPodSpec(t *testing.T) {
 	t.Run("test source", func(t *testing.T) {
 		cl := fake.NewClientBuilder().Build()
 		r := &vertexReconciler{
-			client: cl,
-			scheme: scheme.Scheme,
-			config: fakeConfig,
-			image:  testFlowImage,
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:   cl,
+			scheme:   scheme.Scheme,
+			config:   fakeConfig,
+			image:    testFlowImage,
+			logger:   zaptest.NewLogger(t).Sugar(),
+			recorder: record.NewFakeRecorder(64),
 		}
 		testObj := testSrcVertex.DeepCopy()
 		spec, err := r.buildPodSpec(testObj, testPipeline, fakeIsbSvcConfig, 0)
@@ -211,11 +212,12 @@ func Test_BuildPodSpec(t *testing.T) {
 	t.Run("test source with transformer", func(t *testing.T) {
 		cl := fake.NewClientBuilder().Build()
 		r := &vertexReconciler{
-			client: cl,
-			scheme: scheme.Scheme,
-			config: fakeConfig,
-			image:  testFlowImage,
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:   cl,
+			scheme:   scheme.Scheme,
+			config:   fakeConfig,
+			image:    testFlowImage,
+			logger:   zaptest.NewLogger(t).Sugar(),
+			recorder: record.NewFakeRecorder(64),
 		}
 		testObj := testSrcVertex.DeepCopy()
 		testObj.Spec.Source = &dfv1.Source{
@@ -235,11 +237,12 @@ func Test_BuildPodSpec(t *testing.T) {
 	t.Run("test user-defined source with transformer", func(t *testing.T) {
 		cl := fake.NewClientBuilder().Build()
 		r := &vertexReconciler{
-			client: cl,
-			scheme: scheme.Scheme,
-			config: fakeConfig,
-			image:  testFlowImage,
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:   cl,
+			scheme:   scheme.Scheme,
+			config:   fakeConfig,
+			image:    testFlowImage,
+			logger:   zaptest.NewLogger(t).Sugar(),
+			recorder: record.NewFakeRecorder(64),
 		}
 		testObj := testSrcVertex.DeepCopy()
 		testObj.Spec.Source = &dfv1.Source{
@@ -263,11 +266,12 @@ func Test_BuildPodSpec(t *testing.T) {
 	t.Run("test sink", func(t *testing.T) {
 		cl := fake.NewClientBuilder().Build()
 		r := &vertexReconciler{
-			client: cl,
-			scheme: scheme.Scheme,
-			config: fakeConfig,
-			image:  testFlowImage,
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:   cl,
+			scheme:   scheme.Scheme,
+			config:   fakeConfig,
+			image:    testFlowImage,
+			logger:   zaptest.NewLogger(t).Sugar(),
+			recorder: record.NewFakeRecorder(64),
 		}
 		testObj := testVertex.DeepCopy()
 		testObj.Name = "test-pl-output"
@@ -300,11 +304,12 @@ func Test_BuildPodSpec(t *testing.T) {
 	t.Run("test user defined sink", func(t *testing.T) {
 		cl := fake.NewClientBuilder().Build()
 		r := &vertexReconciler{
-			client: cl,
-			scheme: scheme.Scheme,
-			config: fakeConfig,
-			image:  testFlowImage,
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:   cl,
+			scheme:   scheme.Scheme,
+			config:   fakeConfig,
+			image:    testFlowImage,
+			logger:   zaptest.NewLogger(t).Sugar(),
+			recorder: record.NewFakeRecorder(64),
 		}
 		testObj := testVertex.DeepCopy()
 		testObj.Name = "test-pl-output"
@@ -334,11 +339,12 @@ func Test_BuildPodSpec(t *testing.T) {
 	t.Run("test map udf", func(t *testing.T) {
 		cl := fake.NewClientBuilder().Build()
 		r := &vertexReconciler{
-			client: cl,
-			scheme: scheme.Scheme,
-			config: fakeConfig,
-			image:  testFlowImage,
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:   cl,
+			scheme:   scheme.Scheme,
+			config:   fakeConfig,
+			image:    testFlowImage,
+			logger:   zaptest.NewLogger(t).Sugar(),
+			recorder: record.NewFakeRecorder(64),
 		}
 		testObj := testVertex.DeepCopy()
 		testObj.Spec.UDF = &dfv1.UDF{
@@ -371,11 +377,12 @@ func Test_BuildPodSpec(t *testing.T) {
 	t.Run("test reduce udf", func(t *testing.T) {
 		cl := fake.NewClientBuilder().Build()
 		r := &vertexReconciler{
-			client: cl,
-			scheme: scheme.Scheme,
-			config: fakeConfig,
-			image:  testFlowImage,
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:   cl,
+			scheme:   scheme.Scheme,
+			config:   fakeConfig,
+			image:    testFlowImage,
+			logger:   zaptest.NewLogger(t).Sugar(),
+			recorder: record.NewFakeRecorder(64),
 		}
 		testObj := testVertex.DeepCopy()
 		volSize, _ := resource.ParseQuantity("1Gi")
@@ -425,12 +432,13 @@ func Test_reconcile(t *testing.T) {
 		err = cl.Create(ctx, testPl)
 		assert.Nil(t, err)
 		r := &vertexReconciler{
-			client: cl,
-			scheme: scheme.Scheme,
-			config: fakeConfig,
-			image:  testFlowImage,
-			scaler: scaling.NewScaler(cl),
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:   cl,
+			scheme:   scheme.Scheme,
+			config:   fakeConfig,
+			image:    testFlowImage,
+			scaler:   scaling.NewScaler(cl),
+			logger:   zaptest.NewLogger(t).Sugar(),
+			recorder: record.NewFakeRecorder(64),
 		}
 		testObj := testVertex.DeepCopy()
 		testObj.Spec.Source = &dfv1.Source{
@@ -476,12 +484,13 @@ func Test_reconcile(t *testing.T) {
 		err = cl.Create(ctx, testPl)
 		assert.Nil(t, err)
 		r := &vertexReconciler{
-			client: cl,
-			scheme: scheme.Scheme,
-			config: fakeConfig,
-			image:  testFlowImage,
-			scaler: scaling.NewScaler(cl),
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:   cl,
+			scheme:   scheme.Scheme,
+			config:   fakeConfig,
+			image:    testFlowImage,
+			scaler:   scaling.NewScaler(cl),
+			logger:   zaptest.NewLogger(t).Sugar(),
+			recorder: record.NewFakeRecorder(64),
 		}
 		testObj := testVertex.DeepCopy()
 		testObj.Spec.Sink = &dfv1.Sink{}
@@ -508,12 +517,13 @@ func Test_reconcile(t *testing.T) {
 		err = cl.Create(ctx, testPl)
 		assert.Nil(t, err)
 		r := &vertexReconciler{
-			client: cl,
-			scheme: scheme.Scheme,
-			config: fakeConfig,
-			image:  testFlowImage,
-			scaler: scaling.NewScaler(cl),
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:   cl,
+			scheme:   scheme.Scheme,
+			config:   fakeConfig,
+			image:    testFlowImage,
+			scaler:   scaling.NewScaler(cl),
+			logger:   zaptest.NewLogger(t).Sugar(),
+			recorder: record.NewFakeRecorder(64),
 		}
 		testObj := testVertex.DeepCopy()
 		testObj.Spec.UDF = &dfv1.UDF{
@@ -555,12 +565,13 @@ func Test_reconcile(t *testing.T) {
 		err = cl.Create(ctx, testPl)
 		assert.Nil(t, err)
 		r := &vertexReconciler{
-			client: cl,
-			scheme: scheme.Scheme,
-			config: fakeConfig,
-			image:  testFlowImage,
-			scaler: scaling.NewScaler(cl),
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:   cl,
+			scheme:   scheme.Scheme,
+			config:   fakeConfig,
+			image:    testFlowImage,
+			scaler:   scaling.NewScaler(cl),
+			logger:   zaptest.NewLogger(t).Sugar(),
+			recorder: record.NewFakeRecorder(64),
 		}
 		testObj := testVertex.DeepCopy()
 		testObj.Spec.Sink = &dfv1.Sink{}
@@ -599,12 +610,13 @@ func Test_reconcile(t *testing.T) {
 		err = cl.Create(ctx, testPl)
 		assert.Nil(t, err)
 		r := &vertexReconciler{
-			client: cl,
-			scheme: scheme.Scheme,
-			config: fakeConfig,
-			image:  testFlowImage,
-			scaler: scaling.NewScaler(cl),
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:   cl,
+			scheme:   scheme.Scheme,
+			config:   fakeConfig,
+			image:    testFlowImage,
+			scaler:   scaling.NewScaler(cl),
+			logger:   zaptest.NewLogger(t).Sugar(),
+			recorder: record.NewFakeRecorder(64),
 		}
 		testObj := testVertex.DeepCopy()
 		testObj.Spec.UDF = &dfv1.UDF{
@@ -630,6 +642,14 @@ func Test_reconcileEvents(t *testing.T) {
 	t.Run("test reconcile - isbsvc doesn't exist", func(t *testing.T) {
 		cl := fake.NewClientBuilder().Build()
 		ctx := context.TODO()
+		testIsbSvc := testNativeRedisIsbSvc.DeepCopy()
+		testIsbSvc.Status.MarkConfigured()
+		testIsbSvc.Status.MarkDeployed()
+		err := cl.Create(ctx, testIsbSvc)
+		assert.Nil(t, err)
+		testPl := testPipeline.DeepCopy()
+		err = cl.Create(ctx, testPl)
+		assert.Nil(t, err)
 		r := &vertexReconciler{
 			client:   cl,
 			scheme:   scheme.Scheme,
@@ -640,19 +660,24 @@ func Test_reconcileEvents(t *testing.T) {
 			recorder: record.NewFakeRecorder(64),
 		}
 		testObj := testVertex.DeepCopy()
-		testObj.Spec.InterStepBufferServiceName = "notDefault"
-		_, err := r.reconcile(ctx, testObj)
-		assert.Error(t, err)
-		events := getEvents(r, 1)
-		assert.Equal(t, "Warning ISBSvcNotFound isbsvc notDefault not found", events[0])
+		testObj.Spec.UDF = &dfv1.UDF{
+			Builtin: &dfv1.Function{
+				Name: "cat",
+			},
+		}
+		_, err = r.reconcile(ctx, testObj)
+		assert.NoError(t, err)
+		events := getEvents(r)
+		assert.Contains(t, events, "Normal CreateSvcSuccess Succeeded to create service test-pl-p1-headless")
 	})
 }
 
-func getEvents(reconciler *vertexReconciler, num int) []string {
+func getEvents(reconciler *vertexReconciler) []string {
 	c := reconciler.recorder.(*record.FakeRecorder).Events
-	events := make([]string, num)
-	for i := 0; i < num; i++ {
-		events[i] = <-c
+	close(c)
+	events := make([]string, len(c))
+	for msg := range c {
+		events = append(events, msg)
 	}
 	return events
 }