diff --git a/.goreleaser.yml b/.goreleaser.yml
index d067326ad5d..c835fb25fe5 100644
--- a/.goreleaser.yml
+++ b/.goreleaser.yml
@@ -28,6 +28,7 @@ builds:
       - s390x
     env:
       - CGO_ENABLED=0
+    main: ./cmd/operator/
     ldflags:
       - -s -w -X main.version={{.Tag}}
     flags:
diff --git a/Dockerfile b/Dockerfile
index 52b4b5cb5f5..cfc74b2e27d 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -20,4 +20,4 @@ RUN \
 COPY minio-operator /minio-operator
 COPY logsearchapi-bin /logsearchapi
 
-CMD ["/minio-operator"]
+ENTRYPOINT ["/minio-operator"]
diff --git a/Makefile b/Makefile
index 9e4f89cd761..f0bb72ace58 100644
--- a/Makefile
+++ b/Makefile
@@ -34,7 +34,7 @@ getdeps:
 verify: getdeps govet gotest lint
 
 operator: verify
-	@CGO_ENABLED=0 GOOS=linux go build -trimpath --ldflags $(LDFLAGS) -o minio-operator
+	@CGO_ENABLED=0 GOOS=linux go build -trimpath --ldflags $(LDFLAGS) -o minio-operator ./cmd/operator
 
 docker: operator logsearchapi
 	@docker build --no-cache -t $(TAG) .
diff --git a/cmd/operator/app_commands.go b/cmd/operator/app_commands.go
new file mode 100644
index 00000000000..796a918a082
--- /dev/null
+++ b/cmd/operator/app_commands.go
@@ -0,0 +1,25 @@
+// Copyright (C) 2023, MinIO, Inc.
+//
+// This code is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License, version 3,
+// as published by the Free Software Foundation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License, version 3,
+// along with this program.  If not, see <http://www.gnu.org/licenses/>
+
+package main
+
+import (
+	"github.com/minio/cli"
+)
+
+var appCmds = []cli.Command{
+	controllerCmd,
+	sidecarCmd,
+	validateCmd,
+}
diff --git a/cmd/operator/controller.go b/cmd/operator/controller.go
new file mode 100644
index 00000000000..09befab3369
--- /dev/null
+++ b/cmd/operator/controller.go
@@ -0,0 +1,32 @@
+// Copyright (C) 2023, MinIO, Inc.
+//
+// This code is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License, version 3,
+// as published by the Free Software Foundation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License, version 3,
+// along with this program.  If not, see <http://www.gnu.org/licenses/>
+
+package main
+
+import (
+	"github.com/minio/cli"
+	"github.com/minio/operator/pkg/controller"
+)
+
+// starts the controller
+var controllerCmd = cli.Command{
+	Name:    "controller",
+	Aliases: []string{"ctl"},
+	Usage:   "Start MinIO Operator Controller",
+	Action:  startController,
+}
+
+func startController(ctx *cli.Context) {
+	controller.StartOperator()
+}
diff --git a/cmd/operator/main.go b/cmd/operator/main.go
new file mode 100644
index 00000000000..30ed36b089e
--- /dev/null
+++ b/cmd/operator/main.go
@@ -0,0 +1,130 @@
+// Copyright (C) 2023, MinIO, Inc.
+//
+// This code is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License, version 3,
+// as published by the Free Software Foundation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License, version 3,
+// along with this program.  If not, see <http://www.gnu.org/licenses/>
+
+package main
+
+import (
+	"os"
+	"path/filepath"
+	"sort"
+	"time"
+
+	"github.com/minio/operator/pkg"
+
+	"github.com/minio/cli"
+	"github.com/minio/pkg/console"
+	"github.com/minio/pkg/trie"
+	"github.com/minio/pkg/words"
+)
+
+// Help template for Operator.
+var operatorHelpTemplate = `NAME:
+ {{.Name}} - {{.Usage}}
+
+DESCRIPTION:
+ {{.Description}}
+
+USAGE:
+ {{.HelpName}} {{if .VisibleFlags}}[FLAGS] {{end}}COMMAND{{if .VisibleFlags}}{{end}} [ARGS...]
+
+COMMANDS:
+ {{range .VisibleCommands}}{{join .Names ", "}}{{ "\t" }}{{.Usage}}
+ {{end}}{{if .VisibleFlags}}
+FLAGS:
+ {{range .VisibleFlags}}{{.}}
+ {{end}}{{end}}
+VERSION:
+ {{.Version}}
+`
+
+func newApp(name string) *cli.App {
+	// Collection of console commands currently supported are.
+	var commands []cli.Command
+
+	// Collection of console commands currently supported in a trie tree.
+	commandsTree := trie.NewTrie()
+
+	// registerCommand registers a cli command.
+	registerCommand := func(command cli.Command) {
+		commands = append(commands, command)
+		commandsTree.Insert(command.Name)
+	}
+
+	// register commands
+	for _, cmd := range appCmds {
+		registerCommand(cmd)
+	}
+
+	findClosestCommands := func(command string) []string {
+		var closestCommands []string
+		closestCommands = append(closestCommands, commandsTree.PrefixMatch(command)...)
+
+		sort.Strings(closestCommands)
+		// Suggest other close commands - allow missed, wrongly added and
+		// even transposed characters
+		for _, value := range commandsTree.Walk(commandsTree.Root()) {
+			if sort.SearchStrings(closestCommands, value) < len(closestCommands) {
+				continue
+			}
+			// 2 is arbitrary and represents the max
+			// allowed number of typed errors
+			if words.DamerauLevenshteinDistance(command, value) < 2 {
+				closestCommands = append(closestCommands, value)
+			}
+		}
+
+		return closestCommands
+	}
+
+	cli.HelpFlag = cli.BoolFlag{
+		Name:  "help, h",
+		Usage: "show help",
+	}
+
+	app := cli.NewApp()
+	app.Name = name
+	app.Version = pkg.Version + " - " + pkg.ShortCommitID
+	app.Author = "MinIO, Inc."
+	app.Usage = "MinIO Operator"
+	app.Description = `MinIO Operator automates the orchestration of MinIO Tenants on Kubernetes.`
+	app.Copyright = "(c) 2023 MinIO, Inc."
+	app.Compiled, _ = time.Parse(time.RFC3339, pkg.ReleaseTime)
+	app.Commands = commands
+	app.HideHelpCommand = true // Hide `help, h` command, we already have `minio --help`.
+	app.CustomAppHelpTemplate = operatorHelpTemplate
+	app.CommandNotFound = func(ctx *cli.Context, command string) {
+		console.Printf("‘%s’ is not a console sub-command. See ‘console --help’.\n", command)
+		closestCommands := findClosestCommands(command)
+		if len(closestCommands) > 0 {
+			console.Println()
+			console.Println("Did you mean one of these?")
+			for _, cmd := range closestCommands {
+				console.Printf("\t‘%s’\n", cmd)
+			}
+		}
+		os.Exit(1)
+	}
+
+	return app
+}
+
+func main() {
+	args := os.Args
+	// Set the orchestrator app name.
+	appName := filepath.Base(args[0])
+	// Run the app - exit on error.
+	if err := newApp(appName).Run(args); err != nil {
+		os.Exit(1)
+	}
+}
diff --git a/cmd/operator/sidecar.go b/cmd/operator/sidecar.go
new file mode 100644
index 00000000000..e205afdb07f
--- /dev/null
+++ b/cmd/operator/sidecar.go
@@ -0,0 +1,57 @@
+// Copyright (C) 2023, MinIO, Inc.
+//
+// This code is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License, version 3,
+// as published by the Free Software Foundation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License, version 3,
+// along with this program.  If not, see <http://www.gnu.org/licenses/>
+
+package main
+
+import (
+	"log"
+	"os"
+
+	"github.com/minio/cli"
+	"github.com/minio/operator/pkg/sidecar"
+)
+
+// starts the controller
+var sidecarCmd = cli.Command{
+	Name:    "sidecar",
+	Aliases: []string{"s"},
+	Usage:   "Start MinIO Operator Sidecar",
+	Action:  startSideCar,
+	Flags: []cli.Flag{
+		cli.StringFlag{
+			Name:  "tenant",
+			Value: "",
+			Usage: "name of tenant being validated",
+		},
+		cli.StringFlag{
+			Name:  "config-name",
+			Value: "",
+			Usage: "secret being watched",
+		},
+	},
+}
+
+func startSideCar(ctx *cli.Context) {
+	tenantName := ctx.String("tenant")
+	if tenantName == "" {
+		log.Println("Must pass --tenant flag")
+		os.Exit(1)
+	}
+	configName := ctx.String("config-name")
+	if configName == "" {
+		log.Println("Must pass --config-name flag")
+		os.Exit(1)
+	}
+	sidecar.StartSideCar(tenantName, configName)
+}
diff --git a/cmd/operator/validate.go b/cmd/operator/validate.go
new file mode 100644
index 00000000000..8c6395990ca
--- /dev/null
+++ b/cmd/operator/validate.go
@@ -0,0 +1,42 @@
+// This file is part of MinIO Operator
+// Copyright (c) 2023 MinIO, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package main
+
+import (
+	"github.com/minio/cli"
+	"github.com/minio/operator/pkg/validator"
+)
+
+// starts the controller
+var validateCmd = cli.Command{
+	Name:    "validate",
+	Aliases: []string{"v"},
+	Usage:   "Start MinIO Operator Config Validator",
+	Action:  startValidator,
+	Flags: []cli.Flag{
+		cli.StringFlag{
+			Name:  "tenant",
+			Value: "",
+			Usage: "name of tenant being validated",
+		},
+	},
+}
+
+func startValidator(ctx *cli.Context) {
+	tenantName := ctx.String("tenant")
+	validator.Validate(tenantName)
+}
diff --git a/examples/kustomization/base/tenant.yaml b/examples/kustomization/base/tenant.yaml
index 8cd8ca24ddb..fdc8bde06f4 100644
--- a/examples/kustomization/base/tenant.yaml
+++ b/examples/kustomization/base/tenant.yaml
@@ -203,27 +203,27 @@ spec:
   ## https://kubernetes.io/docs/tasks/tls/managing-tls-in-a-cluster
   requestAutoCert: true
   ## Prometheus setup for MinIO Tenant.
-  prometheus:
-    image: "" # defaults to quay.io/prometheus/prometheus:latest
-    env: [ ]
-    sidecarimage: "" # defaults to alpine
-    initimage: "" # defaults to busybox:1.33.1
-    diskCapacityGB: 1
-    storageClassName: standard
-    annotations: { }
-    labels: { }
-    nodeSelector: { }
-    affinity:
-      nodeAffinity: { }
-      podAffinity: { }
-      podAntiAffinity: { }
-    resources: { }
-    serviceAccountName: ""
-    securityContext:
-      runAsUser: 1000
-      runAsGroup: 1000
-      runAsNonRoot: true
-      fsGroup: 1000
+  #  prometheus:
+  #    image: "" # defaults to quay.io/prometheus/prometheus:latest
+  #    env: [ ]
+  #    sidecarimage: "" # defaults to alpine
+  #    initimage: "" # defaults to busybox:1.33.1
+  #    diskCapacityGB: 1
+  #    storageClassName: standard
+  #    annotations: { }
+  #    labels: { }
+  #    nodeSelector: { }
+  #    affinity:
+  #      nodeAffinity: { }
+  #      podAffinity: { }
+  #      podAntiAffinity: { }
+  #    resources: { }
+  #    serviceAccountName: ""
+  #    securityContext:
+  #      runAsUser: 1000
+  #      runAsGroup: 1000
+  #      runAsNonRoot: true
+  #      fsGroup: 1000
   ## Prometheus Operator's Service Monitor for MinIO Tenant Pods.
   # prometheusOperator:
   #   labels:
diff --git a/go.mod b/go.mod
index 5a3f40f94b3..8b2851212f4 100644
--- a/go.mod
+++ b/go.mod
@@ -22,13 +22,11 @@ require (
 	golang.org/x/time v0.3.0
 	gopkg.in/yaml.v2 v2.4.0
 	k8s.io/api v0.25.4
-	k8s.io/apiextensions-apiserver v0.25.4
 	k8s.io/apimachinery v0.25.4
 	k8s.io/client-go v0.25.4
 	k8s.io/code-generator v0.25.4
 	k8s.io/klog/v2 v2.80.1
 	k8s.io/kubectl v0.25.4
-	sigs.k8s.io/controller-runtime v0.13.1
 )
 
 require (
@@ -169,9 +167,11 @@ require (
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
+	k8s.io/apiextensions-apiserver v0.25.4 // indirect
 	k8s.io/gengo v0.0.0-20220902162205-c0856e24416d // indirect
 	k8s.io/kube-openapi v0.0.0-20221110221610-a28e98eb7c70 // indirect
 	k8s.io/utils v0.0.0-20221128185143-99ec85e7a448 // indirect
+	sigs.k8s.io/controller-runtime v0.13.1 // indirect
 	sigs.k8s.io/json v0.0.0-20220713155537-f223a00ba0e2 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect
 	sigs.k8s.io/yaml v1.3.0 // indirect
diff --git a/helm/operator/templates/cluster-role.yaml b/helm/operator/templates/cluster-role.yaml
index f90749e7ea3..2ff0468aed7 100644
--- a/helm/operator/templates/cluster-role.yaml
+++ b/helm/operator/templates/cluster-role.yaml
@@ -55,6 +55,31 @@ rules:
       - list
       - delete
       - deletecollection
+  - apiGroups:
+      - ""
+    resources:
+      - serviceaccounts
+    verbs:
+      - create
+      - delete
+      - get
+      - list
+      - patch
+      - update
+      - watch
+  - apiGroups:
+      - rbac.authorization.k8s.io
+    resources:
+      - roles
+      - rolebindings
+    verbs:
+      - create
+      - delete
+      - get
+      - list
+      - patch
+      - update
+      - watch
   - apiGroups:
       - apps
     resources:
diff --git a/helm/operator/templates/operator-deployment.yaml b/helm/operator/templates/operator-deployment.yaml
index 3683bd51d05..b48d64cf672 100644
--- a/helm/operator/templates/operator-deployment.yaml
+++ b/helm/operator/templates/operator-deployment.yaml
@@ -44,6 +44,8 @@ spec:
         - name: {{ .Chart.Name }}
           image: "{{ .Values.operator.image.repository }}:{{ .Values.operator.image.tag }}"
           imagePullPolicy: {{ .Values.operator.image.pullPolicy }}
+          args:
+            - controller
           {{- with .Values.operator.env }}
           env:
           {{ toYaml . | nindent 10 }}
@@ -59,6 +61,6 @@ spec:
       {{- toYaml . | nindent 8 }}
       {{- end}}
       {{- with .Values.operator.runtimeClassName }}
-      runtimeClassName: 
+      runtimeClassName:
       {{- toYaml . | nindent 8 }}
   {{- end }}
diff --git a/helm/tenant/values.yaml b/helm/tenant/values.yaml
index 22d54ed0647..ceec9fffe48 100644
--- a/helm/tenant/values.yaml
+++ b/helm/tenant/values.yaml
@@ -5,7 +5,6 @@ secrets:
   # MinIO root user and password
   accessKey: minio
   secretKey: minio123
-
 ## MinIO Tenant Definition
 tenant:
   # Tenant name
@@ -17,10 +16,10 @@ tenant:
     pullPolicy: IfNotPresent
   ## Customize any private registry image pull secret.
   ## currently only one secret registry is supported
-  imagePullSecret: { }
+  imagePullSecret: {}
   ## If a scheduler is specified here, Tenant pods will be dispatched by specified scheduler.
   ## If not specified, the Tenant pods will be dispatched by default scheduler.
-  scheduler: { }
+  scheduler: {}
   ## Secret name that contains additional environment variable configurations.
   ## The secret is expected to have a key named config.env containing environment variables exports.
   configuration:
@@ -40,21 +39,21 @@ tenant:
       ## storageClass specifies the storage class name to be used for this pool
       storageClassName: standard
       ## Used to specify annotations for pods
-      annotations: { }
+      annotations: {}
       ## Used to specify labels for pods
-      labels: { }
+      labels: {}
       ## Used to specify a toleration for a pod
-      tolerations: [ ]
+      tolerations: []
       ## nodeSelector parameters for MinIO Pods. It specifies a map of key-value pairs. For the pod to be
       ## eligible to run on a node, the node must have each of the
       ## indicated key-value pairs as labels.
       ## Read more here: https://kubernetes.io/docs/concepts/configuration/assign-pod-node/
-      nodeSelector: { }
+      nodeSelector: {}
       ## Affinity settings for MinIO pods. Read more about affinity
       ## here: https://kubernetes.io/docs/concepts/configuration/assign-pod-node/#affinity-and-anti-affinity.
-      affinity: { }
+      affinity: {}
       ## Configure resource requests and limits for MinIO containers
-      resources: { }
+      resources: {}
       ## Configure security context
       securityContext:
         runAsUser: 1000
@@ -67,7 +66,7 @@ tenant:
         runAsGroup: 1000
         runAsNonRoot: true
       ## Configure topology constraints
-      topologySpreadConstraints: [ ]
+      topologySpreadConstraints: []
       ## Configure Runtime Class
       # runtimeClassName: ""
   ## Mount path where PV will be mounted inside container(s).
@@ -83,43 +82,43 @@ tenant:
     ## Use this field to provide one or more external CA certificates. This is used by MinIO
     ## to verify TLS connections with other applications:
     ## https://github.com/minio/minio/tree/master/docs/tls/kubernetes#2-create-kubernetes-secret
-    externalCaCertSecret: [ ]
+    externalCaCertSecret: []
     ## Use this field to provide a list of Secrets with external certificates. This can be used to configure
     ## TLS for MinIO Tenant pods. Create secrets as explained here:
     ## https://github.com/minio/minio/tree/master/docs/tls/kubernetes#2-create-kubernetes-secret
-    externalCertSecret: [ ]
+    externalCertSecret: []
     ## Enable automatic Kubernetes based certificate generation and signing as explained in
     ## https://kubernetes.io/docs/tasks/tls/managing-tls-in-a-cluster
     requestAutoCert: true
     ## This field is used only when "requestAutoCert" is set to true. Use this field to set CommonName
     ## for the auto-generated certificate. Internal DNS name for the pod will be used if CommonName is
     ## not provided. DNS name format is *.minio.default.svc.cluster.local
-    certConfig: { }
+    certConfig: {}
   ## MinIO features to enable or disable in the MinIO Tenant
   ## https://github.com/minio/operator/blob/master/docs/crd.adoc#features
   features:
     bucketDNS: false
-    domains: { }
+    domains: {}
   ## List of bucket names to create during tenant provisioning
-  buckets: [ ]
+  buckets: []
   ## List of secret names to use for generating MinIO users during tenant provisioning
-  users: [ ]
+  users: []
   ## PodManagement policy for MinIO Tenant Pods. Can be "OrderedReady" or "Parallel"
   ## Refer https://kubernetes.io/docs/tutorials/stateful-application/basic-stateful-set/#pod-management-policy
   ## for details.
   podManagementPolicy: Parallel
   # Liveness Probe for container liveness. Container will be restarted if the probe fails.
   # Refer https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes.
-  liveness: { }
+  liveness: {}
   # Readiness Probe for container readiness. Container will be removed from service endpoints if the probe fails.
   # Refer https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/
-  readiness: { }
+  readiness: {}
   # Startup Probe for container startup. Container will be restarted if the probe fails.
   # Refer https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/
-  startup: { }
+  startup: {}
   ## exposeServices defines the exposure of the MinIO object storage and Console services.
   ## service is exposed as a loadbalancer in k8s service.
-  exposeServices: { }
+  exposeServices: {}
   # kubernetes service account associated with a specific tenant
   # https://kubernetes.io/docs/tasks/configure-pod-container/configure-service-account/
   serviceAccountName: ""
@@ -137,9 +136,9 @@ tenant:
     quiet: true
   ## serviceMetadata allows passing additional labels and annotations to MinIO and Console specific
   ## services created by the operator.
-  serviceMetadata: { }
+  serviceMetadata: {}
   ## Add environment variables to be set in MinIO container (https://github.com/minio/minio/tree/master/docs/config)
-  env: [ ]
+  env: []
   ## PriorityClassName indicates the Pod priority and hence importance of a Pod relative to other Pods.
   ## This is applied to MinIO pods only.
   ## Refer Kubernetes documentation for details https://kubernetes.io/docs/concepts/configuration/pod-priority-preemption/#priorityclass/
@@ -231,20 +230,20 @@ tenant:
     # When set to true disables the creation of prometheus deployment
     disabled: true
     image: "" # defaults to quay.io/prometheus/prometheus:latest
-    env: [ ]
+    env: []
     sidecarimage: "" # defaults to alpine
     initimage: "" # defaults to busybox:1.33.1
     diskCapacityGB: 1
     storageClassName: standard
-    annotations: { }
-    labels: { }
-    nodeSelector: { }
-    tolerations: [ ]
+    annotations: {}
+    labels: {}
+    nodeSelector: {}
+    tolerations: []
     affinity:
-      nodeAffinity: { }
-      podAffinity: { }
-      podAntiAffinity: { }
-    resources: { }
+      nodeAffinity: {}
+      podAffinity: {}
+      podAntiAffinity: {}
+    resources: {}
     serviceAccountName: ""
     securityContext:
       runAsUser: 1000
@@ -256,25 +255,25 @@ tenant:
     # When set to true disables the creation of logsearch api deployment
     disabled: true
     image: "" # defaults to minio/operator:v4.4.17
-    env: [ ]
-    resources: { }
-    nodeSelector: { }
+    env: []
+    resources: {}
+    nodeSelector: {}
     affinity:
-      nodeAffinity: { }
-      podAffinity: { }
-      podAntiAffinity: { }
-    tolerations: [ ]
-    annotations: { }
-    labels: { }
+      nodeAffinity: {}
+      podAffinity: {}
+      podAntiAffinity: {}
+    tolerations: []
+    annotations: {}
+    labels: {}
     audit:
       diskCapacityGB: 1
     ## Postgres setup for LogSearch API
     db:
       image: "" # defaults to library/postgres
-      env: [ ]
+      env: []
       initimage: "" # defaults to busybox:1.33.1
       volumeClaimTemplate:
-        metadata: { }
+        metadata: {}
         spec:
           storageClassName: standard
           accessModes:
@@ -282,15 +281,15 @@ tenant:
           resources:
             requests:
               storage: 1Gi
-      resources: { }
-      nodeSelector: { }
+      resources: {}
+      nodeSelector: {}
       affinity:
-        nodeAffinity: { }
-        podAffinity: { }
-        podAntiAffinity: { }
-      tolerations: [ ]
-      annotations: { }
-      labels: { }
+        nodeAffinity: {}
+        podAffinity: {}
+        podAntiAffinity: {}
+      tolerations: []
+      annotations: {}
+      labels: {}
       serviceAccountName: ""
       securityContext:
         runAsUser: 999
@@ -303,23 +302,22 @@ tenant:
       runAsGroup: 1000
       runAsNonRoot: true
       fsGroup: 1000
-
 ingress:
   api:
     enabled: false
     ingressClassName: ""
-    labels: { }
-    annotations: { }
-    tls: [ ]
+    labels: {}
+    annotations: {}
+    tls: []
     host: minio.local
     path: /
     pathType: Prefix
   console:
     enabled: false
     ingressClassName: ""
-    labels: { }
-    annotations: { }
-    tls: [ ]
+    labels: {}
+    annotations: {}
+    tls: []
     host: minio-console.local
     path: /
     pathType: Prefix
diff --git a/pkg/apis/minio.min.io/v2/constants.go b/pkg/apis/minio.min.io/v2/constants.go
index b84ccb54d07..c4f61a129b1 100644
--- a/pkg/apis/minio.min.io/v2/constants.go
+++ b/pkg/apis/minio.min.io/v2/constants.go
@@ -48,6 +48,12 @@ const MinIOCertPath = "/tmp/certs"
 // TmpPath /tmp path inside the container file system
 const TmpPath = "/tmp"
 
+// CfgPath is the location of the MinIO Configuration File
+const CfgPath = "/tmp/minio/"
+
+// CfgFile is the Configuration File for MinIO
+const CfgFile = CfgPath + "config.env"
+
 // TenantLabel is applied to all components of a Tenant cluster
 const TenantLabel = "v1.min.io/tenant"
 
diff --git a/pkg/apis/minio.min.io/v2/helper.go b/pkg/apis/minio.min.io/v2/helper.go
index 520c8f0cb5c..51e55f12774 100644
--- a/pkg/apis/minio.min.io/v2/helper.go
+++ b/pkg/apis/minio.min.io/v2/helper.go
@@ -411,6 +411,10 @@ func (t *Tenant) EnsureDefaults() *Tenant {
 			}
 		}
 	}
+	// ServiceAccount
+	if t.Spec.ServiceAccountName == "" {
+		t.Spec.ServiceAccountName = fmt.Sprintf("%s-sa", t.Name)
+	}
 
 	return t
 }
@@ -1338,3 +1342,13 @@ func GetPgImage() string {
 	})
 	return pgDefaultImage
 }
+
+// GetRoleName returns the role name we will use for the tenant
+func (t *Tenant) GetRoleName() string {
+	return fmt.Sprintf("%s-role", t.Name)
+}
+
+// GetBindingName returns the binding name we will use for the tenant
+func (t *Tenant) GetBindingName() string {
+	return fmt.Sprintf("%s-binding", t.Name)
+}
diff --git a/pkg/build-constants.go b/pkg/build-constants.go
new file mode 100644
index 00000000000..65ab34306e6
--- /dev/null
+++ b/pkg/build-constants.go
@@ -0,0 +1,30 @@
+// This file is part of MinIO Operator
+// Copyright (c) 2023 MinIO, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package pkg
+
+var (
+	// Version - the version being released (v prefix stripped)
+	Version = "(dev)"
+	// ReleaseTag - the current git tag
+	ReleaseTag = "(no tag)"
+	// ReleaseTime - current UTC date in RFC3339 format.
+	ReleaseTime = "(no release)"
+	// CommitID - latest commit id.
+	CommitID = "(dev)"
+	// ShortCommitID - first 12 characters from CommitID.
+	ShortCommitID = "(dev)"
+)
diff --git a/pkg/controller/cluster/http_handlers.go b/pkg/controller/cluster/http_handlers.go
index f66f894a781..c18f7bf53c6 100644
--- a/pkg/controller/cluster/http_handlers.go
+++ b/pkg/controller/cluster/http_handlers.go
@@ -17,14 +17,11 @@
 package cluster
 
 import (
-	"context"
 	"fmt"
 	"net/http"
 	"strconv"
 	"strings"
 
-	"github.com/minio/operator/pkg/resources/statefulsets"
-
 	"github.com/gorilla/mux"
 	miniov2 "github.com/minio/operator/pkg/apis/minio.min.io/v2"
 	"github.com/minio/operator/pkg/resources/services"
@@ -119,78 +116,3 @@ func validateBucketName(bucket string) (bool, error) {
 	}
 	return true, nil
 }
-
-// GetenvHandler - GET /webhook/v1/getenv/{namespace}/{name}?key={env}
-func (c *Controller) GetenvHandler(w http.ResponseWriter, r *http.Request) {
-	vars := mux.Vars(r)
-	namespace := vars["namespace"]
-	name := vars["name"]
-	key := vars["key"]
-
-	secret, err := c.kubeClientSet.CoreV1().Secrets(namespace).Get(r.Context(),
-		miniov2.WebhookSecret, metav1.GetOptions{})
-	if err != nil {
-		http.Error(w, err.Error(), http.StatusForbidden)
-		return
-	}
-
-	if err = c.validateRequest(r, secret); err != nil {
-		http.Error(w, err.Error(), http.StatusForbidden)
-		return
-	}
-
-	// Get the Tenant resource with this namespace/name
-	tenant, err := c.minioClientSet.MinioV2().Tenants(namespace).Get(context.Background(), name, metav1.GetOptions{})
-	if err != nil {
-		if k8serrors.IsNotFound(err) {
-			// The Tenant resource may no longer exist, in which case we stop processing.
-			http.Error(w, fmt.Sprintf("Tenant '%s' in work queue no longer exists", key), http.StatusNotFound)
-			return
-		}
-		http.Error(w, err.Error(), http.StatusForbidden)
-		return
-	}
-
-	tenant.EnsureDefaults()
-
-	// Validate the MinIO Tenant
-	if err = tenant.Validate(); err != nil {
-		http.Error(w, err.Error(), http.StatusForbidden)
-		return
-	}
-	// correct all statefulset names by loading them, this will fix their name on the tenant pool names
-	_, err = c.getAllSSForTenant(tenant)
-	if err != nil {
-		http.Error(w, err.Error(), http.StatusForbidden)
-		return
-	}
-
-	switch key {
-	case envMinIOArgs:
-		args := strings.Join(statefulsets.GetContainerArgs(tenant, c.hostsTemplate), " ")
-		klog.Infof("%s value is %s", key, args)
-
-		_, _ = w.Write([]byte(args))
-		w.(http.Flusher).Flush()
-	case envMinIOServiceTarget:
-		schema := "https"
-		if !isOperatorTLS() {
-			schema = "http"
-		}
-		target := fmt.Sprintf("%s://%s:%s%s/%s/%s",
-			schema,
-			fmt.Sprintf("operator.%s.svc.%s",
-				miniov2.GetNSFromFile(),
-				miniov2.GetClusterDomain()),
-			miniov2.WebhookDefaultPort,
-			miniov2.WebhookAPIBucketService,
-			tenant.Namespace,
-			tenant.Name)
-		klog.Infof("%s value is %s", key, target)
-
-		_, _ = w.Write([]byte(target))
-	default:
-		http.Error(w, fmt.Sprintf("%s env key is not supported yet", key), http.StatusBadRequest)
-		return
-	}
-}
diff --git a/pkg/controller/cluster/main-controller.go b/pkg/controller/cluster/main-controller.go
index 57db22368ab..3ac7f52bd2b 100644
--- a/pkg/controller/cluster/main-controller.go
+++ b/pkg/controller/cluster/main-controller.go
@@ -199,6 +199,8 @@ type Controller struct {
 	// time, and makes it easy to ensure we are never processing the same item
 	// simultaneously in two different workers.
 	healthCheckQueue queue.RateLimitingInterface
+	// image being used in the operator deployment
+	operatorImage string
 }
 
 // NewController returns a new sample controller
@@ -213,6 +215,24 @@ func NewController(podName string, namespacesToWatch set.StringSet, kubeClientSe
 	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeClientSet.CoreV1().Events("")})
 	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName})
 
+	// get operator deployment name
+	ns := miniov2.GetNSFromFile()
+	ctx := context.Background()
+	oprImg := DefaultOperatorImage
+	oprDep, err := kubeClientSet.AppsV1().Deployments(ns).Get(ctx, DefaultDeploymentName, metav1.GetOptions{})
+	if err == nil && oprDep != nil {
+		// assume we are the first container, just in case they changed the default name
+		if len(oprDep.Spec.Template.Spec.Containers) > 0 {
+			oprImg = oprDep.Spec.Template.Spec.Containers[0].Image
+		}
+		// attempt to iterate in case there's multiple containers
+		for _, c := range oprDep.Spec.Template.Spec.Containers {
+			if c.Name == "minio-operator" || c.Name == "operator" {
+				oprImg = c.Image
+			}
+		}
+	}
+
 	controller := &Controller{
 		podName:                 podName,
 		namespacesToWatch:       namespacesToWatch,
@@ -232,6 +252,7 @@ func NewController(podName string, namespacesToWatch set.StringSet, kubeClientSe
 		recorder:                recorder,
 		hostsTemplate:           hostsTemplate,
 		operatorVersion:         operatorVersion,
+		operatorImage:           oprImg,
 	}
 
 	// Initialize operator webhook handlers
@@ -670,6 +691,13 @@ func (c *Controller) syncHandler(key string) error {
 	// get combined configurations (tenant.env, tenant.credsSecret and tenant.Configuration) for tenant
 	tenantConfiguration, err := c.getTenantCredentials(ctx, tenant)
 	if err != nil {
+		if errors.Is(err, ErrEmptyRootCredentials) {
+			if _, err2 := c.updateTenantStatus(ctx, tenant, err.Error(), 0); err2 != nil {
+				klog.V(2).Infof(err2.Error())
+			}
+			c.RegisterEvent(ctx, tenant, corev1.EventTypeWarning, "MissingCreds", "Tenant is missing root credentials")
+			return nil
+		}
 		return err
 	}
 	// get existing configuration from config.env
@@ -803,6 +831,11 @@ func (c *Controller) syncHandler(key string) error {
 			}
 		}
 	}
+	// Create Tenant Services Accoutns for Tenant
+	err = c.checkAndCreateServiceAccount(ctx, tenant)
+	if err != nil {
+		return err
+	}
 
 	adminClnt, err := tenant.NewMinIOAdmin(tenantConfiguration, c.getTransport())
 	if err != nil {
@@ -907,7 +940,19 @@ func (c *Controller) syncHandler(key string) error {
 			if tenant, err = c.updateTenantStatus(ctx, tenant, StatusProvisioningStatefulSet, 0); err != nil {
 				return err
 			}
-			ss = statefulsets.NewPool(tenant, secret, skipEnvVars, &pool, &tenant.Status.Pools[i], hlSvc.Name, c.hostsTemplate, c.operatorVersion, isOperatorTLS(), operatorCATLSExists)
+			ss = statefulsets.NewPool(&statefulsets.NewPoolArgs{
+				Tenant:          tenant,
+				WsSecret:        secret,
+				SkipEnvVars:     skipEnvVars,
+				Pool:            &pool,
+				PoolStatus:      &tenant.Status.Pools[i],
+				ServiceName:     hlSvc.Name,
+				HostsTemplate:   c.hostsTemplate,
+				OperatorVersion: c.operatorVersion,
+				OperatorTLS:     isOperatorTLS(),
+				OperatorCATLS:   operatorCATLSExists,
+				OperatorImage:   c.operatorImage,
+			})
 			ss, err = c.kubeClientSet.AppsV1().StatefulSets(tenant.Namespace).Create(ctx, ss, cOpts)
 			if err != nil {
 				return err
@@ -1132,7 +1177,19 @@ func (c *Controller) syncHandler(key string) error {
 
 		for i, pool := range tenant.Spec.Pools {
 			// Now proceed to make the yaml changes for the tenant statefulset.
-			ss := statefulsets.NewPool(tenant, secret, skipEnvVars, &pool, &tenant.Status.Pools[i], hlSvc.Name, c.hostsTemplate, c.operatorVersion, isOperatorTLS(), operatorCATLSExists)
+			ss := statefulsets.NewPool(&statefulsets.NewPoolArgs{
+				Tenant:          tenant,
+				WsSecret:        secret,
+				SkipEnvVars:     skipEnvVars,
+				Pool:            &pool,
+				PoolStatus:      &tenant.Status.Pools[i],
+				ServiceName:     hlSvc.Name,
+				HostsTemplate:   c.hostsTemplate,
+				OperatorVersion: c.operatorVersion,
+				OperatorTLS:     isOperatorTLS(),
+				OperatorCATLS:   operatorCATLSExists,
+				OperatorImage:   c.operatorImage,
+			})
 			if _, err = c.kubeClientSet.AppsV1().StatefulSets(tenant.Namespace).Update(ctx, ss, uOpts); err != nil {
 				return err
 			}
@@ -1172,7 +1229,19 @@ func (c *Controller) syncHandler(key string) error {
 			}
 		}
 		// generated the expected StatefulSet based on the new tenant configuration
-		expectedStatefulSet := statefulsets.NewPool(tenant, secret, skipEnvVars, &pool, &tenant.Status.Pools[i], hlSvc.Name, c.hostsTemplate, c.operatorVersion, isOperatorTLS(), operatorCATLSExists)
+		expectedStatefulSet := statefulsets.NewPool(&statefulsets.NewPoolArgs{
+			Tenant:          tenant,
+			WsSecret:        secret,
+			SkipEnvVars:     skipEnvVars,
+			Pool:            &pool,
+			PoolStatus:      &tenant.Status.Pools[i],
+			ServiceName:     hlSvc.Name,
+			HostsTemplate:   c.hostsTemplate,
+			OperatorVersion: c.operatorVersion,
+			OperatorTLS:     isOperatorTLS(),
+			OperatorCATLS:   operatorCATLSExists,
+			OperatorImage:   c.operatorImage,
+		})
 		// Verify if this pool matches the spec on the tenant (resources, affinity, sidecars, etc)
 		poolMatchesSS, err := poolSSMatchesSpec(expectedStatefulSet, existingStatefulSet)
 		if err != nil {
diff --git a/pkg/controller/cluster/operator.go b/pkg/controller/cluster/operator.go
index 47ca36283a3..5ea611a49cd 100644
--- a/pkg/controller/cluster/operator.go
+++ b/pkg/controller/cluster/operator.go
@@ -51,6 +51,8 @@ const (
 	OperatorTLSSecretName = "operator-tls"
 	// DefaultDeploymentName is the default name of the operator deployment
 	DefaultDeploymentName = "minio-operator"
+	// DefaultOperatorImage is the version fo the operator being used
+	DefaultOperatorImage = "minio/operator:v4.5.8"
 )
 
 var serverCertsManager *xcerts.Manager
diff --git a/pkg/controller/cluster/service-account.go b/pkg/controller/cluster/service-account.go
new file mode 100644
index 00000000000..e838f6f9b89
--- /dev/null
+++ b/pkg/controller/cluster/service-account.go
@@ -0,0 +1,140 @@
+// This file is part of MinIO Operator
+// Copyright (c) 2023 MinIO, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package cluster
+
+import (
+	"context"
+	"fmt"
+
+	miniov2 "github.com/minio/operator/pkg/apis/minio.min.io/v2"
+	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func (c *Controller) checkAndCreateServiceAccount(ctx context.Context, tenant *miniov2.Tenant) error {
+	// check if service account exits
+	sa, err := c.kubeClientSet.CoreV1().ServiceAccounts(tenant.Namespace).Get(ctx, tenant.Spec.ServiceAccountName, v1.GetOptions{})
+	if err != nil {
+		if k8serrors.IsNotFound(err) {
+			// create SA
+			sa, err = c.kubeClientSet.CoreV1().ServiceAccounts(tenant.Namespace).Create(ctx, &corev1.ServiceAccount{
+				ObjectMeta: v1.ObjectMeta{
+					Name:      tenant.Spec.ServiceAccountName,
+					Namespace: tenant.Namespace,
+				},
+			}, v1.CreateOptions{})
+			if err != nil {
+				return err
+			}
+			c.RegisterEvent(ctx, tenant, corev1.EventTypeNormal, "SACreated", "Service Account Created")
+		} else {
+			c.RegisterEvent(ctx, tenant, corev1.EventTypeWarning, "SAFailed", fmt.Sprintf("Service Account could not be created: %s", err.Error()))
+			return err
+		}
+	}
+	// check if role exist
+	role, err := c.kubeClientSet.RbacV1().Roles(tenant.Namespace).Get(ctx, tenant.GetRoleName(), v1.GetOptions{})
+	if err != nil {
+		if k8serrors.IsNotFound(err) {
+			role = getTenantRole(tenant)
+			role, err = c.kubeClientSet.RbacV1().Roles(tenant.Namespace).Create(ctx, role, v1.CreateOptions{})
+			if err != nil {
+				return err
+			}
+			c.RegisterEvent(ctx, tenant, corev1.EventTypeNormal, "RoleCreated", "Role Created")
+		} else {
+			c.RegisterEvent(ctx, tenant, corev1.EventTypeWarning, "RoleFailed", fmt.Sprintf("Role could not be created: %s", err.Error()))
+			return err
+		}
+	}
+	// check rolebinding
+	_, err = c.kubeClientSet.RbacV1().RoleBindings(tenant.Namespace).Get(ctx, tenant.GetBindingName(), v1.GetOptions{})
+	if err != nil {
+		if k8serrors.IsNotFound(err) {
+			_, err = c.kubeClientSet.RbacV1().RoleBindings(tenant.Namespace).Create(ctx, getRoleBinding(tenant, sa, role), v1.CreateOptions{})
+			if err != nil {
+				return err
+			}
+			c.RegisterEvent(ctx, tenant, corev1.EventTypeNormal, "BindingCreated", "Role Binding Created")
+		} else {
+			c.RegisterEvent(ctx, tenant, corev1.EventTypeWarning, "BindingFailed", fmt.Sprintf("Role Binding could not be created: %s", err.Error()))
+			return err
+		}
+	}
+	return nil
+}
+
+func getRoleBinding(tenant *miniov2.Tenant, sa *corev1.ServiceAccount, role *rbacv1.Role) *rbacv1.RoleBinding {
+	return &rbacv1.RoleBinding{
+		ObjectMeta: v1.ObjectMeta{
+			Name:      tenant.GetBindingName(),
+			Namespace: tenant.Namespace,
+		},
+		Subjects: []rbacv1.Subject{
+			{
+				Kind:      rbacv1.ServiceAccountKind,
+				Name:      sa.Name,
+				Namespace: sa.Namespace,
+			},
+		},
+		RoleRef: rbacv1.RoleRef{
+			APIGroup: "rbac.authorization.k8s.io",
+			Kind:     "Role",
+			Name:     role.Name,
+		},
+	}
+}
+
+func getTenantRole(tenant *miniov2.Tenant) *rbacv1.Role {
+	role := rbacv1.Role{
+		ObjectMeta: v1.ObjectMeta{
+			Name:      tenant.GetRoleName(),
+			Namespace: tenant.Namespace,
+		},
+		Rules: []rbacv1.PolicyRule{
+			{
+				APIGroups: []string{
+					"",
+				},
+				Resources: []string{
+					"secrets",
+				},
+				Verbs: []string{
+					"get",
+					"list",
+					"watch",
+				},
+			},
+			{
+				APIGroups: []string{
+					"minio.min.io",
+				},
+				Resources: []string{
+					"tenants",
+				},
+				Verbs: []string{
+					"get",
+					"list",
+					"watch",
+				},
+			},
+		},
+	}
+	return &role
+}
diff --git a/pkg/controller/cluster/tenants.go b/pkg/controller/cluster/tenants.go
index 85e7ad0e8ad..920931a59d6 100644
--- a/pkg/controller/cluster/tenants.go
+++ b/pkg/controller/cluster/tenants.go
@@ -25,6 +25,9 @@ import (
 	miniov2 "github.com/minio/operator/pkg/apis/minio.min.io/v2"
 )
 
+// ErrEmptyRootCredentials is the error returned when we detect missing root credentials
+var ErrEmptyRootCredentials = errors.New("empty tenant credentials")
+
 func (c *Controller) getTenantConfiguration(ctx context.Context, tenant *miniov2.Tenant) (map[string][]byte, error) {
 	tenantConfiguration := map[string][]byte{}
 	// Load tenant configuration from file
@@ -74,7 +77,7 @@ func (c *Controller) getTenantCredentials(ctx context.Context, tenant *miniov2.T
 	}
 
 	if accessKey == "" || secretKey == "" {
-		return tenantConfiguration, errors.New("empty tenant credentials")
+		return tenantConfiguration, ErrEmptyRootCredentials
 	}
 
 	return tenantConfiguration, nil
diff --git a/pkg/controller/cluster/webhook.go b/pkg/controller/cluster/webhook.go
index baa1e864826..6977c4ce708 100644
--- a/pkg/controller/cluster/webhook.go
+++ b/pkg/controller/cluster/webhook.go
@@ -66,11 +66,6 @@ func configureHTTPUpgradeServer(c *Controller) *http.Server {
 func configureWebhookServer(c *Controller) *http.Server {
 	router := mux.NewRouter().SkipClean(true).UseEncodedPath()
 
-	router.Methods(http.MethodGet).
-		Path(miniov2.WebhookAPIGetenv + "/{namespace}/{name:.+}").
-		HandlerFunc(c.GetenvHandler).
-		Queries(restQueries("key")...)
-
 	router.Methods(http.MethodPost).
 		Path(miniov2.WebhookAPIBucketService + "/{namespace}/{name:.+}").
 		HandlerFunc(c.BucketSrvHandler).
diff --git a/main.go b/pkg/controller/controller.go
similarity index 98%
rename from main.go
rename to pkg/controller/controller.go
index 46c9a333545..8999fad0b70 100644
--- a/main.go
+++ b/pkg/controller/controller.go
@@ -12,7 +12,7 @@
 // You should have received a copy of the GNU Affero General Public License, version 3,
 // along with this program.  If not, see <http://www.gnu.org/licenses/>
 
-package main
+package controller
 
 import (
 	"flag"
@@ -23,10 +23,10 @@ import (
 	"syscall"
 	"time"
 
-	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 
 	"github.com/minio/minio-go/v7/pkg/set"
+	"k8s.io/client-go/rest"
 
 	"k8s.io/klog/v2"
 
@@ -67,7 +67,8 @@ func init() {
 	flag.BoolVar(&checkVersion, "version", false, "print version")
 }
 
-func main() {
+// StartOperator starts the MinIO Operator controller
+func StartOperator() {
 	klog.Info("Starting MinIO Operator")
 	// set up signals, so we handle the first shutdown signal gracefully
 	stopCh := setupSignalHandler()
diff --git a/pkg/resources/statefulsets/minio-statefulset.go b/pkg/resources/statefulsets/minio-statefulset.go
index 454bb4c6a66..2749eb31586 100644
--- a/pkg/resources/statefulsets/minio-statefulset.go
+++ b/pkg/resources/statefulsets/minio-statefulset.go
@@ -79,17 +79,6 @@ func minioEnvironmentVars(t *miniov2.Tenant, skipEnvVars map[string][]byte, opVe
 			Name:  "MINIO_UPDATE_MINISIGN_PUBKEY",
 			Value: "RWTx5Zr1tiHQLwG9keckT0c45M3AGeHD6IvimQHpyRywVWGbP1aVSGav",
 		},
-		miniov2.WebhookMinIOArgs: {
-			Name: miniov2.WebhookMinIOArgs,
-			ValueFrom: &corev1.EnvVarSource{
-				SecretKeyRef: &corev1.SecretKeySelector{
-					LocalObjectReference: corev1.LocalObjectReference{
-						Name: miniov2.WebhookSecret,
-					},
-					Key: miniov2.WebhookMinIOArgs,
-				},
-			},
-		},
 		"MINIO_OPERATOR_VERSION": {
 			Name:  "MINIO_OPERATOR_VERSION",
 			Value: opVersion,
@@ -198,7 +187,7 @@ func minioEnvironmentVars(t *miniov2.Tenant, skipEnvVars map[string][]byte, opVe
 	if t.HasConfigurationSecret() {
 		envVarsMap["MINIO_CONFIG_ENV_FILE"] = corev1.EnvVar{
 			Name:  "MINIO_CONFIG_ENV_FILE",
-			Value: miniov2.TmpPath + "/minio-config/config.env",
+			Value: miniov2.CfgFile,
 		}
 	}
 
@@ -279,15 +268,29 @@ func ContainerMatchLabels(t *miniov2.Tenant, pool *miniov2.Pool) *metav1.LabelSe
 	}
 }
 
+// CfgVolumeMount is the volume mount used by `minio`, `sidecar` and `validate-arguments` containers
+var CfgVolumeMount = corev1.VolumeMount{
+	Name:      CfgVol,
+	MountPath: miniov2.CfgPath,
+}
+
+// TmpCfgVolumeMount is the temporary location
+var TmpCfgVolumeMount = corev1.VolumeMount{
+	Name:      "configuration",
+	MountPath: miniov2.TmpPath + "/minio-config",
+}
+
 // Builds the volume mounts for MinIO container.
 func volumeMounts(t *miniov2.Tenant, pool *miniov2.Pool, operatorTLS bool, certVolumeSources []v1.VolumeProjection) (mounts []v1.VolumeMount) {
-	// This is the case where user didn't provide a pool and we deploy a EmptyDir based
-	// single node single drive (FS) MinIO deployment
+	// Default volume name, unless another one was provided
 	name := miniov2.MinIOVolumeName
 	if pool.VolumeClaimTemplate != nil {
 		name = pool.VolumeClaimTemplate.Name
 	}
 
+	// shared configuration Volume
+	mounts = append(mounts, CfgVolumeMount)
+
 	if pool.VolumesPerServer == 1 {
 		mounts = append(mounts, corev1.VolumeMount{
 			Name:      name + strconv.Itoa(0),
@@ -311,13 +314,6 @@ func volumeMounts(t *miniov2.Tenant, pool *miniov2.Pool, operatorTLS bool, certV
 		})
 	}
 
-	if t.HasConfigurationSecret() {
-		mounts = append(mounts, corev1.VolumeMount{
-			Name:      "configuration",
-			MountPath: miniov2.TmpPath + "/minio-config",
-		})
-	}
-
 	return mounts
 }
 
@@ -452,8 +448,38 @@ func poolContainerSecurityContext(pool *miniov2.Pool) *v1.SecurityContext {
 	return &containerSecurityContext
 }
 
+// CfgVol is the name of the configuration volume we will use
+const CfgVol = "cfg-vol"
+
+// NewPoolArgs arguments used to create a new pool
+type NewPoolArgs struct {
+	Tenant          *miniov2.Tenant
+	WsSecret        *v1.Secret
+	SkipEnvVars     map[string][]byte
+	Pool            *miniov2.Pool
+	PoolStatus      *miniov2.PoolStatus
+	ServiceName     string
+	HostsTemplate   string
+	OperatorVersion string
+	OperatorTLS     bool
+	OperatorCATLS   bool
+	OperatorImage   string
+}
+
 // NewPool creates a new StatefulSet for the given Cluster.
-func NewPool(t *miniov2.Tenant, wsSecret *v1.Secret, skipEnvVars map[string][]byte, pool *miniov2.Pool, poolStatus *miniov2.PoolStatus, serviceName, hostsTemplate, operatorVersion string, operatorTLS bool, operatorCATLS bool) *appsv1.StatefulSet {
+func NewPool(args *NewPoolArgs) *appsv1.StatefulSet {
+	t := args.Tenant
+	wsSecret := args.WsSecret
+	skipEnvVars := args.SkipEnvVars
+	pool := args.Pool
+	poolStatus := args.PoolStatus
+	serviceName := args.ServiceName
+	hostsTemplate := args.HostsTemplate
+	operatorVersion := args.OperatorVersion
+	operatorTLS := args.OperatorTLS
+	operatorCATLS := args.OperatorCATLS
+	operatorImage := args.OperatorImage
+
 	var podVolumes []corev1.Volume
 	replicas := pool.Servers
 	var certVolumeSources []corev1.VolumeProjection
@@ -468,6 +494,15 @@ func NewPool(t *miniov2.Tenant, wsSecret *v1.Secret, skipEnvVars map[string][]by
 		{Key: "public.crt", Path: "CAs/kes.crt"},
 	}
 
+	// Create an empty dir volume to share the configuration between the main container and side-car
+
+	podVolumes = append(podVolumes, corev1.Volume{
+		Name: CfgVol,
+		VolumeSource: v1.VolumeSource{
+			EmptyDir: &corev1.EmptyDirVolumeSource{},
+		},
+	})
+
 	// Multiple certificates will be mounted using the following folder structure:
 	//
 	//	certs
@@ -791,6 +826,7 @@ func NewPool(t *miniov2.Tenant, wsSecret *v1.Secret, skipEnvVars map[string][]by
 
 	containers := []corev1.Container{
 		poolMinioServerContainer(t, wsSecret, skipEnvVars, pool, hostsTemplate, operatorVersion, operatorTLS, certVolumeSources),
+		getSideCarContainer(t, operatorImage),
 	}
 
 	// attach any sidecar containers and volumes
@@ -811,6 +847,8 @@ func NewPool(t *miniov2.Tenant, wsSecret *v1.Secret, skipEnvVars map[string][]by
 		unavailable = intstr.FromInt(2)
 	}
 
+	initContainer := getInitContainer(t, operatorImage)
+
 	ss := &appsv1.StatefulSet{
 		ObjectMeta: ssMeta,
 		Spec: appsv1.StatefulSetSpec{
@@ -827,6 +865,9 @@ func NewPool(t *miniov2.Tenant, wsSecret *v1.Secret, skipEnvVars map[string][]by
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: PodMetadata(t, pool),
 				Spec: corev1.PodSpec{
+					InitContainers: []corev1.Container{
+						initContainer,
+					},
 					Containers:                containers,
 					Volumes:                   podVolumes,
 					RestartPolicy:             corev1.RestartPolicyAlways,
@@ -868,3 +909,43 @@ func NewPool(t *miniov2.Tenant, wsSecret *v1.Secret, skipEnvVars map[string][]by
 
 	return ss
 }
+
+func getInitContainer(t *miniov2.Tenant, operatorImage string) v1.Container {
+	initContainer := corev1.Container{
+		Name:  "validate-arguments",
+		Image: operatorImage,
+		Args: []string{
+			"validate",
+			"--tenant",
+			t.Name,
+		},
+		VolumeMounts: []corev1.VolumeMount{
+			CfgVolumeMount,
+		},
+	}
+	if t.HasConfigurationSecret() {
+		initContainer.VolumeMounts = append(initContainer.VolumeMounts, TmpCfgVolumeMount)
+	}
+	return initContainer
+}
+
+func getSideCarContainer(t *miniov2.Tenant, operatorImage string) v1.Container {
+	sidecarContainer := corev1.Container{
+		Name:  "sidecar",
+		Image: operatorImage,
+		Args: []string{
+			"sidecar",
+			"--tenant",
+			t.Name,
+			"--config-name",
+			t.Spec.Configuration.Name,
+		},
+		VolumeMounts: []corev1.VolumeMount{
+			CfgVolumeMount,
+		},
+	}
+	if t.HasConfigurationSecret() {
+		sidecarContainer.VolumeMounts = append(sidecarContainer.VolumeMounts, TmpCfgVolumeMount)
+	}
+	return sidecarContainer
+}
diff --git a/pkg/sidecar/sidecar.go b/pkg/sidecar/sidecar.go
new file mode 100644
index 00000000000..66ef0c15c2f
--- /dev/null
+++ b/pkg/sidecar/sidecar.go
@@ -0,0 +1,217 @@
+// This file is part of MinIO Operator
+// Copyright (c) 2023 MinIO, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package sidecar
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"strings"
+	"time"
+
+	v2 "github.com/minio/operator/pkg/apis/minio.min.io/v2"
+	clientset "github.com/minio/operator/pkg/client/clientset/versioned"
+	minioInformers "github.com/minio/operator/pkg/client/informers/externalversions"
+	v22 "github.com/minio/operator/pkg/client/informers/externalversions/minio.min.io/v2"
+	"github.com/minio/operator/pkg/validator"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/informers"
+	coreinformers "k8s.io/client-go/informers/core/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2"
+)
+
+func init() {
+	log.SetFlags(log.LstdFlags | log.Lshortfile)
+}
+
+// StartSideCar instantiates kube clients and starts the side car controller
+func StartSideCar(tenantName string, secretName string) {
+	log.Println("Starting Sidecar")
+	cfg, err := rest.InClusterConfig()
+	if err != nil {
+		panic(err)
+	}
+
+	if err != nil {
+		klog.Fatalf("Error building kubeconfig: %s", err.Error())
+	}
+
+	kubeClient, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		klog.Fatalf("Error building Kubernetes clientset: %s", err.Error())
+	}
+
+	controllerClient, err := clientset.NewForConfig(cfg)
+	if err != nil {
+		klog.Fatalf("Error building MinIO clientset: %s", err.Error())
+	}
+
+	controller := NewSideCarController(kubeClient, controllerClient, tenantName, secretName)
+
+	stop := make(chan struct{})
+	defer close(stop)
+	err = controller.Run(stop)
+	if err != nil {
+		klog.Fatal(err)
+	}
+	select {}
+}
+
+// Controller is the controller holding the informers used to monitor args and tenant structure
+type Controller struct {
+	kubeClient         *kubernetes.Clientset
+	controllerClient   *clientset.Clientset
+	tenantName         string
+	secretName         string
+	minInformerFactory minioInformers.SharedInformerFactory
+	secretInformer     coreinformers.SecretInformer
+	tenantInformer     v22.TenantInformer
+	namespace          string
+	informerFactory    informers.SharedInformerFactory
+}
+
+// NewSideCarController returns an instance of Controller with the provided clients
+func NewSideCarController(kubeClient *kubernetes.Clientset, controllerClient *clientset.Clientset, tenantName string, secretName string) *Controller {
+	namespace := v2.GetNSFromFile()
+
+	factory := informers.NewSharedInformerFactoryWithOptions(kubeClient, time.Hour*1, informers.WithNamespace(namespace))
+	secretInformer := factory.Core().V1().Secrets()
+
+	minioInformerFactory := minioInformers.NewSharedInformerFactoryWithOptions(controllerClient, time.Hour*1, minioInformers.WithNamespace(namespace))
+	tenantInformer := minioInformerFactory.Minio().V2().Tenants()
+
+	c := &Controller{
+		kubeClient:         kubeClient,
+		controllerClient:   controllerClient,
+		tenantName:         tenantName,
+		namespace:          namespace,
+		secretName:         secretName,
+		minInformerFactory: minioInformerFactory,
+		informerFactory:    factory,
+		tenantInformer:     tenantInformer,
+		secretInformer:     secretInformer,
+	}
+
+	tenantInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		UpdateFunc: func(old, new interface{}) {
+			oldTenant := old.(*v2.Tenant)
+			newTenant := new.(*v2.Tenant)
+			if newTenant.ResourceVersion == oldTenant.ResourceVersion {
+				// Periodic resync will send update events for all known Tenants.
+				// Two different versions of the same Tenant will always have different RVs.
+				return
+			}
+			c.regenCfg(tenantName, namespace)
+		},
+	})
+
+	secretInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		UpdateFunc: func(old, new interface{}) {
+			oldSecret := old.(*corev1.Secret)
+			// ignore anything that is not what we want
+			if oldSecret.Name != secretName {
+				return
+			}
+			newSecret := new.(*corev1.Secret)
+			if newSecret.ResourceVersion == oldSecret.ResourceVersion {
+				// Periodic resync will send update events for all known Tenants.
+				// Two different versions of the same Tenant will always have different RVs.
+				return
+			}
+			data := newSecret.Data["config.env"]
+			// validate root creds in string
+			rootUserMissing := true
+			rootPassMissing := false
+
+			dataStr := string(data)
+			if !strings.Contains(dataStr, "MINIO_ROOT_USER") {
+				rootUserMissing = true
+			}
+			if !strings.Contains(dataStr, "MINIO_ACCESS_KEY") {
+				rootUserMissing = true
+			}
+			if !strings.Contains(dataStr, "MINIO_ROOT_PASSWORD") {
+				rootPassMissing = true
+			}
+			if !strings.Contains(dataStr, "MINIO_SECRET_KEY") {
+				rootPassMissing = true
+			}
+			if rootUserMissing || rootPassMissing {
+				log.Println("Missing root credentials in the configuration.")
+				log.Println("MinIO won't start")
+				os.Exit(1)
+			}
+
+			c.regenCfgWithCfg(tenantName, namespace, string(data))
+		},
+	})
+
+	return c
+}
+
+func (c Controller) regenCfg(tenantName string, namespace string) {
+	rootUserFound, rootPwdFound, fileContents, err := validator.ReadTmpConfig()
+	if err != nil {
+		log.Println(err)
+		return
+	}
+	if !rootUserFound || !rootPwdFound {
+		log.Println("Missing root credentials in the configuration.")
+		log.Println("MinIO won't start")
+		os.Exit(1)
+	}
+	c.regenCfgWithCfg(tenantName, namespace, fileContents)
+}
+
+func (c Controller) regenCfgWithCfg(tenantName string, namespace string, fileContents string) {
+	ctx := context.Background()
+
+	args, err := validator.GetTenantArgs(ctx, c.controllerClient, tenantName, namespace)
+	if err != nil {
+		log.Println(err)
+		return
+	}
+
+	fileContents = fileContents + fmt.Sprintf("export MINIO_ARGS=\"%s\"\n", args)
+
+	err = os.WriteFile(v2.CfgFile, []byte(fileContents), 0o644)
+	if err != nil {
+		log.Println(err)
+	}
+}
+
+// Run starts the informers
+func (c *Controller) Run(stopCh chan struct{}) error {
+	// Starts all the shared minioInformers that have been created by the factory so
+	// far.
+	c.minInformerFactory.Start(stopCh)
+	c.informerFactory.Start(stopCh)
+
+	// wait for the initial synchronization of the local cache.
+	if !cache.WaitForCacheSync(stopCh, c.tenantInformer.Informer().HasSynced) {
+		return fmt.Errorf("Failed to sync")
+	}
+	// wait for the initial synchronization of the local cache.
+	if !cache.WaitForCacheSync(stopCh, c.secretInformer.Informer().HasSynced) {
+		return fmt.Errorf("Failed to sync")
+	}
+	return nil
+}
diff --git a/pkg/validator/validator.go b/pkg/validator/validator.go
new file mode 100644
index 00000000000..d46b51e6862
--- /dev/null
+++ b/pkg/validator/validator.go
@@ -0,0 +1,140 @@
+// This file is part of MinIO Operator
+// Copyright (c) 2023 MinIO, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package validator
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"strings"
+
+	miniov2 "github.com/minio/operator/pkg/apis/minio.min.io/v2"
+	clientset "github.com/minio/operator/pkg/client/clientset/versioned"
+	"github.com/minio/operator/pkg/resources/statefulsets"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/rest"
+	"k8s.io/klog/v2"
+)
+
+// Validate checks the configuration on the seeded configuration and issues a valid one for MinIO to
+// start, however if root credentials are missing, it will exit with error
+func Validate(tenantName string) {
+	rootUserFound, rootPwdFound, fileContents, err := ReadTmpConfig()
+	if err != nil {
+		panic(err)
+	}
+
+	namespace := miniov2.GetNSFromFile()
+
+	cfg, err := rest.InClusterConfig()
+	// If config is passed as a flag use that instead
+	//if kubeconfig != "" {
+	//	cfg, err = clientcmd.BuildConfigFromFlags(masterURL, kubeconfig)
+	//}
+	if err != nil {
+		panic(err)
+	}
+
+	controllerClient, err := clientset.NewForConfig(cfg)
+	if err != nil {
+		klog.Fatalf("Error building MinIO clientset: %s", err.Error())
+	}
+
+	ctx := context.Background()
+
+	args, err := GetTenantArgs(ctx, controllerClient, tenantName, namespace)
+	if err != nil {
+		log.Println(err)
+		os.Exit(1)
+	}
+
+	fileContents = fileContents + fmt.Sprintf("export MINIO_ARGS=\"%s\"\n", args)
+
+	if !rootUserFound || !rootPwdFound {
+		log.Println("Missing root credentials in the configuration.")
+		log.Println("MinIO won't start")
+		os.Exit(1)
+	}
+
+	err = os.WriteFile(miniov2.CfgFile, []byte(fileContents), 0o644)
+	if err != nil {
+		log.Println(err)
+	}
+}
+
+// GetTenantArgs returns the arguments for the tenant based on the tenants they have
+func GetTenantArgs(ctx context.Context, controllerClient *clientset.Clientset, tenantName string, namespace string) (string, error) {
+	// get the only tenant in this namespace
+	tenant, err := controllerClient.MinioV2().Tenants(namespace).Get(ctx, tenantName, metav1.GetOptions{})
+	if err != nil {
+		log.Println(err)
+		return "", err
+	}
+
+	tenant.EnsureDefaults()
+
+	// Validate the MinIO Tenant
+	if err = tenant.Validate(); err != nil {
+		log.Println(err)
+		return "", err
+	}
+
+	args := strings.Join(statefulsets.GetContainerArgs(tenant, ""), " ")
+	return args, err
+}
+
+// ReadTmpConfig reads the seeded configuration from a tmp location
+func ReadTmpConfig() (bool, bool, string, error) {
+	file, err := os.Open("/tmp/minio-config/config.env")
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer file.Close()
+
+	rootUserFound := false
+	rootPwdFound := false
+
+	scanner := bufio.NewScanner(file)
+	newFile := ""
+	for scanner.Scan() {
+		line := scanner.Text()
+		if strings.Contains(line, "MINIO_ROOT_USER") {
+			rootUserFound = true
+		}
+		if strings.Contains(line, "MINIO_ACCESS_KEY") {
+			rootUserFound = true
+		}
+		if strings.Contains(line, "MINIO_ROOT_PASSWORD") {
+			rootPwdFound = true
+		}
+		if strings.Contains(line, "MINIO_SECRET_KEY") {
+			rootPwdFound = true
+		}
+		// We don't allow users to set MINIO_ARGS
+		if strings.Contains(line, "MINIO_ARGS") {
+			log.Println("MINIO_ARGS in config file found. It will be ignored.")
+			continue
+		}
+		newFile = newFile + line + "\n"
+	}
+	if err := scanner.Err(); err != nil {
+		log.Fatal(err)
+	}
+	return rootUserFound, rootPwdFound, newFile, nil
+}
diff --git a/resources/base/cluster-role.yaml b/resources/base/cluster-role.yaml
index f90749e7ea3..2ff0468aed7 100644
--- a/resources/base/cluster-role.yaml
+++ b/resources/base/cluster-role.yaml
@@ -55,6 +55,31 @@ rules:
       - list
       - delete
       - deletecollection
+  - apiGroups:
+      - ""
+    resources:
+      - serviceaccounts
+    verbs:
+      - create
+      - delete
+      - get
+      - list
+      - patch
+      - update
+      - watch
+  - apiGroups:
+      - rbac.authorization.k8s.io
+    resources:
+      - roles
+      - rolebindings
+    verbs:
+      - create
+      - delete
+      - get
+      - list
+      - patch
+      - update
+      - watch
   - apiGroups:
       - apps
     resources:
diff --git a/resources/base/deployment.yaml b/resources/base/deployment.yaml
index de5256b0bf8..a0364add8da 100644
--- a/resources/base/deployment.yaml
+++ b/resources/base/deployment.yaml
@@ -23,6 +23,8 @@ spec:
         - name: minio-operator
           image: minio/operator:v4.5.8
           imagePullPolicy: IfNotPresent
+          args:
+            - controller
           resources:
             requests:
               cpu: 200m
diff --git a/testing/check-logs.sh b/testing/check-logs.sh
index b978849da53..d10ab87d616 100755
--- a/testing/check-logs.sh
+++ b/testing/check-logs.sh
@@ -73,7 +73,7 @@ function main() {
 
     install_operator
 
-    install_tenant
+    install_tenant "logs"
 
 	perform_attempts_to_get_log_api_response
 	if [ $FINAL_RESULT = 1 ]; then
diff --git a/testing/check-prometheus.sh b/testing/check-prometheus.sh
index e05161947d5..129b60a9305 100755
--- a/testing/check-prometheus.sh
+++ b/testing/check-prometheus.sh
@@ -47,7 +47,7 @@ function main() {
 
   install_operator
 
-  install_tenant
+  install_tenant "prometheus"
 
   check_tenant_status tenant-lite storage-lite
 
@@ -57,28 +57,10 @@ function main() {
   wait_on_prometheus_pods
   echo 'end - wait for prometheus to appear'
 
-  echo 'Wait for pod to be ready for port forward'
-  try kubectl wait --namespace tenant-lite \
-    --for=condition=ready pod \
-    --selector=statefulset.kubernetes.io/pod-name=storage-lite-pool-0-0 \
-    --timeout=120s
-
-  echo 'port forward without the hop, directly from the tenant/pod'
-  kubectl port-forward storage-lite-pool-0-0 9443 --namespace tenant-lite &
-
-  echo 'start - wait for port-forward to be completed'
-  sleep 15
-  echo 'end - wait for port-forward to be completed'
-
-  echo 'To display port connections'
-  sudo netstat -tunlp # want to see if 9443 is LISTEN state to proceed
+  echo "make sure there's no rolling restart going"
+  kubectl -n tenant-lite rollout status sts/storage-lite-pool-0
 
-  echo 'start - open and allow port connection'
-  sudo apt install ufw
-  sudo ufw allow http
-  sudo ufw allow https
-  sudo ufw allow 9443/tcp
-  echo 'end - open and allow port connection'
+  port_forward tenant-lite storage-lite storage-lite-console 9443
 
   echo 'Get token from MinIO Console'
   COOKIE=$(
@@ -86,7 +68,7 @@ function main() {
       -H 'content-type: application/json' \
       --data-raw '{"accessKey":"minio","secretKey":"minio123"}' --insecure 2>&1 |
       grep "set-cookie: token=" | sed -e "s/< set-cookie: token=//g" |
-      awk -F ';' '{print $1}'
+      awk -F ';' '{print $  1}'
   )
   echo $COOKIE
 
diff --git a/testing/common.sh b/testing/common.sh
index 83c3b6e41f6..9a61a362f63 100644
--- a/testing/common.sh
+++ b/testing/common.sh
@@ -19,16 +19,21 @@ export CI
 ARCH=`{ case "$(uname -m)" in "x86_64") echo -n "amd64";; "aarch64") echo -n "arm64";; *) echo -n "$(uname -m)";; esac; }`
 OS=$(uname | awk '{print tolower($0)}')
 
-## Make sure to install things if not present already
-sudo curl -#L "https://dl.k8s.io/release/v1.23.1/bin/$OS/$ARCH/kubectl" -o /usr/local/bin/kubectl
-sudo chmod +x /usr/local/bin/kubectl
+DEV_TEST=$OPERATOR_DEV_TEST
 
-sudo curl -#L "https://dl.min.io/client/mc/release/${OS}-${ARCH}/mc" -o /usr/local/bin/mc
-sudo chmod +x /usr/local/bin/mc
+# Set OPERATOR_DEV_TEST to skip downloading these dependencies
+if [[ -z "${DEV_TEST}" ]]; then
+  ## Make sure to install things if not present already
+  sudo curl -#L "https://dl.k8s.io/release/v1.23.1/bin/$OS/$ARCH/kubectl" -o /usr/local/bin/kubectl
+  sudo chmod +x /usr/local/bin/kubectl
 
-## Install yq
-sudo wget -qO /usr/local/bin/yq https://github.com/mikefarah/yq/releases/latest/download/yq_${OS}_${ARCH}
-sudo chmod a+x /usr/local/bin/yq
+  sudo curl -#L "https://dl.min.io/client/mc/release/${OS}-${ARCH}/mc" -o /usr/local/bin/mc
+  sudo chmod +x /usr/local/bin/mc
+
+  ## Install yq
+  sudo wget -qO /usr/local/bin/yq https://github.com/mikefarah/yq/releases/latest/download/yq_${OS}_${ARCH}
+  sudo chmod a+x /usr/local/bin/yq
+fi
 
 yell() { echo "$0: $*" >&2; }
 
@@ -83,6 +88,9 @@ function install_operator() {
     value=minio-operator
   fi
 
+  echo "Scaling down MinIO Operator Deployment"
+  try kubectl -n minio-operator scale deployment minio-operator --replicas=1
+
   # Reusing the wait for both, Kustomize and Helm
   echo "Waiting for k8s api"
   sleep 10
@@ -107,17 +115,22 @@ function install_operator() {
 function install_operator_version() {
   # Obtain release
   version="$1"
-  if [ -z "$version" ]
-  then
+  if [ -z "$version" ]; then
     version=$(curl https://api.github.com/repos/minio/operator/releases/latest | jq --raw-output '.tag_name | "\(.[1:])"')
   fi
   echo "Target operator release: $version"
-  sudo curl -#L "https://github.com/minio/operator/releases/download/v${version}/kubectl-minio_${version}_${OS}_${ARCH}" -o /usr/local/bin/kubectl-minio
-  sudo chmod +x /usr/local/bin/kubectl-minio
+  # Set OPERATOR_DEV_TEST to skip downloading these dependencies
+  if [[ -z "${DEV_TEST}" ]]; then
+    sudo curl -#L "https://github.com/minio/operator/releases/download/v${version}/kubectl-minio_${version}_${OS}_${ARCH}" -o /usr/local/bin/kubectl-minio
+    sudo chmod +x /usr/local/bin/kubectl-minio
+  fi
 
   # Initialize the MinIO Kubernetes Operator
   kubectl minio init
 
+  echo "Scaling down MinIO Operator Deployment"
+  try kubectl -n minio-operator scale deployment minio-operator --replicas=1
+
   # Verify installation of the plugin
   echo "Installed operator release: $(kubectl minio version)"
 
@@ -153,7 +166,16 @@ function install_operator_version() {
 }
 
 function destroy_kind() {
-  kind delete cluster
+  # To allow the execution without killing the cluster at the end of the test
+  # Use below statement to automatically test and kill cluster at the end:
+  # `unset OPERATOR_DEV_TEST`
+  # Use below statement to test and keep cluster alive at the end!:
+  # `export OPERATOR_DEV_TEST="ON"`
+  if [[ -z "${DEV_TEST}" ]]; then
+    echo "Cluster not destroyed due to manual testing"
+  else
+    kind delete cluster
+  fi
 }
 
 function wait_for_resource() {
@@ -208,6 +230,14 @@ function check_tenant_status() {
     --selector=$key=$2 \
     --timeout=300s
 
+  if [ $# -ge 4 ]; then
+    # make sure no rollout is happening
+    try kubectl -n $1 rollout status sts/minio1-pool-0
+  else
+    # make sure no rollout is happening
+    try kubectl -n $1 rollout status sts/$2-pool-0
+  fi
+
   echo "Tenant is created successfully, proceeding to validate 'mc admin info minio/'"
 
   try kubectl get pods --namespace $1
@@ -227,10 +257,9 @@ function check_tenant_status() {
 
 # Install tenant function is being used by deploy-tenant and check-prometheus
 function install_tenant() {
-
-  echo "Check if helm will install the Tenant"
+  # Check if we are going to install helm, lastest in this branch or a particular version
   if [ "$1" = "helm" ]; then
-
+    echo "Installing tenant from Helm"
     echo "This test is intended for helm only not for KES, there is another kes test, so let's remove KES here"
     yq -i eval 'del(.tenant.kes)' "${SCRIPT_DIR}/../helm/tenant/values.yaml"
 
@@ -241,13 +270,34 @@ function install_tenant() {
     value=minio
     try helm install --namespace $namespace \
       --create-namespace tenant ./helm/tenant
-  else
-    namespace=tenant-lite
+  elif [ "$1" = "logs" ]; then
+    namespace="tenant-lite"
     key=v1.min.io/tenant
     value=storage-lite
-    echo "Installing lite tenant"
+    echo "Installing lite tenant from current branch"
+
+    try kubectl apply -k "${SCRIPT_DIR}/../testing/tenant-logs"
+  elif [ "$1" = "prometheus" ]; then
+    namespace="tenant-lite"
+    key=v1.min.io/tenant
+    value=storage-lite
+    echo "Installing lite tenant from current branch"
+
+    try kubectl apply -k "${SCRIPT_DIR}/../testing/tenant-prometheus"
+  elif [ -e $1 ]; then
+    namespace="tenant-lite"
+    key=v1.min.io/tenant
+    value=storage-lite
+    echo "Installing lite tenant from current branch"
 
     try kubectl apply -k "${SCRIPT_DIR}/../testing/tenant"
+  else
+    namespace="tenant-lite"
+    key=v1.min.io/tenant
+    value=storage-lite
+    echo "Installing lite tenant for version $1"
+
+    try kubectl apply -k "github.com/minio/operator/testing/tenant\?ref\=$1"
   fi
 
   echo "Waiting for the tenant statefulset, this indicates the tenant is being fulfilled"
@@ -265,3 +315,44 @@ function install_tenant() {
   echo "Build passes basic tenant creation"
 
 }
+
+# Port forward
+function port_forward() {
+  namespace=$1
+  tenant=$2
+  svc=$3
+  localport=$4
+
+  totalwait=0
+  echo 'Validating tenant pods are ready to serve'
+  for pod in `kubectl --namespace $namespace --selector=v1.min.io/tenant=$tenant get pod -o json |  jq '.items[] | select(.metadata.name|contains("'$tenant'"))| .metadata.name' | sed 's/"//g'`; do
+    while true; do
+      if kubectl --namespace $namespace -c minio logs pod/$pod | grep --quiet 'All MinIO sub-systems initialized successfully'; then
+        echo "$pod is ready to serve" && break
+      fi
+      sleep 5
+      totalwait=$((totalwait + 5))
+      if [ "$totalwait" -gt 305 ]; then
+        echo "Unable to validate pod $pod after 5 minutes, exiting."
+        try false
+      fi
+    done
+  done
+
+  echo "Killing any current port-forward"
+  for pid in $(lsof -i :$localport | awk '{print $2}' | uniq | grep -o '[0-9]*')
+  do
+    if [ -n "$pid" ]
+    then
+      kill -9 $pid
+      echo "Killed previous port-forward process using port $localport: $pid"
+    fi
+  done
+
+  echo "Establishing port-forward"
+  kubectl port-forward service/$svc -n $namespace $localport &
+
+  echo 'start - wait for port-forward to be completed'
+  sleep 15
+  echo 'end - wait for port-forward to be completed'
+}
diff --git a/testing/deploy-tenant-upgrade.sh b/testing/deploy-tenant-upgrade.sh
index 7d79f737efc..ab5e2777e40 100755
--- a/testing/deploy-tenant-upgrade.sh
+++ b/testing/deploy-tenant-upgrade.sh
@@ -51,46 +51,10 @@ function announce_test() {
   echo "## Testing upgrade of Operator from $lower_text to $upper_text ##"
 }
 
-# Port forward
-function port_forward() {
-  totalwait=0
-  echo 'Validating tenant pods are ready to serve'
-  for pod in `kubectl --namespace $namespace --selector=v1.min.io/tenant=$tenant get pod -o json |  jq '.items[] | select(.metadata.name|contains("'$tenant'"))| .metadata.name' | sed 's/"//g'`; do
-    while true; do
-      if kubectl --namespace $namespace logs pod/$pod | grep --quiet 'All MinIO sub-systems initialized successfully'; then
-        echo "$pod is ready to serve" && break
-      fi
-      sleep 5
-      totalwait=$((totalwait + 5))
-      if [ "$totalwait" -gt 305 ]; then
-        echo "Unable to validate pods after 5 minutes, exiting."
-        try false
-      fi
-    done
-  done
-
-  echo "Killing any current port-forward"
-  for pid in $(lsof -i :$localport | awk '{print $2}' | uniq | grep -o '[0-9]*')
-  do
-    if [ -n "$pid" ] 
-    then
-      kill -9 $pid
-      echo "Killed previous port-forward process using port $localport: $pid"
-    fi
-  done
-
-  echo "Establishing port-forward"
-  kubectl port-forward service/$tenant-hl -n $namespace $localport &
-
-  echo 'start - wait for port-forward to be completed'
-  sleep 15
-  echo 'end - wait for port-forward to be completed'
-}
-
 # Preparing tenant for bucket manipulation
 # shellcheck disable=SC2317
 function bootstrap_tenant() {
-  port_forward
+  port_forward $namespace $tenant minio $localport
 
   # Obtain root credentials
   TENANT_CONFIG_SECRET=$(kubectl -n $namespace get tenants $tenant -o jsonpath="{.spec.configuration.name}")
@@ -106,7 +70,7 @@ function bootstrap_tenant() {
 
 # Upload dummy data to tenant bucket
 function upload_dummy_data() {
-  port_forward
+  port_forward $namespace $tenant minio $localport
 
   echo "Uploading dummy data to tenant bucket"
   cp ${SCRIPT_DIR}/deploy-tenant-upgrade.sh ${SCRIPT_DIR}/$dummy
@@ -115,7 +79,7 @@ function upload_dummy_data() {
 
 # Download dummy data from tenant bucket
 function download_dummy_data() {
-  port_forward
+  port_forward $namespace $tenant minio $localport
 
   echo "Download dummy data from tenant bucket"
   mc cp $alias/$bucket/$dummy ${SCRIPT_DIR}/$dummy --insecure
@@ -135,7 +99,7 @@ function main() {
 
   setup_kind
 
-  error=$( {
+  output=$( {
     if [ -n "$lower_version" ]
     then
       # Test specific version of operator
@@ -146,13 +110,10 @@ function main() {
     fi
   } 2>&1 )
   
-  echo "$error"
-  if [ -n "$error" ]
-  then
-    install_operator
-  fi
+  echo "$output"
 
-  install_tenant
+  echo "Installing tenant: $lower_version"
+  install_tenant $lower_version
 
   bootstrap_tenant
 
@@ -166,6 +127,10 @@ function main() {
     # Test current branch
     install_operator
   fi
+
+  # After opreator upgrade, there's a rolling restart
+  echo "Waiting for rolling restart to complete"
+  kubectl -n tenant-lite rollout status sts/storage-lite-pool-0
   
   check_tenant_status tenant-lite storage-lite
 
diff --git a/testing/deploy-tenant.sh b/testing/deploy-tenant.sh
index 8e1fcdfdd6b..22c3aa2e8af 100755
--- a/testing/deploy-tenant.sh
+++ b/testing/deploy-tenant.sh
@@ -31,18 +31,7 @@ function main() {
 
     check_tenant_status tenant-lite storage-lite
 
-    # To allow the execution without killing the cluster at the end of the test
-    # Use below statement to automatically test and kill cluster at the end:
-    # `unset OPERATOR_DEV_TEST`
-    # Use below statement to test and keep cluster alive at the end!:
-    # `export OPERATOR_DEV_TEST="ON"`
-    if [[ -z "${OPERATOR_DEV_TEST}" ]]; then
-        # OPERATOR_DEV_TEST is not defined, hence destroy_kind
-        echo "Cluster will be destroyed for automated testing"
-        destroy_kind
-    else
-        echo "Cluster will remain alive for manual testing"
-    fi
+    destroy_kind
 }
 
 main "$@"
diff --git a/testing/tenant-logs/kustomization.yaml b/testing/tenant-logs/kustomization.yaml
new file mode 100644
index 00000000000..cd7a79e761e
--- /dev/null
+++ b/testing/tenant-logs/kustomization.yaml
@@ -0,0 +1,9 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+
+resources:
+  - ../../examples/kustomization/tenant-lite
+
+patchesStrategicMerge:
+  - tenant.yaml
+
diff --git a/testing/tenant-logs/tenant.yaml b/testing/tenant-logs/tenant.yaml
new file mode 100644
index 00000000000..527fd9c51ca
--- /dev/null
+++ b/testing/tenant-logs/tenant.yaml
@@ -0,0 +1,30 @@
+apiVersion: minio.min.io/v2
+kind: Tenant
+metadata:
+  name: storage
+  namespace: minio-tenant
+spec:
+  log:
+    image: minio/operator:noop
+    audit:
+      diskCapacityGB: 1
+    ## Postgres setup for LogSearch API
+    db:
+      volumeClaimTemplate:
+        spec:
+          storageClassName: standard
+          accessModes:
+            - ReadWriteOnce
+          resources:
+            requests:
+              storage: 1Gi
+      securityContext:
+        runAsUser: 999
+        runAsGroup: 999
+        runAsNonRoot: true
+        fsGroup: 999
+    securityContext:
+      runAsUser: 1000
+      runAsGroup: 1000
+      runAsNonRoot: true
+      fsGroup: 1000
diff --git a/testing/tenant-prometheus/kustomization.yaml b/testing/tenant-prometheus/kustomization.yaml
new file mode 100644
index 00000000000..cd7a79e761e
--- /dev/null
+++ b/testing/tenant-prometheus/kustomization.yaml
@@ -0,0 +1,9 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+
+resources:
+  - ../../examples/kustomization/tenant-lite
+
+patchesStrategicMerge:
+  - tenant.yaml
+
diff --git a/testing/tenant-prometheus/tenant.yaml b/testing/tenant-prometheus/tenant.yaml
new file mode 100644
index 00000000000..ed735c938f9
--- /dev/null
+++ b/testing/tenant-prometheus/tenant.yaml
@@ -0,0 +1,27 @@
+apiVersion: minio.min.io/v2
+kind: Tenant
+metadata:
+  name: storage
+  namespace: minio-tenant
+spec:
+  prometheus:
+    image: "" # defaults to quay.io/prometheus/prometheus:latest
+    env: [ ]
+    sidecarimage: "" # defaults to alpine
+    initimage: "" # defaults to busybox:1.33.1
+    diskCapacityGB: 1
+    storageClassName: standard
+    annotations: { }
+    labels: { }
+    nodeSelector: { }
+    affinity:
+      nodeAffinity: { }
+      podAffinity: { }
+      podAntiAffinity: { }
+    resources: { }
+    serviceAccountName: ""
+    securityContext:
+      runAsUser: 1000
+      runAsGroup: 1000
+      runAsNonRoot: true
+      fsGroup: 1000
diff --git a/testing/tenant/kustomization.yaml b/testing/tenant/kustomization.yaml
index cd7a79e761e..c5127779718 100644
--- a/testing/tenant/kustomization.yaml
+++ b/testing/tenant/kustomization.yaml
@@ -3,7 +3,3 @@ kind: Kustomization
 
 resources:
   - ../../examples/kustomization/tenant-lite
-
-patchesStrategicMerge:
-  - tenant.yaml
-
diff --git a/testing/tenant/tenant.yaml b/testing/tenant/tenant.yaml
deleted file mode 100644
index 9f42c5fea8e..00000000000
--- a/testing/tenant/tenant.yaml
+++ /dev/null
@@ -1,8 +0,0 @@
-apiVersion: minio.min.io/v2
-kind: Tenant
-metadata:
-  name: storage
-  namespace: minio-tenant
-spec:
-  log:
-    image: minio/operator:noop