Repository: numaproj/numaflow
Pull request #1291 (2023-10-30 06:15:28)

chore: set stream.replicas=1 when in non-cluster mode
When jetstream is not in cluster mode, automatically sets the stream/KV replica to 1. @joelcomp1 

```golang
	if r.isbSvc.Spec.JetStream.GetReplicas() < 3 {
		// Replica can not > 1 with non-cluster mode
		v.Set("otbucket.replicas", 1)
		v.Set("procbucket.replicas", 1)
		v.Set("stream.replicas", 1)
	}
```

<!--

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.

-->


joelcomp1: So is that KV store configuration different then the one here: https://github.com/numaproj/numaflow/blob/86df4a84083613cf3912f35ada400434ce1b200c/pkg/isbsvc/jetstream_service.go#L100

must be?  
whynowy: > So is that KV store configuration different then the one here:
> 
> https://github.com/numaproj/numaflow/blob/86df4a84083613cf3912f35ada400434ce1b200c/pkg/isbsvc/jetstream_service.go#L100
> 
> must be?

We can use the stream.replicas for sideinput KV as well for now, can always revisit later one.

Patch:
diff --git a/pkg/reconciler/isbsvc/controller.go b/pkg/reconciler/isbsvc/controller.go
index 4206f2ee..87b2c754 100644
--- a/pkg/reconciler/isbsvc/controller.go
+++ b/pkg/reconciler/isbsvc/controller.go
@@ -52,60 +52,60 @@ func NewReconciler(client client.Client, kubeClient kubernetes.Interface, scheme
 }
 
 func (r *interStepBufferServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	isbs := &dfv1.InterStepBufferService{}
-	if err := r.client.Get(ctx, req.NamespacedName, isbs); err != nil {
+	isbSvc := &dfv1.InterStepBufferService{}
+	if err := r.client.Get(ctx, req.NamespacedName, isbSvc); err != nil {
 		if apierrors.IsNotFound(err) {
 			return reconcile.Result{}, nil
 		}
-		r.logger.Errorw("Unable to get ISBS", zap.Any("request", req), zap.Error(err))
+		r.logger.Errorw("Unable to get ISB Service", zap.Any("request", req), zap.Error(err))
 		return ctrl.Result{}, err
 	}
-	log := r.logger.With("namespace", isbs.Namespace).With("isbsvc", isbs.Name)
-	isbsCopy := isbs.DeepCopy()
-	reconcileErr := r.reconcile(ctx, isbsCopy)
+	log := r.logger.With("namespace", isbSvc.Namespace).With("isbsvc", isbSvc.Name)
+	isbSvcCopy := isbSvc.DeepCopy()
+	reconcileErr := r.reconcile(ctx, isbSvcCopy)
 	if reconcileErr != nil {
 		log.Errorw("Reconcile error", zap.Error(reconcileErr))
 	}
-	if r.needsUpdate(isbs, isbsCopy) {
+	if r.needsUpdate(isbSvc, isbSvcCopy) {
 		// Update with a DeepCopy because .Status will be cleaned up.
-		if err := r.client.Update(ctx, isbsCopy.DeepCopy()); err != nil {
+		if err := r.client.Update(ctx, isbSvcCopy.DeepCopy()); err != nil {
 			return reconcile.Result{}, err
 		}
 	}
-	if err := r.client.Status().Update(ctx, isbsCopy); err != nil {
+	if err := r.client.Status().Update(ctx, isbSvcCopy); err != nil {
 		return reconcile.Result{}, err
 	}
 	return ctrl.Result{}, reconcileErr
 }
 
 // reconcile does the real logic
-func (r *interStepBufferServiceReconciler) reconcile(ctx context.Context, isbs *dfv1.InterStepBufferService) error {
-	log := r.logger.With("namespace", isbs.Namespace).With("isbs", isbs.Name)
-	if !isbs.DeletionTimestamp.IsZero() {
-		log.Info("Deleting isbs")
-		if controllerutil.ContainsFinalizer(isbs, finalizerName) {
+func (r *interStepBufferServiceReconciler) reconcile(ctx context.Context, isbSvc *dfv1.InterStepBufferService) error {
+	log := r.logger.With("namespace", isbSvc.Namespace).With("isbsvc", isbSvc.Name)
+	if !isbSvc.DeletionTimestamp.IsZero() {
+		log.Info("Deleting isbsvc")
+		if controllerutil.ContainsFinalizer(isbSvc, finalizerName) {
 			// Finalizer logic should be added here.
-			if err := installer.Uninstall(ctx, isbs, r.client, r.kubeClient, r.config, log); err != nil {
+			if err := installer.Uninstall(ctx, isbSvc, r.client, r.kubeClient, r.config, log); err != nil {
 				log.Errorw("Failed to uninstall", zap.Error(err))
 				return err
 			}
-			controllerutil.RemoveFinalizer(isbs, finalizerName)
+			controllerutil.RemoveFinalizer(isbSvc, finalizerName)
 		}
 		return nil
 	}
-	if needsFinalizer(isbs) {
-		controllerutil.AddFinalizer(isbs, finalizerName)
+	if needsFinalizer(isbSvc) {
+		controllerutil.AddFinalizer(isbSvc, finalizerName)
 	}
 
-	isbs.Status.InitConditions()
-	if err := ValidateInterStepBufferService(isbs); err != nil {
+	isbSvc.Status.InitConditions()
+	if err := ValidateInterStepBufferService(isbSvc); err != nil {
 		log.Errorw("Validation failed", zap.Error(err))
-		isbs.Status.MarkNotConfigured("InvalidSpec", err.Error())
+		isbSvc.Status.MarkNotConfigured("InvalidSpec", err.Error())
 		return err
 	} else {
-		isbs.Status.MarkConfigured()
+		isbSvc.Status.MarkConfigured()
 	}
-	return installer.Install(ctx, isbs, r.client, r.kubeClient, r.config, log)
+	return installer.Install(ctx, isbSvc, r.client, r.kubeClient, r.config, log)
 }
 
 func (r *interStepBufferServiceReconciler) needsUpdate(old, new *dfv1.InterStepBufferService) bool {
@@ -118,11 +118,11 @@ func (r *interStepBufferServiceReconciler) needsUpdate(old, new *dfv1.InterStepB
 	return false
 }
 
-func needsFinalizer(isbs *dfv1.InterStepBufferService) bool {
-	if isbs.Spec.Redis != nil && isbs.Spec.Redis.Native != nil && isbs.Spec.Redis.Native.Persistence != nil {
+func needsFinalizer(isbSvc *dfv1.InterStepBufferService) bool {
+	if isbSvc.Spec.Redis != nil && isbSvc.Spec.Redis.Native != nil && isbSvc.Spec.Redis.Native.Persistence != nil {
 		return true
 	}
-	if isbs.Spec.JetStream != nil && isbs.Spec.JetStream.Persistence != nil {
+	if isbSvc.Spec.JetStream != nil && isbSvc.Spec.JetStream.Persistence != nil {
 		return true
 	}
 	return false
diff --git a/pkg/reconciler/isbsvc/installer/external_redis.go b/pkg/reconciler/isbsvc/installer/external_redis.go
index 19eceb80..62da20d7 100644
--- a/pkg/reconciler/isbsvc/installer/external_redis.go
+++ b/pkg/reconciler/isbsvc/installer/external_redis.go
@@ -25,26 +25,26 @@ import (
 )
 
 type externalRedisInstaller struct {
-	isbs   *dfv1.InterStepBufferService
+	isbSvc *dfv1.InterStepBufferService
 	logger *zap.SugaredLogger
 }
 
-func NewExternalRedisInstaller(isbs *dfv1.InterStepBufferService, logger *zap.SugaredLogger) Installer {
+func NewExternalRedisInstaller(isbSvc *dfv1.InterStepBufferService, logger *zap.SugaredLogger) Installer {
 	return &externalRedisInstaller{
-		isbs:   isbs,
-		logger: logger.With("isbs", isbs.Name),
+		isbSvc: isbSvc,
+		logger: logger.With("isbsvc", isbSvc.Name),
 	}
 }
 
 func (eri *externalRedisInstaller) Install(ctx context.Context) (*dfv1.BufferServiceConfig, error) {
-	if eri.isbs.Spec.Redis == nil || eri.isbs.Spec.Redis.External == nil {
+	if eri.isbSvc.Spec.Redis == nil || eri.isbSvc.Spec.Redis.External == nil {
 		return nil, fmt.Errorf("invalid InterStepBufferService spec, no external config")
 	}
-	eri.isbs.Status.SetType(dfv1.ISBSvcTypeRedis)
-	eri.isbs.Status.MarkConfigured()
-	eri.isbs.Status.MarkDeployed()
+	eri.isbSvc.Status.SetType(dfv1.ISBSvcTypeRedis)
+	eri.isbSvc.Status.MarkConfigured()
+	eri.isbSvc.Status.MarkDeployed()
 	eri.logger.Info("Using external redis config")
-	return &dfv1.BufferServiceConfig{Redis: eri.isbs.Spec.Redis.External}, nil
+	return &dfv1.BufferServiceConfig{Redis: eri.isbSvc.Spec.Redis.External}, nil
 }
 
 func (eri *externalRedisInstaller) Uninstall(ctx context.Context) error {
diff --git a/pkg/reconciler/isbsvc/installer/installer.go b/pkg/reconciler/isbsvc/installer/installer.go
index 6b2ad1a1..3a1e135d 100644
--- a/pkg/reconciler/isbsvc/installer/installer.go
+++ b/pkg/reconciler/isbsvc/installer/installer.go
@@ -36,8 +36,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) error {
+	installer, err := getInstaller(isbSvc, client, kubeClient, config, logger)
 	if err != nil {
 		logger.Errorw("failed to get an installer", zap.Error(err))
 		return err
@@ -47,28 +47,28 @@ func Install(ctx context.Context, isbsvc *dfv1.InterStepBufferService, client cl
 		logger.Errorw("installation error", zap.Error(err))
 		return err
 	}
-	isbsvc.Status.Config = *bufferConfig
+	isbSvc.Status.Config = *bufferConfig
 	return nil
 }
 
 // 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) (Installer, error) {
 	labels := map[string]string{
 		dfv1.KeyPartOf:     dfv1.Project,
 		dfv1.KeyManagedBy:  dfv1.ControllerISBSvc,
 		dfv1.KeyComponent:  dfv1.ComponentISBSvc,
-		dfv1.KeyISBSvcName: isbsvc.Name,
+		dfv1.KeyISBSvcName: isbSvc.Name,
 	}
-	if redis := isbsvc.Spec.Redis; redis != nil {
+	if redis := isbSvc.Spec.Redis; redis != nil {
 		labels[dfv1.KeyISBSvcType] = string(dfv1.ISBSvcTypeRedis)
 		if redis.External != nil {
-			return NewExternalRedisInstaller(isbsvc, logger), 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), nil
 		}
-	} else if js := isbsvc.Spec.JetStream; js != 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), nil
 	}
 	return nil, fmt.Errorf("invalid isb service spec")
 }
@@ -81,8 +81,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) error {
+	installer, err := getInstaller(isbSvc, client, kubeClient, config, logger)
 	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 798d75da..5ebee712 100644
--- a/pkg/reconciler/isbsvc/installer/jetstream.go
+++ b/pkg/reconciler/isbsvc/installer/jetstream.go
@@ -57,39 +57,45 @@ var (
 type jetStreamInstaller struct {
 	client     client.Client
 	kubeClient kubernetes.Interface
-	isbs       *dfv1.InterStepBufferService
+	isbSvc     *dfv1.InterStepBufferService
 	config     *reconciler.GlobalConfig
 	labels     map[string]string
 	logger     *zap.SugaredLogger
 }
 
-func NewJetStreamInstaller(client client.Client, kubeClient kubernetes.Interface, isbs *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) Installer {
 	return &jetStreamInstaller{
 		client:     client,
 		kubeClient: kubeClient,
-		isbs:       isbs,
+		isbSvc:     isbSvc,
 		config:     config,
 		labels:     labels,
-		logger:     logger.With("isbs", isbs.Name),
+		logger:     logger.With("isbsvc", isbSvc.Name),
 	}
 }
 
 func (r *jetStreamInstaller) Install(ctx context.Context) (*dfv1.BufferServiceConfig, error) {
-	if js := r.isbs.Spec.JetStream; js == nil {
-		return nil, fmt.Errorf("invalid jetstream isbs spec")
+	if js := r.isbSvc.Spec.JetStream; js == nil {
+		return nil, fmt.Errorf("invalid jetstream ISB Service spec")
 	}
-	r.isbs.Status.SetType(dfv1.ISBSvcTypeJetStream)
+	r.isbSvc.Status.SetType(dfv1.ISBSvcTypeJetStream)
 	// merge
 	v := viper.New()
 	v.SetConfigType("yaml")
 	if err := v.ReadConfig(bytes.NewBufferString(r.config.ISBSvc.JetStream.BufferConfig)); err != nil {
 		return nil, fmt.Errorf("invalid jetstream buffer config in global configuration, %w", err)
 	}
-	if x := r.isbs.Spec.JetStream.BufferConfig; x != nil {
+	if x := r.isbSvc.Spec.JetStream.BufferConfig; x != nil {
 		if err := v.MergeConfig(bytes.NewBufferString(*x)); err != nil {
 			return nil, fmt.Errorf("failed to merge customized buffer config, %w", err)
 		}
 	}
+	if r.isbSvc.Spec.JetStream.GetReplicas() < 3 {
+		// Replica can not > 1 with non-cluster mode
+		v.Set("otbucket.replicas", 1)
+		v.Set("procbucket.replicas", 1)
+		v.Set("stream.replicas", 1)
+	}
 	b, err := yaml.Marshal(v.AllSettings())
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal merged buffer config, %w", err)
@@ -97,52 +103,52 @@ 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.isbs.Status.MarkDeployFailed("JetStreamAuthSecretsFailed", err.Error())
+		r.isbSvc.Status.MarkDeployFailed("JetStreamAuthSecretsFailed", err.Error())
 		return nil, err
 	}
 	if err := r.createConfigMap(ctx); err != nil {
 		r.logger.Errorw("Failed to create jetstream ConfigMap", zap.Error(err))
-		r.isbs.Status.MarkDeployFailed("JetStreamConfigMapFailed", err.Error())
+		r.isbSvc.Status.MarkDeployFailed("JetStreamConfigMapFailed", err.Error())
 		return nil, err
 	}
 	if err := r.createService(ctx); err != nil {
 		r.logger.Errorw("Failed to create jetstream Service", zap.Error(err))
-		r.isbs.Status.MarkDeployFailed("JetStreamServiceFailed", err.Error())
+		r.isbSvc.Status.MarkDeployFailed("JetStreamServiceFailed", err.Error())
 		return nil, err
 	}
 	if err := r.createStatefulSet(ctx); err != nil {
 		r.logger.Errorw("Failed to create jetstream StatefulSet", zap.Error(err))
-		r.isbs.Status.MarkDeployFailed("JetStreamStatefulSetFailed", err.Error())
+		r.isbSvc.Status.MarkDeployFailed("JetStreamStatefulSetFailed", err.Error())
 		return nil, err
 	}
-	r.isbs.Status.MarkDeployed()
+	r.isbSvc.Status.MarkDeployed()
 	return &dfv1.BufferServiceConfig{
 		JetStream: &dfv1.JetStreamConfig{
-			URL: fmt.Sprintf("nats://%s.%s.svc:%s", generateJetStreamServiceName(r.isbs), r.isbs.Namespace, strconv.Itoa(int(clientPort))),
+			URL: fmt.Sprintf("nats://%s.%s.svc:%s", generateJetStreamServiceName(r.isbSvc), r.isbSvc.Namespace, strconv.Itoa(int(clientPort))),
 			Auth: &dfv1.NatsAuth{
 				Basic: &dfv1.BasicAuth{
 					User: &corev1.SecretKeySelector{
 						LocalObjectReference: corev1.LocalObjectReference{
-							Name: generateJetStreamClientAuthSecretName(r.isbs),
+							Name: generateJetStreamClientAuthSecretName(r.isbSvc),
 						},
 						Key: dfv1.JetStreamClientAuthSecretUserKey,
 					},
 					Password: &corev1.SecretKeySelector{
 						LocalObjectReference: corev1.LocalObjectReference{
-							Name: generateJetStreamClientAuthSecretName(r.isbs),
+							Name: generateJetStreamClientAuthSecretName(r.isbSvc),
 						},
 						Key: dfv1.JetStreamClientAuthSecretPasswordKey,
 					},
 				},
 			},
 			StreamConfig: string(b),
-			TLSEnabled:   r.isbs.Spec.JetStream.TLS,
+			TLSEnabled:   r.isbSvc.Spec.JetStream.TLS,
 		},
 	}, nil
 }
 
 func (r *jetStreamInstaller) createService(ctx context.Context) error {
-	spec := r.isbs.Spec.JetStream.GetServiceSpec(dfv1.GetJetStreamServiceSpecReq{
+	spec := r.isbSvc.Spec.JetStream.GetServiceSpec(dfv1.GetJetStreamServiceSpecReq{
 		Labels:      r.labels,
 		MetricsPort: metricsPort,
 		ClusterPort: clusterPort,
@@ -152,14 +158,14 @@ func (r *jetStreamInstaller) createService(ctx context.Context) error {
 	hash := sharedutil.MustHash(spec)
 	obj := &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: r.isbs.Namespace,
-			Name:      generateJetStreamServiceName(r.isbs),
+			Namespace: r.isbSvc.Namespace,
+			Name:      generateJetStreamServiceName(r.isbSvc),
 			Labels:    r.labels,
 			Annotations: map[string]string{
 				dfv1.KeyHash: hash,
 			},
 			OwnerReferences: []metav1.OwnerReference{
-				*metav1.NewControllerRef(r.isbs.GetObjectMeta(), dfv1.ISBGroupVersionKind),
+				*metav1.NewControllerRef(r.isbSvc.GetObjectMeta(), dfv1.ISBGroupVersionKind),
 			},
 		},
 		Spec: spec,
@@ -188,12 +194,12 @@ func (r *jetStreamInstaller) createService(ctx context.Context) error {
 }
 
 func (r *jetStreamInstaller) createStatefulSet(ctx context.Context) error {
-	jsVersion, err := r.config.GetJetStreamVersion(r.isbs.Spec.JetStream.Version)
+	jsVersion, err := r.config.GetJetStreamVersion(r.isbSvc.Spec.JetStream.Version)
 	if err != nil {
 		return fmt.Errorf("failed to get jetstream version, err: %w", err)
 	}
-	spec := r.isbs.Spec.JetStream.GetStatefulSetSpec(dfv1.GetJetStreamStatefulSetSpecReq{
-		ServiceName:                generateJetStreamServiceName(r.isbs),
+	spec := r.isbSvc.Spec.JetStream.GetStatefulSetSpec(dfv1.GetJetStreamStatefulSetSpecReq{
+		ServiceName:                generateJetStreamServiceName(r.isbSvc),
 		Labels:                     r.labels,
 		NatsImage:                  jsVersion.NatsImage,
 		MetricsExporterImage:       jsVersion.MetricsExporterImage,
@@ -202,23 +208,23 @@ func (r *jetStreamInstaller) createStatefulSet(ctx context.Context) error {
 		MonitorPort:                monitorPort,
 		ClientPort:                 clientPort,
 		MetricsPort:                metricsPort,
-		ServerAuthSecretName:       generateJetStreamServerSecretName(r.isbs),
-		ServerEncryptionSecretName: generateJetStreamServerSecretName(r.isbs),
-		ConfigMapName:              generateJetStreamConfigMapName(r.isbs),
-		PvcNameIfNeeded:            generateJetStreamPVCName(r.isbs),
+		ServerAuthSecretName:       generateJetStreamServerSecretName(r.isbSvc),
+		ServerEncryptionSecretName: generateJetStreamServerSecretName(r.isbSvc),
+		ConfigMapName:              generateJetStreamConfigMapName(r.isbSvc),
+		PvcNameIfNeeded:            generateJetStreamPVCName(r.isbSvc),
 		StartCommand:               jsVersion.StartCommand,
 	})
 	hash := sharedutil.MustHash(spec)
 	obj := &appv1.StatefulSet{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: r.isbs.Namespace,
-			Name:      generateJetStreamStatefulSetName(r.isbs),
+			Namespace: r.isbSvc.Namespace,
+			Name:      generateJetStreamStatefulSetName(r.isbSvc),
 			Labels:    r.labels,
 			Annotations: map[string]string{
 				dfv1.KeyHash: hash,
 			},
 			OwnerReferences: []metav1.OwnerReference{
-				*metav1.NewControllerRef(r.isbs.GetObjectMeta(), dfv1.ISBGroupVersionKind),
+				*metav1.NewControllerRef(r.isbSvc.GetObjectMeta(), dfv1.ISBGroupVersionKind),
 			},
 		},
 		Spec: spec,
@@ -249,7 +255,7 @@ func (r *jetStreamInstaller) createStatefulSet(ctx context.Context) error {
 func (r *jetStreamInstaller) createSecrets(ctx context.Context) error {
 	oldServerObjExisting, oldClientObjExisting := true, true
 
-	oldSObj, err := r.kubeClient.CoreV1().Secrets(r.isbs.Namespace).Get(ctx, generateJetStreamServerSecretName(r.isbs), metav1.GetOptions{})
+	oldSObj, err := r.kubeClient.CoreV1().Secrets(r.isbSvc.Namespace).Get(ctx, generateJetStreamServerSecretName(r.isbSvc), metav1.GetOptions{})
 	if err != nil {
 		if apierrors.IsNotFound(err) {
 			oldServerObjExisting = false
@@ -258,7 +264,7 @@ func (r *jetStreamInstaller) createSecrets(ctx context.Context) error {
 		}
 	}
 
-	oldCObj, err := r.kubeClient.CoreV1().Secrets(r.isbs.Namespace).Get(ctx, generateJetStreamClientAuthSecretName(r.isbs), metav1.GetOptions{})
+	oldCObj, err := r.kubeClient.CoreV1().Secrets(r.isbSvc.Namespace).Get(ctx, generateJetStreamClientAuthSecretName(r.isbSvc), metav1.GetOptions{})
 	if err != nil {
 		if apierrors.IsNotFound(err) {
 			oldClientObjExisting = false
@@ -290,7 +296,7 @@ func (r *jetStreamInstaller) createSecrets(ctx context.Context) error {
 	jsPass := sharedutil.RandomString(16)
 	sysPassword := sharedutil.RandomString(24)
 	authTlsConfig := ""
-	if r.isbs.Spec.JetStream.TLS {
+	if r.isbSvc.Spec.JetStream.TLS {
 		authTlsConfig = `tls {
   cert_file: "/etc/nats-config/server-cert.pem"
   key_file: "/etc/nats-config/server-key.pem"
@@ -318,8 +324,8 @@ func (r *jetStreamInstaller) createSecrets(ctx context.Context) error {
 	certOrg := "io.numaproj"
 	// Generate server Cert
 	hosts := []string{
-		fmt.Sprintf("%s.%s.svc.cluster.local", generateJetStreamServiceName(r.isbs), r.isbs.Namespace),
-		fmt.Sprintf("%s.%s.svc", generateJetStreamServiceName(r.isbs), r.isbs.Namespace),
+		fmt.Sprintf("%s.%s.svc.cluster.local", generateJetStreamServiceName(r.isbSvc), r.isbSvc.Namespace),
+		fmt.Sprintf("%s.%s.svc", generateJetStreamServiceName(r.isbSvc), r.isbSvc.Namespace),
 	}
 	serverKeyPEM, serverCertPEM, caCertPEM, err := tls.CreateCerts(certOrg, hosts, time.Now().Add(10*365*24*time.Hour), true, false)
 	if err != nil {
@@ -327,8 +333,8 @@ func (r *jetStreamInstaller) createSecrets(ctx context.Context) error {
 	}
 	// Generate cluster Cert
 	clusterNodeHosts := []string{
-		fmt.Sprintf("*.%s.%s.svc.cluster.local", generateJetStreamServiceName(r.isbs), r.isbs.Namespace),
-		fmt.Sprintf("*.%s.%s.svc", generateJetStreamServiceName(r.isbs), r.isbs.Namespace),
+		fmt.Sprintf("*.%s.%s.svc.cluster.local", generateJetStreamServiceName(r.isbSvc), r.isbSvc.Namespace),
+		fmt.Sprintf("*.%s.%s.svc", generateJetStreamServiceName(r.isbSvc), r.isbSvc.Namespace),
 	}
 	clusterKeyPEM, clusterCertPEM, clusterCACertPEM, err := tls.CreateCerts(certOrg, clusterNodeHosts, time.Now().Add(10*365*24*time.Hour), true, true)
 	if err != nil {
@@ -337,11 +343,11 @@ func (r *jetStreamInstaller) createSecrets(ctx context.Context) error {
 
 	serverObj := &corev1.Secret{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: r.isbs.Namespace,
-			Name:      generateJetStreamServerSecretName(r.isbs),
+			Namespace: r.isbSvc.Namespace,
+			Name:      generateJetStreamServerSecretName(r.isbSvc),
 			Labels:    r.labels,
 			OwnerReferences: []metav1.OwnerReference{
-				*metav1.NewControllerRef(r.isbs.GetObjectMeta(), dfv1.ISBGroupVersionKind),
+				*metav1.NewControllerRef(r.isbSvc.GetObjectMeta(), dfv1.ISBGroupVersionKind),
 			},
 		},
 		Type: corev1.SecretTypeOpaque,
@@ -359,11 +365,11 @@ func (r *jetStreamInstaller) createSecrets(ctx context.Context) error {
 
 	clientAuthObj := &corev1.Secret{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: r.isbs.Namespace,
-			Name:      generateJetStreamClientAuthSecretName(r.isbs),
+			Namespace: r.isbSvc.Namespace,
+			Name:      generateJetStreamClientAuthSecretName(r.isbSvc),
 			Labels:    r.labels,
 			OwnerReferences: []metav1.OwnerReference{
-				*metav1.NewControllerRef(r.isbs.GetObjectMeta(), dfv1.ISBGroupVersionKind),
+				*metav1.NewControllerRef(r.isbSvc.GetObjectMeta(), dfv1.ISBGroupVersionKind),
 			},
 		},
 		Type: corev1.SecretTypeOpaque,
@@ -387,19 +393,19 @@ func (r *jetStreamInstaller) createSecrets(ctx context.Context) error {
 
 func (r *jetStreamInstaller) createConfigMap(ctx context.Context) error {
 	data := make(map[string]string)
-	svcName := generateJetStreamServiceName(r.isbs)
-	ssName := generateJetStreamStatefulSetName(r.isbs)
-	replicas := r.isbs.Spec.JetStream.GetReplicas()
+	svcName := generateJetStreamServiceName(r.isbSvc)
+	ssName := generateJetStreamStatefulSetName(r.isbSvc)
+	replicas := r.isbSvc.Spec.JetStream.GetReplicas()
 	routes := []string{}
 	for j := 0; j < replicas; j++ {
-		routes = append(routes, fmt.Sprintf("nats://%s-%s.%s.%s.svc:%s", ssName, strconv.Itoa(j), svcName, r.isbs.Namespace, strconv.Itoa(int(clusterPort))))
+		routes = append(routes, fmt.Sprintf("nats://%s-%s.%s.%s.svc:%s", ssName, strconv.Itoa(j), svcName, r.isbSvc.Namespace, strconv.Itoa(int(clusterPort))))
 	}
 	encryptionSettings := ""
-	if r.isbs.Spec.JetStream.Encryption {
+	if r.isbSvc.Spec.JetStream.Encryption {
 		encryptionSettings = "key: $JS_KEY"
 	}
 	clusterTLSConfig := ""
-	if r.isbs.Spec.JetStream.TLS {
+	if r.isbSvc.Spec.JetStream.TLS {
 		clusterTLSConfig = `
   tls {
     cert_file: "/etc/nats-config/cluster-server-cert.pem"
@@ -414,7 +420,7 @@ func (r *jetStreamInstaller) createConfigMap(ctx context.Context) error {
 	if err := v.ReadConfig(bytes.NewBufferString(r.config.ISBSvc.JetStream.Settings)); err != nil {
 		return fmt.Errorf("invalid jetstream settings in global configuration, %w", err)
 	}
-	if x := r.isbs.Spec.JetStream.Settings; x != nil {
+	if x := r.isbSvc.Spec.JetStream.Settings; x != nil {
 		if err := v.MergeConfig(bytes.NewBufferString(*x)); err != nil {
 			return fmt.Errorf("failed to merge customized jetstream settings, %w", err)
 		}
@@ -439,7 +445,7 @@ func (r *jetStreamInstaller) createConfigMap(ctx context.Context) error {
 		EncryptionSettings string
 		TLSConfig          string
 	}{
-		ClusterName:        r.isbs.Name,
+		ClusterName:        r.isbSvc.Name,
 		MonitorPort:        strconv.Itoa(int(monitorPort)),
 		ClusterPort:        strconv.Itoa(int(clusterPort)),
 		ClientPort:         strconv.Itoa(int(clientPort)),
@@ -457,14 +463,14 @@ func (r *jetStreamInstaller) createConfigMap(ctx context.Context) error {
 	hash := sharedutil.MustHash(data)
 	obj := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: r.isbs.Namespace,
-			Name:      generateJetStreamConfigMapName(r.isbs),
+			Namespace: r.isbSvc.Namespace,
+			Name:      generateJetStreamConfigMapName(r.isbSvc),
 			Labels:    r.labels,
 			Annotations: map[string]string{
 				dfv1.KeyHash: hash,
 			},
 			OwnerReferences: []metav1.OwnerReference{
-				*metav1.NewControllerRef(r.isbs.GetObjectMeta(), dfv1.ISBGroupVersionKind),
+				*metav1.NewControllerRef(r.isbSvc.GetObjectMeta(), dfv1.ISBGroupVersionKind),
 			},
 		},
 		Data: data,
@@ -520,7 +526,7 @@ func (r *jetStreamInstaller) uninstallPVCs(ctx context.Context) error {
 func (r *jetStreamInstaller) getPVCs(ctx context.Context) ([]corev1.PersistentVolumeClaim, error) {
 	pvcl := &corev1.PersistentVolumeClaimList{}
 	err := r.client.List(ctx, pvcl, &client.ListOptions{
-		Namespace:     r.isbs.Namespace,
+		Namespace:     r.isbSvc.Namespace,
 		LabelSelector: labels.SelectorFromSet(r.labels),
 	})
 	if err != nil {
@@ -529,26 +535,26 @@ func (r *jetStreamInstaller) getPVCs(ctx context.Context) ([]corev1.PersistentVo
 	return pvcl.Items, nil
 }
 
-func generateJetStreamServerSecretName(isbs *dfv1.InterStepBufferService) string {
-	return fmt.Sprintf("isbsvc-%s-js-server", isbs.Name)
+func generateJetStreamServerSecretName(isbSvc *dfv1.InterStepBufferService) string {
+	return fmt.Sprintf("isbsvc-%s-js-server", isbSvc.Name)
 }
 
-func generateJetStreamClientAuthSecretName(isbs *dfv1.InterStepBufferService) string {
-	return fmt.Sprintf("isbsvc-%s-js-client-auth", isbs.Name)
+func generateJetStreamClientAuthSecretName(isbSvc *dfv1.InterStepBufferService) string {
+	return fmt.Sprintf("isbsvc-%s-js-client-auth", isbSvc.Name)
 }
 
-func generateJetStreamServiceName(isbs *dfv1.InterStepBufferService) string {
-	return fmt.Sprintf("isbsvc-%s-js-svc", isbs.Name)
+func generateJetStreamServiceName(isbSvc *dfv1.InterStepBufferService) string {
+	return fmt.Sprintf("isbsvc-%s-js-svc", isbSvc.Name)
 }
 
-func generateJetStreamStatefulSetName(isbs *dfv1.InterStepBufferService) string {
-	return fmt.Sprintf("isbsvc-%s-js", isbs.Name)
+func generateJetStreamStatefulSetName(isbSvc *dfv1.InterStepBufferService) string {
+	return fmt.Sprintf("isbsvc-%s-js", isbSvc.Name)
 }
 
-func generateJetStreamConfigMapName(isbs *dfv1.InterStepBufferService) string {
-	return fmt.Sprintf("isbsvc-%s-js-config", isbs.Name)
+func generateJetStreamConfigMapName(isbSvc *dfv1.InterStepBufferService) string {
+	return fmt.Sprintf("isbsvc-%s-js-config", isbSvc.Name)
 }
 
-func generateJetStreamPVCName(isbs *dfv1.InterStepBufferService) string {
-	return fmt.Sprintf("isbsvc-%s-js-vol", isbs.Name)
+func generateJetStreamPVCName(isbSvc *dfv1.InterStepBufferService) string {
+	return fmt.Sprintf("isbsvc-%s-js-vol", isbSvc.Name)
 }
diff --git a/pkg/reconciler/isbsvc/installer/native_redis.go b/pkg/reconciler/isbsvc/installer/native_redis.go
index e6deada2..aef4dd38 100644
--- a/pkg/reconciler/isbsvc/installer/native_redis.go
+++ b/pkg/reconciler/isbsvc/installer/native_redis.go
@@ -55,79 +55,79 @@ var (
 type redisInstaller struct {
 	client     client.Client
 	kubeClient kubernetes.Interface
-	isbs       *dfv1.InterStepBufferService
+	isbSvc     *dfv1.InterStepBufferService
 	config     *reconciler.GlobalConfig
 	labels     map[string]string
 	logger     *zap.SugaredLogger
 }
 
-func NewNativeRedisInstaller(client client.Client, kubeClient kubernetes.Interface, isbs *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) Installer {
 	return &redisInstaller{
 		client:     client,
 		kubeClient: kubeClient,
-		isbs:       isbs,
+		isbSvc:     isbSvc,
 		config:     config,
 		labels:     labels,
-		logger:     logger.With("isbs", isbs.Name),
+		logger:     logger.With("isbsvc", isbSvc.Name),
 	}
 }
 
 func (r *redisInstaller) Install(ctx context.Context) (*dfv1.BufferServiceConfig, error) {
-	if redis := r.isbs.Spec.Redis; redis == nil {
-		return nil, fmt.Errorf("invalid native redis isbs spec")
+	if redis := r.isbSvc.Spec.Redis; redis == nil {
+		return nil, fmt.Errorf("invalid native redis isb svc spec")
 	}
-	r.isbs.Status.SetType(dfv1.ISBSvcTypeRedis)
+	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.isbs.Status.MarkDeployFailed("RedisScriptsConfigMapFailed", 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.isbs.Status.MarkDeployFailed("RedisHealthConfigMapFailed", 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.isbs.Status.MarkDeployFailed("RedisConfConfigMapFailed", err.Error())
+		r.isbSvc.Status.MarkDeployFailed("RedisConfConfigMapFailed", err.Error())
 		return nil, err
 	}
 	if err := r.createAuthCredentialSecret(ctx); err != nil {
 		r.logger.Errorw("Failed to create Redis password", zap.Error(err))
-		r.isbs.Status.MarkDeployFailed("RedisPasswordSecretFailed", err.Error())
+		r.isbSvc.Status.MarkDeployFailed("RedisPasswordSecretFailed", err.Error())
 		return nil, err
 	}
 	if err := r.createRedisService(ctx); err != nil {
 		r.logger.Errorw("Failed to create Redis Service", zap.Error(err))
-		r.isbs.Status.MarkDeployFailed("RedisServiceFailed", 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.isbs.Status.MarkDeployFailed("RedisHeadlessServiceFailed", 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.isbs.Status.MarkDeployFailed("RedisStatefulSetFailed", err.Error())
+		r.isbSvc.Status.MarkDeployFailed("RedisStatefulSetFailed", err.Error())
 		return nil, err
 	}
 
-	r.isbs.Status.MarkDeployed()
+	r.isbSvc.Status.MarkDeployed()
 	return &dfv1.BufferServiceConfig{
 		Redis: &dfv1.RedisConfig{
-			SentinelURL: fmt.Sprintf("%s.%s.svc:%v", generateRedisServiceName(r.isbs), r.isbs.Namespace, sentinelPort),
+			SentinelURL: fmt.Sprintf("%s.%s.svc:%v", generateRedisServiceName(r.isbSvc), r.isbSvc.Namespace, sentinelPort),
 			MasterName:  dfv1.DefaultRedisSentinelMasterName,
 			User:        defaultUser,
 			Password: &corev1.SecretKeySelector{
 				LocalObjectReference: corev1.LocalObjectReference{
-					Name: generateRedisCredentialSecretName(r.isbs),
+					Name: generateRedisCredentialSecretName(r.isbSvc),
 				},
 				Key: dfv1.RedisAuthSecretKey,
 			},
 			SentinelPassword: &corev1.SecretKeySelector{
 				LocalObjectReference: corev1.LocalObjectReference{
-					Name: generateRedisCredentialSecretName(r.isbs),
+					Name: generateRedisCredentialSecretName(r.isbSvc),
 				},
 				Key: dfv1.RedisAuthSecretKey,
 			},
@@ -139,11 +139,11 @@ func (r *redisInstaller) createAuthCredentialSecret(ctx context.Context) error {
 	password := sharedutil.RandomString(16)
 	obj := &corev1.Secret{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: r.isbs.Namespace,
-			Name:      generateRedisCredentialSecretName(r.isbs),
+			Namespace: r.isbSvc.Namespace,
+			Name:      generateRedisCredentialSecretName(r.isbSvc),
 			Labels:    r.labels,
 			OwnerReferences: []metav1.OwnerReference{
-				*metav1.NewControllerRef(r.isbs.GetObjectMeta(), dfv1.ISBGroupVersionKind),
+				*metav1.NewControllerRef(r.isbSvc.GetObjectMeta(), dfv1.ISBGroupVersionKind),
 			},
 		},
 		Type: corev1.SecretTypeOpaque,
@@ -151,7 +151,7 @@ func (r *redisInstaller) createAuthCredentialSecret(ctx context.Context) error {
 			dfv1.RedisAuthSecretKey: []byte(password),
 		},
 	}
-	if _, err := r.kubeClient.CoreV1().Secrets(r.isbs.Namespace).Get(ctx, generateRedisCredentialSecretName(r.isbs), metav1.GetOptions{}); err != nil {
+	if _, err := r.kubeClient.CoreV1().Secrets(r.isbSvc.Namespace).Get(ctx, generateRedisCredentialSecretName(r.isbSvc), metav1.GetOptions{}); err != nil {
 		if apierrors.IsNotFound(err) {
 			if err := r.client.Create(ctx, obj); err != nil {
 				return fmt.Errorf("failed to create redis auth credential secret, err: %w", err)
@@ -178,11 +178,11 @@ func (r *redisInstaller) createScriptsConfigMap(ctx context.Context) error {
 		RedisPort           int32
 		SentinelPort        int32
 	}{
-		ServiceName:         generateRedisServiceName(r.isbs),
-		HeadlessServiceName: generateRedisHeadlessServiceName(r.isbs),
-		StatefulSetName:     generateRedisStatefulSetName(r.isbs),
-		Namespace:           r.isbs.Namespace,
-		Replicas:            r.isbs.Spec.Redis.Native.GetReplicas(),
+		ServiceName:         generateRedisServiceName(r.isbSvc),
+		HeadlessServiceName: generateRedisHeadlessServiceName(r.isbSvc),
+		StatefulSetName:     generateRedisStatefulSetName(r.isbSvc),
+		Namespace:           r.isbSvc.Namespace,
+		Replicas:            r.isbSvc.Spec.Redis.Native.GetReplicas(),
 		RedisPort:           redisPort,
 		SentinelPort:        sentinelPort,
 	}
@@ -200,14 +200,14 @@ func (r *redisInstaller) createScriptsConfigMap(ctx context.Context) error {
 	hash := sharedutil.MustHash(data)
 	obj := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: r.isbs.Namespace,
-			Name:      generateScriptsConfigMapName(r.isbs),
+			Namespace: r.isbSvc.Namespace,
+			Name:      generateScriptsConfigMapName(r.isbSvc),
 			Labels:    r.labels,
 			Annotations: map[string]string{
 				dfv1.KeyHash: hash,
 			},
 			OwnerReferences: []metav1.OwnerReference{
-				*metav1.NewControllerRef(r.isbs.GetObjectMeta(), dfv1.ISBGroupVersionKind),
+				*metav1.NewControllerRef(r.isbSvc.GetObjectMeta(), dfv1.ISBGroupVersionKind),
 			},
 		},
 		Data: data,
@@ -249,14 +249,14 @@ func (r *redisInstaller) createHealthConfigMap(ctx context.Context) error {
 	hash := sharedutil.MustHash(data)
 	obj := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: r.isbs.Namespace,
-			Name:      generateHealthConfigMapName(r.isbs),
+			Namespace: r.isbSvc.Namespace,
+			Name:      generateHealthConfigMapName(r.isbSvc),
 			Labels:    r.labels,
 			Annotations: map[string]string{
 				dfv1.KeyHash: hash,
 			},
 			OwnerReferences: []metav1.OwnerReference{
-				*metav1.NewControllerRef(r.isbs.GetObjectMeta(), dfv1.ISBGroupVersionKind),
+				*metav1.NewControllerRef(r.isbSvc.GetObjectMeta(), dfv1.ISBGroupVersionKind),
 			},
 		},
 		Data: data,
@@ -301,7 +301,7 @@ func (r *redisInstaller) createConfConfigMap(ctx context.Context) error {
 			sentinelConf = x.Sentinel
 		}
 	}
-	if x := r.isbs.Spec.Redis.Native.Settings; x != nil {
+	if x := r.isbSvc.Spec.Redis.Native.Settings; x != nil {
 		if x.Redis != "" {
 			redisConf = x.Redis
 		}
@@ -328,12 +328,12 @@ func (r *redisInstaller) createConfConfigMap(ctx context.Context) error {
 		Quorum              int
 	}{
 		SentinelPort:        sentinelPort,
-		StatefulSetName:     generateRedisStatefulSetName(r.isbs),
-		HeadlessServiceName: generateRedisHeadlessServiceName(r.isbs),
-		Namespace:           r.isbs.Namespace,
+		StatefulSetName:     generateRedisStatefulSetName(r.isbSvc),
+		HeadlessServiceName: generateRedisHeadlessServiceName(r.isbSvc),
+		Namespace:           r.isbSvc.Namespace,
 		SentinelSettings:    sentinelConf,
 		RedisPort:           redisPort,
-		Quorum:              r.isbs.Spec.Redis.Native.GetReplicas() / 2,
+		Quorum:              r.isbSvc.Spec.Redis.Native.GetReplicas() / 2,
 	}); err != nil {
 		return fmt.Errorf("failed to parse sentinel config template, error: %w", err)
 	}
@@ -375,14 +375,14 @@ func (r *redisInstaller) createConfConfigMap(ctx context.Context) error {
 	hash := sharedutil.MustHash(data)
 	obj := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: r.isbs.Namespace,
-			Name:      generateRedisConfigMapName(r.isbs),
+			Namespace: r.isbSvc.Namespace,
+			Name:      generateRedisConfigMapName(r.isbSvc),
 			Labels:    r.labels,
 			Annotations: map[string]string{
 				dfv1.KeyHash: hash,
 			},
 			OwnerReferences: []metav1.OwnerReference{
-				*metav1.NewControllerRef(r.isbs.GetObjectMeta(), dfv1.ISBGroupVersionKind),
+				*metav1.NewControllerRef(r.isbSvc.GetObjectMeta(), dfv1.ISBGroupVersionKind),
 			},
 		},
 		Data: data,
@@ -411,7 +411,7 @@ func (r *redisInstaller) createConfConfigMap(ctx context.Context) error {
 }
 
 func (r *redisInstaller) createRedisService(ctx context.Context) error {
-	spec := r.isbs.Spec.Redis.Native.GetServiceSpec(dfv1.GetRedisServiceSpecReq{
+	spec := r.isbSvc.Spec.Redis.Native.GetServiceSpec(dfv1.GetRedisServiceSpecReq{
 		Labels:                r.labels,
 		RedisContainerPort:    redisPort,
 		SentinelContainerPort: sentinelPort,
@@ -419,14 +419,14 @@ func (r *redisInstaller) createRedisService(ctx context.Context) error {
 	hash := sharedutil.MustHash(spec)
 	obj := &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: r.isbs.Namespace,
-			Name:      generateRedisServiceName(r.isbs),
+			Namespace: r.isbSvc.Namespace,
+			Name:      generateRedisServiceName(r.isbSvc),
 			Labels:    r.labels,
 			Annotations: map[string]string{
 				dfv1.KeyHash: hash,
 			},
 			OwnerReferences: []metav1.OwnerReference{
-				*metav1.NewControllerRef(r.isbs.GetObjectMeta(), dfv1.ISBGroupVersionKind),
+				*metav1.NewControllerRef(r.isbSvc.GetObjectMeta(), dfv1.ISBGroupVersionKind),
 			},
 		},
 		Spec: spec,
@@ -455,7 +455,7 @@ func (r *redisInstaller) createRedisService(ctx context.Context) error {
 }
 
 func (r *redisInstaller) createRedisHeadlessService(ctx context.Context) error {
-	spec := r.isbs.Spec.Redis.Native.GetHeadlessServiceSpec(dfv1.GetRedisServiceSpecReq{
+	spec := r.isbSvc.Spec.Redis.Native.GetHeadlessServiceSpec(dfv1.GetRedisServiceSpecReq{
 		Labels:                r.labels,
 		RedisContainerPort:    redisPort,
 		SentinelContainerPort: sentinelPort,
@@ -463,14 +463,14 @@ func (r *redisInstaller) createRedisHeadlessService(ctx context.Context) error {
 	hash := sharedutil.MustHash(spec)
 	obj := &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: r.isbs.Namespace,
-			Name:      generateRedisHeadlessServiceName(r.isbs),
+			Namespace: r.isbSvc.Namespace,
+			Name:      generateRedisHeadlessServiceName(r.isbSvc),
 			Labels:    r.labels,
 			Annotations: map[string]string{
 				dfv1.KeyHash: hash,
 			},
 			OwnerReferences: []metav1.OwnerReference{
-				*metav1.NewControllerRef(r.isbs.GetObjectMeta(), dfv1.ISBGroupVersionKind),
+				*metav1.NewControllerRef(r.isbSvc.GetObjectMeta(), dfv1.ISBGroupVersionKind),
 			},
 		},
 		Spec: spec,
@@ -499,12 +499,12 @@ func (r *redisInstaller) createRedisHeadlessService(ctx context.Context) error {
 }
 
 func (r *redisInstaller) createStatefulSet(ctx context.Context) error {
-	redisVersion, err := r.config.GetRedisVersion(r.isbs.Spec.Redis.Native.Version)
+	redisVersion, err := r.config.GetRedisVersion(r.isbSvc.Spec.Redis.Native.Version)
 	if err != nil {
 		return fmt.Errorf("failed to get redis version, err: %w", err)
 	}
-	spec := r.isbs.Spec.Redis.Native.GetStatefulSetSpec(dfv1.GetRedisStatefulSetSpecReq{
-		ServiceName:               generateRedisHeadlessServiceName(r.isbs),
+	spec := r.isbSvc.Spec.Redis.Native.GetStatefulSetSpec(dfv1.GetRedisStatefulSetSpecReq{
+		ServiceName:               generateRedisHeadlessServiceName(r.isbSvc),
 		Labels:                    r.labels,
 		RedisImage:                redisVersion.RedisImage,
 		SentinelImage:             redisVersion.SentinelImage,
@@ -513,24 +513,24 @@ func (r *redisInstaller) createStatefulSet(ctx context.Context) error {
 		RedisContainerPort:        redisPort,
 		SentinelContainerPort:     sentinelPort,
 		RedisMetricsContainerPort: redisMetricsPort,
-		CredentialSecretName:      generateRedisCredentialSecretName(r.isbs),
+		CredentialSecretName:      generateRedisCredentialSecretName(r.isbSvc),
 		TLSEnabled:                false,
-		PvcNameIfNeeded:           generateRedisPVCName(r.isbs),
-		ConfConfigMapName:         generateRedisConfigMapName(r.isbs),
-		HealthConfigMapName:       generateHealthConfigMapName(r.isbs),
-		ScriptsConfigMapName:      generateScriptsConfigMapName(r.isbs),
+		PvcNameIfNeeded:           generateRedisPVCName(r.isbSvc),
+		ConfConfigMapName:         generateRedisConfigMapName(r.isbSvc),
+		HealthConfigMapName:       generateHealthConfigMapName(r.isbSvc),
+		ScriptsConfigMapName:      generateScriptsConfigMapName(r.isbSvc),
 	})
 	hash := sharedutil.MustHash(spec)
 	obj := &appv1.StatefulSet{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: r.isbs.Namespace,
-			Name:      generateRedisStatefulSetName(r.isbs),
+			Namespace: r.isbSvc.Namespace,
+			Name:      generateRedisStatefulSetName(r.isbSvc),
 			Labels:    r.labels,
 			Annotations: map[string]string{
 				dfv1.KeyHash: hash,
 			},
 			OwnerReferences: []metav1.OwnerReference{
-				*metav1.NewControllerRef(r.isbs.GetObjectMeta(), dfv1.ISBGroupVersionKind),
+				*metav1.NewControllerRef(r.isbSvc.GetObjectMeta(), dfv1.ISBGroupVersionKind),
 			},
 		},
 		Spec: spec,
@@ -586,7 +586,7 @@ func (r *redisInstaller) uninstallPVCs(ctx context.Context) error {
 func (r *redisInstaller) getPVCs(ctx context.Context) ([]corev1.PersistentVolumeClaim, error) {
 	pvcl := &corev1.PersistentVolumeClaimList{}
 	err := r.client.List(ctx, pvcl, &client.ListOptions{
-		Namespace:     r.isbs.Namespace,
+		Namespace:     r.isbSvc.Namespace,
 		LabelSelector: labels.SelectorFromSet(r.labels),
 	})
 	if err != nil {
@@ -595,34 +595,34 @@ func (r *redisInstaller) getPVCs(ctx context.Context) ([]corev1.PersistentVolume
 	return pvcl.Items, nil
 }
 
-func generateRedisServiceName(isbs *dfv1.InterStepBufferService) string {
-	return fmt.Sprintf("isbsvc-%s-redis-svc", isbs.Name)
+func generateRedisServiceName(isbSvc *dfv1.InterStepBufferService) string {
+	return fmt.Sprintf("isbsvc-%s-redis-svc", isbSvc.Name)
 }
 
-func generateRedisHeadlessServiceName(isbs *dfv1.InterStepBufferService) string {
-	return fmt.Sprintf("isbsvc-%s-redis-svc-headless", isbs.Name)
+func generateRedisHeadlessServiceName(isbSvc *dfv1.InterStepBufferService) string {
+	return fmt.Sprintf("isbsvc-%s-redis-svc-headless", isbSvc.Name)
 }
 
-func generateRedisConfigMapName(isbs *dfv1.InterStepBufferService) string {
-	return fmt.Sprintf("isbsvc-%s-redis-config", isbs.Name)
+func generateRedisConfigMapName(isbSvc *dfv1.InterStepBufferService) string {
+	return fmt.Sprintf("isbsvc-%s-redis-config", isbSvc.Name)
 }
 
-func generateScriptsConfigMapName(isbs *dfv1.InterStepBufferService) string {
-	return fmt.Sprintf("isbsvc-%s-redis-scripts", isbs.Name)
+func generateScriptsConfigMapName(isbSvc *dfv1.InterStepBufferService) string {
+	return fmt.Sprintf("isbsvc-%s-redis-scripts", isbSvc.Name)
 }
 
-func generateHealthConfigMapName(isbs *dfv1.InterStepBufferService) string {
-	return fmt.Sprintf("isbsvc-%s-redis-health", isbs.Name)
+func generateHealthConfigMapName(isbSvc *dfv1.InterStepBufferService) string {
+	return fmt.Sprintf("isbsvc-%s-redis-health", isbSvc.Name)
 }
 
-func generateRedisStatefulSetName(isbs *dfv1.InterStepBufferService) string {
-	return fmt.Sprintf("isbsvc-%s-redis", isbs.Name)
+func generateRedisStatefulSetName(isbSvc *dfv1.InterStepBufferService) string {
+	return fmt.Sprintf("isbsvc-%s-redis", isbSvc.Name)
 }
 
-func generateRedisCredentialSecretName(isbs *dfv1.InterStepBufferService) string {
-	return fmt.Sprintf("isbsvc-%s-redis-auth", isbs.Name)
+func generateRedisCredentialSecretName(isbSvc *dfv1.InterStepBufferService) string {
+	return fmt.Sprintf("isbsvc-%s-redis-auth", isbSvc.Name)
 }
 
-func generateRedisPVCName(isbs *dfv1.InterStepBufferService) string {
-	return fmt.Sprintf("isbsvc-%s-redis-vol", isbs.Name)
+func generateRedisPVCName(isbSvc *dfv1.InterStepBufferService) string {
+	return fmt.Sprintf("isbsvc-%s-redis-vol", isbSvc.Name)
 }


Test patch:
diff --git a/pkg/reconciler/isbsvc/installer/external_redis_test.go b/pkg/reconciler/isbsvc/installer/external_redis_test.go
index 4f788398..c257ac12 100644
--- a/pkg/reconciler/isbsvc/installer/external_redis_test.go
+++ b/pkg/reconciler/isbsvc/installer/external_redis_test.go
@@ -29,7 +29,7 @@ func TestExternalRedisInstallation(t *testing.T) {
 		badIsbs := testExternalRedisIsbSvc.DeepCopy()
 		badIsbs.Spec.Redis = nil
 		installer := &externalRedisInstaller{
-			isbs:   badIsbs,
+			isbSvc: badIsbs,
 			logger: zaptest.NewLogger(t).Sugar(),
 		}
 		_, err := installer.Install(context.TODO())
@@ -39,7 +39,7 @@ func TestExternalRedisInstallation(t *testing.T) {
 	t.Run("good installation", func(t *testing.T) {
 		goodIsbs := testExternalRedisIsbSvc.DeepCopy()
 		installer := &externalRedisInstaller{
-			isbs:   goodIsbs,
+			isbSvc: goodIsbs,
 			logger: zaptest.NewLogger(t).Sugar(),
 		}
 		c, err := installer.Install(context.TODO())
@@ -52,7 +52,7 @@ func TestExternalRedisInstallation(t *testing.T) {
 func TestExternalRedisUninstallation(t *testing.T) {
 	obj := testExternalRedisIsbSvc.DeepCopy()
 	installer := &externalRedisInstaller{
-		isbs:   obj,
+		isbSvc: obj,
 		logger: zaptest.NewLogger(t).Sugar(),
 	}
 	err := installer.Uninstall(context.TODO())
diff --git a/pkg/reconciler/isbsvc/installer/jetstream_test.go b/pkg/reconciler/isbsvc/installer/jetstream_test.go
index 0b3a8815..bdfdbb09 100644
--- a/pkg/reconciler/isbsvc/installer/jetstream_test.go
+++ b/pkg/reconciler/isbsvc/installer/jetstream_test.go
@@ -43,7 +43,7 @@ func TestJetStreamBadInstallation(t *testing.T) {
 		badIsbs.Spec.JetStream = nil
 		installer := &jetStreamInstaller{
 			client: fake.NewClientBuilder().Build(),
-			isbs:   badIsbs,
+			isbSvc: badIsbs,
 			config: fakeConfig,
 			labels: testLabels,
 			logger: zaptest.NewLogger(t).Sugar(),
@@ -74,7 +74,7 @@ func TestJetStreamCreateObjects(t *testing.T) {
 	i := &jetStreamInstaller{
 		client:     cl,
 		kubeClient: k8sfake.NewSimpleClientset(),
-		isbs:       testJetStreamIsbSvc,
+		isbSvc:     testJetStreamIsbSvc,
 		config:     fakeConfig,
 		labels:     testLabels,
 		logger:     zaptest.NewLogger(t).Sugar(),
@@ -82,7 +82,7 @@ func TestJetStreamCreateObjects(t *testing.T) {
 
 	t.Run("test create sts", func(t *testing.T) {
 		testObj := testJetStreamIsbSvc.DeepCopy()
-		i.isbs = testObj
+		i.isbSvc = testObj
 		err := i.createStatefulSet(ctx)
 		assert.NoError(t, err)
 		sts := &appv1.StatefulSet{}
@@ -98,7 +98,7 @@ func TestJetStreamCreateObjects(t *testing.T) {
 
 	t.Run("test create svc", func(t *testing.T) {
 		testObj := testJetStreamIsbSvc.DeepCopy()
-		i.isbs = testObj
+		i.isbSvc = testObj
 		err := i.createService(ctx)
 		assert.NoError(t, err)
 		svc := &corev1.Service{}
@@ -110,7 +110,7 @@ func TestJetStreamCreateObjects(t *testing.T) {
 
 	t.Run("test create auth secrets", func(t *testing.T) {
 		testObj := testJetStreamIsbSvc.DeepCopy()
-		i.isbs = testObj
+		i.isbSvc = testObj
 		err := i.createSecrets(ctx)
 		assert.NoError(t, err)
 		s := &corev1.Secret{}
@@ -135,7 +135,7 @@ func TestJetStreamCreateObjects(t *testing.T) {
 
 	t.Run("test create configmap", func(t *testing.T) {
 		testObj := testJetStreamIsbSvc.DeepCopy()
-		i.isbs = testObj
+		i.isbSvc = testObj
 		err := i.createConfigMap(ctx)
 		assert.NoError(t, err)
 		c := &corev1.ConfigMap{}
@@ -152,7 +152,7 @@ func Test_JetStreamInstall_Uninstall(t *testing.T) {
 	i := &jetStreamInstaller{
 		client:     cl,
 		kubeClient: k8sfake.NewSimpleClientset(),
-		isbs:       testJetStreamIsbSvc,
+		isbSvc:     testJetStreamIsbSvc,
 		config:     fakeConfig,
 		labels:     testLabels,
 		logger:     zaptest.NewLogger(t).Sugar(),
diff --git a/pkg/reconciler/isbsvc/installer/native_redis_test.go b/pkg/reconciler/isbsvc/installer/native_redis_test.go
index 0b9e9f65..7f72e640 100644
--- a/pkg/reconciler/isbsvc/installer/native_redis_test.go
+++ b/pkg/reconciler/isbsvc/installer/native_redis_test.go
@@ -43,7 +43,7 @@ func TestNativeRedisBadInstallation(t *testing.T) {
 		badIsbs.Spec.Redis = nil
 		installer := &redisInstaller{
 			client: fake.NewClientBuilder().Build(),
-			isbs:   badIsbs,
+			isbSvc: badIsbs,
 			config: fakeConfig,
 			labels: testLabels,
 			logger: zaptest.NewLogger(t).Sugar(),
@@ -78,7 +78,7 @@ func TestNativeRedisCreateObjects(t *testing.T) {
 	i := &redisInstaller{
 		client:     cl,
 		kubeClient: k8sfake.NewSimpleClientset(),
-		isbs:       testNativeRedisIsbSvc,
+		isbSvc:     testNativeRedisIsbSvc,
 		config:     fakeConfig,
 		labels:     testLabels,
 		logger:     zaptest.NewLogger(t).Sugar(),
@@ -86,7 +86,7 @@ func TestNativeRedisCreateObjects(t *testing.T) {
 
 	t.Run("test create sts", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
-		i.isbs = testObj
+		i.isbSvc = testObj
 		err := i.createStatefulSet(ctx)
 		assert.NoError(t, err)
 		sts := &appv1.StatefulSet{}
@@ -102,7 +102,7 @@ func TestNativeRedisCreateObjects(t *testing.T) {
 
 	t.Run("test create redis svc", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
-		i.isbs = testObj
+		i.isbSvc = testObj
 		err := i.createRedisService(ctx)
 		assert.NoError(t, err)
 		svc := &corev1.Service{}
@@ -114,7 +114,7 @@ func TestNativeRedisCreateObjects(t *testing.T) {
 
 	t.Run("test create redis headless svc", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
-		i.isbs = testObj
+		i.isbSvc = testObj
 		err := i.createRedisHeadlessService(ctx)
 		assert.NoError(t, err)
 		svc := &corev1.Service{}
@@ -126,7 +126,7 @@ func TestNativeRedisCreateObjects(t *testing.T) {
 
 	t.Run("test create redis auth secret", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
-		i.isbs = testObj
+		i.isbSvc = testObj
 		err := i.createAuthCredentialSecret(ctx)
 		assert.NoError(t, err)
 		s := &corev1.Secret{}
@@ -138,7 +138,7 @@ func TestNativeRedisCreateObjects(t *testing.T) {
 
 	t.Run("test create redis config", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
-		i.isbs = testObj
+		i.isbSvc = testObj
 		err := i.createConfConfigMap(ctx)
 		assert.NoError(t, err)
 		c := &corev1.ConfigMap{}
@@ -150,7 +150,7 @@ func TestNativeRedisCreateObjects(t *testing.T) {
 
 	t.Run("test create redis scripts config", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
-		i.isbs = testObj
+		i.isbSvc = testObj
 		err := i.createScriptsConfigMap(ctx)
 		assert.NoError(t, err)
 		c := &corev1.ConfigMap{}
@@ -162,7 +162,7 @@ func TestNativeRedisCreateObjects(t *testing.T) {
 
 	t.Run("test create redis health config", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
-		i.isbs = testObj
+		i.isbSvc = testObj
 		err := i.createHealthConfigMap(ctx)
 		assert.NoError(t, err)
 		c := &corev1.ConfigMap{}
@@ -179,7 +179,7 @@ func Test_NativeRedisInstall_Uninstall(t *testing.T) {
 	i := &redisInstaller{
 		client:     cl,
 		kubeClient: k8sfake.NewSimpleClientset(),
-		isbs:       testNativeRedisIsbSvc,
+		isbSvc:     testNativeRedisIsbSvc,
 		config:     fakeConfig,
 		labels:     testLabels,
 		logger:     zaptest.NewLogger(t).Sugar(),