Repository: numaproj/numaflow
Pull request #192 (2022-09-29 21:56:19)

refactor: move controllers package to pkg/reconciler
Signed-off-by: Derek Wang <whynowy@gmail.com>

Explain what this PR does.

<!--

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/cmd/commands/controller.go b/cmd/commands/controller.go
index bdc48940..249bfc38 100644
--- a/cmd/commands/controller.go
+++ b/cmd/commands/controller.go
@@ -1,7 +1,7 @@
 package commands
 
 import (
-	ctrlcmd "github.com/numaproj/numaflow/controllers/cmd"
+	ctrlcmd "github.com/numaproj/numaflow/pkg/reconciler/cmd"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"github.com/spf13/cobra"
 )
diff --git a/docs/development.md b/docs/development.md
index f914d046..682a2686 100644
--- a/docs/development.md
+++ b/docs/development.md
@@ -16,7 +16,7 @@ This doc explains how to set up a development environment for Numaflow.
 
 ```shell
 # Create a cluster with default name k3s-default
-k3d cluster create -i rancher/k3s:v1.21.7-k3s1
+k3d cluster create -i rancher/k3s:v1.24.4-k3s1
 
 # Get kubeconfig for the cluster
 k3d kubeconfig get k3s-default
diff --git a/docs/numaflow-controller-config.yaml b/docs/numaflow-controller-config.yaml
index e6cb59cc..9832559c 100644
--- a/docs/numaflow-controller-config.yaml
+++ b/docs/numaflow-controller-config.yaml
@@ -4,14 +4,6 @@ metadata:
   name: numaflow-controller-config
 data:
   controller-config.yaml: |
-    udf:
-      # UDF responds the messages in what Content-Type. Available options: application/msgpack, application/json.
-      # Defaults to application/msgpack
-      contentType: application/msgpack
-    sink:
-      # UDSink accepts and responds the messages in what Content-Type. Available options: application/msgpack, application/json.
-      # Defaults to application/msgpack
-      contentType: application/msgpack
     isbsvc:
       redis:
         # Default Redis settings, could be overridden by InterStepBufferService specs
@@ -104,3 +96,8 @@ data:
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: nats-server
+        - version: 2.9.1
+          natsImage: nats:2.9.1
+          metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
+          configReloaderImage: natsio/nats-server-config-reloader:0.7.0
+          startCommand: /nats-server
diff --git a/controllers/cmd/start.go b/pkg/reconciler/cmd/start.go
similarity index 95%
rename from controllers/cmd/start.go
rename to pkg/reconciler/cmd/start.go
index 2c4d5ace..e09bf394 100644
--- a/controllers/cmd/start.go
+++ b/pkg/reconciler/cmd/start.go
@@ -20,12 +20,12 @@ import (
 	"reflect"
 
 	numaflow "github.com/numaproj/numaflow"
-	"github.com/numaproj/numaflow/controllers"
-	isbsvcctrl "github.com/numaproj/numaflow/controllers/isbsvc"
-	plctrl "github.com/numaproj/numaflow/controllers/pipeline"
-	vertexctrl "github.com/numaproj/numaflow/controllers/vertex"
-	"github.com/numaproj/numaflow/controllers/vertex/scaling"
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/reconciler"
+	isbsvcctrl "github.com/numaproj/numaflow/pkg/reconciler/isbsvc"
+	plctrl "github.com/numaproj/numaflow/pkg/reconciler/pipeline"
+	vertexctrl "github.com/numaproj/numaflow/pkg/reconciler/vertex"
+	"github.com/numaproj/numaflow/pkg/reconciler/vertex/scaling"
 	logging "github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"go.uber.org/zap"
@@ -42,7 +42,7 @@ import (
 
 func Start(namespaced bool, managedNamespace string) {
 	logger := logging.NewLogger().Named("controller-manager")
-	config, err := controllers.LoadConfig(func(err error) {
+	config, err := reconciler.LoadConfig(func(err error) {
 		logger.Errorf("Failed to reload global configuration file", zap.Error(err))
 	})
 	if err != nil {
diff --git a/controllers/config.go b/pkg/reconciler/config.go
similarity index 94%
rename from controllers/config.go
rename to pkg/reconciler/config.go
index 93fc71f5..a57950f1 100644
--- a/controllers/config.go
+++ b/pkg/reconciler/config.go
@@ -1,4 +1,4 @@
-package controllers
+package reconciler
 
 import (
 	"fmt"
@@ -7,18 +7,13 @@ import (
 	"github.com/spf13/viper"
 )
 
+// GlobalConfig is the configuration for the controllers, it is
+// supposed to be populated from the configmap attached to the
+// controller manager.
 type GlobalConfig struct {
 	ISBSvc *ISBSvcConfig `json:"isbsvc"`
 }
 
-type UDFConfig struct {
-	ContentType string `json:"contentType"`
-}
-
-type SinkConfig struct {
-	ContentType string `json:"contentType"`
-}
-
 type ISBSvcConfig struct {
 	Redis     *RedisConfig     `json:"redis"`
 	JetStream *JetStreamConfig `json:"jetstream"`
diff --git a/pkg/reconciler/doc.go b/pkg/reconciler/doc.go
new file mode 100644
index 00000000..20cabc51
--- /dev/null
+++ b/pkg/reconciler/doc.go
@@ -0,0 +1,27 @@
+/*
+
+
+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 defines implementations of the Reconciler interface
+// defined at sigs.k8s.io/controller-runtime/pkg/reconcile.Reconciler. They
+// implement the basic workhorse functionality of controllers, which include
+// an InterStepBufferService controller, a Pipeline controller, and a Vertex
+// controller.
+//
+// Despite the implementation of the controllers, this package also implements
+// a Start() function to watch corresponding Kubernetes resources for those
+// controllers, it is supposed to be called at the time the controller manager
+// service starts.
+package reconciler
diff --git a/controllers/isbsvc/controller.go b/pkg/reconciler/isbsvc/controller.go
similarity index 93%
rename from controllers/isbsvc/controller.go
rename to pkg/reconciler/isbsvc/controller.go
index 7afc8e1b..ceac53fc 100644
--- a/controllers/isbsvc/controller.go
+++ b/pkg/reconciler/isbsvc/controller.go
@@ -3,9 +3,9 @@ package isbsvc
 import (
 	"context"
 
-	"github.com/numaproj/numaflow/controllers"
-	"github.com/numaproj/numaflow/controllers/isbsvc/installer"
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/reconciler"
+	"github.com/numaproj/numaflow/pkg/reconciler/isbsvc/installer"
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -25,11 +25,11 @@ type interStepBufferServiceReconciler struct {
 	client client.Client
 	scheme *runtime.Scheme
 
-	config *controllers.GlobalConfig
+	config *reconciler.GlobalConfig
 	logger *zap.SugaredLogger
 }
 
-func NewReconciler(client client.Client, scheme *runtime.Scheme, config *controllers.GlobalConfig, logger *zap.SugaredLogger) reconcile.Reconciler {
+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}
 }
 
diff --git a/controllers/isbsvc/installer/assets/jetstream/nats.conf b/pkg/reconciler/isbsvc/installer/assets/jetstream/nats.conf
similarity index 100%
rename from controllers/isbsvc/installer/assets/jetstream/nats.conf
rename to pkg/reconciler/isbsvc/installer/assets/jetstream/nats.conf
diff --git a/controllers/isbsvc/installer/assets/jetstream/server-auth.conf b/pkg/reconciler/isbsvc/installer/assets/jetstream/server-auth.conf
similarity index 100%
rename from controllers/isbsvc/installer/assets/jetstream/server-auth.conf
rename to pkg/reconciler/isbsvc/installer/assets/jetstream/server-auth.conf
diff --git a/controllers/isbsvc/installer/assets/redis/config/master.conf b/pkg/reconciler/isbsvc/installer/assets/redis/config/master.conf
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/config/master.conf
rename to pkg/reconciler/isbsvc/installer/assets/redis/config/master.conf
diff --git a/controllers/isbsvc/installer/assets/redis/config/redis.conf b/pkg/reconciler/isbsvc/installer/assets/redis/config/redis.conf
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/config/redis.conf
rename to pkg/reconciler/isbsvc/installer/assets/redis/config/redis.conf
diff --git a/controllers/isbsvc/installer/assets/redis/config/replica.conf b/pkg/reconciler/isbsvc/installer/assets/redis/config/replica.conf
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/config/replica.conf
rename to pkg/reconciler/isbsvc/installer/assets/redis/config/replica.conf
diff --git a/controllers/isbsvc/installer/assets/redis/config/sentinel.conf b/pkg/reconciler/isbsvc/installer/assets/redis/config/sentinel.conf
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/config/sentinel.conf
rename to pkg/reconciler/isbsvc/installer/assets/redis/config/sentinel.conf
diff --git a/controllers/isbsvc/installer/assets/redis/health/parse_sentinels.awk b/pkg/reconciler/isbsvc/installer/assets/redis/health/parse_sentinels.awk
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/health/parse_sentinels.awk
rename to pkg/reconciler/isbsvc/installer/assets/redis/health/parse_sentinels.awk
diff --git a/controllers/isbsvc/installer/assets/redis/health/ping_liveness_local.sh b/pkg/reconciler/isbsvc/installer/assets/redis/health/ping_liveness_local.sh
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/health/ping_liveness_local.sh
rename to pkg/reconciler/isbsvc/installer/assets/redis/health/ping_liveness_local.sh
diff --git a/controllers/isbsvc/installer/assets/redis/health/ping_liveness_local_and_master.sh b/pkg/reconciler/isbsvc/installer/assets/redis/health/ping_liveness_local_and_master.sh
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/health/ping_liveness_local_and_master.sh
rename to pkg/reconciler/isbsvc/installer/assets/redis/health/ping_liveness_local_and_master.sh
diff --git a/controllers/isbsvc/installer/assets/redis/health/ping_liveness_master.sh b/pkg/reconciler/isbsvc/installer/assets/redis/health/ping_liveness_master.sh
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/health/ping_liveness_master.sh
rename to pkg/reconciler/isbsvc/installer/assets/redis/health/ping_liveness_master.sh
diff --git a/controllers/isbsvc/installer/assets/redis/health/ping_readiness_local.sh b/pkg/reconciler/isbsvc/installer/assets/redis/health/ping_readiness_local.sh
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/health/ping_readiness_local.sh
rename to pkg/reconciler/isbsvc/installer/assets/redis/health/ping_readiness_local.sh
diff --git a/controllers/isbsvc/installer/assets/redis/health/ping_readiness_local_and_master.sh b/pkg/reconciler/isbsvc/installer/assets/redis/health/ping_readiness_local_and_master.sh
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/health/ping_readiness_local_and_master.sh
rename to pkg/reconciler/isbsvc/installer/assets/redis/health/ping_readiness_local_and_master.sh
diff --git a/controllers/isbsvc/installer/assets/redis/health/ping_readiness_master.sh b/pkg/reconciler/isbsvc/installer/assets/redis/health/ping_readiness_master.sh
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/health/ping_readiness_master.sh
rename to pkg/reconciler/isbsvc/installer/assets/redis/health/ping_readiness_master.sh
diff --git a/controllers/isbsvc/installer/assets/redis/health/ping_sentinel.sh b/pkg/reconciler/isbsvc/installer/assets/redis/health/ping_sentinel.sh
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/health/ping_sentinel.sh
rename to pkg/reconciler/isbsvc/installer/assets/redis/health/ping_sentinel.sh
diff --git a/controllers/isbsvc/installer/assets/redis/scripts/prestop-redis.sh b/pkg/reconciler/isbsvc/installer/assets/redis/scripts/prestop-redis.sh
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/scripts/prestop-redis.sh
rename to pkg/reconciler/isbsvc/installer/assets/redis/scripts/prestop-redis.sh
diff --git a/controllers/isbsvc/installer/assets/redis/scripts/prestop-sentinel.sh b/pkg/reconciler/isbsvc/installer/assets/redis/scripts/prestop-sentinel.sh
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/scripts/prestop-sentinel.sh
rename to pkg/reconciler/isbsvc/installer/assets/redis/scripts/prestop-sentinel.sh
diff --git a/controllers/isbsvc/installer/assets/redis/scripts/start-node.sh b/pkg/reconciler/isbsvc/installer/assets/redis/scripts/start-node.sh
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/scripts/start-node.sh
rename to pkg/reconciler/isbsvc/installer/assets/redis/scripts/start-node.sh
diff --git a/controllers/isbsvc/installer/assets/redis/scripts/start-sentinel.sh b/pkg/reconciler/isbsvc/installer/assets/redis/scripts/start-sentinel.sh
similarity index 100%
rename from controllers/isbsvc/installer/assets/redis/scripts/start-sentinel.sh
rename to pkg/reconciler/isbsvc/installer/assets/redis/scripts/start-sentinel.sh
diff --git a/controllers/isbsvc/installer/external_redis.go b/pkg/reconciler/isbsvc/installer/external_redis.go
similarity index 100%
rename from controllers/isbsvc/installer/external_redis.go
rename to pkg/reconciler/isbsvc/installer/external_redis.go
diff --git a/controllers/isbsvc/installer/external_redis_test.go b/pkg/reconciler/isbsvc/installer/external_redis_test.go
similarity index 100%
rename from controllers/isbsvc/installer/external_redis_test.go
rename to pkg/reconciler/isbsvc/installer/external_redis_test.go
diff --git a/controllers/isbsvc/installer/installer.go b/pkg/reconciler/isbsvc/installer/installer.go
similarity index 88%
rename from controllers/isbsvc/installer/installer.go
rename to pkg/reconciler/isbsvc/installer/installer.go
index 73c400d2..e6ee1d85 100644
--- a/controllers/isbsvc/installer/installer.go
+++ b/pkg/reconciler/isbsvc/installer/installer.go
@@ -4,8 +4,8 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/numaproj/numaflow/controllers"
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/reconciler"
 	"go.uber.org/zap"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -19,7 +19,7 @@ type Installer interface {
 }
 
 // Install function installs the ISBS
-func Install(ctx context.Context, isbsvc *dfv1.InterStepBufferService, client client.Client, config *controllers.GlobalConfig, logger *zap.SugaredLogger) error {
+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)
 	if err != nil {
 		logger.Errorw("failed to get an installer", zap.Error(err))
@@ -35,7 +35,7 @@ func Install(ctx context.Context, isbsvc *dfv1.InterStepBufferService, client cl
 }
 
 // GetInstaller returns Installer implementation
-func getInstaller(isbsvc *dfv1.InterStepBufferService, client client.Client, config *controllers.GlobalConfig, logger *zap.SugaredLogger) (Installer, error) {
+func getInstaller(isbsvc *dfv1.InterStepBufferService, client client.Client, config *reconciler.GlobalConfig, logger *zap.SugaredLogger) (Installer, error) {
 	labels := map[string]string{
 		dfv1.KeyPartOf:     dfv1.Project,
 		dfv1.KeyManagedBy:  dfv1.ControllerISBSvc,
@@ -64,7 +64,7 @@ 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 *controllers.GlobalConfig, logger *zap.SugaredLogger) error {
+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)
 	if err != nil {
 		logger.Errorw("Failed to get an installer", zap.Error(err))
diff --git a/controllers/isbsvc/installer/jetstream.go b/pkg/reconciler/isbsvc/installer/jetstream.go
similarity index 98%
rename from controllers/isbsvc/installer/jetstream.go
rename to pkg/reconciler/isbsvc/installer/jetstream.go
index e831cc5a..7d73a99d 100644
--- a/controllers/isbsvc/installer/jetstream.go
+++ b/pkg/reconciler/isbsvc/installer/jetstream.go
@@ -10,8 +10,8 @@ import (
 	"text/template"
 	"time"
 
-	"github.com/numaproj/numaflow/controllers"
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/reconciler"
 	"github.com/numaproj/numaflow/pkg/shared/tls"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"github.com/spf13/viper"
@@ -41,12 +41,12 @@ var (
 type jetStreamInstaller struct {
 	client client.Client
 	isbs   *dfv1.InterStepBufferService
-	config *controllers.GlobalConfig
+	config *reconciler.GlobalConfig
 	labels map[string]string
 	logger *zap.SugaredLogger
 }
 
-func NewJetStreamInstaller(client client.Client, isbs *dfv1.InterStepBufferService, config *controllers.GlobalConfig, labels map[string]string, logger *zap.SugaredLogger) Installer {
+func NewJetStreamInstaller(client client.Client, isbs *dfv1.InterStepBufferService, config *reconciler.GlobalConfig, labels map[string]string, logger *zap.SugaredLogger) Installer {
 	return &jetStreamInstaller{
 		client: client,
 		isbs:   isbs,
diff --git a/controllers/isbsvc/installer/jetstream_test.go b/pkg/reconciler/isbsvc/installer/jetstream_test.go
similarity index 100%
rename from controllers/isbsvc/installer/jetstream_test.go
rename to pkg/reconciler/isbsvc/installer/jetstream_test.go
diff --git a/controllers/isbsvc/installer/native_redis.go b/pkg/reconciler/isbsvc/installer/native_redis.go
similarity index 99%
rename from controllers/isbsvc/installer/native_redis.go
rename to pkg/reconciler/isbsvc/installer/native_redis.go
index 1243bada..2a26501c 100644
--- a/controllers/isbsvc/installer/native_redis.go
+++ b/pkg/reconciler/isbsvc/installer/native_redis.go
@@ -7,8 +7,8 @@ import (
 	"fmt"
 	"text/template"
 
-	"github.com/numaproj/numaflow/controllers"
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/reconciler"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"go.uber.org/zap"
 	appv1 "k8s.io/api/apps/v1"
@@ -38,12 +38,12 @@ var (
 type redisInstaller struct {
 	client client.Client
 	isbs   *dfv1.InterStepBufferService
-	config *controllers.GlobalConfig
+	config *reconciler.GlobalConfig
 	labels map[string]string
 	logger *zap.SugaredLogger
 }
 
-func NewNativeRedisInstaller(client client.Client, isbs *dfv1.InterStepBufferService, config *controllers.GlobalConfig, labels map[string]string, logger *zap.SugaredLogger) Installer {
+func NewNativeRedisInstaller(client client.Client, isbs *dfv1.InterStepBufferService, config *reconciler.GlobalConfig, labels map[string]string, logger *zap.SugaredLogger) Installer {
 	return &redisInstaller{
 		client: client,
 		isbs:   isbs,
diff --git a/controllers/isbsvc/installer/native_redis_test.go b/pkg/reconciler/isbsvc/installer/native_redis_test.go
similarity index 100%
rename from controllers/isbsvc/installer/native_redis_test.go
rename to pkg/reconciler/isbsvc/installer/native_redis_test.go
diff --git a/controllers/isbsvc/validate.go b/pkg/reconciler/isbsvc/validate.go
similarity index 100%
rename from controllers/isbsvc/validate.go
rename to pkg/reconciler/isbsvc/validate.go
diff --git a/controllers/isbsvc/validate_test.go b/pkg/reconciler/isbsvc/validate_test.go
similarity index 100%
rename from controllers/isbsvc/validate_test.go
rename to pkg/reconciler/isbsvc/validate_test.go
diff --git a/controllers/pipeline/controller.go b/pkg/reconciler/pipeline/controller.go
similarity index 99%
rename from controllers/pipeline/controller.go
rename to pkg/reconciler/pipeline/controller.go
index f5dfd5c5..25ab6ab1 100644
--- a/controllers/pipeline/controller.go
+++ b/pkg/reconciler/pipeline/controller.go
@@ -23,9 +23,9 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
-	"github.com/numaproj/numaflow/controllers"
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	daemonclient "github.com/numaproj/numaflow/pkg/daemon/client"
+	"github.com/numaproj/numaflow/pkg/reconciler"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 )
@@ -39,12 +39,12 @@ type pipelineReconciler struct {
 	client client.Client
 	scheme *runtime.Scheme
 
-	config *controllers.GlobalConfig
+	config *reconciler.GlobalConfig
 	image  string
 	logger *zap.SugaredLogger
 }
 
-func NewReconciler(client client.Client, scheme *runtime.Scheme, config *controllers.GlobalConfig, image string, logger *zap.SugaredLogger) reconcile.Reconciler {
+func NewReconciler(client client.Client, scheme *runtime.Scheme, config *reconciler.GlobalConfig, image string, logger *zap.SugaredLogger) reconcile.Reconciler {
 	return &pipelineReconciler{client: client, scheme: scheme, config: config, image: image, logger: logger}
 }
 
diff --git a/controllers/pipeline/validate.go b/pkg/reconciler/pipeline/validate.go
similarity index 100%
rename from controllers/pipeline/validate.go
rename to pkg/reconciler/pipeline/validate.go
diff --git a/controllers/pipeline/validate_test.go b/pkg/reconciler/pipeline/validate_test.go
similarity index 100%
rename from controllers/pipeline/validate_test.go
rename to pkg/reconciler/pipeline/validate_test.go
diff --git a/controllers/vertex/controller.go b/pkg/reconciler/vertex/controller.go
similarity index 97%
rename from controllers/vertex/controller.go
rename to pkg/reconciler/vertex/controller.go
index 59ea521e..933e5694 100644
--- a/controllers/vertex/controller.go
+++ b/pkg/reconciler/vertex/controller.go
@@ -7,9 +7,9 @@ import (
 	"strings"
 	"time"
 
-	"github.com/numaproj/numaflow/controllers"
-	"github.com/numaproj/numaflow/controllers/vertex/scaling"
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/reconciler"
+	"github.com/numaproj/numaflow/pkg/reconciler/vertex/scaling"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"go.uber.org/zap"
@@ -30,14 +30,14 @@ type vertexReconciler struct {
 	client client.Client
 	scheme *runtime.Scheme
 
-	config *controllers.GlobalConfig
+	config *reconciler.GlobalConfig
 	image  string
 	logger *zap.SugaredLogger
 
 	scaler *scaling.Scaler
 }
 
-func NewReconciler(client client.Client, scheme *runtime.Scheme, config *controllers.GlobalConfig, image string, scaler *scaling.Scaler, logger *zap.SugaredLogger) reconcile.Reconciler {
+func NewReconciler(client client.Client, scheme *runtime.Scheme, config *reconciler.GlobalConfig, image string, scaler *scaling.Scaler, logger *zap.SugaredLogger) reconcile.Reconciler {
 	return &vertexReconciler{client: client, scheme: scheme, config: config, image: image, scaler: scaler, logger: logger}
 }
 


Test patch:
diff --git a/controllers/isbsvc/controller_test.go b/pkg/reconciler/isbsvc/controller_test.go
similarity index 96%
rename from controllers/isbsvc/controller_test.go
rename to pkg/reconciler/isbsvc/controller_test.go
index 0fbb4d09..fe027e37 100644
--- a/controllers/isbsvc/controller_test.go
+++ b/pkg/reconciler/isbsvc/controller_test.go
@@ -4,8 +4,8 @@ import (
 	"context"
 	"testing"
 
-	"github.com/numaproj/numaflow/controllers"
 	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,10 +45,10 @@ var (
 		},
 	}
 
-	fakeConfig = &controllers.GlobalConfig{
-		ISBSvc: &controllers.ISBSvcConfig{
-			Redis: &controllers.RedisConfig{
-				Versions: []controllers.RedisVersion{
+	fakeConfig = &reconciler.GlobalConfig{
+		ISBSvc: &reconciler.ISBSvcConfig{
+			Redis: &reconciler.RedisConfig{
+				Versions: []reconciler.RedisVersion{
 					{
 						Version:            testVersion,
 						RedisImage:         testImage,
@@ -57,8 +57,8 @@ var (
 					},
 				},
 			},
-			JetStream: &controllers.JetStreamConfig{
-				Versions: []controllers.JetStreamVersion{
+			JetStream: &reconciler.JetStreamConfig{
+				Versions: []reconciler.JetStreamVersion{
 					{
 						Version:              testVersion,
 						NatsImage:            testJSImage,
diff --git a/controllers/isbsvc/installer/installer_test.go b/pkg/reconciler/isbsvc/installer/installer_test.go
similarity index 95%
rename from controllers/isbsvc/installer/installer_test.go
rename to pkg/reconciler/isbsvc/installer/installer_test.go
index 978ca130..ac9b57a3 100644
--- a/controllers/isbsvc/installer/installer_test.go
+++ b/pkg/reconciler/isbsvc/installer/installer_test.go
@@ -4,8 +4,8 @@ import (
 	"context"
 	"testing"
 
-	"github.com/numaproj/numaflow/controllers"
 	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"
@@ -71,10 +71,10 @@ var (
 		},
 	}
 
-	fakeConfig = &controllers.GlobalConfig{
-		ISBSvc: &controllers.ISBSvcConfig{
-			Redis: &controllers.RedisConfig{
-				Versions: []controllers.RedisVersion{
+	fakeConfig = &reconciler.GlobalConfig{
+		ISBSvc: &reconciler.ISBSvcConfig{
+			Redis: &reconciler.RedisConfig{
+				Versions: []reconciler.RedisVersion{
 					{
 						Version:            testVersion,
 						RedisImage:         testImage,
@@ -83,8 +83,8 @@ var (
 					},
 				},
 			},
-			JetStream: &controllers.JetStreamConfig{
-				Versions: []controllers.JetStreamVersion{
+			JetStream: &reconciler.JetStreamConfig{
+				Versions: []reconciler.JetStreamVersion{
 					{
 						Version:              testVersion,
 						NatsImage:            testJSImage,
diff --git a/controllers/pipeline/controller_test.go b/pkg/reconciler/pipeline/controller_test.go
similarity index 97%
rename from controllers/pipeline/controller_test.go
rename to pkg/reconciler/pipeline/controller_test.go
index 366966f1..352b8506 100644
--- a/controllers/pipeline/controller_test.go
+++ b/pkg/reconciler/pipeline/controller_test.go
@@ -5,8 +5,8 @@ import (
 	"testing"
 
 	"github.com/goccy/go-json"
-	"github.com/numaproj/numaflow/controllers"
 	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"
@@ -44,10 +44,10 @@ var (
 		},
 	}
 
-	fakeConfig = &controllers.GlobalConfig{
-		ISBSvc: &controllers.ISBSvcConfig{
-			Redis: &controllers.RedisConfig{
-				Versions: []controllers.RedisVersion{
+	fakeConfig = &reconciler.GlobalConfig{
+		ISBSvc: &reconciler.ISBSvcConfig{
+			Redis: &reconciler.RedisConfig{
+				Versions: []reconciler.RedisVersion{
 					{
 						Version:            testVersion,
 						RedisImage:         testImage,
diff --git a/controllers/vertex/controller_test.go b/pkg/reconciler/vertex/controller_test.go
similarity index 98%
rename from controllers/vertex/controller_test.go
rename to pkg/reconciler/vertex/controller_test.go
index 51b703a8..6e0a60cd 100644
--- a/controllers/vertex/controller_test.go
+++ b/pkg/reconciler/vertex/controller_test.go
@@ -6,9 +6,9 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/numaproj/numaflow/controllers"
-	"github.com/numaproj/numaflow/controllers/vertex/scaling"
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/reconciler"
+	"github.com/numaproj/numaflow/pkg/reconciler/vertex/scaling"
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap/zaptest"
 	appv1 "k8s.io/api/apps/v1"
@@ -47,10 +47,10 @@ var (
 		},
 	}
 
-	fakeConfig = &controllers.GlobalConfig{
-		ISBSvc: &controllers.ISBSvcConfig{
-			Redis: &controllers.RedisConfig{
-				Versions: []controllers.RedisVersion{
+	fakeConfig = &reconciler.GlobalConfig{
+		ISBSvc: &reconciler.ISBSvcConfig{
+			Redis: &reconciler.RedisConfig{
+				Versions: []reconciler.RedisVersion{
 					{
 						Version:            testVersion,
 						RedisImage:         testImage,