Repository: numaproj/numaflow
Pull request #542 (2023-02-10 18:23:21)

feat: remove secret watch privilege dependency
Giving Secret watch privilege is kind of sensitive to some users, this PR remove the secret watch dependency from the controller.

<!--

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.

-->


whynowy: Remain unchanged.

This is only for the auto generated secrets for Isb Serviy.

> what is our strategy now for managing secrets?



Patch:
diff --git a/config/advanced-install/namespaced-controller-wo-crds.yaml b/config/advanced-install/namespaced-controller-wo-crds.yaml
index 0d185c5c..a50175b1 100644
--- a/config/advanced-install/namespaced-controller-wo-crds.yaml
+++ b/config/advanced-install/namespaced-controller-wo-crds.yaml
@@ -36,7 +36,6 @@ rules:
   - pods
   - pods/exec
   - configmaps
-  - secrets
   - services
   - persistentvolumeclaims
   verbs:
@@ -47,6 +46,17 @@ rules:
   - update
   - patch
   - delete
+- apiGroups:
+  - ""
+  resources:
+  - secrets
+  verbs:
+  - create
+  - get
+  - list
+  - update
+  - patch
+  - delete
 - apiGroups:
   - apps
   resources:
diff --git a/config/cluster-install/rbac/controller-manager/numaflow-cluster-role.yaml b/config/cluster-install/rbac/controller-manager/numaflow-cluster-role.yaml
index 3311c77e..4df4efc2 100644
--- a/config/cluster-install/rbac/controller-manager/numaflow-cluster-role.yaml
+++ b/config/cluster-install/rbac/controller-manager/numaflow-cluster-role.yaml
@@ -31,7 +31,6 @@ rules:
       - pods
       - pods/exec
       - configmaps
-      - secrets
       - services
       - persistentvolumeclaims
     verbs:
@@ -42,6 +41,17 @@ rules:
       - update
       - patch
       - delete
+  - apiGroups:
+      - ""
+    resources:
+      - secrets
+    verbs:
+      - create
+      - get
+      - list
+      - update
+      - patch
+      - delete
   - apiGroups:
       - apps
     resources:
diff --git a/config/install.yaml b/config/install.yaml
index ae3094f7..5480a00a 100644
--- a/config/install.yaml
+++ b/config/install.yaml
@@ -11207,7 +11207,6 @@ rules:
   - pods
   - pods/exec
   - configmaps
-  - secrets
   - services
   - persistentvolumeclaims
   verbs:
@@ -11218,6 +11217,17 @@ rules:
   - update
   - patch
   - delete
+- apiGroups:
+  - ""
+  resources:
+  - secrets
+  verbs:
+  - create
+  - get
+  - list
+  - update
+  - patch
+  - delete
 - apiGroups:
   - apps
   resources:
diff --git a/config/namespace-install.yaml b/config/namespace-install.yaml
index 2ff3527f..6fa0317e 100644
--- a/config/namespace-install.yaml
+++ b/config/namespace-install.yaml
@@ -11120,7 +11120,6 @@ rules:
   - pods
   - pods/exec
   - configmaps
-  - secrets
   - services
   - persistentvolumeclaims
   verbs:
@@ -11131,6 +11130,17 @@ rules:
   - update
   - patch
   - delete
+- apiGroups:
+  - ""
+  resources:
+  - secrets
+  verbs:
+  - create
+  - get
+  - list
+  - update
+  - patch
+  - delete
 - apiGroups:
   - apps
   resources:
diff --git a/config/namespace-install/rbac/controller-manager/numaflow-role.yaml b/config/namespace-install/rbac/controller-manager/numaflow-role.yaml
index 721daa2b..975793e3 100644
--- a/config/namespace-install/rbac/controller-manager/numaflow-role.yaml
+++ b/config/namespace-install/rbac/controller-manager/numaflow-role.yaml
@@ -31,7 +31,6 @@ rules:
       - pods
       - pods/exec
       - configmaps
-      - secrets
       - services
       - persistentvolumeclaims
     verbs:
@@ -42,6 +41,17 @@ rules:
       - update
       - patch
       - delete
+  - apiGroups:
+      - ""
+    resources:
+      - secrets
+    verbs:
+      - create
+      - get
+      - list
+      - update
+      - patch
+      - delete
   - apiGroups:
       - apps
     resources:
diff --git a/pkg/isb/stores/simplebuffer/options.go b/pkg/isb/stores/simplebuffer/options.go
index d083926d..df9ea98d 100644
--- a/pkg/isb/stores/simplebuffer/options.go
+++ b/pkg/isb/stores/simplebuffer/options.go
@@ -1,3 +1,19 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package simplebuffer
 
 import "time"
diff --git a/pkg/metrics/health_checker.go b/pkg/metrics/health_checker.go
index 0f257bee..031b0eb8 100644
--- a/pkg/metrics/health_checker.go
+++ b/pkg/metrics/health_checker.go
@@ -1,3 +1,19 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package metrics
 
 import "context"
diff --git a/pkg/reconciler/cmd/start.go b/pkg/reconciler/cmd/start.go
index 6cbdcc05..b73245e4 100644
--- a/pkg/reconciler/cmd/start.go
+++ b/pkg/reconciler/cmd/start.go
@@ -31,6 +31,7 @@ import (
 	"go.uber.org/zap"
 	appv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/event"
@@ -61,11 +62,14 @@ func Start(namespaced bool, managedNamespace string) {
 	if namespaced {
 		opts.Namespace = managedNamespace
 	}
-	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), opts)
+	restConfig := ctrl.GetConfigOrDie()
+	mgr, err := ctrl.NewManager(restConfig, opts)
 	if err != nil {
 		logger.Fatalw("Unable to get a controller-runtime manager", zap.Error(err))
 	}
 
+	kubeClient := kubernetes.NewForConfigOrDie(restConfig)
+
 	// Readiness probe
 	if err := mgr.AddReadyzCheck("readiness", healthz.Ping); err != nil {
 		logger.Fatalw("Unable add a readiness check", zap.Error(err))
@@ -80,7 +84,7 @@ func Start(namespaced bool, managedNamespace string) {
 	}
 
 	isbSvcController, err := controller.New(dfv1.ControllerISBSvc, mgr, controller.Options{
-		Reconciler: isbsvcctrl.NewReconciler(mgr.GetClient(), mgr.GetScheme(), config, logger),
+		Reconciler: isbsvcctrl.NewReconciler(mgr.GetClient(), kubeClient, mgr.GetScheme(), config, logger),
 	})
 	if err != nil {
 		logger.Fatalw("Unable to set up ISB controller", zap.Error(err))
@@ -98,11 +102,6 @@ func Start(namespaced bool, managedNamespace string) {
 		logger.Fatalw("Unable to watch ConfigMaps", zap.Error(err))
 	}
 
-	// Watch Secrets with ResourceVersion changes, and enqueue owning InterStepBuffer key
-	if err := isbSvcController.Watch(&source.Kind{Type: &corev1.Secret{}}, &handler.EnqueueRequestForOwner{OwnerType: &dfv1.InterStepBufferService{}, IsController: true}, predicate.ResourceVersionChangedPredicate{}); err != nil {
-		logger.Fatalw("Unable to watch Secrets", zap.Error(err))
-	}
-
 	// Watch StatefulSets with Generation changes, and enqueue owning InterStepBuffer key
 	if err := isbSvcController.Watch(&source.Kind{Type: &appv1.StatefulSet{}}, &handler.EnqueueRequestForOwner{OwnerType: &dfv1.InterStepBufferService{}, IsController: true}, predicate.GenerationChangedPredicate{}); err != nil {
 		logger.Fatalw("Unable to watch StatefulSets", zap.Error(err))
diff --git a/pkg/reconciler/isbsvc/controller.go b/pkg/reconciler/isbsvc/controller.go
index 18365f74..4206f2ee 100644
--- a/pkg/reconciler/isbsvc/controller.go
+++ b/pkg/reconciler/isbsvc/controller.go
@@ -26,6 +26,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
@@ -38,15 +39,16 @@ const (
 
 // interStepBufferReconciler reconciles an Inter-Step Buffer Service object.
 type interStepBufferServiceReconciler struct {
-	client client.Client
-	scheme *runtime.Scheme
+	client     client.Client
+	kubeClient kubernetes.Interface
+	scheme     *runtime.Scheme
 
 	config *reconciler.GlobalConfig
 	logger *zap.SugaredLogger
 }
 
-func NewReconciler(client client.Client, scheme *runtime.Scheme, config *reconciler.GlobalConfig, logger *zap.SugaredLogger) reconcile.Reconciler {
-	return &interStepBufferServiceReconciler{client: client, scheme: scheme, config: config, logger: logger}
+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 (r *interStepBufferServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
@@ -83,7 +85,7 @@ func (r *interStepBufferServiceReconciler) reconcile(ctx context.Context, isbs *
 		log.Info("Deleting isbs")
 		if controllerutil.ContainsFinalizer(isbs, finalizerName) {
 			// Finalizer logic should be added here.
-			if err := installer.Uninstall(ctx, isbs, r.client, r.config, log); err != nil {
+			if err := installer.Uninstall(ctx, isbs, r.client, r.kubeClient, r.config, log); err != nil {
 				log.Errorw("Failed to uninstall", zap.Error(err))
 				return err
 			}
@@ -103,7 +105,7 @@ func (r *interStepBufferServiceReconciler) reconcile(ctx context.Context, isbs *
 	} else {
 		isbs.Status.MarkConfigured()
 	}
-	return installer.Install(ctx, isbs, r.client, r.config, log)
+	return installer.Install(ctx, isbs, r.client, r.kubeClient, r.config, log)
 }
 
 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 adfff8a5..6b2ad1a1 100644
--- a/pkg/reconciler/isbsvc/installer/installer.go
+++ b/pkg/reconciler/isbsvc/installer/installer.go
@@ -23,6 +23,7 @@ import (
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/reconciler"
 	"go.uber.org/zap"
+	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -35,8 +36,8 @@ type Installer interface {
 }
 
 // Install function installs the ISBS
-func Install(ctx context.Context, isbsvc *dfv1.InterStepBufferService, client client.Client, config *reconciler.GlobalConfig, logger *zap.SugaredLogger) error {
-	installer, err := getInstaller(isbsvc, client, 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
@@ -51,7 +52,7 @@ func Install(ctx context.Context, isbsvc *dfv1.InterStepBufferService, client cl
 }
 
 // GetInstaller returns Installer implementation
-func getInstaller(isbsvc *dfv1.InterStepBufferService, client client.Client, 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,
@@ -63,11 +64,11 @@ func getInstaller(isbsvc *dfv1.InterStepBufferService, client client.Client, con
 		if redis.External != nil {
 			return NewExternalRedisInstaller(isbsvc, logger), nil
 		} else if redis.Native != nil {
-			return NewNativeRedisInstaller(client, isbsvc, config, labels, logger), nil
+			return NewNativeRedisInstaller(client, kubeClient, isbsvc, config, labels, logger), nil
 		}
 	} else if js := isbsvc.Spec.JetStream; js != nil {
 		labels[dfv1.KeyISBSvcType] = string(dfv1.ISBSvcTypeJetStream)
-		return NewJetStreamInstaller(client, isbsvc, config, labels, logger), nil
+		return NewJetStreamInstaller(client, kubeClient, isbsvc, config, labels, logger), nil
 	}
 	return nil, fmt.Errorf("invalid isb service spec")
 }
@@ -80,8 +81,8 @@ func getInstaller(isbsvc *dfv1.InterStepBufferService, client client.Client, con
 // 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, config *reconciler.GlobalConfig, logger *zap.SugaredLogger) error {
-	installer, err := getInstaller(isbsvc, client, 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 41ba3b7b..70b5d298 100644
--- a/pkg/reconciler/isbsvc/installer/jetstream.go
+++ b/pkg/reconciler/isbsvc/installer/jetstream.go
@@ -37,7 +37,7 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/yaml"
 )
@@ -55,20 +55,22 @@ var (
 )
 
 type jetStreamInstaller struct {
-	client client.Client
-	isbs   *dfv1.InterStepBufferService
-	config *reconciler.GlobalConfig
-	labels map[string]string
-	logger *zap.SugaredLogger
+	client     client.Client
+	kubeClient kubernetes.Interface
+	isbs       *dfv1.InterStepBufferService
+	config     *reconciler.GlobalConfig
+	labels     map[string]string
+	logger     *zap.SugaredLogger
 }
 
-func NewJetStreamInstaller(client client.Client, isbs *dfv1.InterStepBufferService, config *reconciler.GlobalConfig, labels map[string]string, logger *zap.SugaredLogger) Installer {
+func NewJetStreamInstaller(client client.Client, kubeClient kubernetes.Interface, isbs *dfv1.InterStepBufferService, config *reconciler.GlobalConfig, labels map[string]string, logger *zap.SugaredLogger) Installer {
 	return &jetStreamInstaller{
-		client: client,
-		isbs:   isbs,
-		config: config,
-		labels: labels,
-		logger: logger.With("isbs", isbs.Name),
+		client:     client,
+		kubeClient: kubeClient,
+		isbs:       isbs,
+		config:     config,
+		labels:     labels,
+		logger:     logger.With("isbs", isbs.Name),
 	}
 }
 
@@ -247,8 +249,8 @@ func (r *jetStreamInstaller) createStatefulSet(ctx context.Context) error {
 func (r *jetStreamInstaller) createSecrets(ctx context.Context) error {
 	oldServerObjExisting, oldClientObjExisting := true, true
 
-	oldSObj := &corev1.Secret{}
-	if err := r.client.Get(ctx, types.NamespacedName{Namespace: r.isbs.Namespace, Name: generateJetStreamServerSecretName(r.isbs)}, oldSObj); err != nil {
+	oldSObj, err := r.kubeClient.CoreV1().Secrets(r.isbs.Namespace).Get(ctx, generateJetStreamServerSecretName(r.isbs), metav1.GetOptions{})
+	if err != nil {
 		if apierrors.IsNotFound(err) {
 			oldServerObjExisting = false
 		} else {
@@ -256,8 +258,8 @@ func (r *jetStreamInstaller) createSecrets(ctx context.Context) error {
 		}
 	}
 
-	oldCObj := &corev1.Secret{}
-	if err := r.client.Get(ctx, types.NamespacedName{Namespace: r.isbs.Namespace, Name: generateJetStreamClientAuthSecretName(r.isbs)}, oldCObj); err != nil {
+	oldCObj, err := r.kubeClient.CoreV1().Secrets(r.isbs.Namespace).Get(ctx, generateJetStreamClientAuthSecretName(r.isbs), metav1.GetOptions{})
+	if err != nil {
 		if apierrors.IsNotFound(err) {
 			oldClientObjExisting = false
 		} else {
diff --git a/pkg/reconciler/isbsvc/installer/native_redis.go b/pkg/reconciler/isbsvc/installer/native_redis.go
index 33cb9b8d..24d31ec7 100644
--- a/pkg/reconciler/isbsvc/installer/native_redis.go
+++ b/pkg/reconciler/isbsvc/installer/native_redis.go
@@ -32,6 +32,7 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -52,20 +53,22 @@ var (
 )
 
 type redisInstaller struct {
-	client client.Client
-	isbs   *dfv1.InterStepBufferService
-	config *reconciler.GlobalConfig
-	labels map[string]string
-	logger *zap.SugaredLogger
+	client     client.Client
+	kubeClient kubernetes.Interface
+	isbs       *dfv1.InterStepBufferService
+	config     *reconciler.GlobalConfig
+	labels     map[string]string
+	logger     *zap.SugaredLogger
 }
 
-func NewNativeRedisInstaller(client client.Client, isbs *dfv1.InterStepBufferService, config *reconciler.GlobalConfig, labels map[string]string, logger *zap.SugaredLogger) Installer {
+func NewNativeRedisInstaller(client client.Client, kubeClient kubernetes.Interface, isbs *dfv1.InterStepBufferService, config *reconciler.GlobalConfig, labels map[string]string, logger *zap.SugaredLogger) Installer {
 	return &redisInstaller{
-		client: client,
-		isbs:   isbs,
-		config: config,
-		labels: labels,
-		logger: logger.With("isbs", isbs.Name),
+		client:     client,
+		kubeClient: kubeClient,
+		isbs:       isbs,
+		config:     config,
+		labels:     labels,
+		logger:     logger.With("isbs", isbs.Name),
 	}
 }
 
@@ -148,8 +151,7 @@ func (r *redisInstaller) createAuthCredentialSecret(ctx context.Context) error {
 			dfv1.RedisAuthSecretKey: []byte(password),
 		},
 	}
-	old := &corev1.Secret{}
-	if err := r.client.Get(ctx, client.ObjectKeyFromObject(obj), old); err != nil {
+	if _, err := r.kubeClient.CoreV1().Secrets(r.isbs.Namespace).Get(ctx, generateRedisCredentialSecretName(r.isbs), 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)
diff --git a/pkg/reduce/metrics.go b/pkg/reduce/metrics.go
index 0e62133e..8c566563 100644
--- a/pkg/reduce/metrics.go
+++ b/pkg/reduce/metrics.go
@@ -1,3 +1,19 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package reduce
 
 import (
diff --git a/pkg/reduce/pbq/metrics.go b/pkg/reduce/pbq/metrics.go
index cc600d82..e5decc0c 100644
--- a/pkg/reduce/pbq/metrics.go
+++ b/pkg/reduce/pbq/metrics.go
@@ -1,3 +1,19 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package pbq
 
 import (
diff --git a/pkg/reduce/pnf/metrics.go b/pkg/reduce/pnf/metrics.go
index 0598ce7e..9425ea64 100644
--- a/pkg/reduce/pnf/metrics.go
+++ b/pkg/reduce/pnf/metrics.go
@@ -1,3 +1,19 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package pnf
 
 import (
diff --git a/pkg/reduce/readloop/metrics.go b/pkg/reduce/readloop/metrics.go
index ac2886da..c41e896b 100644
--- a/pkg/reduce/readloop/metrics.go
+++ b/pkg/reduce/readloop/metrics.go
@@ -1,3 +1,19 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package readloop
 
 import (


Test patch:
diff --git a/pkg/reconciler/isbsvc/controller_test.go b/pkg/reconciler/isbsvc/controller_test.go
index e3a3db88..9230cd6b 100644
--- a/pkg/reconciler/isbsvc/controller_test.go
+++ b/pkg/reconciler/isbsvc/controller_test.go
@@ -28,6 +28,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	k8sfake "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/kubernetes/scheme"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
@@ -107,7 +108,8 @@ func init() {
 
 func Test_NewReconciler(t *testing.T) {
 	cl := fake.NewClientBuilder().Build()
-	r := NewReconciler(cl, scheme.Scheme, fakeConfig, zaptest.NewLogger(t).Sugar())
+	kubeClient := k8sfake.NewSimpleClientset()
+	r := NewReconciler(cl, kubeClient, scheme.Scheme, fakeConfig, zaptest.NewLogger(t).Sugar())
 	_, ok := r.(*interStepBufferServiceReconciler)
 	assert.True(t, ok)
 }
@@ -118,10 +120,11 @@ func TestReconcileNativeRedis(t *testing.T) {
 		ctx := context.TODO()
 		cl := fake.NewClientBuilder().Build()
 		r := &interStepBufferServiceReconciler{
-			client: cl,
-			scheme: scheme.Scheme,
-			config: fakeConfig,
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:     cl,
+			kubeClient: k8sfake.NewSimpleClientset(),
+			scheme:     scheme.Scheme,
+			config:     fakeConfig,
+			logger:     zaptest.NewLogger(t).Sugar(),
 		}
 		err := r.reconcile(ctx, testIsb)
 		assert.NoError(t, err)
@@ -157,10 +160,11 @@ func TestReconcileJetStream(t *testing.T) {
 		ctx := context.TODO()
 		cl := fake.NewClientBuilder().Build()
 		r := &interStepBufferServiceReconciler{
-			client: cl,
-			scheme: scheme.Scheme,
-			config: fakeConfig,
-			logger: zaptest.NewLogger(t).Sugar(),
+			client:     cl,
+			kubeClient: k8sfake.NewSimpleClientset(),
+			scheme:     scheme.Scheme,
+			config:     fakeConfig,
+			logger:     zaptest.NewLogger(t).Sugar(),
 		}
 		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 42c7b957..b6b85b0e 100644
--- a/pkg/reconciler/isbsvc/installer/installer_test.go
+++ b/pkg/reconciler/isbsvc/installer/installer_test.go
@@ -27,6 +27,7 @@ import (
 	appv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8sfake "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/kubernetes/scheme"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
@@ -121,7 +122,7 @@ func init() {
 
 func TestGetInstaller(t *testing.T) {
 	t.Run("get native redis installer", func(t *testing.T) {
-		installer, err := getInstaller(testNativeRedisIsbSvc, nil, fakeConfig, zaptest.NewLogger(t).Sugar())
+		installer, err := getInstaller(testNativeRedisIsbSvc, nil, nil, fakeConfig, zaptest.NewLogger(t).Sugar())
 		assert.NoError(t, err)
 		assert.NotNil(t, installer)
 		_, ok := installer.(*redisInstaller)
@@ -129,7 +130,7 @@ func TestGetInstaller(t *testing.T) {
 	})
 
 	t.Run("get jetstream installer", func(t *testing.T) {
-		installer, err := getInstaller(testJetStreamIsbSvc, nil, fakeConfig, zaptest.NewLogger(t).Sugar())
+		installer, err := getInstaller(testJetStreamIsbSvc, nil, nil, fakeConfig, zaptest.NewLogger(t).Sugar())
 		assert.NoError(t, err)
 		assert.NotNil(t, installer)
 		_, ok := installer.(*jetStreamInstaller)
@@ -137,7 +138,7 @@ func TestGetInstaller(t *testing.T) {
 	})
 
 	t.Run("get external redis installer", func(t *testing.T) {
-		installer, err := getInstaller(testExternalRedisIsbSvc, nil, fakeConfig, zaptest.NewLogger(t).Sugar())
+		installer, err := getInstaller(testExternalRedisIsbSvc, nil, nil, fakeConfig, zaptest.NewLogger(t).Sugar())
 		assert.NoError(t, err)
 		assert.NotNil(t, installer)
 		_, ok := installer.(*externalRedisInstaller)
@@ -147,25 +148,26 @@ func TestGetInstaller(t *testing.T) {
 	t.Run("test error", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
 		testObj.Spec.Redis = nil
-		_, err := getInstaller(testObj, nil, fakeConfig, zaptest.NewLogger(t).Sugar())
+		_, err := getInstaller(testObj, nil, nil, fakeConfig, zaptest.NewLogger(t).Sugar())
 		assert.Error(t, err)
 	})
 }
 
 func TestInstall(t *testing.T) {
 	cl := fake.NewClientBuilder().Build()
+	kubeClient := k8sfake.NewSimpleClientset()
 	ctx := context.TODO()
 	t.Run("test redis error", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
 		testObj.Spec.Redis = nil
-		err := Install(ctx, testObj, cl, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Install(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
 		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, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Install(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
 		assert.NoError(t, err)
 		assert.True(t, testObj.Status.IsReady())
 		assert.NotNil(t, testObj.Status.Config.Redis)
@@ -178,14 +180,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, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Install(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
 		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, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Install(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
 		assert.NoError(t, err)
 		assert.True(t, testObj.Status.IsReady())
 		assert.NotNil(t, testObj.Status.Config.JetStream)
@@ -200,32 +202,33 @@ func TestInstall(t *testing.T) {
 
 func TestUnInstall(t *testing.T) {
 	cl := fake.NewClientBuilder().Build()
+	kubeClient := k8sfake.NewSimpleClientset()
 	ctx := context.TODO()
 	t.Run("test redis error", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
 		testObj.Spec.Redis = nil
-		err := Uninstall(ctx, testObj, cl, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Uninstall(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
 		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, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Uninstall(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
 		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, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Uninstall(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
 		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, fakeConfig, zaptest.NewLogger(t).Sugar())
+		err := Uninstall(ctx, testObj, cl, kubeClient, fakeConfig, zaptest.NewLogger(t).Sugar())
 		assert.NoError(t, err)
 	})
 }
diff --git a/pkg/reconciler/isbsvc/installer/jetstream_test.go b/pkg/reconciler/isbsvc/installer/jetstream_test.go
index 51fe57e5..0b3a8815 100644
--- a/pkg/reconciler/isbsvc/installer/jetstream_test.go
+++ b/pkg/reconciler/isbsvc/installer/jetstream_test.go
@@ -26,6 +26,7 @@ import (
 	appv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
+	k8sfake "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/kubernetes/scheme"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
@@ -71,11 +72,12 @@ func TestJetStreamCreateObjects(t *testing.T) {
 	cl := fake.NewClientBuilder().Build()
 	ctx := context.TODO()
 	i := &jetStreamInstaller{
-		client: cl,
-		isbs:   testJetStreamIsbSvc,
-		config: fakeConfig,
-		labels: testLabels,
-		logger: zaptest.NewLogger(t).Sugar(),
+		client:     cl,
+		kubeClient: k8sfake.NewSimpleClientset(),
+		isbs:       testJetStreamIsbSvc,
+		config:     fakeConfig,
+		labels:     testLabels,
+		logger:     zaptest.NewLogger(t).Sugar(),
 	}
 
 	t.Run("test create sts", func(t *testing.T) {
@@ -148,11 +150,12 @@ func Test_JetStreamInstall_Uninstall(t *testing.T) {
 	cl := fake.NewClientBuilder().Build()
 	ctx := context.TODO()
 	i := &jetStreamInstaller{
-		client: cl,
-		isbs:   testJetStreamIsbSvc,
-		config: fakeConfig,
-		labels: testLabels,
-		logger: zaptest.NewLogger(t).Sugar(),
+		client:     cl,
+		kubeClient: k8sfake.NewSimpleClientset(),
+		isbs:       testJetStreamIsbSvc,
+		config:     fakeConfig,
+		labels:     testLabels,
+		logger:     zaptest.NewLogger(t).Sugar(),
 	}
 	t.Run("test install", func(t *testing.T) {
 		c, err := i.Install(ctx)
diff --git a/pkg/reconciler/isbsvc/installer/native_redis_test.go b/pkg/reconciler/isbsvc/installer/native_redis_test.go
index 88ff202b..0b9e9f65 100644
--- a/pkg/reconciler/isbsvc/installer/native_redis_test.go
+++ b/pkg/reconciler/isbsvc/installer/native_redis_test.go
@@ -26,6 +26,7 @@ import (
 	appv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
+	k8sfake "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/kubernetes/scheme"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
@@ -75,11 +76,12 @@ func TestNativeRedisCreateObjects(t *testing.T) {
 	cl := fake.NewClientBuilder().Build()
 	ctx := context.TODO()
 	i := &redisInstaller{
-		client: cl,
-		isbs:   testNativeRedisIsbSvc,
-		config: fakeConfig,
-		labels: testLabels,
-		logger: zaptest.NewLogger(t).Sugar(),
+		client:     cl,
+		kubeClient: k8sfake.NewSimpleClientset(),
+		isbs:       testNativeRedisIsbSvc,
+		config:     fakeConfig,
+		labels:     testLabels,
+		logger:     zaptest.NewLogger(t).Sugar(),
 	}
 
 	t.Run("test create sts", func(t *testing.T) {
@@ -175,11 +177,12 @@ func Test_NativeRedisInstall_Uninstall(t *testing.T) {
 	cl := fake.NewClientBuilder().Build()
 	ctx := context.TODO()
 	i := &redisInstaller{
-		client: cl,
-		isbs:   testNativeRedisIsbSvc,
-		config: fakeConfig,
-		labels: testLabels,
-		logger: zaptest.NewLogger(t).Sugar(),
+		client:     cl,
+		kubeClient: k8sfake.NewSimpleClientset(),
+		isbs:       testNativeRedisIsbSvc,
+		config:     fakeConfig,
+		labels:     testLabels,
+		logger:     zaptest.NewLogger(t).Sugar(),
 	}
 	t.Run("test install", func(t *testing.T) {
 		c, err := i.Install(ctx)
diff --git a/pkg/reduce/pbq/store/memory/stores_test.go b/pkg/reduce/pbq/store/memory/stores_test.go
index bf617cb9..fecc9cc0 100644
--- a/pkg/reduce/pbq/store/memory/stores_test.go
+++ b/pkg/reduce/pbq/store/memory/stores_test.go
@@ -1,3 +1,19 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package memory
 
 import (
diff --git a/pkg/reduce/pbq/store/wal/stores_test.go b/pkg/reduce/pbq/store/wal/stores_test.go
index 69ed5892..41d69291 100644
--- a/pkg/reduce/pbq/store/wal/stores_test.go
+++ b/pkg/reduce/pbq/store/wal/stores_test.go
@@ -1,3 +1,19 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package wal
 
 import (
diff --git a/pkg/reduce/readloop/readloop_test.go b/pkg/reduce/readloop/readloop_test.go
index a3388204..67fc2361 100644
--- a/pkg/reduce/readloop/readloop_test.go
+++ b/pkg/reduce/readloop/readloop_test.go
@@ -1,3 +1,19 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package readloop
 
 import (
diff --git a/pkg/window/strategy/sliding/sliding_benchmark_test.go b/pkg/window/strategy/sliding/sliding_benchmark_test.go
index 2af089a9..878018d4 100644
--- a/pkg/window/strategy/sliding/sliding_benchmark_test.go
+++ b/pkg/window/strategy/sliding/sliding_benchmark_test.go
@@ -1,3 +1,19 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package sliding
 
 import (