Repository: rook/rook
Pull request #8584 (2021-08-24 13:44:33)

core: convert util.NewSet() to sets.NewString()
Converting util.NewSet() instance to use sets.NewString() instance

Closes: https://github.com/rook/rook/issues/8479
Signed-off-by: parth-gr <paarora@redhat.com>

<!-- Please take a look at our [Contributing](https://rook.io/docs/rook/master/development-flow.html)
documentation before submitting a Pull Request!
Thank you for contributing to Rook! -->

**Description of your changes:**

**Which issue is resolved by this Pull Request:**
Resolves #

**Checklist:**

- [ ] **Commit Message Formatting**: Commit titles and messages follow guidelines in the [developer guide](https://rook.io/docs/rook/master/development-flow.html#commit-structure).
- [ ] **Skip Tests for Docs**: Add the flag for skipping the build if this is only a documentation change. See [here](https://github.com/rook/rook/blob/master/INSTALL.md#skip-ci) for the flag.
- [ ] **Skip Unrelated Tests**: Add a flag to run tests for a specific storage provider. See [test options](https://github.com/rook/rook/blob/master/INSTALL.md#test-storage-provider).
- [ ] Reviewed the developer guide on [Submitting a Pull Request](https://rook.io/docs/rook/master/development-flow.html#submitting-a-pull-request)
- [ ] Documentation has been updated, if necessary.
- [ ] Unit tests have been added, if necessary.
- [ ] Integration tests have been added, if necessary.
- [ ] Pending release notes updated with breaking and/or notable changes, if necessary.
- [ ] Upgrade from previous release is tested and upgrade user guide is updated, if necessary.
- [ ] Code generation (`make codegen`) has been run to update object specifications, if necessary.


parth-gr: By the changes for `createConfig` string Instance
The Unit Test fails by throwing an error `failed to get node node23 orchestration status, will try again: configmaps "rook-ceph-osd-node23-status" not found`

It is because `Config Map` is not being returned from this call.
 https://github.com/rook/rook/blob/6582bac8e71240de85b3f4d6cdcda3572f665559/pkg/operator/ceph/cluster/osd/status_test.go#L79
BlaineEXE: @parth-gr In response to your question, I would suggest debugging the test and stepping through the code to understand the behavior. In VisualStudio Code with the Go extension installed, there should be options to run tests hovering over test definitions. When A test is failing unexpectedly, I use `debug test` to investigate more deeply.

parth-gr: > Seems straightforward and LGTM. One thing though: should we remove the `pkg/util.set.go` file/code?

yes we are doing that.

Patch:
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 0a42af3c6..c55ae1524 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -57,29 +57,24 @@ jobs:
 
   linux-build-all:
     runs-on: ubuntu-18.04
-    strategy:
-      fail-fast: false
-      matrix:
-        go-version : ['1.16', '1.17']
     steps:
     - name: checkout
       uses: actions/checkout@v2
       with:
         fetch-depth: 0
 
-    - name: setup golang ${{ matrix.go-version }}
+    - name: setup golang
       uses: actions/setup-go@v2
       with:
-        go-version: ${{ matrix.go-version }}
+        go-version: 1.16
 
     - name: set up QEMU
       uses: docker/setup-qemu-action@master
       with:
         platforms: all
 
-    - name: build.all rook with go ${{ matrix.go-version }}
-      run: |
-        tests/scripts/github-action-helper.sh build_rook_all
+    - name: build.all rook
+      run: tests/scripts/github-action-helper.sh build_rook_all
 
     - name: setup tmate session for debugging
       if: failure()
diff --git a/build/crds/crds.go b/build/crds/crds.go
index 01bee7092..1bcaee88b 100644
--- a/build/crds/crds.go
+++ b/build/crds/crds.go
@@ -1,4 +1,3 @@
-//go:build crds
 // +build crds
 
 /*
diff --git a/build/makelib/golang.mk b/build/makelib/golang.mk
index 6f9b12f53..38296d70c 100644
--- a/build/makelib/golang.mk
+++ b/build/makelib/golang.mk
@@ -48,7 +48,7 @@ GO_TEST_FLAGS ?=
 # ====================================================================================
 # Setup go environment
 
-GO_SUPPORTED_VERSIONS ?= 1.16|1.17
+GO_SUPPORTED_VERSIONS ?= 1.16
 
 GO_PACKAGES := $(foreach t,$(GO_SUBDIRS),$(GO_PROJECT)/$(t)/...)
 GO_INTEGRATION_TEST_PACKAGES := $(foreach t,$(GO_INTEGRATION_TESTS_SUBDIRS),$(GO_PROJECT)/$(t)/integration)
@@ -155,11 +155,8 @@ go.vet:
 	@CGO_ENABLED=0 $(GOHOST) vet $(GO_COMMON_FLAGS) $(GO_PACKAGES) $(GO_INTEGRATION_TEST_PACKAGES)
 
 .PHONY: go.fmt
-# ignore deepcopy generated files since the tool hardcoded the header with a "// +build" which in Golang 1.17 makes it fail gofmt since "////go:build" is preferred
-# see: https://github.com/kubernetes/gengo/blob/master/examples/deepcopy-gen/generators/deepcopy.go#L136
-# https://github.com/kubernetes/gengo/pull/210
 go.fmt: $(GOFMT)
-	@gofmt_out=$$(find $(GO_SUBDIRS) -name "*.go" -not -name "*.deepcopy.go" | xargs $(GOFMT) -s -d -e 2>&1) && [ -z "$${gofmt_out}" ] || (echo "$${gofmt_out}" 1>&2; exit 1)
+	@gofmt_out=$$($(GOFMT) -s -d -e $(GO_SUBDIRS) $(GO_INTEGRATION_TESTS_SUBDIRS) 2>&1) && [ -z "$${gofmt_out}" ] || (echo "$${gofmt_out}" 1>&2; exit 1)
 
 go.validate: go.vet go.fmt
 
diff --git a/cluster/charts/rook-ceph/templates/clusterrole.yaml b/cluster/charts/rook-ceph/templates/clusterrole.yaml
index d9565da51..2218ec33d 100644
--- a/cluster/charts/rook-ceph/templates/clusterrole.yaml
+++ b/cluster/charts/rook-ceph/templates/clusterrole.yaml
@@ -471,3 +471,12 @@ rules:
   verbs:
   - use
 {{- end }}
+---
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: rook-ceph-admission-controller-role
+rules:
+  - apiGroups: ["ceph.rook.io"]
+    resources: ["*"]
+    verbs: ["get", "watch", "list"]
diff --git a/cluster/charts/rook-ceph/templates/clusterrolebinding.yaml b/cluster/charts/rook-ceph/templates/clusterrolebinding.yaml
index d3c22ca74..36d539a25 100644
--- a/cluster/charts/rook-ceph/templates/clusterrolebinding.yaml
+++ b/cluster/charts/rook-ceph/templates/clusterrolebinding.yaml
@@ -260,6 +260,20 @@ subjects:
     name: rook-ceph-mgr
     namespace: {{ .Release.Namespace }}
 ---
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: rook-ceph-admission-controller-rolebinding
+subjects:
+  - kind: ServiceAccount
+    name: rook-ceph-admission-controller
+    apiGroup: ""
+    namespace: {{ .Release.Namespace }}
+roleRef:
+  kind: ClusterRole
+  name: rook-ceph-admission-controller-role
+  apiGroup: rbac.authorization.k8s.io
+---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: RoleBinding
 metadata:
diff --git a/cluster/charts/rook-ceph/templates/deployment.yaml b/cluster/charts/rook-ceph/templates/deployment.yaml
index 7efb599eb..c6402c129 100644
--- a/cluster/charts/rook-ceph/templates/deployment.yaml
+++ b/cluster/charts/rook-ceph/templates/deployment.yaml
@@ -313,6 +313,20 @@ spec:
           valueFrom:
             fieldRef:
               fieldPath: metadata.namespace
+{{- if .Values.admissionController }}
+{{- if .Values.admissionController.tolerations }}
+        - name: ADMISSION_CONTROLLER_TOLERATIONS
+          value: {{ toYaml .Values.admissionController.tolerations | quote }}
+{{- end }}
+{{- if .Values.admissionController.nodeAffinity }}
+        - name: ADMISSION_CONTROLLER_NODE_AFFINITY
+          value: {{ .Values.admissionController.nodeAffinity }}
+{{- end }}
+{{- end }}
+{{- if .Values.unreachableNodeTolerationSeconds }}
+        - name: ROOK_UNREACHABLE_NODE_TOLERATION_SECONDS
+          value: {{ .Values.unreachableNodeTolerationSeconds | quote }}
+{{- end }}
 {{- if .Values.resources }}
         resources:
 {{ toYaml .Values.resources | indent 10 }}
diff --git a/cluster/charts/rook-ceph/templates/serviceaccount.yaml b/cluster/charts/rook-ceph/templates/serviceaccount.yaml
index f9317cb0a..5d3687f67 100644
--- a/cluster/charts/rook-ceph/templates/serviceaccount.yaml
+++ b/cluster/charts/rook-ceph/templates/serviceaccount.yaml
@@ -77,6 +77,12 @@ metadata:
   namespace:  {{  .Release.Namespace }}
 {{ template "imagePullSecrets" . }}
 ---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: rook-ceph-admission-controller
+  namespace: {{  .Release.Namespace }}
+---
 # Service account for the purge osd job
 apiVersion: v1
 kind: ServiceAccount
@@ -84,4 +90,4 @@ metadata:
   name: rook-ceph-purge-osd
   namespace: {{ .Release.Namespace }}
 {{ template "imagePullSecrets" . }}
----
+---
\ No newline at end of file
diff --git a/cluster/examples/kubernetes/ceph/common.yaml b/cluster/examples/kubernetes/ceph/common.yaml
index 15ee9a612..aed387d5e 100644
--- a/cluster/examples/kubernetes/ceph/common.yaml
+++ b/cluster/examples/kubernetes/ceph/common.yaml
@@ -32,6 +32,35 @@ subjects:
 # OLM: END OBJECTBUCKET ROLEBINDING
 # OLM: BEGIN OPERATOR ROLE
 ---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: rook-ceph-admission-controller
+  namespace: rook-ceph # namespace:operator
+---
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: rook-ceph-admission-controller-role
+rules:
+  - apiGroups: ["ceph.rook.io"]
+    resources: ["*"]
+    verbs: ["get", "watch", "list"]
+---
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: rook-ceph-admission-controller-rolebinding
+subjects:
+  - kind: ServiceAccount
+    name: rook-ceph-admission-controller
+    apiGroup: ""
+    namespace: rook-ceph # namespace:operator
+roleRef:
+  kind: ClusterRole
+  name: rook-ceph-admission-controller-role
+  apiGroup: rbac.authorization.k8s.io
+---
 # The cluster role for managing all the cluster-specific resources in a namespace
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
diff --git a/pkg/operator/ceph/csi/template/cephfs/csi-cephfsplugin-provisioner-dep.yaml b/cluster/examples/kubernetes/ceph/csi/template/cephfs/csi-cephfsplugin-provisioner-dep.yaml
similarity index 100%
rename from pkg/operator/ceph/csi/template/cephfs/csi-cephfsplugin-provisioner-dep.yaml
rename to cluster/examples/kubernetes/ceph/csi/template/cephfs/csi-cephfsplugin-provisioner-dep.yaml
diff --git a/pkg/operator/ceph/csi/template/cephfs/csi-cephfsplugin-svc.yaml b/cluster/examples/kubernetes/ceph/csi/template/cephfs/csi-cephfsplugin-svc.yaml
similarity index 100%
rename from pkg/operator/ceph/csi/template/cephfs/csi-cephfsplugin-svc.yaml
rename to cluster/examples/kubernetes/ceph/csi/template/cephfs/csi-cephfsplugin-svc.yaml
diff --git a/pkg/operator/ceph/csi/template/cephfs/csi-cephfsplugin.yaml b/cluster/examples/kubernetes/ceph/csi/template/cephfs/csi-cephfsplugin.yaml
similarity index 100%
rename from pkg/operator/ceph/csi/template/cephfs/csi-cephfsplugin.yaml
rename to cluster/examples/kubernetes/ceph/csi/template/cephfs/csi-cephfsplugin.yaml
diff --git a/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin-provisioner-dep.yaml b/cluster/examples/kubernetes/ceph/csi/template/rbd/csi-rbdplugin-provisioner-dep.yaml
similarity index 100%
rename from pkg/operator/ceph/csi/template/rbd/csi-rbdplugin-provisioner-dep.yaml
rename to cluster/examples/kubernetes/ceph/csi/template/rbd/csi-rbdplugin-provisioner-dep.yaml
diff --git a/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin-svc.yaml b/cluster/examples/kubernetes/ceph/csi/template/rbd/csi-rbdplugin-svc.yaml
similarity index 100%
rename from pkg/operator/ceph/csi/template/rbd/csi-rbdplugin-svc.yaml
rename to cluster/examples/kubernetes/ceph/csi/template/rbd/csi-rbdplugin-svc.yaml
diff --git a/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin.yaml b/cluster/examples/kubernetes/ceph/csi/template/rbd/csi-rbdplugin.yaml
similarity index 100%
rename from pkg/operator/ceph/csi/template/rbd/csi-rbdplugin.yaml
rename to cluster/examples/kubernetes/ceph/csi/template/rbd/csi-rbdplugin.yaml
diff --git a/cluster/examples/kubernetes/ceph/operator-openshift.yaml b/cluster/examples/kubernetes/ceph/operator-openshift.yaml
index 930554d0f..5e7ce8d30 100644
--- a/cluster/examples/kubernetes/ceph/operator-openshift.yaml
+++ b/cluster/examples/kubernetes/ceph/operator-openshift.yaml
@@ -405,6 +405,18 @@ data:
   # The timeout value (in seconds) of Ceph commands. It should be >= 1. If this variable is not set or is an invalid value, it's default to 15.
   ROOK_CEPH_COMMANDS_TIMEOUT_SECONDS: "15"
   # CSI_VOLUME_REPLICATION_IMAGE: "quay.io/csiaddons/volumereplication-operator:v0.1.0"
+
+  # (Optional) Admission controller NodeAffinity.
+  # ADMISSION_CONTROLLER_NODE_AFFINITY: "role=storage-node; storage=rook, ceph"
+  # (Optional) Admission controller tolerations list. Put here list of taints you want to tolerate in YAML format.
+  # Admission controller would be best to start on the same nodes as other ceph daemons.
+  # ADMISSION_CONTROLLER_TOLERATIONS: |
+  #   - effect: NoSchedule
+  #     key: node-role.kubernetes.io/controlplane
+  #     operator: Exists
+  #   - effect: NoExecute
+  #     key: node-role.kubernetes.io/etcd
+  #     operator: Exists
 ---
 # The deployment for the rook operator
 # OLM: BEGIN OPERATOR DEPLOYMENT
@@ -436,12 +448,6 @@ spec:
               name: rook-config
             - mountPath: /etc/ceph
               name: default-config-dir
-            - mountPath: /etc/webhook
-              name: webhook-cert
-          ports:
-            - containerPort: 9443
-              name: https-webhook
-              protocol: TCP
           env:
             - name: ROOK_CURRENT_NAMESPACE_ONLY
               value: "false"
@@ -571,6 +577,4 @@ spec:
           emptyDir: {}
         - name: default-config-dir
           emptyDir: {}
-        - name: webhook-cert
-          emptyDir: {}
 # OLM: END OPERATOR DEPLOYMENT
diff --git a/cluster/examples/kubernetes/ceph/operator.yaml b/cluster/examples/kubernetes/ceph/operator.yaml
index 0a3d70594..3393c2237 100644
--- a/cluster/examples/kubernetes/ceph/operator.yaml
+++ b/cluster/examples/kubernetes/ceph/operator.yaml
@@ -329,6 +329,18 @@ data:
   # Enable volume replication controller
   CSI_ENABLE_VOLUME_REPLICATION: "false"
   # CSI_VOLUME_REPLICATION_IMAGE: "quay.io/csiaddons/volumereplication-operator:v0.1.0"
+
+  # (Optional) Admission controller NodeAffinity.
+  # ADMISSION_CONTROLLER_NODE_AFFINITY: "role=storage-node; storage=rook, ceph"
+  # (Optional) Admission controller tolerations list. Put here list of taints you want to tolerate in YAML format.
+  # Admission controller would be best to start on the same nodes as other ceph daemons.
+  # ADMISSION_CONTROLLER_TOLERATIONS: |
+  #   - effect: NoSchedule
+  #     key: node-role.kubernetes.io/controlplane
+  #     operator: Exists
+  #   - effect: NoExecute
+  #     key: node-role.kubernetes.io/etcd
+  #     operator: Exists
 ---
 # OLM: BEGIN OPERATOR DEPLOYMENT
 apiVersion: apps/v1
@@ -359,12 +371,6 @@ spec:
               name: rook-config
             - mountPath: /etc/ceph
               name: default-config-dir
-            - mountPath: /etc/webhook
-              name: webhook-cert
-          ports:
-            - containerPort: 9443
-              name: https-webhook
-              protocol: TCP
           env:
             # If the operator should only watch for cluster CRDs in the same namespace, set this to "true".
             # If this is not set to true, the operator will watch for cluster CRDs in all namespaces.
@@ -516,6 +522,4 @@ spec:
           emptyDir: {}
         - name: default-config-dir
           emptyDir: {}
-        - name: webhook-cert
-          emptyDir: {}
 # OLM: END OPERATOR DEPLOYMENT
diff --git a/cmd/rook/ceph/admission.go b/cmd/rook/ceph/admission.go
new file mode 100644
index 000000000..8f7945e4e
--- /dev/null
+++ b/cmd/rook/ceph/admission.go
@@ -0,0 +1,43 @@
+/*
+Copyright 2020 The Rook Authors. All rights reserved.
+
+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 ceph
+
+import (
+	"github.com/rook/rook/cmd/rook/rook"
+	operator "github.com/rook/rook/pkg/operator/ceph"
+	"github.com/spf13/cobra"
+)
+
+var (
+	admissionCmd = &cobra.Command{
+		Use:   "admission-controller",
+		Short: "Starts admission controller",
+	}
+)
+
+func init() {
+	admissionCmd.Run = startAdmissionController
+}
+
+func startAdmissionController(cmd *cobra.Command, args []string) {
+	rook.SetLogLevel()
+	rook.LogStartupInfo(admissionCmd.Flags())
+	err := operator.StartAdmissionController()
+	if err != nil {
+		rook.TerminateFatal(err)
+	}
+}
diff --git a/cmd/rook/ceph/ceph.go b/cmd/rook/ceph/ceph.go
index 705ac0b1b..56a18828f 100644
--- a/cmd/rook/ceph/ceph.go
+++ b/cmd/rook/ceph/ceph.go
@@ -58,6 +58,7 @@ func init() {
 	Cmd.AddCommand(cleanUpCmd,
 		operatorCmd,
 		agentCmd,
+		admissionCmd,
 		osdCmd,
 		mgrCmd,
 		configCmd)
diff --git a/cmd/rook/ceph/operator.go b/cmd/rook/ceph/operator.go
index 3407a1a7f..e17db78f4 100644
--- a/cmd/rook/ceph/operator.go
+++ b/cmd/rook/ceph/operator.go
@@ -22,6 +22,7 @@ import (
 	"github.com/rook/rook/pkg/daemon/ceph/agent/flexvolume/attachment"
 	operator "github.com/rook/rook/pkg/operator/ceph"
 	cluster "github.com/rook/rook/pkg/operator/ceph/cluster"
+	"github.com/rook/rook/pkg/operator/ceph/csi"
 
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
@@ -41,6 +42,14 @@ https://github.com/rook/rook`,
 }
 
 func init() {
+	// csi deployment templates
+	operatorCmd.Flags().StringVar(&csi.RBDPluginTemplatePath, "csi-rbd-plugin-template-path", csi.DefaultRBDPluginTemplatePath, "path to ceph-csi rbd plugin template")
+
+	operatorCmd.Flags().StringVar(&csi.RBDProvisionerDepTemplatePath, "csi-rbd-provisioner-dep-template-path", csi.DefaultRBDProvisionerDepTemplatePath, "path to ceph-csi rbd provisioner deployment template")
+
+	operatorCmd.Flags().StringVar(&csi.CephFSPluginTemplatePath, "csi-cephfs-plugin-template-path", csi.DefaultCephFSPluginTemplatePath, "path to ceph-csi cephfs plugin template")
+	operatorCmd.Flags().StringVar(&csi.CephFSProvisionerDepTemplatePath, "csi-cephfs-provisioner-dep-template-path", csi.DefaultCephFSProvisionerDepTemplatePath, "path to ceph-csi cephfs provisioner deployment template")
+
 	operatorCmd.Flags().BoolVar(&cluster.EnableMachineDisruptionBudget, "enable-machine-disruption-budget", false, "enable fencing controllers")
 
 	flags.SetFlagsFromEnv(operatorCmd.Flags(), rook.RookEnvVarPrefix)
@@ -75,7 +84,7 @@ func startOperator(cmd *cobra.Command, args []string) error {
 	op := operator.New(context, volumeAttachment, rookImage, serviceAccountName)
 	err = op.Run()
 	if err != nil {
-		rook.TerminateFatal(errors.Wrap(err, "failed to run operator"))
+		rook.TerminateFatal(errors.Wrap(err, "failed to run operator\n"))
 	}
 
 	return nil
diff --git a/go.mod b/go.mod
index 65881bb00..94b76473f 100644
--- a/go.mod
+++ b/go.mod
@@ -15,7 +15,7 @@ require (
 	github.com/google/uuid v1.1.2
 	github.com/hashicorp/vault/api v1.0.5-0.20200902155336-f9d5ce5a171a
 	github.com/k8snetworkplumbingwg/network-attachment-definition-client v1.1.0
-	github.com/kube-object-storage/lib-bucket-provisioner v0.0.0-20210818162813-3eee31c01875
+	github.com/kube-object-storage/lib-bucket-provisioner v0.0.0-20210311161930-4bea5edaff58
 	github.com/libopenstorage/secrets v0.0.0-20210709082113-dde442ea20ec
 	github.com/openshift/cluster-api v0.0.0-20191129101638-b09907ac6668
 	github.com/openshift/machine-api-operator v0.2.1-0.20190903202259-474e14e4965a
diff --git a/go.sum b/go.sum
index fef8d50ba..5acd43305 100644
--- a/go.sum
+++ b/go.sum
@@ -719,8 +719,8 @@ github.com/kr/pty v1.1.5/go.mod h1:9r2w37qlBe7rQ6e1fg1S/9xpWHSnaqNdHD3WcMdbPDA=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
-github.com/kube-object-storage/lib-bucket-provisioner v0.0.0-20210818162813-3eee31c01875 h1:jX3VXgmNOye8XYKjwcTVXcBYcPv3jj657fwX8DN/HiM=
-github.com/kube-object-storage/lib-bucket-provisioner v0.0.0-20210818162813-3eee31c01875/go.mod h1:XpQ9HGG9uF5aJCBP+s6w5kSiyTIVSqCV8+XAE4qms5E=
+github.com/kube-object-storage/lib-bucket-provisioner v0.0.0-20210311161930-4bea5edaff58 h1:O9m6tfyhjr3F3yKGsgGMp1+seFyTRovmGMmbcNp6GSo=
+github.com/kube-object-storage/lib-bucket-provisioner v0.0.0-20210311161930-4bea5edaff58/go.mod h1:XpQ9HGG9uF5aJCBP+s6w5kSiyTIVSqCV8+XAE4qms5E=
 github.com/kubernetes-csi/csi-lib-utils v0.9.1 h1:sGq6ifVujfMSkfTsMZip44Ttv8SDXvsBlFk9GdYl/b8=
 github.com/kubernetes-csi/csi-lib-utils v0.9.1/go.mod h1:8E2jVUX9j3QgspwHXa6LwyN7IHQDjW9jX3kwoWnSC+M=
 github.com/kubernetes-csi/external-snapshotter/client/v4 v4.0.0/go.mod h1:YBCo4DoEeDndqvAn6eeu0vWM7QdXmHEeI9cFWplmBys=
diff --git a/images/ceph/Dockerfile b/images/ceph/Dockerfile
index 53f26151d..5e66b9bbb 100644
--- a/images/ceph/Dockerfile
+++ b/images/ceph/Dockerfile
@@ -23,6 +23,7 @@ RUN curl --fail -sSL -o /tini https://github.com/krallin/tini/releases/download/
     chmod +x /tini
 
 COPY rook rookflex toolbox.sh set-ceph-debug-level /usr/local/bin/
+COPY ceph-csi /etc/ceph-csi
 COPY ceph-monitoring /etc/ceph-monitoring
 COPY rook-external /etc/rook-external/
 COPY ceph-csv-templates /etc/ceph-csv-templates
diff --git a/images/ceph/Makefile b/images/ceph/Makefile
index eea0bb3f0..267fa68e8 100755
--- a/images/ceph/Makefile
+++ b/images/ceph/Makefile
@@ -60,6 +60,7 @@ do.build:
 	@cp set-ceph-debug-level $(TEMP)
 	@cp $(OUTPUT_DIR)/bin/linux_$(GOARCH)/rook $(TEMP)
 	@cp $(OUTPUT_DIR)/bin/linux_$(GOARCH)/rookflex $(TEMP)
+	@cp -r ../../cluster/examples/kubernetes/ceph/csi/template $(TEMP)/ceph-csi
 	@cp -r ../../cluster/examples/kubernetes/ceph/monitoring $(TEMP)/ceph-monitoring
 	@mkdir -p $(TEMP)/rook-external/test-data
 	@cp ../../cluster/examples/kubernetes/ceph/create-external-cluster-resources.* $(TEMP)/rook-external/
diff --git a/pkg/operator/ceph/cluster/operator_watchers.go b/pkg/operator/ceph/cluster/operator_watchers.go
index f76765d45..ec3ca60c7 100644
--- a/pkg/operator/ceph/cluster/operator_watchers.go
+++ b/pkg/operator/ceph/cluster/operator_watchers.go
@@ -27,8 +27,6 @@ import (
 )
 
 // StartOperatorSettingsWatch starts the operator settings watcher
-// TODO: use controller runtime, so we can use a Context to cancel the watch instead of using another channel
-// The cache package mentions that it'd be nice to use a Context for cancellation too
 func (c *ClusterController) StartOperatorSettingsWatch(stopCh chan struct{}) {
 	operatorNamespace := os.Getenv(k8sutil.PodNamespaceEnvVar)
 	// watch for "rook-ceph-operator-config" ConfigMap
diff --git a/pkg/operator/ceph/cluster/osd/create.go b/pkg/operator/ceph/cluster/osd/create.go
index 30b200a1f..eaba45506 100644
--- a/pkg/operator/ceph/cluster/osd/create.go
+++ b/pkg/operator/ceph/cluster/osd/create.go
@@ -25,17 +25,17 @@ import (
 	osdconfig "github.com/rook/rook/pkg/operator/ceph/cluster/osd/config"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
-	"github.com/rook/rook/pkg/util"
 	v1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/version"
 )
 
 type createConfig struct {
 	cluster                  *Cluster
 	provisionConfig          *provisionConfig
-	awaitingStatusConfigMaps *util.Set      // These status configmaps were created for OSD prepare jobs
-	finishedStatusConfigMaps *util.Set      // Status configmaps are added here as provisioning is completed for them
+	awaitingStatusConfigMaps sets.String    // These status configmaps were created for OSD prepare jobs
+	finishedStatusConfigMaps sets.String    // Status configmaps are added here as provisioning is completed for them
 	deployments              *existenceList // these OSDs have existing deployments
 }
 
@@ -49,27 +49,27 @@ var (
 
 func (c *Cluster) newCreateConfig(
 	provisionConfig *provisionConfig,
-	awaitingStatusConfigMaps *util.Set,
+	awaitingStatusConfigMaps sets.String,
 	deployments *existenceList,
 ) *createConfig {
 	if awaitingStatusConfigMaps == nil {
-		awaitingStatusConfigMaps = util.NewSet()
+		awaitingStatusConfigMaps = sets.NewString()
 	}
 	return &createConfig{
 		c,
 		provisionConfig,
 		awaitingStatusConfigMaps,
-		util.NewSet(),
+		sets.NewString(),
 		deployments,
 	}
 }
 
 func (c *createConfig) progress() (completed, initial int) {
-	return c.finishedStatusConfigMaps.Count(), c.awaitingStatusConfigMaps.Count()
+	return c.finishedStatusConfigMaps.Len(), c.awaitingStatusConfigMaps.Len()
 }
 
 func (c *createConfig) doneCreating() bool {
-	return c.awaitingStatusConfigMaps.Count() == c.finishedStatusConfigMaps.Count()
+	return c.awaitingStatusConfigMaps.Len() == c.finishedStatusConfigMaps.Len()
 }
 
 func (c *createConfig) createNewOSDsFromStatus(
@@ -77,13 +77,13 @@ func (c *createConfig) createNewOSDsFromStatus(
 	nodeOrPVCName string,
 	errs *provisionErrors,
 ) {
-	if !c.awaitingStatusConfigMaps.Contains(statusConfigMapName(nodeOrPVCName)) {
+	if !c.awaitingStatusConfigMaps.Has(statusConfigMapName(nodeOrPVCName)) {
 		// If there is a dangling OSD prepare configmap from another reconcile, don't process it
 		logger.Infof("not creating deployments for OSD prepare results found in ConfigMap %q which was not created for the latest storage spec", statusConfigMapName(nodeOrPVCName))
 		return
 	}
 
-	if c.finishedStatusConfigMaps.Contains(statusConfigMapName(nodeOrPVCName)) {
+	if c.finishedStatusConfigMaps.Has(statusConfigMapName(nodeOrPVCName)) {
 		// If we have already processed this configmap, don't process it again
 		logger.Infof("not creating deployments for OSD prepare results found in ConfigMap %q which was already processed", statusConfigMapName(nodeOrPVCName))
 		return
@@ -115,7 +115,7 @@ func (c *createConfig) createNewOSDsFromStatus(
 
 // Call this if createNewOSDsFromStatus() isn't going to be called (like for a failed status)
 func (c *createConfig) doneWithStatus(nodeOrPVCName string) {
-	c.finishedStatusConfigMaps.Add(statusConfigMapName(nodeOrPVCName))
+	c.finishedStatusConfigMaps.Insert(statusConfigMapName(nodeOrPVCName))
 }
 
 // Returns a set of all the awaitingStatusConfigMaps that will be updated by provisioning jobs.
@@ -124,34 +124,34 @@ func (c *createConfig) doneWithStatus(nodeOrPVCName string) {
 //
 // Creation of prepare jobs is most directly related to creating new OSDs. And we want to keep all
 // usage of awaitingStatusConfigMaps in this file.
-func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig, errs *provisionErrors) (*util.Set, error) {
+func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig, errs *provisionErrors) (sets.String, error) {
 	// Parsing storageClassDeviceSets and parsing it to volume sources
 	c.prepareStorageClassDeviceSets(errs)
 
 	// no valid VolumeSource is ready to run an osd
 	if len(c.deviceSets) == 0 {
 		logger.Info("no storageClassDeviceSets defined to configure OSDs on PVCs")
-		return util.NewSet(), nil
+		return sets.NewString(), nil
 	}
 
 	// Check k8s version
 	k8sVersion, err := k8sutil.GetK8SVersion(c.context.Clientset)
 	if err != nil {
 		errs.addError("failed to provision OSDs on PVCs. user has specified storageClassDeviceSets, but the Kubernetes version could not be determined. minimum Kubernetes version required: 1.13.0. %v", err)
-		return util.NewSet(), nil
+		return sets.NewString(), nil
 	}
 	if !k8sVersion.AtLeast(version.MustParseSemantic("v1.13.0")) {
 		errs.addError("failed to provision OSDs on PVCs. user has specified storageClassDeviceSets, but the Kubernetes version is not supported. user must update Kubernetes version. minimum Kubernetes version required: 1.13.0. version detected: %s", k8sVersion.String())
-		return util.NewSet(), nil
+		return sets.NewString(), nil
 	}
 
 	existingDeployments, err := c.getExistingOSDDeploymentsOnPVCs()
 	if err != nil {
 		errs.addError("failed to provision OSDs on PVCs. failed to query existing OSD deployments on PVCs. %v", err)
-		return util.NewSet(), nil
+		return sets.NewString(), nil
 	}
 
-	awaitingStatusConfigMaps := util.NewSet()
+	awaitingStatusConfigMaps := sets.NewString()
 	for _, volume := range c.deviceSets {
 		// Check whether we need to cancel the orchestration
 		if err := opcontroller.CheckForCancelledOrchestration(c.context); err != nil {
@@ -233,7 +233,7 @@ func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig, errs *provi
 		}
 
 		// Skip OSD prepare if deployment already exists for the PVC
-		if existingDeployments.Contains(dataSource.ClaimName) {
+		if existingDeployments.Has(dataSource.ClaimName) {
 			logger.Debugf("skipping OSD prepare job creation for PVC %q because OSD daemon using the PVC already exists", osdProps.crushHostname)
 			continue
 		}
@@ -251,7 +251,7 @@ func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig, errs *provi
 		// record the name of the status configmap that will eventually receive results from the
 		// OSD provisioning job we just created. This will help us determine when we are done
 		// processing the results of provisioning jobs.
-		awaitingStatusConfigMaps.Add(cmName)
+		awaitingStatusConfigMaps.Insert(cmName)
 	}
 
 	return awaitingStatusConfigMaps, nil
@@ -263,10 +263,10 @@ func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig, errs *provi
 //
 // Creation of prepare jobs is most directly related to creating new OSDs. And we want to keep all
 // usage of awaitingStatusConfigMaps in this file.
-func (c *Cluster) startProvisioningOverNodes(config *provisionConfig, errs *provisionErrors) (*util.Set, error) {
+func (c *Cluster) startProvisioningOverNodes(config *provisionConfig, errs *provisionErrors) (sets.String, error) {
 	if !c.spec.Storage.UseAllNodes && len(c.spec.Storage.Nodes) == 0 {
 		logger.Info("no nodes are defined for configuring OSDs on raw devices")
-		return util.NewSet(), nil
+		return sets.NewString(), nil
 	}
 
 	if c.spec.Storage.UseAllNodes {
@@ -278,7 +278,7 @@ func (c *Cluster) startProvisioningOverNodes(config *provisionConfig, errs *prov
 		hostnameMap, err := k8sutil.GetNodeHostNames(c.context.Clientset)
 		if err != nil {
 			errs.addError("failed to provision OSDs on nodes. failed to get node hostnames. %v", err)
-			return util.NewSet(), nil
+			return sets.NewString(), nil
 		}
 		c.spec.Storage.Nodes = nil
 		for _, hostname := range hostnameMap {
@@ -300,15 +300,15 @@ func (c *Cluster) startProvisioningOverNodes(config *provisionConfig, errs *prov
 	// no valid node is ready to run an osd
 	if len(validNodes) == 0 {
 		logger.Warningf("no valid nodes available to run osds on nodes in namespace %q", c.clusterInfo.Namespace)
-		return util.NewSet(), nil
+		return sets.NewString(), nil
 	}
 
 	if len(c.spec.DataDirHostPath) == 0 {
 		errs.addError("failed to provision OSDs on nodes. user has specified valid nodes for storage, but dataDirHostPath is empty. user must set CephCluster dataDirHostPath")
-		return util.NewSet(), nil
+		return sets.NewString(), nil
 	}
 
-	awaitingStatusConfigMaps := util.NewSet()
+	awaitingStatusConfigMaps := sets.NewString()
 	for _, node := range c.ValidStorage.Nodes {
 		// Check whether we need to cancel the orchestration
 		if err := opcontroller.CheckForCancelledOrchestration(c.context); err != nil {
@@ -353,7 +353,7 @@ func (c *Cluster) startProvisioningOverNodes(config *provisionConfig, errs *prov
 		// record the name of the status configmap that will eventually receive results from the
 		// OSD provisioning job we just created. This will help us determine when we are done
 		// processing the results of provisioning jobs.
-		awaitingStatusConfigMaps.Add(cmName)
+		awaitingStatusConfigMaps.Insert(cmName)
 	}
 
 	return awaitingStatusConfigMaps, nil
diff --git a/pkg/operator/ceph/cluster/osd/osd.go b/pkg/operator/ceph/cluster/osd/osd.go
index c131d51ab..50cc160e3 100644
--- a/pkg/operator/ceph/cluster/osd/osd.go
+++ b/pkg/operator/ceph/cluster/osd/osd.go
@@ -37,11 +37,11 @@ import (
 	"github.com/rook/rook/pkg/operator/ceph/controller"
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
-	"github.com/rook/rook/pkg/util"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
 )
 
 var (
@@ -199,21 +199,21 @@ func (c *Cluster) Start() error {
 	updateConfig := c.newUpdateConfig(config, updateQueue, deployments)
 
 	// prepare for creating new OSDs
-	statusConfigMaps := util.NewSet()
+	statusConfigMaps := sets.NewString()
 
 	logger.Info("start provisioning the OSDs on PVCs, if needed")
 	pvcConfigMaps, err := c.startProvisioningOverPVCs(config, errs)
 	if err != nil {
 		return err
 	}
-	statusConfigMaps.AddSet(pvcConfigMaps)
+	statusConfigMaps = statusConfigMaps.Union(pvcConfigMaps)
 
 	logger.Info("start provisioning the OSDs on nodes, if needed")
 	nodeConfigMaps, err := c.startProvisioningOverNodes(config, errs)
 	if err != nil {
 		return err
 	}
-	statusConfigMaps.AddSet(nodeConfigMaps.Copy())
+	statusConfigMaps = statusConfigMaps.Union(nodeConfigMaps)
 
 	createConfig := c.newCreateConfig(config, statusConfigMaps, deployments)
 
@@ -239,7 +239,7 @@ func (c *Cluster) Start() error {
 	return nil
 }
 
-func (c *Cluster) getExistingOSDDeploymentsOnPVCs() (*util.Set, error) {
+func (c *Cluster) getExistingOSDDeploymentsOnPVCs() (sets.String, error) {
 	ctx := context.TODO()
 	listOpts := metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s,%s", k8sutil.AppAttr, AppName, OSDOverPVCLabelKey)}
 
@@ -248,10 +248,10 @@ func (c *Cluster) getExistingOSDDeploymentsOnPVCs() (*util.Set, error) {
 		return nil, errors.Wrap(err, "failed to query existing OSD deployments")
 	}
 
-	result := util.NewSet()
+	result := sets.NewString()
 	for _, deployment := range deployments.Items {
 		if pvcID, ok := deployment.Labels[OSDOverPVCLabelKey]; ok {
-			result.Add(pvcID)
+			result.Insert(pvcID)
 		}
 	}
 
diff --git a/pkg/operator/ceph/cr_manager.go b/pkg/operator/ceph/cr_manager.go
index 1055de8ed..0c51311e2 100644
--- a/pkg/operator/ceph/cr_manager.go
+++ b/pkg/operator/ceph/cr_manager.go
@@ -20,6 +20,7 @@ import (
 	"context"
 
 	"github.com/pkg/errors"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/apis/rook.io/v1alpha2"
 	"github.com/rook/rook/pkg/operator/ceph/cluster"
 	"github.com/rook/rook/pkg/operator/ceph/disruption/controllerconfig"
@@ -27,15 +28,9 @@ import (
 
 	mapiv1 "github.com/openshift/cluster-api/pkg/apis/machine/v1beta1"
 	healthchecking "github.com/openshift/machine-api-operator/pkg/apis/healthchecking/v1alpha1"
-	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
-	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client/config"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
-	"sigs.k8s.io/controller-runtime/pkg/webhook"
-)
-
-const (
-	certDir = "/etc/webhook"
 )
 
 var (
@@ -48,11 +43,7 @@ var (
 	}
 )
 
-var (
-	webhookResources = []webhook.Validator{&cephv1.CephCluster{}, &cephv1.CephBlockPool{}, &cephv1.CephObjectStore{}}
-)
-
-func (o *Operator) startManager(context context.Context, namespaceToWatch string, mgrErrorCh chan error) {
+func (o *Operator) startManager(namespaceToWatch string, context context.Context, mgrErrorCh chan error) {
 	logger.Info("setting up schemes")
 	// Setup Scheme for all resources
 	scheme := runtime.NewScheme()
@@ -69,11 +60,16 @@ func (o *Operator) startManager(context context.Context, namespaceToWatch string
 		LeaderElection: false,
 		Namespace:      namespaceToWatch,
 		Scheme:         scheme,
-		CertDir:        certDir,
 	}
 
 	logger.Info("setting up the controller-runtime manager")
-	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), mgrOpts)
+	kubeConfig, err := config.GetConfig()
+	if err != nil {
+		mgrErrorCh <- errors.Wrap(err, "failed to get client config for controller-runtime manager")
+		return
+	}
+
+	mgr, err := manager.New(kubeConfig, mgrOpts)
 	if err != nil {
 		mgrErrorCh <- errors.Wrap(err, "failed to set up overall controller-runtime manager")
 		return
@@ -81,12 +77,11 @@ func (o *Operator) startManager(context context.Context, namespaceToWatch string
 
 	// options to pass to the controllers
 	controllerOpts := &controllerconfig.Context{
-		RookImage:         o.config.Image,
+		RookImage:         o.rookImage,
 		ClusterdContext:   o.context,
-		OperatorNamespace: o.config.OperatorNamespace,
+		OperatorNamespace: o.operatorNamespace,
 		ReconcileCanaries: &controllerconfig.LockingBool{},
 	}
-
 	// Add the registered controllers to the manager (entrypoint for controllers)
 	err = cluster.AddToManager(mgr, controllerOpts, o.clusterController)
 	if err != nil {
@@ -94,29 +89,6 @@ func (o *Operator) startManager(context context.Context, namespaceToWatch string
 		return
 	}
 
-	// Add Webhooks
-	// TODO: this needs a callback with a watcher for this secret so we can quickly enable the webhook if it changes
-	isPresent, err := isSecretPresent(context, o.context)
-	if err != nil {
-		mgrErrorCh <- errors.Wrap(err, "failed to retrieve admission webhook secret")
-		return
-	}
-	if isPresent {
-		err := createWebhookService(o.context)
-		if err != nil {
-			mgrErrorCh <- errors.Wrap(err, "failed to create admission webhook service")
-			return
-		}
-		logger.Info("setting up admission webhooks")
-		for _, resource := range webhookResources {
-			err = ctrl.NewWebhookManagedBy(mgr).For(resource).Complete()
-			if err != nil {
-				mgrErrorCh <- errors.Wrapf(err, "failed to register webhook for %q", resource.GetObjectKind().GroupVersionKind().Kind)
-				return
-			}
-		}
-	}
-
 	logger.Info("starting the controller-runtime manager")
 	if err := mgr.Start(context); err != nil {
 		mgrErrorCh <- errors.Wrap(err, "failed to run the controller-runtime manager")
diff --git a/pkg/operator/ceph/csi/betav1csidriver.go b/pkg/operator/ceph/csi/betav1csidriver.go
index ec3996e0c..3d33ecd14 100644
--- a/pkg/operator/ceph/csi/betav1csidriver.go
+++ b/pkg/operator/ceph/csi/betav1csidriver.go
@@ -72,8 +72,6 @@ func (d beta1CsiDriver) createCSIDriverInfo(ctx context.Context, clientset kuber
 	// As FSGroupPolicy field is immutable, should be set only during create time.
 	// if the request is to change the FSGroupPolicy, we are deleting the CSIDriver object and creating it.
 	if driver.Spec.FSGroupPolicy != nil && csiDriver.Spec.FSGroupPolicy != nil && *driver.Spec.FSGroupPolicy != *csiDriver.Spec.FSGroupPolicy {
-		d.csiClient = csidrivers
-		d.csiDriver = csiDriver
 		return d.reCreateCSIDriverInfo(ctx)
 	}
 
@@ -90,16 +88,18 @@ func (d beta1CsiDriver) createCSIDriverInfo(ctx context.Context, clientset kuber
 }
 
 func (d beta1CsiDriver) reCreateCSIDriverInfo(ctx context.Context) error {
-	err := d.csiClient.Delete(ctx, d.csiDriver.Name, metav1.DeleteOptions{})
+	csiDriver := d.csiDriver
+	csiClient := d.csiClient
+	err := csiClient.Delete(ctx, csiDriver.Name, metav1.DeleteOptions{})
 	if err != nil {
-		return errors.Wrapf(err, "failed to delete CSIDriver object for driver %q", d.csiDriver.Name)
+		return errors.Wrapf(err, "failed to delete CSIDriver object for driver %q", csiDriver.Name)
 	}
-	logger.Infof("CSIDriver object deleted for driver %q", d.csiDriver.Name)
-	_, err = d.csiClient.Create(ctx, d.csiDriver, metav1.CreateOptions{})
+	logger.Infof("CSIDriver object deleted for driver %q", csiDriver.Name)
+	_, err = csiClient.Create(ctx, csiDriver, metav1.CreateOptions{})
 	if err != nil {
-		return errors.Wrapf(err, "failed to recreate CSIDriver object for driver %q", d.csiDriver.Name)
+		return errors.Wrapf(err, "failed to recreate CSIDriver object for driver %q", csiDriver.Name)
 	}
-	logger.Infof("CSIDriver object recreated for driver %q", d.csiDriver.Name)
+	logger.Infof("CSIDriver object recreated for driver %q", csiDriver.Name)
 	return nil
 }
 
@@ -108,7 +108,7 @@ func (d beta1CsiDriver) deleteCSIDriverInfo(ctx context.Context, clientset kuber
 	err := clientset.StorageV1beta1().CSIDrivers().Delete(ctx, name, metav1.DeleteOptions{})
 	if err != nil {
 		if apierrors.IsNotFound(err) {
-			logger.Debugf("%q CSIDriver not found; skipping deletion.", name)
+			logger.Debug("%q CSIDriver not found; skipping deletion.", name)
 			return nil
 		}
 	}
diff --git a/pkg/operator/ceph/csi/csidriver.go b/pkg/operator/ceph/csi/csidriver.go
index ec373e872..f30267462 100644
--- a/pkg/operator/ceph/csi/csidriver.go
+++ b/pkg/operator/ceph/csi/csidriver.go
@@ -72,8 +72,6 @@ func (d v1CsiDriver) createCSIDriverInfo(ctx context.Context, clientset kubernet
 	// As FSGroupPolicy field is immutable, should be set only during create time.
 	// if the request is to change the FSGroupPolicy, we are deleting the CSIDriver object and creating it.
 	if driver.Spec.FSGroupPolicy != nil && csiDriver.Spec.FSGroupPolicy != nil && *driver.Spec.FSGroupPolicy != *csiDriver.Spec.FSGroupPolicy {
-		d.csiClient = csidrivers
-		d.csiDriver = csiDriver
 		return d.reCreateCSIDriverInfo(ctx)
 	}
 
@@ -90,16 +88,18 @@ func (d v1CsiDriver) createCSIDriverInfo(ctx context.Context, clientset kubernet
 }
 
 func (d v1CsiDriver) reCreateCSIDriverInfo(ctx context.Context) error {
-	err := d.csiClient.Delete(ctx, d.csiDriver.Name, metav1.DeleteOptions{})
+	csiDriver := d.csiDriver
+	csiClient := d.csiClient
+	err := csiClient.Delete(ctx, csiDriver.Name, metav1.DeleteOptions{})
 	if err != nil {
-		return errors.Wrapf(err, "failed to delete CSIDriver object for driver %q", d.csiDriver.Name)
+		return errors.Wrapf(err, "failed to delete CSIDriver object for driver %q", csiDriver.Name)
 	}
-	logger.Infof("CSIDriver object deleted for driver %q", d.csiDriver.Name)
-	_, err = d.csiClient.Create(ctx, d.csiDriver, metav1.CreateOptions{})
+	logger.Infof("CSIDriver object deleted for driver %q", csiDriver.Name)
+	_, err = csiClient.Create(ctx, d.csiDriver, metav1.CreateOptions{})
 	if err != nil {
-		return errors.Wrapf(err, "failed to recreate CSIDriver object for driver %q", d.csiDriver.Name)
+		return errors.Wrapf(err, "failed to recreate CSIDriver object for driver %q", csiDriver.Name)
 	}
-	logger.Infof("CSIDriver object recreated for driver %q", d.csiDriver.Name)
+	logger.Infof("CSIDriver object recreated for driver %q", csiDriver.Name)
 	return nil
 }
 
@@ -108,7 +108,7 @@ func (d v1CsiDriver) deleteCSIDriverInfo(ctx context.Context, clientset kubernet
 	err := clientset.StorageV1().CSIDrivers().Delete(ctx, name, metav1.DeleteOptions{})
 	if err != nil {
 		if apierrors.IsNotFound(err) {
-			logger.Debugf("%s CSIDriver not found; skipping deletion.", name)
+			logger.Debug("%s CSIDriver not found; skipping deletion.", name)
 			return nil
 		}
 	}
diff --git a/pkg/operator/ceph/csi/spec.go b/pkg/operator/ceph/csi/spec.go
index b798c8807..bc3a7bc51 100644
--- a/pkg/operator/ceph/csi/spec.go
+++ b/pkg/operator/ceph/csi/spec.go
@@ -18,7 +18,6 @@ package csi
 
 import (
 	"context"
-	_ "embed"
 	"fmt"
 	"strconv"
 	"strings"
@@ -88,6 +87,13 @@ var (
 	CephFSDriverName string
 	RBDDriverName    string
 
+	// template paths
+	RBDPluginTemplatePath         string
+	RBDProvisionerDepTemplatePath string
+
+	CephFSPluginTemplatePath         string
+	CephFSProvisionerDepTemplatePath string
+
 	// configuration map for csi
 	ConfigName = "rook-ceph-csi-config"
 	ConfigKey  = "csi-cluster-config-json"
@@ -109,22 +115,6 @@ var (
 	DefaultSnapshotterImage       = "k8s.gcr.io/sig-storage/csi-snapshotter:v4.1.1"
 	DefaultResizerImage           = "k8s.gcr.io/sig-storage/csi-resizer:v1.2.0"
 	DefaultVolumeReplicationImage = "quay.io/csiaddons/volumereplication-operator:v0.1.0"
-
-	// Local package template path for RBD
-	//go:embed template/rbd/csi-rbdplugin.yaml
-	RBDPluginTemplatePath string
-	//go:embed template/rbd/csi-rbdplugin-provisioner-dep.yaml
-	RBDProvisionerDepTemplatePath string
-	//go:embed template/rbd/csi-rbdplugin-svc.yaml
-	RBDPluginServiceTemplatePath string
-
-	// Local package template path for CephFS
-	//go:embed template/cephfs/csi-cephfsplugin.yaml
-	CephFSPluginTemplatePath string
-	//go:embed template/cephfs/csi-cephfsplugin-provisioner-dep.yaml
-	CephFSProvisionerDepTemplatePath string
-	//go:embed template/cephfs/csi-cephfsplugin-svc.yaml
-	CephFSPluginServiceTemplatePath string
 )
 
 const (
@@ -164,6 +154,15 @@ const (
 	// kubelet directory path
 	DefaultKubeletDirPath = "/var/lib/kubelet"
 
+	// template
+	DefaultRBDPluginTemplatePath         = "/etc/ceph-csi/rbd/csi-rbdplugin.yaml"
+	DefaultRBDProvisionerDepTemplatePath = "/etc/ceph-csi/rbd/csi-rbdplugin-provisioner-dep.yaml"
+	DefaultRBDPluginServiceTemplatePath  = "/etc/ceph-csi/rbd/csi-rbdplugin-svc.yaml"
+
+	DefaultCephFSPluginTemplatePath         = "/etc/ceph-csi/cephfs/csi-cephfsplugin.yaml"
+	DefaultCephFSProvisionerDepTemplatePath = "/etc/ceph-csi/cephfs/csi-cephfsplugin-provisioner-dep.yaml"
+	DefaultCephFSPluginServiceTemplatePath  = "/etc/ceph-csi/cephfs/csi-cephfsplugin-svc.yaml"
+
 	// grpc metrics and liveness port for cephfs  and rbd
 	DefaultCephFSGRPCMerticsPort     uint16 = 9091
 	DefaultCephFSLivenessMerticsPort uint16 = 9081
@@ -192,6 +191,7 @@ func CSIEnabled() bool {
 }
 
 func validateCSIParam() error {
+
 	if len(CSIParam.CSIPluginImage) == 0 {
 		return errors.New("missing csi rbd plugin image")
 	}
@@ -205,6 +205,23 @@ func validateCSIParam() error {
 		return errors.New("missing csi attacher image")
 	}
 
+	if EnableRBD {
+		if len(RBDPluginTemplatePath) == 0 {
+			return errors.New("missing rbd plugin template path")
+		}
+		if len(RBDProvisionerDepTemplatePath) == 0 {
+			return errors.New("missing rbd provisioner template path")
+		}
+	}
+
+	if EnableCephFS {
+		if len(CephFSPluginTemplatePath) == 0 {
+			return errors.New("missing cephfs plugin template path")
+		}
+		if len(CephFSProvisionerDepTemplatePath) == 0 {
+			return errors.New("missing ceph provisioner template path")
+		}
+	}
 	return nil
 }
 
@@ -413,11 +430,12 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 			return errors.Wrap(err, "failed to load rbd provisioner deployment template")
 		}
 
-		rbdService, err = templateToService("rbd-service", RBDPluginServiceTemplatePath, tp)
+		rbdService, err = templateToService("rbd-service", DefaultRBDPluginServiceTemplatePath, tp)
 		if err != nil {
 			return errors.Wrap(err, "failed to load rbd plugin service template")
 		}
 		rbdService.Namespace = namespace
+		logger.Info("successfully started CSI Ceph RBD")
 	}
 	if EnableCephFS {
 		cephfsPlugin, err = templateToDaemonSet("cephfsplugin", CephFSPluginTemplatePath, tp)
@@ -430,11 +448,12 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 			return errors.Wrap(err, "failed to load rbd provisioner deployment template")
 		}
 
-		cephfsService, err = templateToService("cephfs-service", CephFSPluginServiceTemplatePath, tp)
+		cephfsService, err = templateToService("cephfs-service", DefaultCephFSPluginServiceTemplatePath, tp)
 		if err != nil {
 			return errors.Wrap(err, "failed to load cephfs plugin service template")
 		}
 		cephfsService.Namespace = namespace
+		logger.Info("successfully started CSI CephFS driver")
 	}
 
 	// get common provisioner tolerations and node affinity
@@ -497,7 +516,6 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 			return errors.Wrapf(err, "failed to start rbd provisioner deployment: %+v", rbdProvisionerDeployment)
 		}
 		k8sutil.AddRookVersionLabelToDeployment(rbdProvisionerDeployment)
-		logger.Info("successfully started CSI Ceph RBD driver")
 	}
 
 	if rbdService != nil {
@@ -566,7 +584,6 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 			return errors.Wrapf(err, "failed to start cephfs provisioner deployment: %+v", cephfsProvisionerDeployment)
 		}
 		k8sutil.AddRookVersionLabelToDeployment(cephfsProvisionerDeployment)
-		logger.Info("successfully started CSI CephFS driver")
 	}
 	if cephfsService != nil {
 		err = ownerInfo.SetControllerReference(cephfsService)
diff --git a/pkg/operator/ceph/csi/util.go b/pkg/operator/ceph/csi/util.go
index a79d3fd60..4d3b2bf01 100644
--- a/pkg/operator/ceph/csi/util.go
+++ b/pkg/operator/ceph/csi/util.go
@@ -18,6 +18,8 @@ package csi
 
 import (
 	"bytes"
+	"io/ioutil"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"text/template"
@@ -32,10 +34,15 @@ import (
 	"k8s.io/client-go/kubernetes"
 )
 
-func loadTemplate(name, templateData string, p templateParam) (string, error) {
+func loadTemplate(name, templatePath string, p templateParam) (string, error) {
+	b, err := ioutil.ReadFile(filepath.Clean(templatePath))
+	if err != nil {
+		return "", err
+	}
+	data := string(b)
 	var writer bytes.Buffer
 	t := template.New(name)
-	t, err := t.Parse(templateData)
+	t, err = t.Parse(data)
 	if err != nil {
 		return "", errors.Wrapf(err, "failed to parse template %v", name)
 	}
@@ -43,9 +50,9 @@ func loadTemplate(name, templateData string, p templateParam) (string, error) {
 	return writer.String(), err
 }
 
-func templateToService(name, templateData string, p templateParam) (*corev1.Service, error) {
+func templateToService(name, templatePath string, p templateParam) (*corev1.Service, error) {
 	var svc corev1.Service
-	t, err := loadTemplate(name, templateData, p)
+	t, err := loadTemplate(name, templatePath, p)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to load service template")
 	}
@@ -57,9 +64,9 @@ func templateToService(name, templateData string, p templateParam) (*corev1.Serv
 	return &svc, nil
 }
 
-func templateToDaemonSet(name, templateData string, p templateParam) (*apps.DaemonSet, error) {
+func templateToDaemonSet(name, templatePath string, p templateParam) (*apps.DaemonSet, error) {
 	var ds apps.DaemonSet
-	t, err := loadTemplate(name, templateData, p)
+	t, err := loadTemplate(name, templatePath, p)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to load daemonset template")
 	}
@@ -71,18 +78,18 @@ func templateToDaemonSet(name, templateData string, p templateParam) (*apps.Daem
 	return &ds, nil
 }
 
-func templateToDeployment(name, templateData string, p templateParam) (*apps.Deployment, error) {
-	var dep apps.Deployment
-	t, err := loadTemplate(name, templateData, p)
+func templateToDeployment(name, templatePath string, p templateParam) (*apps.Deployment, error) {
+	var ds apps.Deployment
+	t, err := loadTemplate(name, templatePath, p)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to load deployment template")
 	}
 
-	err = yaml.Unmarshal([]byte(t), &dep)
+	err = yaml.Unmarshal([]byte(t), &ds)
 	if err != nil {
-		return nil, errors.Wrap(err, "failed to unmarshal deployment template")
+		return nil, errors.Wrapf(err, "failed to unmarshal deployment template")
 	}
-	return &dep, nil
+	return &ds, nil
 }
 
 func applyResourcesToContainers(clientset kubernetes.Interface, key string, podspec *corev1.PodSpec) {
diff --git a/pkg/operator/ceph/object/bucket/provisioner.go b/pkg/operator/ceph/object/bucket/provisioner.go
index 81eaeba4a..29bed4651 100644
--- a/pkg/operator/ceph/object/bucket/provisioner.go
+++ b/pkg/operator/ceph/object/bucket/provisioner.go
@@ -554,8 +554,8 @@ func (p *Provisioner) deleteOBCResourceLogError(bucketname string) {
 // Check for additional options mentioned in OBC and set them accordingly
 func (p Provisioner) setAdditionalSettings(options *apibkt.BucketOptions) error {
 	quotaEnabled := true
-	maxObjects := MaxObjectQuota(options.ObjectBucketClaim.Spec.AdditionalConfig)
-	maxSize := MaxSizeQuota(options.ObjectBucketClaim.Spec.AdditionalConfig)
+	maxObjects := MaxObjectQuota(options)
+	maxSize := MaxSizeQuota(options)
 	if maxObjects == "" && maxSize == "" {
 		return nil
 	}
@@ -665,69 +665,3 @@ func (p *Provisioner) setAdminOpsAPIClient() error {
 
 	return nil
 }
-func (p Provisioner) updateAdditionalSettings(ob *bktv1alpha1.ObjectBucket) error {
-	var maxObjectsInt64 int64
-	var maxSizeInt64 int64
-	var err error
-	var quotaEnabled bool
-	maxObjects := MaxObjectQuota(ob.Spec.Endpoint.AdditionalConfigData)
-	maxSize := MaxSizeQuota(ob.Spec.Endpoint.AdditionalConfigData)
-	if maxObjects != "" {
-		maxObjectsInt, err := strconv.Atoi(maxObjects)
-		if err != nil {
-			return errors.Wrap(err, "failed to convert maxObjects to integer")
-		}
-		maxObjectsInt64 = int64(maxObjectsInt)
-	}
-	if maxSize != "" {
-		maxSizeInt64, err = maxSizeToInt64(maxSize)
-		if err != nil {
-			return errors.Wrapf(err, "failed to parse maxSize quota for user %q", p.cephUserName)
-		}
-	}
-	objectUser, err := p.adminOpsClient.GetUser(context.TODO(), admin.User{ID: ob.Spec.Connection.AdditionalState[cephUser]})
-	if err != nil {
-		return errors.Wrapf(err, "failed to fetch user %q", p.cephUserName)
-	}
-	if *objectUser.UserQuota.Enabled &&
-		(maxObjects == "" || maxObjectsInt64 < 0) &&
-		(maxSize == "" || maxSizeInt64 < 0) {
-		quotaEnabled = false
-		err = p.adminOpsClient.SetUserQuota(context.TODO(), admin.QuotaSpec{UID: p.cephUserName, Enabled: &quotaEnabled})
-		if err != nil {
-			return errors.Wrapf(err, "failed to disable quota to user %q", p.cephUserName)
-		}
-		return nil
-	}
-
-	quotaEnabled = true
-	quotaSpec := admin.QuotaSpec{UID: p.cephUserName, Enabled: &quotaEnabled}
-
-	//MaxObject is modified
-	if maxObjects != "" && (maxObjectsInt64 != *objectUser.UserQuota.MaxObjects) {
-		quotaSpec.MaxObjects = &maxObjectsInt64
-	}
-
-	//MaxSize is modified
-	if maxSize != "" && (maxSizeInt64 != *objectUser.UserQuota.MaxSize) {
-		quotaSpec.MaxSize = &maxSizeInt64
-	}
-	err = p.adminOpsClient.SetUserQuota(context.TODO(), quotaSpec)
-	if err != nil {
-		return errors.Wrapf(err, "failed to update quota to user %q", p.cephUserName)
-	}
-
-	return nil
-}
-
-// Update is sent when only there is modification to AdditionalConfig field in OBC
-func (p Provisioner) Update(ob *bktv1alpha1.ObjectBucket) error {
-	logger.Debugf("Update event for OB: %+v", ob)
-
-	err := p.initializeDeleteOrRevoke(ob)
-	if err != nil {
-		return err
-	}
-
-	return p.updateAdditionalSettings(ob)
-}
diff --git a/pkg/operator/ceph/object/bucket/util.go b/pkg/operator/ceph/object/bucket/util.go
index 9fd4ed6ef..71faa16cf 100644
--- a/pkg/operator/ceph/object/bucket/util.go
+++ b/pkg/operator/ceph/object/bucket/util.go
@@ -23,6 +23,7 @@ import (
 	"github.com/coreos/pkg/capnslog"
 	bktv1alpha1 "github.com/kube-object-storage/lib-bucket-provisioner/pkg/apis/objectbucket.io/v1alpha1"
 	"github.com/kube-object-storage/lib-bucket-provisioner/pkg/provisioner"
+	apibkt "github.com/kube-object-storage/lib-bucket-provisioner/pkg/provisioner/api"
 	"github.com/pkg/errors"
 	storagev1 "k8s.io/api/storage/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
@@ -116,10 +117,10 @@ func randomString(n int) string {
 	return string(b)
 }
 
-func MaxObjectQuota(AdditionalConfig map[string]string) string {
-	return AdditionalConfig["maxObjects"]
+func MaxObjectQuota(options *apibkt.BucketOptions) string {
+	return options.ObjectBucketClaim.Spec.AdditionalConfig["maxObjects"]
 }
 
-func MaxSizeQuota(AdditionalConfig map[string]string) string {
-	return AdditionalConfig["maxSize"]
+func MaxSizeQuota(options *apibkt.BucketOptions) string {
+	return options.ObjectBucketClaim.Spec.AdditionalConfig["maxSize"]
 }
diff --git a/pkg/operator/ceph/operator.go b/pkg/operator/ceph/operator.go
index 06603fe6b..334afcdf2 100644
--- a/pkg/operator/ceph/operator.go
+++ b/pkg/operator/ceph/operator.go
@@ -68,33 +68,28 @@ var (
 
 // Operator type for managing storage
 type Operator struct {
-	context   *clusterd.Context
-	resources []k8sutil.CustomResource
-	config    *Config
+	context           *clusterd.Context
+	resources         []k8sutil.CustomResource
+	operatorNamespace string
+	rookImage         string
+	securityAccount   string
 	// The custom resource that is global to the kubernetes cluster.
 	// The cluster is global because you create multiple clusters in k8s
 	clusterController     *cluster.ClusterController
 	delayedDaemonsStarted bool
 }
 
-type Config struct {
-	OperatorNamespace string
-	Image             string
-	ServiceAccount    string
-}
-
 // New creates an operator instance
-func New(context *clusterd.Context, volumeAttachmentWrapper attachment.Attachment, rookImage, serviceAccount string) *Operator {
+func New(context *clusterd.Context, volumeAttachmentWrapper attachment.Attachment, rookImage, securityAccount string) *Operator {
 	schemes := []k8sutil.CustomResource{opcontroller.ClusterResource, attachment.VolumeResource}
 
+	operatorNamespace := os.Getenv(k8sutil.PodNamespaceEnvVar)
 	o := &Operator{
-		context:   context,
-		resources: schemes,
-		config: &Config{
-			OperatorNamespace: os.Getenv(k8sutil.PodNamespaceEnvVar),
-			Image:             rookImage,
-			ServiceAccount:    serviceAccount,
-		},
+		context:           context,
+		resources:         schemes,
+		operatorNamespace: operatorNamespace,
+		rookImage:         rookImage,
+		securityAccount:   securityAccount,
 	}
 	operatorConfigCallbacks := []func() error{
 		o.updateDrivers,
@@ -131,11 +126,10 @@ func (o *Operator) updateOperatorLogLevel() error {
 // Run the operator instance
 func (o *Operator) Run() error {
 
-	if o.config.OperatorNamespace == "" {
+	if o.operatorNamespace == "" {
 		return errors.Errorf("rook operator namespace is not provided. expose it via downward API in the rook operator manifest file using environment variable %q", k8sutil.PodNamespaceEnvVar)
 	}
 
-	// TODO: needs a callback?
 	opcontroller.SetCephCommandsTimeout(o.context)
 
 	// Initialize signal handler and context
@@ -144,15 +138,20 @@ func (o *Operator) Run() error {
 
 	rookDiscover := discover.New(o.context.Clientset)
 	if opcontroller.DiscoveryDaemonEnabled(o.context) {
-		if err := rookDiscover.Start(o.config.OperatorNamespace, o.config.Image, o.config.ServiceAccount, true); err != nil {
+		if err := rookDiscover.Start(o.operatorNamespace, o.rookImage, o.securityAccount, true); err != nil {
 			return errors.Wrap(err, "failed to start device discovery daemonset")
 		}
 	} else {
-		if err := rookDiscover.Stop(stopContext, o.config.OperatorNamespace); err != nil {
+		if err := rookDiscover.Stop(stopContext, o.operatorNamespace); err != nil {
 			return errors.Wrap(err, "failed to stop device discovery daemonset")
 		}
 	}
 
+	logger.Debug("checking for admission controller secrets")
+	err := StartControllerIfSecretPresent(stopContext, o.context, o.rookImage)
+	if err != nil {
+		return errors.Wrap(err, "failed to start webhook")
+	}
 	serverVersion, err := o.context.Clientset.Discovery().ServerVersion()
 	if err != nil {
 		return errors.Wrap(err, "failed to get server version")
@@ -179,7 +178,7 @@ func (o *Operator) Run() error {
 	var namespaceToWatch string
 	if os.Getenv("ROOK_CURRENT_NAMESPACE_ONLY") == "true" {
 		logger.Infof("watching the current namespace for a ceph cluster CR")
-		namespaceToWatch = o.config.OperatorNamespace
+		namespaceToWatch = o.operatorNamespace
 	} else {
 		logger.Infof("watching all namespaces for ceph cluster CRs")
 		namespaceToWatch = v1.NamespaceAll
@@ -187,7 +186,7 @@ func (o *Operator) Run() error {
 
 	// Start the controller-runtime Manager.
 	mgrErrorChan := make(chan error)
-	go o.startManager(stopContext, namespaceToWatch, mgrErrorChan)
+	go o.startManager(namespaceToWatch, stopContext, mgrErrorChan)
 
 	// Start the operator setting watcher
 	go o.clusterController.StartOperatorSettingsWatch(stopChan)
@@ -199,8 +198,8 @@ func (o *Operator) Run() error {
 			logger.Infof("shutdown signal received, exiting... %v", stopContext.Err())
 			o.cleanup(stopChan)
 			return nil
-
 		case err := <-mgrErrorChan:
+			logger.Errorf("gave up to run the operator. %v", err)
 			o.cleanup(stopChan)
 			return err
 		}
@@ -229,13 +228,13 @@ func (o *Operator) updateDrivers() error {
 		return nil
 	}
 
-	if o.config.OperatorNamespace == "" {
+	if o.operatorNamespace == "" {
 		return errors.Errorf("rook operator namespace is not provided. expose it via downward API in the rook operator manifest file using environment variable %s", k8sutil.PodNamespaceEnvVar)
 	}
 
 	if opcontroller.FlexDriverEnabled(o.context) {
 		rookAgent := agent.New(o.context.Clientset)
-		if err := rookAgent.Start(o.config.OperatorNamespace, o.config.Image, o.config.ServiceAccount); err != nil {
+		if err := rookAgent.Start(o.operatorNamespace, o.rookImage, o.securityAccount); err != nil {
 			return errors.Wrap(err, "error starting agent daemonset")
 		}
 	}
@@ -253,7 +252,7 @@ func (o *Operator) updateDrivers() error {
 		return nil
 	}
 
-	ownerRef, err := getDeploymentOwnerReference(o.context.Clientset, o.config.OperatorNamespace)
+	ownerRef, err := getDeploymentOwnerReference(o.context.Clientset, o.operatorNamespace)
 	if err != nil {
 		logger.Warningf("could not find deployment owner reference to assign to csi drivers. %v", err)
 	}
@@ -262,15 +261,15 @@ func (o *Operator) updateDrivers() error {
 		ownerRef.BlockOwnerDeletion = &blockOwnerDeletion
 	}
 
-	ownerInfo := k8sutil.NewOwnerInfoWithOwnerRef(ownerRef, o.config.OperatorNamespace)
+	ownerInfo := k8sutil.NewOwnerInfoWithOwnerRef(ownerRef, o.operatorNamespace)
 	// create an empty config map. config map will be filled with data
 	// later when clusters have mons
-	err = csi.CreateCsiConfigMap(o.config.OperatorNamespace, o.context.Clientset, ownerInfo)
+	err = csi.CreateCsiConfigMap(o.operatorNamespace, o.context.Clientset, ownerInfo)
 	if err != nil {
 		return errors.Wrap(err, "failed creating csi config map")
 	}
 
-	go csi.ValidateAndConfigureDrivers(o.context, o.config.OperatorNamespace, o.config.Image, o.config.ServiceAccount, serverVersion, ownerInfo)
+	go csi.ValidateAndConfigureDrivers(o.context, o.operatorNamespace, o.rookImage, o.securityAccount, serverVersion, ownerInfo)
 	return nil
 }
 
diff --git a/pkg/operator/ceph/server.go b/pkg/operator/ceph/server.go
new file mode 100644
index 000000000..60ed2df63
--- /dev/null
+++ b/pkg/operator/ceph/server.go
@@ -0,0 +1,68 @@
+/*
+Copyright 2020 The Rook Authors. All rights reserved.
+
+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 operator
+
+import (
+	"github.com/pkg/errors"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/webhook"
+)
+
+var (
+	scheme    = runtime.NewScheme()
+	resources = []webhook.Validator{&cephv1.CephCluster{}, &cephv1.CephBlockPool{}, &cephv1.CephObjectStore{}}
+)
+
+const (
+	// Default directory where TLS certs are stored
+	certDir = "/etc/webhook"
+	// Default port for server
+	port = 8079
+)
+
+// StartAdmissionController will start the server
+func StartAdmissionController() error {
+	logger.Infof("starting the webhook for backend ceph")
+	err := cephv1.AddToScheme(scheme)
+	if err != nil {
+		return errors.Wrap(err, "failed to add to scheme")
+	}
+	opts := ctrl.Options{
+		Scheme:  scheme,
+		Port:    port,
+		CertDir: certDir,
+	}
+	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), opts)
+	if err != nil {
+		return errors.Wrap(err, "failed to create manager")
+	}
+	for _, resource := range resources {
+		err = ctrl.NewWebhookManagedBy(mgr).For(resource).Complete()
+		if err != nil {
+			return errors.Wrap(err, "failed to register webhooks")
+		}
+	}
+	logger.Info("starting webhook server")
+	err = mgr.Start(ctrl.SetupSignalHandler())
+	if err != nil {
+		return errors.Wrap(err, "failed to start server")
+	}
+
+	return nil
+}
diff --git a/pkg/operator/ceph/webhook.go b/pkg/operator/ceph/webhook.go
index d67577a77..89f1b54fc 100644
--- a/pkg/operator/ceph/webhook.go
+++ b/pkg/operator/ceph/webhook.go
@@ -18,23 +18,31 @@ package operator
 
 import (
 	"context"
-	"fmt"
-	"io/ioutil"
 	"os"
 
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/clusterd"
+	"github.com/rook/rook/pkg/operator/ceph/controller"
+	"github.com/rook/rook/pkg/operator/ceph/csi"
 	"github.com/rook/rook/pkg/operator/k8sutil"
+	v1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
-	"sigs.k8s.io/controller-runtime/pkg/webhook"
+	"k8s.io/client-go/kubernetes"
 )
 
 const (
-	appName       = "rook-ceph-admission-controller"
-	tlsPort int32 = 443
+	appName                                  = "rook-ceph-admission-controller"
+	secretVolumeName                         = "webhook-certificates" // #nosec G101 This is just a var name, not a real secret
+	serviceAccountName                       = "rook-ceph-admission-controller"
+	portName                                 = "webhook-api"
+	servicePort                        int32 = 443
+	serverPort                         int32 = 8079
+	tlsDir                                   = "/etc/webhook"
+	admissionControllerTolerationsEnv        = "ADMISSION_CONTROLLER_TOLERATIONS"
+	admissionControllerNodeAffinityEnv       = "ADMISSION_CONTROLLER_NODE_AFFINITY"
 )
 
 var (
@@ -42,30 +50,16 @@ var (
 )
 
 func isSecretPresent(ctx context.Context, context *clusterd.Context) (bool, error) {
-	logger.Infof("looking for admission webhook secret %q", appName)
-	s, err := context.Clientset.CoreV1().Secrets(namespace).Get(ctx, appName, metav1.GetOptions{})
+	logger.Infof("looking for secret %q", appName)
+	_, err := context.Clientset.CoreV1().Secrets(namespace).Get(ctx, appName, metav1.GetOptions{})
 	if err != nil {
 		// If secret is not found. All good ! Proceed with rook without admission controllers
 		if apierrors.IsNotFound(err) {
-			logger.Infof("admission webhook secret %q not found. proceeding without the admission controller", appName)
+			logger.Infof("secret %q not found. proceeding without the admission controller", appName)
 			return false, nil
 		}
 		return false, err
 	}
-
-	// Search for any previous admission controller deployment and if so removing it
-	logger.Debug("searching for old admission controller deployment")
-	removeOldAdmissionControllerDeployment(ctx, context)
-
-	logger.Infof("admission webhook secret %q found", appName)
-	for k, data := range s.Data {
-		path := fmt.Sprintf("%s/%s", certDir, k)
-		err := ioutil.WriteFile(path, data, 0400)
-		if err != nil {
-			return false, errors.Wrapf(err, "failed to write secret content to file %q", path)
-		}
-	}
-
 	return true, nil
 }
 
@@ -78,44 +72,189 @@ func createWebhookService(context *clusterd.Context) error {
 		Spec: corev1.ServiceSpec{
 			Ports: []corev1.ServicePort{
 				{
-					Port: tlsPort,
+					Port: servicePort,
 					TargetPort: intstr.IntOrString{
-						IntVal: int32(webhook.DefaultPort),
+						IntVal: serverPort,
 					},
 				},
 			},
+
 			Selector: map[string]string{
-				k8sutil.AppAttr: "rook-ceph-operator",
+				k8sutil.AppAttr: appName,
 			},
 		},
 	}
 
 	_, err := k8sutil.CreateOrUpdateService(context.Clientset, namespace, &webhookService)
-	if err != nil {
+	if err != nil && !apierrors.IsAlreadyExists(err) {
 		return err
 	}
+	return nil
+}
 
+// StartControllerIfSecretPresent will initialize the webhook if secret is detected
+func StartControllerIfSecretPresent(ctx context.Context, context *clusterd.Context, admissionImage string) error {
+	isPresent, err := isSecretPresent(ctx, context)
+	if err != nil {
+		return errors.Wrap(err, "failed to retrieve secret")
+	}
+	if isPresent {
+		err = initWebhook(ctx, context, admissionImage)
+		if err != nil {
+			return errors.Wrap(err, "failed to initialize webhook")
+		}
+	}
 	return nil
 }
 
-func removeOldAdmissionControllerDeployment(ctx context.Context, context *clusterd.Context) {
-	opts := metav1.ListOptions{LabelSelector: fmt.Sprintf("app=%s", appName)}
-	d, err := context.Clientset.AppsV1().Deployments(namespace).List(ctx, opts)
+func initWebhook(ctx context.Context, context *clusterd.Context, admissionImage string) error {
+	// At this point volume should be mounted, so proceed with creating the service and validatingwebhookconfig
+	err := createWebhookService(context)
+	if err != nil {
+		return errors.Wrap(err, "failed to create service")
+	}
+	err = createWebhookDeployment(ctx, context, admissionImage)
+	if err != nil {
+		return errors.Wrap(err, "failed to create deployment")
+	}
+	return nil
+}
+
+func createWebhookDeployment(ctx context.Context, context *clusterd.Context, admissionImage string) error {
+	logger.Info("creating admission controller pods")
+	admission_parameters := []string{"ceph",
+		"admission-controller"}
+	secretVolume := getSecretVolume()
+	secretVolumeMount := getSecretVolumeMount()
+
+	antiAffinity := csi.GetPodAntiAffinity(k8sutil.AppAttr, appName)
+	admissionControllerDeployment := getDeployment(ctx, context, secretVolume, antiAffinity, admissionImage, admission_parameters, secretVolumeMount)
+
+	_, err := k8sutil.CreateOrUpdateDeployment(context.Clientset, &admissionControllerDeployment)
+	if err != nil {
+		return errors.Wrap(err, "failed to create admission-controller deployment")
+	}
+
+	return nil
+}
+
+func getDeployment(ctx context.Context, context *clusterd.Context, secretVolume corev1.Volume, antiAffinity corev1.PodAntiAffinity,
+	admissionImage string, admission_parameters []string, secretVolumeMount corev1.VolumeMount) v1.Deployment {
+	var replicas int32 = 2
+	nodes, err := context.Clientset.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
+	if err == nil {
+		if len(nodes.Items) == 1 {
+			replicas = 1
+		}
+	} else {
+		logger.Errorf("failed to get nodes. Defaulting the number of replicas of admission controller pods to 2. %v", err)
+	}
+
+	admissionControllerDeployment := v1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      appName,
+			Namespace: namespace,
+		},
+		Spec: v1.DeploymentSpec{
+			Replicas: &replicas,
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{
+					k8sutil.AppAttr: appName,
+				},
+			},
+			Template: corev1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      appName,
+					Namespace: namespace,
+					Labels: map[string]string{
+						k8sutil.AppAttr: appName,
+					},
+				},
+				Spec: corev1.PodSpec{
+					Volumes: []corev1.Volume{
+						secretVolume,
+					},
+					Containers: []corev1.Container{
+						{
+							Name:  appName,
+							Image: admissionImage,
+							Args:  admission_parameters,
+							Ports: []corev1.ContainerPort{
+								{
+									Name:          portName,
+									ContainerPort: serverPort,
+								},
+							},
+							VolumeMounts: []corev1.VolumeMount{
+								secretVolumeMount,
+							},
+						},
+					},
+					ServiceAccountName: serviceAccountName,
+					Affinity: &corev1.Affinity{
+						PodAntiAffinity: &antiAffinity,
+						NodeAffinity:    getNodeAffinity(context.Clientset),
+					},
+					Tolerations: getTolerations(context.Clientset),
+				},
+			},
+		},
+	}
+	return admissionControllerDeployment
+}
+
+func getSecretVolumeMount() corev1.VolumeMount {
+	secretVolumeMount := corev1.VolumeMount{
+		Name:      secretVolumeName,
+		ReadOnly:  true,
+		MountPath: tlsDir,
+	}
+	return secretVolumeMount
+}
+
+func getSecretVolume() corev1.Volume {
+	secretVolume := corev1.Volume{
+		Name: secretVolumeName,
+		VolumeSource: corev1.VolumeSource{
+			Secret: &corev1.SecretVolumeSource{
+				SecretName: appName,
+			},
+		},
+	}
+	return secretVolume
+}
+
+func getTolerations(clientset kubernetes.Interface) []corev1.Toleration {
+	// Add toleration if any
+	tolerations := []corev1.Toleration{}
+	tolerationsRaw, err := k8sutil.GetOperatorSetting(clientset, controller.OperatorSettingConfigMapName, admissionControllerTolerationsEnv, "")
 	if err != nil {
-		logger.Warningf("failed to get old admission controller deployment. %v", err)
-		return
-	}
-
-	if len(d.Items) > 0 {
-		for _, deploy := range d.Items {
-			var gracePeriod int64
-			propagation := metav1.DeletePropagationForeground
-			deleteOpts := metav1.DeleteOptions{GracePeriodSeconds: &gracePeriod, PropagationPolicy: &propagation}
-			if err = context.Clientset.AppsV1().Deployments(namespace).Delete(ctx, deploy.Name, deleteOpts); err != nil {
-				logger.Warningf("failed to delete admission controller deployment %q. %v", deploy.Name, err)
-				return
-			}
+		logger.Warningf("toleration will be empty because failed to read the setting. %v", err)
+		return tolerations
+	}
+	tolerations, err = k8sutil.YamlToTolerations(tolerationsRaw)
+	if err != nil {
+		logger.Warningf("toleration will be empty because failed to parse the setting %q. %v", tolerationsRaw, err)
+		return tolerations
+	}
+	return tolerations
+}
+
+func getNodeAffinity(clientset kubernetes.Interface) *corev1.NodeAffinity {
+	// Add NodeAffinity if any
+	v1NodeAffinity := &corev1.NodeAffinity{}
+	nodeAffinity, err := k8sutil.GetOperatorSetting(clientset, controller.OperatorSettingConfigMapName, admissionControllerNodeAffinityEnv, "")
+	if err != nil {
+		// nodeAffinity will be empty by default in case of error
+		logger.Warningf("node affinity will be empty because failed to read the setting. %v", err)
+		return v1NodeAffinity
+	}
+	if nodeAffinity != "" {
+		v1NodeAffinity, err = k8sutil.GenerateNodeAffinity(nodeAffinity)
+		if err != nil {
+			logger.Warningf("node affinity will be empty because failed to parse the setting %q. %v", nodeAffinity, err)
+			return v1NodeAffinity
 		}
-		logger.Info("successfully removed old admission controller deployment. please remove old service account, cluster role and bindings manually")
 	}
+	return v1NodeAffinity
 }
diff --git a/pkg/util/set.go b/pkg/util/set.go
deleted file mode 100644
index 2801b59fa..000000000
--- a/pkg/util/set.go
+++ /dev/null
@@ -1,169 +0,0 @@
-/*
-Copyright 2016 The Rook Authors. All rights reserved.
-
-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 util
-
-type Set struct {
-	values map[string]struct{}
-}
-
-// Create a new empty set
-func NewSet() *Set {
-	set := &Set{}
-	set.values = make(map[string]struct{})
-	return set
-}
-
-// Create a new set from the array
-func CreateSet(values []string) *Set {
-	set := &Set{}
-	set.values = make(map[string]struct{})
-	for _, value := range values {
-		set.add(value)
-	}
-	return set
-}
-
-// Create a copy of the set
-func (s *Set) Copy() *Set {
-	set := NewSet()
-	for value := range s.values {
-		set.values[value] = struct{}{}
-	}
-
-	return set
-}
-
-// Subtract the subset from the set
-func (s *Set) Subtract(subset *Set) {
-	// Iterate over each element in the set to see if it's in the subset
-	for value := range s.values {
-		if _, ok := subset.values[value]; ok {
-			delete(s.values, value)
-		}
-	}
-}
-
-// Add a value to the set. Returns true if the value was added, false if it already exists.
-func (s *Set) Add(newValue string) bool {
-	if _, ok := s.values[newValue]; !ok {
-		s.add(newValue)
-		return true
-	}
-
-	// The value is already in the set
-	return false
-}
-
-// Remove a value from the set. Returns true if the value was removed, false if it does not exist.
-func (s *Set) Remove(oldValue string) bool {
-	if _, ok := s.values[oldValue]; ok {
-		delete(s.values, oldValue)
-		return true
-	}
-
-	// The value is not in the set
-	return false
-}
-
-// Add the value to the set
-func (s *Set) add(value string) {
-	s.values[value] = struct{}{}
-}
-
-// Check whether a value is already contained in the set
-func (s *Set) Contains(value string) bool {
-	_, ok := s.values[value]
-	return ok
-}
-
-// Iterate over the items in the set
-func (s *Set) Iter() <-chan string {
-	channel := make(chan string)
-	go func() {
-		for value := range s.values {
-			channel <- value
-		}
-		close(channel)
-	}()
-	return channel
-}
-
-// Get the count of items in the set
-func (s *Set) Count() int {
-	return len(s.values)
-}
-
-// Add other set items
-func (s *Set) AddSet(other *Set) {
-	for value := range other.Iter() {
-		s.add(value)
-	}
-}
-
-// Add multiple items more efficiently
-func (s *Set) AddMultiple(values []string) {
-	for _, value := range values {
-		s.add(value)
-	}
-}
-
-// Check if two sets contain the same elements
-func (s *Set) Equals(other *Set) bool {
-	if s.Count() != other.Count() {
-		return false
-	}
-
-	for value := range s.Iter() {
-		if !other.Contains(value) {
-			return false
-		}
-	}
-
-	return true
-}
-
-// Convert the set to an array
-func (s *Set) ToSlice() []string {
-	values := []string{}
-	for value := range s.values {
-		values = append(values, value)
-	}
-
-	return values
-}
-
-// find items in the left slice that are not in the right slice
-func SetDifference(left, right []string) *Set {
-	result := NewSet()
-	for _, leftItem := range left {
-		foundItem := false
-
-		// search for the left item in the right set
-		for _, rightItem := range right {
-			if leftItem == rightItem {
-				foundItem = true
-				break
-			}
-		}
-
-		if !foundItem {
-			result.Add(leftItem)
-		}
-	}
-
-	return result
-}


Test patch:
diff --git a/pkg/operator/ceph/cluster/osd/create_test.go b/pkg/operator/ceph/cluster/osd/create_test.go
index e79938b1b..5272da8ea 100644
--- a/pkg/operator/ceph/cluster/osd/create_test.go
+++ b/pkg/operator/ceph/cluster/osd/create_test.go
@@ -27,7 +27,6 @@ import (
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/test"
-	"github.com/rook/rook/pkg/util"
 	"github.com/stretchr/testify/assert"
 	"github.com/tevino/abool"
 	corev1 "k8s.io/api/core/v1"
@@ -35,6 +34,7 @@ import (
 	apiresource "k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/kubernetes/fake"
 	k8stesting "k8s.io/client-go/testing"
 )
@@ -93,7 +93,7 @@ func Test_createNewOSDsFromStatus(t *testing.T) {
 
 	spec := cephv1.ClusterSpec{}
 	var status *OrchestrationStatus
-	awaitingStatusConfigMaps := util.NewSet()
+	awaitingStatusConfigMaps := sets.NewString()
 
 	var c *Cluster
 	var createConfig *createConfig
@@ -102,10 +102,10 @@ func Test_createNewOSDsFromStatus(t *testing.T) {
 		// none of this code should ever add or remove deployments from the existence list
 		assert.Equal(t, 3, deployments.Len())
 		// Simulate environment where provision jobs were created for node0, node2, pvc1, and pvc2
-		awaitingStatusConfigMaps = util.NewSet()
-		awaitingStatusConfigMaps.AddMultiple([]string{
+		awaitingStatusConfigMaps = sets.NewString()
+		awaitingStatusConfigMaps.Insert(
 			statusNameNode0, statusNameNode2,
-			statusNamePVC1, statusNamePVC2})
+			statusNamePVC1, statusNamePVC2)
 		createCallsOnNode = createCallsOnNode[:0]
 		createCallsOnPVC = createCallsOnPVC[:0]
 		errs = newProvisionErrors()
@@ -128,9 +128,9 @@ func Test_createNewOSDsFromStatus(t *testing.T) {
 		assert.Len(t, createCallsOnNode, 0)
 		assert.Len(t, createCallsOnPVC, 0)
 		// status map should have been marked completed
-		assert.Equal(t, 4, awaitingStatusConfigMaps.Count())
-		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Count())
-		assert.True(t, createConfig.finishedStatusConfigMaps.Contains(statusNameNode0))
+		assert.Equal(t, 4, awaitingStatusConfigMaps.Len())
+		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Len())
+		assert.True(t, createConfig.finishedStatusConfigMaps.Has(statusNameNode0))
 	})
 
 	t.Run("test: node: create all OSDs on node when all do not exist", func(t *testing.T) {
@@ -146,9 +146,9 @@ func Test_createNewOSDsFromStatus(t *testing.T) {
 		assert.ElementsMatch(t, createCallsOnNode, []int{0, 1, 2})
 		assert.Len(t, createCallsOnPVC, 0)
 		// status map should have been marked completed
-		assert.Equal(t, 4, awaitingStatusConfigMaps.Count())
-		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Count())
-		assert.True(t, createConfig.finishedStatusConfigMaps.Contains(statusNameNode2))
+		assert.Equal(t, 4, awaitingStatusConfigMaps.Len())
+		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Len())
+		assert.True(t, createConfig.finishedStatusConfigMaps.Has(statusNameNode2))
 	})
 
 	t.Run("node: create only nonexistent OSDs on node when some already exist", func(t *testing.T) {
@@ -167,9 +167,9 @@ func Test_createNewOSDsFromStatus(t *testing.T) {
 		assert.ElementsMatch(t, createCallsOnNode, []int{5, 7})
 		assert.Len(t, createCallsOnPVC, 0)
 		// status map should have been marked completed
-		assert.Equal(t, 4, awaitingStatusConfigMaps.Count())
-		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Count())
-		assert.True(t, createConfig.finishedStatusConfigMaps.Contains(statusNameNode0))
+		assert.Equal(t, 4, awaitingStatusConfigMaps.Len())
+		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Len())
+		assert.True(t, createConfig.finishedStatusConfigMaps.Has(statusNameNode0))
 	})
 
 	t.Run("node: skip creating OSDs for status configmaps that weren't created for this reconcile", func(t *testing.T) {
@@ -185,8 +185,8 @@ func Test_createNewOSDsFromStatus(t *testing.T) {
 		assert.ElementsMatch(t, createCallsOnNode, []int{})
 		assert.Len(t, createCallsOnPVC, 0)
 		// status map should NOT have been marked completed
-		assert.Equal(t, 4, awaitingStatusConfigMaps.Count())
-		assert.Equal(t, 0, createConfig.finishedStatusConfigMaps.Count())
+		assert.Equal(t, 4, awaitingStatusConfigMaps.Len())
+		assert.Equal(t, 0, createConfig.finishedStatusConfigMaps.Len())
 	})
 
 	t.Run("node: errors reported if OSDs fail to create", func(t *testing.T) {
@@ -203,9 +203,9 @@ func Test_createNewOSDsFromStatus(t *testing.T) {
 		assert.ElementsMatch(t, createCallsOnNode, []int{0, 1, 2})
 		assert.Len(t, createCallsOnPVC, 0)
 		// status map should have been marked completed
-		assert.Equal(t, 4, awaitingStatusConfigMaps.Count())
-		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Count())
-		assert.True(t, createConfig.finishedStatusConfigMaps.Contains(statusNameNode0))
+		assert.Equal(t, 4, awaitingStatusConfigMaps.Len())
+		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Len())
+		assert.True(t, createConfig.finishedStatusConfigMaps.Has(statusNameNode0))
 		induceFailureCreatingOSD = -1 // off
 	})
 
@@ -220,9 +220,9 @@ func Test_createNewOSDsFromStatus(t *testing.T) {
 		assert.Len(t, createCallsOnNode, 0)
 		assert.Len(t, createCallsOnPVC, 0)
 		// status map should have been marked completed
-		assert.Equal(t, 4, awaitingStatusConfigMaps.Count())
-		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Count())
-		assert.True(t, createConfig.finishedStatusConfigMaps.Contains(statusNamePVC1))
+		assert.Equal(t, 4, awaitingStatusConfigMaps.Len())
+		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Len())
+		assert.True(t, createConfig.finishedStatusConfigMaps.Has(statusNamePVC1))
 	})
 
 	t.Run("pvc: create all OSDs on pvc when all do not exist", func(t *testing.T) {
@@ -238,9 +238,9 @@ func Test_createNewOSDsFromStatus(t *testing.T) {
 		assert.ElementsMatch(t, createCallsOnPVC, []int{0, 1, 2})
 		assert.Len(t, createCallsOnNode, 0)
 		// status map should have been marked completed
-		assert.Equal(t, 4, awaitingStatusConfigMaps.Count())
-		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Count())
-		assert.True(t, createConfig.finishedStatusConfigMaps.Contains(statusNamePVC2))
+		assert.Equal(t, 4, awaitingStatusConfigMaps.Len())
+		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Len())
+		assert.True(t, createConfig.finishedStatusConfigMaps.Has(statusNamePVC2))
 	})
 
 	t.Run("pvc: create only nonexistent OSDs on pvc when some already exist", func(t *testing.T) {
@@ -259,9 +259,9 @@ func Test_createNewOSDsFromStatus(t *testing.T) {
 		assert.ElementsMatch(t, createCallsOnPVC, []int{5, 7})
 		assert.Len(t, createCallsOnNode, 0)
 		// status map should have been marked completed
-		assert.Equal(t, 4, awaitingStatusConfigMaps.Count())
-		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Count())
-		assert.True(t, createConfig.finishedStatusConfigMaps.Contains(statusNamePVC1))
+		assert.Equal(t, 4, awaitingStatusConfigMaps.Len())
+		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Len())
+		assert.True(t, createConfig.finishedStatusConfigMaps.Has(statusNamePVC1))
 	})
 
 	t.Run("pvc: skip creating OSDs for status configmaps that weren't created for this reconcile", func(t *testing.T) {
@@ -277,8 +277,8 @@ func Test_createNewOSDsFromStatus(t *testing.T) {
 		assert.ElementsMatch(t, createCallsOnPVC, []int{})
 		assert.Len(t, createCallsOnNode, 0)
 		// no status maps should have been marked completed
-		assert.Equal(t, 4, awaitingStatusConfigMaps.Count())
-		assert.Equal(t, 0, createConfig.finishedStatusConfigMaps.Count())
+		assert.Equal(t, 4, awaitingStatusConfigMaps.Len())
+		assert.Equal(t, 0, createConfig.finishedStatusConfigMaps.Len())
 	})
 
 	t.Run("pvc: errors reported if OSDs fail to create", func(t *testing.T) {
@@ -295,9 +295,9 @@ func Test_createNewOSDsFromStatus(t *testing.T) {
 		assert.ElementsMatch(t, createCallsOnPVC, []int{0, 1, 2})
 		assert.Len(t, createCallsOnNode, 0)
 		// status map should have been marked completed
-		assert.Equal(t, 4, awaitingStatusConfigMaps.Count())
-		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Count())
-		assert.True(t, createConfig.finishedStatusConfigMaps.Contains(statusNamePVC1))
+		assert.Equal(t, 4, awaitingStatusConfigMaps.Len())
+		assert.Equal(t, 1, createConfig.finishedStatusConfigMaps.Len())
+		assert.True(t, createConfig.finishedStatusConfigMaps.Has(statusNamePVC1))
 		induceFailureCreatingOSD = -1 // off
 	})
 }
@@ -323,7 +323,7 @@ func Test_startProvisioningOverPVCs(t *testing.T) {
 	var errs *provisionErrors
 	var c *Cluster
 	var config *provisionConfig
-	var awaitingStatusConfigMaps *util.Set
+	var awaitingStatusConfigMaps sets.String
 	var err error
 	doSetup := func() {
 		test.SetFakeKubernetesVersion(clientset, fakeK8sVersion) // PVCs require k8s version v1.13+
@@ -341,7 +341,7 @@ func Test_startProvisioningOverPVCs(t *testing.T) {
 		doSetup()
 		awaitingStatusConfigMaps, err = c.startProvisioningOverPVCs(config, errs)
 		assert.NoError(t, err)
-		assert.Zero(t, awaitingStatusConfigMaps.Count())
+		assert.Zero(t, awaitingStatusConfigMaps.Len())
 		assert.Zero(t, errs.len())
 		// no result configmaps should have been created
 		cms, err := clientset.CoreV1().ConfigMaps(namespace).List(context.TODO(), metav1.ListOptions{})
@@ -366,7 +366,7 @@ func Test_startProvisioningOverPVCs(t *testing.T) {
 		doSetup()
 		awaitingStatusConfigMaps, err = c.startProvisioningOverPVCs(config, errs)
 		assert.NoError(t, err)
-		assert.Zero(t, awaitingStatusConfigMaps.Count())
+		assert.Zero(t, awaitingStatusConfigMaps.Len())
 		assert.Zero(t, errs.len()) // this was not a problem with a single job but with ALL jobs
 		// no result configmaps should have been created
 		cms, err := clientset.CoreV1().ConfigMaps(namespace).List(context.TODO(), metav1.ListOptions{})
@@ -391,7 +391,7 @@ func Test_startProvisioningOverPVCs(t *testing.T) {
 		doSetup()
 		awaitingStatusConfigMaps, err = c.startProvisioningOverPVCs(config, errs)
 		assert.NoError(t, err)
-		assert.Equal(t, 2, awaitingStatusConfigMaps.Count())
+		assert.Equal(t, 2, awaitingStatusConfigMaps.Len())
 		assert.Zero(t, errs.len())
 		cms, err := clientset.CoreV1().ConfigMaps(namespace).List(context.TODO(), metav1.ListOptions{})
 		assert.NoError(t, err)
@@ -403,7 +403,7 @@ func Test_startProvisioningOverPVCs(t *testing.T) {
 		doSetup()
 		awaitingStatusConfigMaps, err = c.startProvisioningOverPVCs(config, errs)
 		assert.NoError(t, err)
-		assert.Equal(t, 2, awaitingStatusConfigMaps.Count())
+		assert.Equal(t, 2, awaitingStatusConfigMaps.Len())
 		assert.Zero(t, errs.len())
 		cms, err := clientset.CoreV1().ConfigMaps(namespace).List(context.TODO(), metav1.ListOptions{})
 		assert.NoError(t, err)
@@ -417,7 +417,7 @@ func Test_startProvisioningOverPVCs(t *testing.T) {
 		doSetup()
 		awaitingStatusConfigMaps, err = c.startProvisioningOverPVCs(config, errs)
 		assert.NoError(t, err)
-		assert.Equal(t, 0, awaitingStatusConfigMaps.Count())
+		assert.Equal(t, 0, awaitingStatusConfigMaps.Len())
 		assert.Equal(t, 1, errs.len())
 		cms, err := clientset.CoreV1().ConfigMaps(namespace).List(context.TODO(), metav1.ListOptions{})
 		assert.NoError(t, err)
@@ -433,7 +433,7 @@ func Test_startProvisioningOverPVCs(t *testing.T) {
 		awaitingStatusConfigMaps, err = c.startProvisioningOverPVCs(config, errs)
 		assert.Error(t, err)
 		assert.Zero(t, errs.len())
-		assert.Zero(t, awaitingStatusConfigMaps.Count())
+		assert.Zero(t, awaitingStatusConfigMaps.Len())
 		requestCancelOrchestration.UnSet()
 	})
 
@@ -453,7 +453,7 @@ func Test_startProvisioningOverPVCs(t *testing.T) {
 		doSetup()
 		awaitingStatusConfigMaps, err = c.startProvisioningOverPVCs(config, errs)
 		assert.NoError(t, err)
-		assert.Equal(t, 0, awaitingStatusConfigMaps.Count())
+		assert.Equal(t, 0, awaitingStatusConfigMaps.Len())
 		assert.Equal(t, 1, errs.len())
 		cms, err := clientset.CoreV1().ConfigMaps(namespace).List(context.TODO(), metav1.ListOptions{})
 		assert.NoError(t, err)
@@ -490,7 +490,7 @@ func Test_startProvisioningOverNodes(t *testing.T) {
 	var errs *provisionErrors
 	var c *Cluster
 	var config *provisionConfig
-	var prepareJobsRun *util.Set
+	var prepareJobsRun sets.String
 	var err error
 	var cms *corev1.ConfigMapList
 	doSetup := func() {
@@ -508,7 +508,7 @@ func Test_startProvisioningOverNodes(t *testing.T) {
 		doSetup()
 		prepareJobsRun, err = c.startProvisioningOverNodes(config, errs)
 		assert.NoError(t, err)
-		assert.Zero(t, prepareJobsRun.Count())
+		assert.Zero(t, prepareJobsRun.Len())
 		assert.Zero(t, errs.len())
 		// no result configmaps should have been created
 		cms, err = clientset.CoreV1().ConfigMaps(namespace).List(context.TODO(), metav1.ListOptions{})
@@ -532,7 +532,7 @@ func Test_startProvisioningOverNodes(t *testing.T) {
 		doSetup()
 		prepareJobsRun, err = c.startProvisioningOverNodes(config, errs)
 		assert.NoError(t, err)
-		assert.Zero(t, prepareJobsRun.Count())
+		assert.Zero(t, prepareJobsRun.Len())
 		assert.Equal(t, 1, errs.len()) // this was not a problem with a single job but with ALL jobs
 		// no result configmaps should have been created
 		cms, err = clientset.CoreV1().ConfigMaps(namespace).List(context.TODO(), metav1.ListOptions{})
@@ -549,7 +549,7 @@ func Test_startProvisioningOverNodes(t *testing.T) {
 		assert.Zero(t, errs.len())
 		assert.ElementsMatch(t,
 			[]string{statusNameNode0, statusNameNode1, statusNameNode2},
-			prepareJobsRun.ToSlice(),
+			prepareJobsRun.List(),
 		)
 		// all result configmaps should have been created
 		cms, err = clientset.CoreV1().ConfigMaps(namespace).List(context.TODO(), metav1.ListOptions{})
@@ -568,7 +568,7 @@ func Test_startProvisioningOverNodes(t *testing.T) {
 		assert.Zero(t, errs.len())
 		assert.ElementsMatch(t,
 			[]string{statusNameNode0, statusNameNode1, statusNameNode2},
-			prepareJobsRun.ToSlice(),
+			prepareJobsRun.List(),
 		)
 	})
 
@@ -595,7 +595,7 @@ func Test_startProvisioningOverNodes(t *testing.T) {
 		assert.Zero(t, errs.len())
 		assert.ElementsMatch(t,
 			[]string{statusNameNode0, statusNameNode2},
-			prepareJobsRun.ToSlice(),
+			prepareJobsRun.List(),
 		)
 	})
 
@@ -605,7 +605,7 @@ func Test_startProvisioningOverNodes(t *testing.T) {
 		prepareJobsRun, err = c.startProvisioningOverNodes(config, errs)
 		assert.Error(t, err)
 		assert.Zero(t, errs.len())
-		assert.Zero(t, prepareJobsRun.Count())
+		assert.Zero(t, prepareJobsRun.Len())
 		requestCancelOrchestration.UnSet()
 	})
 
@@ -626,7 +626,7 @@ func Test_startProvisioningOverNodes(t *testing.T) {
 		prepareJobsRun, err = c.startProvisioningOverNodes(config, errs)
 		assert.NoError(t, err)
 		assert.Zero(t, errs.len())
-		assert.Zero(t, prepareJobsRun.Count())
+		assert.Zero(t, prepareJobsRun.Len())
 	})
 
 	t.Run("failures running prepare jobs", func(t *testing.T) {
@@ -671,13 +671,13 @@ func Test_startProvisioningOverNodes(t *testing.T) {
 		assert.Equal(t, 1, errs.len())
 		assert.ElementsMatch(t,
 			[]string{statusNameNode0},
-			prepareJobsRun.ToSlice(),
+			prepareJobsRun.List(),
 		)
 		// with a fresh clientset, only the one results ConfigMap should exist
 		cms, err = clientset.CoreV1().ConfigMaps(namespace).List(context.TODO(), metav1.ListOptions{})
 		assert.NoError(t, err)
 		assert.Len(t, cms.Items, 1)
-		assert.Equal(t, prepareJobsRun.ToSlice()[0], cms.Items[0].Name)
+		assert.Equal(t, prepareJobsRun.List()[0], cms.Items[0].Name)
 	})
 }
 
diff --git a/pkg/operator/ceph/csi/spec_test.go b/pkg/operator/ceph/csi/spec_test.go
index fbd035750..272fcd417 100644
--- a/pkg/operator/ceph/csi/spec_test.go
+++ b/pkg/operator/ceph/csi/spec_test.go
@@ -27,6 +27,11 @@ import (
 )
 
 func TestStartCSI(t *testing.T) {
+	RBDPluginTemplatePath = "csi-rbdplugin.yaml"
+	RBDProvisionerDepTemplatePath = "csi-rbdplugin-provisioner-dep.yaml"
+	CephFSPluginTemplatePath = "csi-cephfsplugin.yaml"
+	CephFSProvisionerDepTemplatePath = "csi-cephfsplugin-provisioner-dep.yaml"
+
 	CSIParam = Param{
 		CSIPluginImage:   "image",
 		RegistrarImage:   "image",
diff --git a/pkg/operator/ceph/csi/util_test.go b/pkg/operator/ceph/csi/util_test.go
index 840307e75..d211afd97 100644
--- a/pkg/operator/ceph/csi/util_test.go
+++ b/pkg/operator/ceph/csi/util_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package csi
 
 import (
+	"io/ioutil"
 	"os"
 	"testing"
 
@@ -24,21 +25,97 @@ import (
 	"k8s.io/client-go/kubernetes/fake"
 )
 
+var (
+	testDSTemplate = []byte(`
+kind: DaemonSet
+apiVersion: apps/v1
+metadata:
+  name: test-label
+  namespace: {{ .Namespace }}
+spec:
+  selector:
+    matchLabels:
+      app: test-label
+  template:
+    metadata:
+      labels:
+        app: test-label
+    spec:
+      serviceAccount: test-sa
+      containers:
+        - name: registrar
+          image: {{ .RegistrarImage }}
+        - name: rbdplugin
+          image: {{ .CSIPluginImage }}
+        - name: cephfsplugin
+          image: {{ .CSIPluginImage }}
+`)
+	testDepTemplate = []byte(`
+kind: Deployment
+apiVersion: apps/v1
+metadata:
+  name: test-label
+  namespace: {{ .Namespace }}
+spec:
+  replicas: {{ .ProvisionerReplicas }}
+  selector:
+    matchLabels:
+      app: test-label
+  template:
+    metadata:
+      labels:
+        app: test-label
+    spec:
+      serviceAccount: test-sa
+      containers:
+        - name: csi-attacher
+          image: {{ .AttacherImage }}
+        - name: csi-snapshotter
+          image: {{ .SnapshotterImage }}
+        - name: csi-resizer
+          image: {{ .ResizerImage }}
+        - name: csi-provisioner
+          image: {{ .ProvisionerImage }}
+        - name: csi-cephfsplugin
+          image: {{ .CSIPluginImage }}
+`)
+)
+
 func TestDaemonSetTemplate(t *testing.T) {
+	tmp, err := ioutil.TempFile("", "yaml")
+	assert.Nil(t, err)
+
+	defer os.Remove(tmp.Name())
+
+	_, err = tmp.Write(testDSTemplate)
+	assert.Nil(t, err)
+	err = tmp.Close()
+	assert.Nil(t, err)
+
 	tp := templateParam{
 		Param:     CSIParam,
 		Namespace: "foo",
 	}
-	_, err := templateToDaemonSet("test-ds", RBDPluginTemplatePath, tp)
+	_, err = templateToDaemonSet("test-ds", tmp.Name(), tp)
 	assert.Nil(t, err)
 }
 
 func TestDeploymentTemplate(t *testing.T) {
+	tmp, err := ioutil.TempFile("", "yaml")
+	assert.Nil(t, err)
+
+	defer os.Remove(tmp.Name())
+
+	_, err = tmp.Write(testDepTemplate)
+	assert.Nil(t, err)
+	err = tmp.Close()
+	assert.Nil(t, err)
+
 	tp := templateParam{
 		Param:     CSIParam,
 		Namespace: "foo",
 	}
-	_, err := templateToDeployment("test-dep", RBDProvisionerDepTemplatePath, tp)
+	_, err = templateToDeployment("test-dep", tmp.Name(), tp)
 	assert.Nil(t, err)
 }
 
diff --git a/pkg/operator/ceph/webhook_test.go b/pkg/operator/ceph/webhook_test.go
new file mode 100644
index 000000000..0b186ebf4
--- /dev/null
+++ b/pkg/operator/ceph/webhook_test.go
@@ -0,0 +1,134 @@
+/*
+Copyright 2020 The Rook Authors. All rights reserved.
+
+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 operator
+
+import (
+	"context"
+	"os"
+	"testing"
+
+	"github.com/rook/rook/pkg/operator/ceph/controller"
+	"github.com/stretchr/testify/assert"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+const (
+	testNamespace = "test-namespace"
+)
+
+func TestGetTolerations(t *testing.T) {
+	ctx := context.TODO()
+	clientset := fake.NewSimpleClientset()
+	os.Setenv("POD_NAMESPACE", testNamespace)
+
+	// No setting results in the default value.
+	cm := &v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      controller.OperatorSettingConfigMapName,
+			Namespace: testNamespace,
+		},
+		Data: map[string]string{},
+	}
+	_, err := clientset.CoreV1().ConfigMaps(testNamespace).Create(ctx, cm, metav1.CreateOptions{})
+	assert.NoError(t, err)
+	tolerations := getTolerations(clientset)
+	expected := []v1.Toleration{}
+	assert.Equal(t, expected, tolerations)
+
+	// The invalid setting results in the default value.
+	cm.Data = map[string]string{
+		admissionControllerTolerationsEnv: "",
+	}
+	_, err = clientset.CoreV1().ConfigMaps(testNamespace).Update(ctx, cm, metav1.UpdateOptions{})
+	assert.NoError(t, err)
+	tolerations = getTolerations(clientset)
+	assert.Equal(t, expected, tolerations)
+
+	// Correct setting result in the desired value.
+	cm.Data = map[string]string{
+		admissionControllerTolerationsEnv: `
+- effect: NoSchedule
+  key: node-role.kubernetes.io/controlplane
+  operator: Exists`,
+	}
+	_, err = clientset.CoreV1().ConfigMaps(testNamespace).Update(ctx, cm, metav1.UpdateOptions{})
+	assert.NoError(t, err)
+	tolerations = getTolerations(clientset)
+	expected = []v1.Toleration{
+		{
+			Effect:   "NoSchedule",
+			Key:      "node-role.kubernetes.io/controlplane",
+			Operator: v1.TolerationOpExists,
+		},
+	}
+	assert.Equal(t, expected, tolerations)
+}
+
+func TestGetNodeAffinity(t *testing.T) {
+	ctx := context.TODO()
+	clientset := fake.NewSimpleClientset()
+	os.Setenv("POD_NAMESPACE", testNamespace)
+
+	// No setting results in the default value.
+	cm := &v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      controller.OperatorSettingConfigMapName,
+			Namespace: testNamespace,
+		},
+		Data: map[string]string{},
+	}
+	_, err := clientset.CoreV1().ConfigMaps(testNamespace).Create(ctx, cm, metav1.CreateOptions{})
+	assert.NoError(t, err)
+	nodeAffinity := getNodeAffinity(clientset)
+	expected := &v1.NodeAffinity{}
+	assert.Equal(t, expected, nodeAffinity)
+
+	// The invalid setting results in the default value.
+	cm.Data = map[string]string{
+		admissionControllerNodeAffinityEnv: "",
+	}
+	_, err = clientset.CoreV1().ConfigMaps(testNamespace).Update(ctx, cm, metav1.UpdateOptions{})
+	assert.NoError(t, err)
+	nodeAffinity = getNodeAffinity(clientset)
+	assert.Equal(t, expected, nodeAffinity)
+
+	// Correct setting result in the desired value.
+	cm.Data = map[string]string{
+		admissionControllerNodeAffinityEnv: "role=storage-node",
+	}
+	_, err = clientset.CoreV1().ConfigMaps(testNamespace).Update(ctx, cm, metav1.UpdateOptions{})
+	assert.NoError(t, err)
+	nodeAffinity = getNodeAffinity(clientset)
+	expected = &v1.NodeAffinity{
+		RequiredDuringSchedulingIgnoredDuringExecution: &v1.NodeSelector{
+			NodeSelectorTerms: []v1.NodeSelectorTerm{
+				{
+					MatchExpressions: []v1.NodeSelectorRequirement{
+						{
+							Key:      "role",
+							Operator: v1.NodeSelectorOpIn,
+							Values:   []string{"storage-node"},
+						},
+					},
+				},
+			},
+		},
+	}
+	assert.Equal(t, expected, nodeAffinity)
+}
diff --git a/pkg/util/set_test.go b/pkg/util/set_test.go
deleted file mode 100644
index 3f4323743..000000000
--- a/pkg/util/set_test.go
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
-Copyright 2016 The Rook Authors. All rights reserved.
-
-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 util
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestSubtract(t *testing.T) {
-	set := CreateSet([]string{"a", "b", "x", "y", "z"})
-	subset := CreateSet([]string{"b", "z"})
-	set.Subtract(subset)
-	assert.Equal(t, 3, set.Count())
-	assert.True(t, set.Contains("a"))
-	assert.False(t, set.Contains("b"))
-	assert.True(t, set.Contains("x"))
-	assert.True(t, set.Contains("y"))
-	assert.False(t, set.Contains("z"))
-}
-
-func TestSubtractEmptySet(t *testing.T) {
-	// Both sets empty
-	set := NewSet()
-	subset := NewSet()
-	set.Subtract(subset)
-	assert.Equal(t, 0, set.Count())
-
-	// Subset is empty
-	set = CreateSet([]string{"1", "2"})
-	set.Subtract(subset)
-	assert.Equal(t, 2, set.Count())
-}
-
-func TestAddSingle(t *testing.T) {
-	set := NewSet()
-	assert.True(t, set.Add("foo"))
-	assert.False(t, set.Add("foo"))
-
-	assert.Equal(t, 1, set.Count())
-	assert.True(t, set.Contains("foo"))
-	assert.False(t, set.Contains("bar"))
-
-	assert.True(t, set.Add("bar"))
-	assert.Equal(t, 2, set.Count())
-	assert.True(t, set.Contains("foo"))
-	assert.True(t, set.Contains("bar"))
-	assert.False(t, set.Contains("baz"))
-}
-
-func TestAddMultiple(t *testing.T) {
-	set := NewSet()
-	set.AddMultiple([]string{"a", "b", "z"})
-	assert.Equal(t, 3, set.Count())
-	assert.True(t, set.Contains("a"))
-	assert.True(t, set.Contains("b"))
-	assert.False(t, set.Contains("c"))
-	assert.True(t, set.Contains("z"))
-}
-
-func TestToSlice(t *testing.T) {
-	set := CreateSet([]string{"1", "2", "3"})
-	arr := set.ToSlice()
-	assert.Equal(t, 3, len(arr))
-
-	// Empty set
-	set = CreateSet([]string{})
-	setSlice := set.ToSlice()
-	assert.NotNil(t, setSlice)
-	assert.Equal(t, 0, len(setSlice))
-}
-
-func TestCopy(t *testing.T) {
-	set := CreateSet([]string{"x", "y", "z"})
-	copySet := set.Copy()
-	assert.Equal(t, 3, copySet.Count())
-	assert.True(t, copySet.Contains("x"))
-	assert.True(t, copySet.Contains("y"))
-	assert.True(t, copySet.Contains("z"))
-	assert.False(t, copySet.Contains("a"))
-}
-
-func TestIter(t *testing.T) {
-	set := CreateSet([]string{"a", "b", "c", "x", "y", "z"})
-	count := 0
-	for range set.Iter() {
-		count++
-	}
-	assert.Equal(t, 6, count)
-}
-
-func TestSetEquals(t *testing.T) {
-	set := CreateSet([]string{"a", "b"})
-	assert.True(t, set.Equals(CreateSet([]string{"a", "b"})))
-	assert.False(t, set.Equals(CreateSet([]string{"a", "b", "c"})))
-	assert.False(t, set.Equals(CreateSet([]string{"a"})))
-	assert.False(t, set.Equals(CreateSet([]string{"a", "x"})))
-
-	set = CreateSet([]string{})
-	assert.True(t, set.Equals(CreateSet([]string{})))
-	assert.False(t, set.Equals(CreateSet([]string{"a"})))
-}
diff --git a/tests/framework/clients/bucket.go b/tests/framework/clients/bucket.go
index 8f3edeb86..70e73f8b5 100644
--- a/tests/framework/clients/bucket.go
+++ b/tests/framework/clients/bucket.go
@@ -53,10 +53,6 @@ func (b *BucketOperation) DeleteObc(obcName string, storageClassName string, buc
 	return b.k8sh.ResourceOperation("delete", b.manifests.GetOBC(obcName, storageClassName, bucketName, maxObject, createBucket))
 }
 
-func (b *BucketOperation) UpdateObc(obcName string, storageClassName string, bucketName string, maxObject string, createBucket bool) error {
-	return b.k8sh.ResourceOperation("apply", b.manifests.GetOBC(obcName, storageClassName, bucketName, maxObject, createBucket))
-}
-
 // CheckOBC, returns true if the obc, secret and configmap are all in the "check" state,
 // and returns false if any of these resources are not in the "check" state.
 // Check state values:
@@ -127,10 +123,3 @@ func (b *BucketOperation) GetSecretKey(obcName string) (string, error) {
 	return string(decode), nil
 
 }
-
-// Checks whether MaxObject is updated for ob
-func (b *BucketOperation) CheckOBMaxObject(obcName, maxobject string) bool {
-	obName, _ := b.k8sh.GetResource("obc", obcName, "--output", "jsonpath={.spec.objectBucketName}")
-	fetchMaxObject, _ := b.k8sh.GetResource("ob", obName, "--output", "jsonpath={.spec.endpoint.additionalConfig.maxObjects}")
-	return maxobject == fetchMaxObject
-}
diff --git a/tests/framework/installer/ceph_installer.go b/tests/framework/installer/ceph_installer.go
index ec6d60a51..9346c3712 100644
--- a/tests/framework/installer/ceph_installer.go
+++ b/tests/framework/installer/ceph_installer.go
@@ -466,6 +466,14 @@ func (h *CephInstaller) installRookOperator() (bool, error) {
 		return false, err
 	}
 
+	// disable admission controller test for Kubernetes version older than v1.16.0
+	if h.settings.EnableAdmissionController && !utils.IsPlatformOpenShift() && h.k8shelper.VersionAtLeast("v1.16.0") {
+		if !h.k8shelper.IsPodInExpectedState("rook-ceph-admission-controller", h.settings.OperatorNamespace, "Running") {
+			assert.Fail(h.T(), "admission controller is not running")
+			return false, errors.Errorf("admission controller is not running")
+		}
+	}
+
 	discovery, err := h.k8shelper.Clientset.AppsV1().DaemonSets(h.settings.OperatorNamespace).Get(ctx, "rook-discover", metav1.GetOptions{})
 	if startDiscovery {
 		assert.NoError(h.T(), err)
diff --git a/tests/integration/ceph_base_object_test.go b/tests/integration/ceph_base_object_test.go
index cb54a0165..6cc912f90 100644
--- a/tests/integration/ceph_base_object_test.go
+++ b/tests/integration/ceph_base_object_test.go
@@ -48,12 +48,10 @@ var (
 	ObjectKey1             = "rookObj1"
 	ObjectKey2             = "rookObj2"
 	ObjectKey3             = "rookObj3"
-	ObjectKey4             = "rookObj4"
 	contentType            = "plain/text"
 	obcName                = "smoke-delete-bucket"
 	region                 = "us-east-1"
 	maxObject              = "2"
-	newMaxObject           = "3"
 	bucketStorageClassName = "rook-smoke-delete-bucket"
 )
 
@@ -282,27 +280,11 @@ func testObjectStoreOperations(s suite.Suite, helper *clients.TestClient, k8sh *
 			assert.Error(s.T(), poErr)
 		})
 
-		t.Run("test update quota on OBC bucket", func(t *testing.T) {
-			poErr := helper.BucketClient.UpdateObc(obcName, bucketStorageClassName, bucketname, newMaxObject, true)
-			assert.Nil(s.T(), poErr)
-			updated := utils.Retry(5, 2*time.Second, "OBC is updated", func() bool {
-				return helper.BucketClient.CheckOBMaxObject(obcName, newMaxObject)
-			})
-			assert.True(s.T(), updated)
-			logger.Infof("Testing the updated object limit")
-			_, poErr = s3client.PutObjectInBucket(bucketname, ObjBody, ObjectKey3, contentType)
-			assert.NoError(s.T(), poErr)
-			_, poErr = s3client.PutObjectInBucket(bucketname, ObjBody, ObjectKey4, contentType)
-			assert.Error(s.T(), poErr)
-		})
-
 		t.Run("delete objects on OBC bucket", func(t *testing.T) {
 			_, delobjErr := s3client.DeleteObjectInBucket(bucketname, ObjectKey1)
 			assert.Nil(s.T(), delobjErr)
 			_, delobjErr = s3client.DeleteObjectInBucket(bucketname, ObjectKey2)
 			assert.Nil(s.T(), delobjErr)
-			_, delobjErr = s3client.DeleteObjectInBucket(bucketname, ObjectKey3)
-			assert.Nil(s.T(), delobjErr)
 			logger.Info("Objects deleted on bucket successfully")
 		})
 	})