Repository: numaproj/numaflow
Pull request #1499 (2024-02-11 00:53:11)

chore: load controller config with lock
Fix controller config auto-reloading concurrency problem.

<!--

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/config/advanced-install/namespaced-controller-wo-crds.yaml b/config/advanced-install/namespaced-controller-wo-crds.yaml
index b603aae5..e380a7a0 100644
--- a/config/advanced-install/namespaced-controller-wo-crds.yaml
+++ b/config/advanced-install/namespaced-controller-wo-crds.yaml
@@ -168,6 +168,11 @@ data:
           sentinelImage: bitnami/redis-sentinel:7.0.11-debian-11-r3
           redisExporterImage: bitnami/redis-exporter:1.50.0-debian-11-r4
           initContainerImage: debian:latest
+        - version: 7.0.15
+          redisImage: bitnami/redis:7.0.15-debian-11-r2
+          sentinelImage: bitnami/redis-sentinel:7.0.15-debian-11-r2
+          redisExporterImage: bitnami/redis-exporter:1.56.0-debian-11-r2
+          initContainerImage: debian:latest
       jetstream:
         # Default JetStream settings, could be overridden by InterStepBufferService specs
         settings: |
diff --git a/config/base/controller-manager/numaflow-controller-config.yaml b/config/base/controller-manager/numaflow-controller-config.yaml
index 566695dd..a2b171a7 100644
--- a/config/base/controller-manager/numaflow-controller-config.yaml
+++ b/config/base/controller-manager/numaflow-controller-config.yaml
@@ -39,6 +39,11 @@ data:
           sentinelImage: bitnami/redis-sentinel:7.0.11-debian-11-r3
           redisExporterImage: bitnami/redis-exporter:1.50.0-debian-11-r4
           initContainerImage: debian:latest
+        - version: 7.0.15
+          redisImage: bitnami/redis:7.0.15-debian-11-r2
+          sentinelImage: bitnami/redis-sentinel:7.0.15-debian-11-r2
+          redisExporterImage: bitnami/redis-exporter:1.56.0-debian-11-r2
+          initContainerImage: debian:latest
       jetstream:
         # Default JetStream settings, could be overridden by InterStepBufferService specs
         settings: |
diff --git a/config/install.yaml b/config/install.yaml
index 753d1c62..29d28b62 100644
--- a/config/install.yaml
+++ b/config/install.yaml
@@ -16360,6 +16360,11 @@ data:
           sentinelImage: bitnami/redis-sentinel:7.0.11-debian-11-r3
           redisExporterImage: bitnami/redis-exporter:1.50.0-debian-11-r4
           initContainerImage: debian:latest
+        - version: 7.0.15
+          redisImage: bitnami/redis:7.0.15-debian-11-r2
+          sentinelImage: bitnami/redis-sentinel:7.0.15-debian-11-r2
+          redisExporterImage: bitnami/redis-exporter:1.56.0-debian-11-r2
+          initContainerImage: debian:latest
       jetstream:
         # Default JetStream settings, could be overridden by InterStepBufferService specs
         settings: |
diff --git a/config/namespace-install.yaml b/config/namespace-install.yaml
index 35a5b1a2..2763fe11 100644
--- a/config/namespace-install.yaml
+++ b/config/namespace-install.yaml
@@ -16263,6 +16263,11 @@ data:
           sentinelImage: bitnami/redis-sentinel:7.0.11-debian-11-r3
           redisExporterImage: bitnami/redis-exporter:1.50.0-debian-11-r4
           initContainerImage: debian:latest
+        - version: 7.0.15
+          redisImage: bitnami/redis:7.0.15-debian-11-r2
+          sentinelImage: bitnami/redis-sentinel:7.0.15-debian-11-r2
+          redisExporterImage: bitnami/redis-exporter:1.56.0-debian-11-r2
+          initContainerImage: debian:latest
       jetstream:
         # Default JetStream settings, could be overridden by InterStepBufferService specs
         settings: |
diff --git a/pkg/reconciler/config.go b/pkg/reconciler/config.go
index 5eac1227..16d368ab 100644
--- a/pkg/reconciler/config.go
+++ b/pkg/reconciler/config.go
@@ -18,6 +18,7 @@ package reconciler
 
 import (
 	"fmt"
+	"sync"
 
 	"github.com/fsnotify/fsnotify"
 	"github.com/spf13/viper"
@@ -30,6 +31,11 @@ import (
 // supposed to be populated from the configmap attached to the
 // controller manager.
 type GlobalConfig struct {
+	conf *config
+	lock *sync.RWMutex
+}
+
+type config struct {
 	Defaults *DefaultConfig `json:"defaults"`
 	ISBSvc   *ISBSvcConfig  `json:"isbsvc"`
 }
@@ -77,7 +83,27 @@ type JetStreamVersion struct {
 	StartCommand         string `json:"startCommand"`
 }
 
-func (g *GlobalConfig) GetDefaultContainerResources() corev1.ResourceRequirements {
+// Get controller scope default config
+func (g *GlobalConfig) GetDefaults() DefaultConfig {
+	g.lock.RLock()
+	defer g.lock.RUnlock()
+	if g.conf.Defaults != nil {
+		return *g.conf.Defaults
+	}
+	return DefaultConfig{}
+}
+
+// Get controller scope ISB Service config
+func (g *GlobalConfig) GetISBSvcConfig() ISBSvcConfig {
+	g.lock.RLock()
+	defer g.lock.RUnlock()
+	if g.conf.ISBSvc != nil {
+		return *g.conf.ISBSvc
+	}
+	return ISBSvcConfig{}
+}
+
+func (dc DefaultConfig) GetDefaultContainerResources() corev1.ResourceRequirements {
 	// the standard resources used by the `init` and `main` containers.
 	defaultResources := corev1.ResourceRequirements{
 		Limits: corev1.ResourceList{},
@@ -87,12 +113,12 @@ func (g *GlobalConfig) GetDefaultContainerResources() corev1.ResourceRequirement
 		},
 	}
 
-	if g.Defaults == nil || g.Defaults.ContainerResources == "" {
+	if dc.ContainerResources == "" {
 		return defaultResources
 	}
 
 	var resourceConfig corev1.ResourceRequirements
-	err := yaml.Unmarshal([]byte(g.Defaults.ContainerResources), &resourceConfig)
+	err := yaml.Unmarshal([]byte(dc.ContainerResources), &resourceConfig)
 	if err != nil {
 		panic(fmt.Errorf("failed to unmarshal default container resources, %w", err))
 	}
@@ -100,11 +126,11 @@ func (g *GlobalConfig) GetDefaultContainerResources() corev1.ResourceRequirement
 	return resourceConfig
 }
 
-func (g *GlobalConfig) GetRedisVersion(version string) (*RedisVersion, error) {
-	if g.ISBSvc == nil || g.ISBSvc.Redis == nil || len(g.ISBSvc.Redis.Versions) == 0 {
+func (isc ISBSvcConfig) GetRedisVersion(version string) (*RedisVersion, error) {
+	if isc.Redis == nil || len(isc.Redis.Versions) == 0 {
 		return nil, fmt.Errorf("no redis configuration found")
 	}
-	for _, r := range g.ISBSvc.Redis.Versions {
+	for _, r := range isc.Redis.Versions {
 		if r.Version == version {
 			return &r, nil
 		}
@@ -112,11 +138,11 @@ func (g *GlobalConfig) GetRedisVersion(version string) (*RedisVersion, error) {
 	return nil, fmt.Errorf("no redis configuration found for %q", version)
 }
 
-func (g *GlobalConfig) GetJetStreamVersion(version string) (*JetStreamVersion, error) {
-	if g.ISBSvc.JetStream == nil || len(g.ISBSvc.JetStream.Versions) == 0 {
+func (isc ISBSvcConfig) GetJetStreamVersion(version string) (*JetStreamVersion, error) {
+	if isc.JetStream == nil || len(isc.JetStream.Versions) == 0 {
 		return nil, fmt.Errorf("no jetstream configuration found")
 	}
-	for _, r := range g.ISBSvc.JetStream.Versions {
+	for _, r := range isc.JetStream.Versions {
 		if r.Version == version {
 			return &r, nil
 		}
@@ -133,17 +159,26 @@ func LoadConfig(onErrorReloading func(error)) (*GlobalConfig, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to load configuration file. %w", err)
 	}
-	r := &GlobalConfig{}
-	err = v.Unmarshal(r)
+	r := &GlobalConfig{
+		lock: new(sync.RWMutex),
+	}
+	conf := &config{}
+	err = v.Unmarshal(conf)
 	if err != nil {
 		return nil, fmt.Errorf("failed unmarshal configuration file. %w", err)
 	}
+	r.conf = conf
 	v.WatchConfig()
 	v.OnConfigChange(func(e fsnotify.Event) {
-		err = v.Unmarshal(r)
+		cf := &config{}
+		err = v.Unmarshal(cf)
 		if err != nil {
 			onErrorReloading(err)
+			return
 		}
+		r.lock.Lock()
+		defer r.lock.Unlock()
+		r.conf = cf
 	})
 	return r, nil
 }
diff --git a/pkg/reconciler/isbsvc/installer/jetstream.go b/pkg/reconciler/isbsvc/installer/jetstream.go
index 5f706be7..3d390bd1 100644
--- a/pkg/reconciler/isbsvc/installer/jetstream.go
+++ b/pkg/reconciler/isbsvc/installer/jetstream.go
@@ -85,7 +85,7 @@ func (r *jetStreamInstaller) Install(ctx context.Context) (*dfv1.BufferServiceCo
 	// merge
 	v := viper.New()
 	v.SetConfigType("yaml")
-	if err := v.ReadConfig(bytes.NewBufferString(r.config.ISBSvc.JetStream.BufferConfig)); err != nil {
+	if err := v.ReadConfig(bytes.NewBufferString(r.config.GetISBSvcConfig().JetStream.BufferConfig)); err != nil {
 		return nil, fmt.Errorf("invalid jetstream buffer config in global configuration, %w", err)
 	}
 	if x := r.isbSvc.Spec.JetStream.BufferConfig; x != nil {
@@ -203,7 +203,7 @@ func (r *jetStreamInstaller) createService(ctx context.Context) error {
 }
 
 func (r *jetStreamInstaller) createStatefulSet(ctx context.Context) error {
-	jsVersion, err := r.config.GetJetStreamVersion(r.isbSvc.Spec.JetStream.Version)
+	jsVersion, err := r.config.GetISBSvcConfig().GetJetStreamVersion(r.isbSvc.Spec.JetStream.Version)
 	if err != nil {
 		return fmt.Errorf("failed to get jetstream version, err: %w", err)
 	}
@@ -223,7 +223,7 @@ func (r *jetStreamInstaller) createStatefulSet(ctx context.Context) error {
 		ConfigMapName:              generateJetStreamConfigMapName(r.isbSvc),
 		PvcNameIfNeeded:            generateJetStreamPVCName(r.isbSvc),
 		StartCommand:               jsVersion.StartCommand,
-		DefaultResources:           r.config.GetDefaultContainerResources(),
+		DefaultResources:           r.config.GetDefaults().GetDefaultContainerResources(),
 	})
 	hash := sharedutil.MustHash(spec)
 	obj := &appv1.StatefulSet{
@@ -430,7 +430,7 @@ func (r *jetStreamInstaller) createConfigMap(ctx context.Context) error {
 	// Merge Nats settings
 	v := viper.New()
 	v.SetConfigType("yaml")
-	if err := v.ReadConfig(bytes.NewBufferString(r.config.ISBSvc.JetStream.Settings)); err != nil {
+	if err := v.ReadConfig(bytes.NewBufferString(r.config.GetISBSvcConfig().JetStream.Settings)); err != nil {
 		return fmt.Errorf("invalid jetstream settings in global configuration, %w", err)
 	}
 	if x := r.isbSvc.Spec.JetStream.Settings; x != nil {
diff --git a/pkg/reconciler/isbsvc/installer/native_redis.go b/pkg/reconciler/isbsvc/installer/native_redis.go
index 3c50330b..04f7cc8b 100644
--- a/pkg/reconciler/isbsvc/installer/native_redis.go
+++ b/pkg/reconciler/isbsvc/installer/native_redis.go
@@ -300,7 +300,7 @@ func (r *redisInstaller) createHealthConfigMap(ctx context.Context) error {
 func (r *redisInstaller) createConfConfigMap(ctx context.Context) error {
 	data := make(map[string]string)
 	redisConf, masterConf, replicaConf, sentinelConf := "", "", "", ""
-	if x := r.config.ISBSvc.Redis.Settings; x != nil {
+	if x := r.config.GetISBSvcConfig().Redis.Settings; x != nil {
 		if x.Redis != "" {
 			redisConf = x.Redis
 		}
@@ -518,7 +518,7 @@ func (r *redisInstaller) createRedisHeadlessService(ctx context.Context) error {
 }
 
 func (r *redisInstaller) createStatefulSet(ctx context.Context) error {
-	redisVersion, err := r.config.GetRedisVersion(r.isbSvc.Spec.Redis.Native.Version)
+	redisVersion, err := r.config.GetISBSvcConfig().GetRedisVersion(r.isbSvc.Spec.Redis.Native.Version)
 	if err != nil {
 		return fmt.Errorf("failed to get redis version, err: %w", err)
 	}
@@ -539,7 +539,7 @@ func (r *redisInstaller) createStatefulSet(ctx context.Context) error {
 		ConfConfigMapName:         generateRedisConfigMapName(r.isbSvc),
 		HealthConfigMapName:       generateHealthConfigMapName(r.isbSvc),
 		ScriptsConfigMapName:      generateScriptsConfigMapName(r.isbSvc),
-		DefaultResources:          r.config.GetDefaultContainerResources(),
+		DefaultResources:          r.config.GetDefaults().GetDefaultContainerResources(),
 	})
 	hash := sharedutil.MustHash(spec)
 	obj := &appv1.StatefulSet{
diff --git a/pkg/reconciler/pipeline/controller.go b/pkg/reconciler/pipeline/controller.go
index 7ec863d1..6e34f6c5 100644
--- a/pkg/reconciler/pipeline/controller.go
+++ b/pkg/reconciler/pipeline/controller.go
@@ -381,7 +381,7 @@ func (r *pipelineReconciler) createOrUpdateDaemonDeployment(ctx context.Context,
 		Image:            r.image,
 		PullPolicy:       corev1.PullPolicy(sharedutil.LookupEnvStringOr(dfv1.EnvImagePullPolicy, "")),
 		Env:              envs,
-		DefaultResources: r.config.GetDefaultContainerResources(),
+		DefaultResources: r.config.GetDefaults().GetDefaultContainerResources(),
 	}
 	deploy, err := pl.GetDaemonDeploymentObj(req)
 	if err != nil {
@@ -449,7 +449,7 @@ func (r *pipelineReconciler) createOrUpdateSIMDeployments(ctx context.Context, p
 		Image:            r.image,
 		PullPolicy:       corev1.PullPolicy(sharedutil.LookupEnvStringOr(dfv1.EnvImagePullPolicy, "")),
 		Env:              envs,
-		DefaultResources: r.config.GetDefaultContainerResources(),
+		DefaultResources: r.config.GetDefaults().GetDefaultContainerResources(),
 	}
 
 	newObjs, err := pl.GetSideInputsManagerDeployments(req)
diff --git a/pkg/reconciler/vertex/controller.go b/pkg/reconciler/vertex/controller.go
index 96562b3e..4223adf3 100644
--- a/pkg/reconciler/vertex/controller.go
+++ b/pkg/reconciler/vertex/controller.go
@@ -358,7 +358,7 @@ func (r *vertexReconciler) buildPodSpec(vertex *dfv1.Vertex, pl *dfv1.Pipeline,
 		PullPolicy:          corev1.PullPolicy(sharedutil.LookupEnvStringOr(dfv1.EnvImagePullPolicy, "")),
 		Env:                 envs,
 		SideInputsStoreName: pl.GetSideInputsStoreName(),
-		DefaultResources:    r.config.GetDefaultContainerResources(),
+		DefaultResources:    r.config.GetDefaults().GetDefaultContainerResources(),
 	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to generate pod spec, error: %w", err)


Test patch:
diff --git a/pkg/reconciler/config_test.go b/pkg/reconciler/config_test.go
index 60900d68..5befdc02 100644
--- a/pkg/reconciler/config_test.go
+++ b/pkg/reconciler/config_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 reconciler
 
 import (
@@ -55,12 +71,9 @@ func TestGlobalConfig_GetDefaultContainerResources(t *testing.T) {
 		},
 	}
 	for _, tt := range tests {
-		g := &GlobalConfig{
-			Defaults: tt.fields.Defaults,
-		}
-		got := g.GetDefaultContainerResources()
+		got := tt.fields.Defaults.GetDefaultContainerResources()
 		if !reflect.DeepEqual(got, tt.want) {
-			t.Errorf("%q. GlobalConfig.GetDefaultContainerResources() = %v, want %v", tt.name, got, tt.want)
+			t.Errorf("%q.GetDefaultContainerResources() = %v, want %v", tt.name, got, tt.want)
 		}
 	}
 }
diff --git a/pkg/reconciler/isbsvc/controller_test.go b/pkg/reconciler/isbsvc/controller_test.go
index c2ab161e..a8b2ca7d 100644
--- a/pkg/reconciler/isbsvc/controller_test.go
+++ b/pkg/reconciler/isbsvc/controller_test.go
@@ -64,26 +64,24 @@ var (
 		},
 	}
 
-	fakeConfig = &reconciler.GlobalConfig{
-		ISBSvc: &reconciler.ISBSvcConfig{
-			Redis: &reconciler.RedisConfig{
-				Versions: []reconciler.RedisVersion{
-					{
-						Version:            testVersion,
-						RedisImage:         testImage,
-						SentinelImage:      testSImage,
-						RedisExporterImage: testRedisExporterImage,
-					},
+	fakeGlobalISBSvcConfig = &reconciler.ISBSvcConfig{
+		Redis: &reconciler.RedisConfig{
+			Versions: []reconciler.RedisVersion{
+				{
+					Version:            testVersion,
+					RedisImage:         testImage,
+					SentinelImage:      testSImage,
+					RedisExporterImage: testRedisExporterImage,
 				},
 			},
-			JetStream: &reconciler.JetStreamConfig{
-				Versions: []reconciler.JetStreamVersion{
-					{
-						Version:              testVersion,
-						NatsImage:            testJSImage,
-						ConfigReloaderImage:  testJSReloaderImage,
-						MetricsExporterImage: testJSMetricsImage,
-					},
+		},
+		JetStream: &reconciler.JetStreamConfig{
+			Versions: []reconciler.JetStreamVersion{
+				{
+					Version:              testVersion,
+					NatsImage:            testJSImage,
+					ConfigReloaderImage:  testJSReloaderImage,
+					MetricsExporterImage: testJSMetricsImage,
 				},
 			},
 		},
@@ -111,7 +109,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(), record.NewFakeRecorder(64))
+	r := NewReconciler(cl, kubeClient, scheme.Scheme, reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig), zaptest.NewLogger(t).Sugar(), record.NewFakeRecorder(64))
 	_, ok := r.(*interStepBufferServiceReconciler)
 	assert.True(t, ok)
 }
@@ -125,7 +123,7 @@ func TestReconcileNativeRedis(t *testing.T) {
 			client:     cl,
 			kubeClient: k8sfake.NewSimpleClientset(),
 			scheme:     scheme.Scheme,
-			config:     fakeConfig,
+			config:     reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 			logger:     zaptest.NewLogger(t).Sugar(),
 			recorder:   record.NewFakeRecorder(64),
 		}
@@ -166,7 +164,7 @@ func TestReconcileJetStream(t *testing.T) {
 			client:     cl,
 			kubeClient: k8sfake.NewSimpleClientset(),
 			scheme:     scheme.Scheme,
-			config:     fakeConfig,
+			config:     reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 			logger:     zaptest.NewLogger(t).Sugar(),
 			recorder:   record.NewFakeRecorder(64),
 		}
@@ -206,7 +204,7 @@ func TestNeedsUpdate(t *testing.T) {
 		r := &interStepBufferServiceReconciler{
 			client: cl,
 			scheme: scheme.Scheme,
-			config: fakeConfig,
+			config: reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 			logger: zaptest.NewLogger(t).Sugar(),
 		}
 		assert.False(t, r.needsUpdate(nativeRedisIsbs, testIsbs))
@@ -226,7 +224,7 @@ func TestNeedsUpdate(t *testing.T) {
 		r := &interStepBufferServiceReconciler{
 			client: cl,
 			scheme: scheme.Scheme,
-			config: fakeConfig,
+			config: reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 			logger: zaptest.NewLogger(t).Sugar(),
 		}
 		assert.False(t, r.needsUpdate(nativeRedisIsbs, testIsbs))
diff --git a/pkg/reconciler/isbsvc/installer/installer_test.go b/pkg/reconciler/isbsvc/installer/installer_test.go
index 796cd139..427d3513 100644
--- a/pkg/reconciler/isbsvc/installer/installer_test.go
+++ b/pkg/reconciler/isbsvc/installer/installer_test.go
@@ -89,26 +89,24 @@ var (
 		},
 	}
 
-	fakeConfig = &reconciler.GlobalConfig{
-		ISBSvc: &reconciler.ISBSvcConfig{
-			Redis: &reconciler.RedisConfig{
-				Versions: []reconciler.RedisVersion{
-					{
-						Version:            testVersion,
-						RedisImage:         testImage,
-						SentinelImage:      testSImage,
-						RedisExporterImage: testRedisExporterImage,
-					},
+	fakeGlobalISBSvcConfig = &reconciler.ISBSvcConfig{
+		Redis: &reconciler.RedisConfig{
+			Versions: []reconciler.RedisVersion{
+				{
+					Version:            testVersion,
+					RedisImage:         testImage,
+					SentinelImage:      testSImage,
+					RedisExporterImage: testRedisExporterImage,
 				},
 			},
-			JetStream: &reconciler.JetStreamConfig{
-				Versions: []reconciler.JetStreamVersion{
-					{
-						Version:              testVersion,
-						NatsImage:            testJSImage,
-						ConfigReloaderImage:  testJSReloaderImage,
-						MetricsExporterImage: testJSMetricsImage,
-					},
+		},
+		JetStream: &reconciler.JetStreamConfig{
+			Versions: []reconciler.JetStreamVersion{
+				{
+					Version:              testVersion,
+					NatsImage:            testJSImage,
+					ConfigReloaderImage:  testJSReloaderImage,
+					MetricsExporterImage: testJSMetricsImage,
 				},
 			},
 		},
@@ -122,6 +120,9 @@ func init() {
 }
 
 func TestGetInstaller(t *testing.T) {
+
+	fakeConfig := reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig)
+
 	t.Run("get native redis installer", func(t *testing.T) {
 		installer, err := getInstaller(testNativeRedisIsbSvc, nil, nil, fakeConfig, zaptest.NewLogger(t).Sugar(), nil)
 		assert.NoError(t, err)
@@ -157,6 +158,7 @@ func TestGetInstaller(t *testing.T) {
 func TestInstall(t *testing.T) {
 	cl := fake.NewClientBuilder().Build()
 	kubeClient := k8sfake.NewSimpleClientset()
+	fakeConfig := reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig)
 	ctx := context.TODO()
 	t.Run("test redis error", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
@@ -204,6 +206,7 @@ func TestInstall(t *testing.T) {
 func TestUnInstall(t *testing.T) {
 	cl := fake.NewClientBuilder().Build()
 	kubeClient := k8sfake.NewSimpleClientset()
+	fakeConfig := reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig)
 	ctx := context.TODO()
 	t.Run("test redis error", func(t *testing.T) {
 		testObj := testNativeRedisIsbSvc.DeepCopy()
diff --git a/pkg/reconciler/isbsvc/installer/jetstream_test.go b/pkg/reconciler/isbsvc/installer/jetstream_test.go
index 81f0749f..c527a41e 100644
--- a/pkg/reconciler/isbsvc/installer/jetstream_test.go
+++ b/pkg/reconciler/isbsvc/installer/jetstream_test.go
@@ -21,6 +21,7 @@ import (
 	"testing"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/reconciler"
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap/zaptest"
 	appv1 "k8s.io/api/apps/v1"
@@ -45,7 +46,7 @@ func TestJetStreamBadInstallation(t *testing.T) {
 		installer := &jetStreamInstaller{
 			client:   fake.NewClientBuilder().Build(),
 			isbSvc:   badIsbs,
-			config:   fakeConfig,
+			config:   reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 			labels:   testLabels,
 			logger:   zaptest.NewLogger(t).Sugar(),
 			recorder: record.NewFakeRecorder(64),
@@ -77,7 +78,7 @@ func TestJetStreamCreateObjects(t *testing.T) {
 		client:     cl,
 		kubeClient: k8sfake.NewSimpleClientset(),
 		isbSvc:     testJetStreamIsbSvc,
-		config:     fakeConfig,
+		config:     reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 		labels:     testLabels,
 		logger:     zaptest.NewLogger(t).Sugar(),
 		recorder:   record.NewFakeRecorder(64),
@@ -156,7 +157,7 @@ func Test_JetStreamInstall_Uninstall(t *testing.T) {
 		client:     cl,
 		kubeClient: k8sfake.NewSimpleClientset(),
 		isbSvc:     testJetStreamIsbSvc,
-		config:     fakeConfig,
+		config:     reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 		labels:     testLabels,
 		logger:     zaptest.NewLogger(t).Sugar(),
 		recorder:   record.NewFakeRecorder(64),
diff --git a/pkg/reconciler/isbsvc/installer/native_redis_test.go b/pkg/reconciler/isbsvc/installer/native_redis_test.go
index 3a336853..482926f9 100644
--- a/pkg/reconciler/isbsvc/installer/native_redis_test.go
+++ b/pkg/reconciler/isbsvc/installer/native_redis_test.go
@@ -21,6 +21,7 @@ import (
 	"testing"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/reconciler"
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap/zaptest"
 	appv1 "k8s.io/api/apps/v1"
@@ -45,7 +46,7 @@ func TestNativeRedisBadInstallation(t *testing.T) {
 		installer := &redisInstaller{
 			client:   fake.NewClientBuilder().Build(),
 			isbSvc:   badIsbs,
-			config:   fakeConfig,
+			config:   reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 			labels:   testLabels,
 			logger:   zaptest.NewLogger(t).Sugar(),
 			recorder: record.NewFakeRecorder(64),
@@ -81,7 +82,7 @@ func TestNativeRedisCreateObjects(t *testing.T) {
 		client:     cl,
 		kubeClient: k8sfake.NewSimpleClientset(),
 		isbSvc:     testNativeRedisIsbSvc,
-		config:     fakeConfig,
+		config:     reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 		labels:     testLabels,
 		logger:     zaptest.NewLogger(t).Sugar(),
 		recorder:   record.NewFakeRecorder(64),
@@ -183,7 +184,7 @@ func Test_NativeRedisInstall_Uninstall(t *testing.T) {
 		client:     cl,
 		kubeClient: k8sfake.NewSimpleClientset(),
 		isbSvc:     testNativeRedisIsbSvc,
-		config:     fakeConfig,
+		config:     reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 		labels:     testLabels,
 		logger:     zaptest.NewLogger(t).Sugar(),
 		recorder:   record.NewFakeRecorder(64),
diff --git a/pkg/reconciler/pipeline/controller_test.go b/pkg/reconciler/pipeline/controller_test.go
index d0745a0d..89a174e2 100644
--- a/pkg/reconciler/pipeline/controller_test.go
+++ b/pkg/reconciler/pipeline/controller_test.go
@@ -64,21 +64,18 @@ var (
 		},
 	}
 
-	fakeConfig = &reconciler.GlobalConfig{
-		ISBSvc: &reconciler.ISBSvcConfig{
-			Redis: &reconciler.RedisConfig{
-				Versions: []reconciler.RedisVersion{
-					{
-						Version:            testVersion,
-						RedisImage:         testImage,
-						SentinelImage:      testSImage,
-						RedisExporterImage: testRedisExporterImage,
-					},
+	fakeGlobalISBSvcConfig = &reconciler.ISBSvcConfig{
+		Redis: &reconciler.RedisConfig{
+			Versions: []reconciler.RedisVersion{
+				{
+					Version:            testVersion,
+					RedisImage:         testImage,
+					SentinelImage:      testSImage,
+					RedisExporterImage: testRedisExporterImage,
 				},
 			},
 		},
 	}
-
 	fakeIsbSvcConfig = dfv1.BufferServiceConfig{
 		Redis: &dfv1.RedisConfig{
 			URL:         "xxx",
@@ -110,7 +107,7 @@ func init() {
 
 func Test_NewReconciler(t *testing.T) {
 	cl := fake.NewClientBuilder().Build()
-	r := NewReconciler(cl, scheme.Scheme, fakeConfig, testFlowImage, zaptest.NewLogger(t).Sugar(), record.NewFakeRecorder(64))
+	r := NewReconciler(cl, scheme.Scheme, reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig), testFlowImage, zaptest.NewLogger(t).Sugar(), record.NewFakeRecorder(64))
 	_, ok := r.(*pipelineReconciler)
 	assert.True(t, ok)
 }
@@ -127,7 +124,7 @@ func Test_reconcile(t *testing.T) {
 		r := &pipelineReconciler{
 			client:   cl,
 			scheme:   scheme.Scheme,
-			config:   fakeConfig,
+			config:   reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 			image:    testFlowImage,
 			logger:   zaptest.NewLogger(t).Sugar(),
 			recorder: record.NewFakeRecorder(64),
@@ -148,6 +145,8 @@ func Test_reconcile(t *testing.T) {
 }
 
 func Test_reconcileEvents(t *testing.T) {
+
+	fakeConfig := reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig)
 	t.Run("test reconcile - invalid name", func(t *testing.T) {
 		cl := fake.NewClientBuilder().Build()
 		ctx := context.TODO()
@@ -235,7 +234,7 @@ func Test_pauseAndResumePipeline(t *testing.T) {
 		r := &pipelineReconciler{
 			client:   cl,
 			scheme:   scheme.Scheme,
-			config:   fakeConfig,
+			config:   reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 			image:    testFlowImage,
 			logger:   zaptest.NewLogger(t).Sugar(),
 			recorder: record.NewFakeRecorder(64),
@@ -272,7 +271,7 @@ func Test_pauseAndResumePipeline(t *testing.T) {
 		r := &pipelineReconciler{
 			client:   cl,
 			scheme:   scheme.Scheme,
-			config:   fakeConfig,
+			config:   reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 			image:    testFlowImage,
 			logger:   zaptest.NewLogger(t).Sugar(),
 			recorder: record.NewFakeRecorder(64),
@@ -480,7 +479,7 @@ func Test_cleanupBuffers(t *testing.T) {
 	r := &pipelineReconciler{
 		client: cl,
 		scheme: scheme.Scheme,
-		config: fakeConfig,
+		config: reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 		image:  testFlowImage,
 		logger: zaptest.NewLogger(t).Sugar(),
 	}
@@ -522,7 +521,7 @@ func TestCreateOrUpdateDaemon(t *testing.T) {
 	r := &pipelineReconciler{
 		client:   cl,
 		scheme:   scheme.Scheme,
-		config:   fakeConfig,
+		config:   reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 		image:    testFlowImage,
 		logger:   zaptest.NewLogger(t).Sugar(),
 		recorder: record.NewFakeRecorder(64),
@@ -557,7 +556,7 @@ func Test_createOrUpdateSIMDeployments(t *testing.T) {
 	r := &pipelineReconciler{
 		client:   cl,
 		scheme:   scheme.Scheme,
-		config:   fakeConfig,
+		config:   reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 		image:    testFlowImage,
 		logger:   zaptest.NewLogger(t).Sugar(),
 		recorder: record.NewFakeRecorder(64),
diff --git a/pkg/reconciler/test_utils.go b/pkg/reconciler/test_utils.go
new file mode 100644
index 00000000..d2354cdf
--- /dev/null
+++ b/pkg/reconciler/test_utils.go
@@ -0,0 +1,32 @@
+/*
+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 reconciler
+
+import (
+	"sync"
+	"testing"
+)
+
+func FakeGlobalConfig(t *testing.T, isbSvcConfig *ISBSvcConfig) *GlobalConfig {
+	t.Helper()
+	return &GlobalConfig{
+		conf: &config{
+			ISBSvc: isbSvcConfig,
+		},
+		lock: new(sync.RWMutex),
+	}
+}
diff --git a/pkg/reconciler/vertex/controller_test.go b/pkg/reconciler/vertex/controller_test.go
index f74c4625..dfcdf669 100644
--- a/pkg/reconciler/vertex/controller_test.go
+++ b/pkg/reconciler/vertex/controller_test.go
@@ -65,16 +65,14 @@ var (
 		},
 	}
 
-	fakeConfig = &reconciler.GlobalConfig{
-		ISBSvc: &reconciler.ISBSvcConfig{
-			Redis: &reconciler.RedisConfig{
-				Versions: []reconciler.RedisVersion{
-					{
-						Version:            testVersion,
-						RedisImage:         testImage,
-						SentinelImage:      testSImage,
-						RedisExporterImage: testRedisExporterImage,
-					},
+	fakeGlobalISBSvcConfig = &reconciler.ISBSvcConfig{
+		Redis: &reconciler.RedisConfig{
+			Versions: []reconciler.RedisVersion{
+				{
+					Version:            testVersion,
+					RedisImage:         testImage,
+					SentinelImage:      testSImage,
+					RedisExporterImage: testRedisExporterImage,
 				},
 			},
 		},
@@ -170,12 +168,14 @@ func init() {
 
 func Test_NewReconciler(t *testing.T) {
 	cl := fake.NewClientBuilder().Build()
-	r := NewReconciler(cl, scheme.Scheme, fakeConfig, testFlowImage, scaling.NewScaler(cl), zaptest.NewLogger(t).Sugar(), record.NewFakeRecorder(64))
+	r := NewReconciler(cl, scheme.Scheme, reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig), testFlowImage, scaling.NewScaler(cl), zaptest.NewLogger(t).Sugar(), record.NewFakeRecorder(64))
 	_, ok := r.(*vertexReconciler)
 	assert.True(t, ok)
 }
 
 func Test_BuildPodSpec(t *testing.T) {
+
+	fakeConfig := reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig)
 	t.Run("test source", func(t *testing.T) {
 		cl := fake.NewClientBuilder().Build()
 		r := &vertexReconciler{
@@ -420,6 +420,7 @@ func Test_BuildPodSpec(t *testing.T) {
 }
 
 func Test_reconcile(t *testing.T) {
+	fakeConfig := reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig)
 	t.Run("test reconcile source", func(t *testing.T) {
 		cl := fake.NewClientBuilder().Build()
 		ctx := context.TODO()
@@ -653,7 +654,7 @@ func Test_reconcileEvents(t *testing.T) {
 		r := &vertexReconciler{
 			client:   cl,
 			scheme:   scheme.Scheme,
-			config:   fakeConfig,
+			config:   reconciler.FakeGlobalConfig(t, fakeGlobalISBSvcConfig),
 			image:    testFlowImage,
 			scaler:   scaling.NewScaler(cl),
 			logger:   zaptest.NewLogger(t).Sugar(),