diff --git a/.circleci/config.yml b/.circleci/config.yml
index 4653e676d5..46c437c89f 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -70,7 +70,8 @@ commands:
         type: string
       consul-k8s-image:
         type: string
-        default: "docker.mirror.hashicorp.services/hashicorpdev/consul-k8s-control-plane:latest"
+        #default: "docker.mirror.hashicorp.services/hashicorpdev/consul-k8s-control-plane:latest"
+        default: "kyleschochenmaier/consul-k8s-acls"
       go-path:
         type: string
         default: "/home/circleci/.go_workspace"
diff --git a/charts/consul/templates/connect-inject-clusterrole.yaml b/charts/consul/templates/connect-inject-clusterrole.yaml
index 9d01420363..ec8b288fda 100644
--- a/charts/consul/templates/connect-inject-clusterrole.yaml
+++ b/charts/consul/templates/connect-inject-clusterrole.yaml
@@ -1,4 +1,3 @@
-{{- if or (and (ne (.Values.connectInject.enabled | toString) "-") .Values.connectInject.enabled) (and (eq (.Values.connectInject.enabled | toString) "-") .Values.global.enabled) }}
 # The ClusterRole to enable the Connect injector to get, list, watch and patch MutatingWebhookConfiguration.
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
@@ -55,4 +54,3 @@ rules:
   verbs:
     - get
 {{- end }}
-{{- end }}
diff --git a/charts/consul/templates/connect-inject-clusterrolebinding.yaml b/charts/consul/templates/connect-inject-clusterrolebinding.yaml
index 64bff8269f..3a9d041852 100644
--- a/charts/consul/templates/connect-inject-clusterrolebinding.yaml
+++ b/charts/consul/templates/connect-inject-clusterrolebinding.yaml
@@ -1,4 +1,3 @@
-{{- if or (and (ne (.Values.connectInject.enabled | toString) "-") .Values.connectInject.enabled) (and (eq (.Values.connectInject.enabled | toString) "-") .Values.global.enabled) }}
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
@@ -17,4 +16,3 @@ subjects:
 - kind: ServiceAccount
   name: {{ template "consul.fullname" . }}-connect-injector
   namespace: {{ .Release.Namespace }}
-{{- end }}
diff --git a/charts/consul/templates/connect-inject-serviceaccount.yaml b/charts/consul/templates/connect-inject-serviceaccount.yaml
index 250b23d6c3..c95136914a 100644
--- a/charts/consul/templates/connect-inject-serviceaccount.yaml
+++ b/charts/consul/templates/connect-inject-serviceaccount.yaml
@@ -1,4 +1,3 @@
-{{- if or (and (ne (.Values.connectInject.enabled | toString) "-") .Values.connectInject.enabled) (and (eq (.Values.connectInject.enabled | toString) "-") .Values.global.enabled) }}
 apiVersion: v1
 kind: ServiceAccount
 metadata:
@@ -20,4 +19,3 @@ imagePullSecrets:
   - name: {{ .name }}
 {{- end }}
 {{- end }}
-{{- end }}
diff --git a/charts/consul/templates/controller-deployment.yaml b/charts/consul/templates/controller-deployment.yaml
index e5ed0d74f5..52234f5a68 100644
--- a/charts/consul/templates/controller-deployment.yaml
+++ b/charts/consul/templates/controller-deployment.yaml
@@ -47,16 +47,52 @@ spec:
     spec:
       {{- if or .Values.global.acls.manageSystemACLs (and .Values.global.tls.enabled .Values.global.tls.enableAutoEncrypt) }}
       initContainers:
+      {{- if (and .Values.global.tls.enabled .Values.global.tls.enableAutoEncrypt) }}
+      {{- include "consul.getAutoEncryptClientCA" . | nindent 6 }}
+      {{- end }}
       {{- if .Values.global.acls.manageSystemACLs }}
       - name: controller-acl-init
+        env:
+          - name: HOST_IP
+            valueFrom:
+              fieldRef:
+                fieldPath: status.hostIP
+          {{- if .Values.global.tls.enabled }}
+          - name: CONSUL_CACERT
+            value: /consul/tls/ca/tls.crt
+          {{- end }}
+          - name: CONSUL_HTTP_ADDR
+            {{- if .Values.global.tls.enabled }}
+            value: https://$(HOST_IP):8501
+            {{- else }}
+            value: http://$(HOST_IP):8500
+            {{- end }}
         image: {{ .Values.global.imageK8S }}
+        volumeMounts:
+        - mountPath: /consul/login
+          name: consul-data
+          readOnly: false
+        {{- if .Values.global.tls.enabled }}
+        {{- if .Values.global.tls.enableAutoEncrypt }}
+        - name: consul-auto-encrypt-ca-cert
+        {{- else }}
+        - name: consul-ca-cert
+        {{- end }}
+          mountPath: /consul/tls/ca
+          readOnly: true
+        {{- end }}
         command:
           - "/bin/sh"
           - "-ec"
           - |
             consul-k8s-control-plane acl-init \
-              -secret-name="{{ template "consul.fullname" . }}-controller-acl-token" \
-              -k8s-namespace={{ .Release.Namespace }}
+              -component-name=controller \
+              -acl-auth-method={{ template "consul.fullname" . }}-k8s-component-auth-method \
+              {{- if .Values.global.adminPartitions.enabled }}
+              -partition={{ .Values.global.adminPartitions.name }} \
+              {{- end }}
+              -log-level={{ default .Values.global.logLevel .Values.controller.logLevel }} \
+              -log-json={{ .Values.global.logJSON }} \
         resources:
           requests:
             memory: "25Mi"
@@ -65,9 +101,6 @@ spec:
             memory: "25Mi"
             cpu: "50m"
       {{- end }}
-      {{- if (and .Values.global.tls.enabled .Values.global.tls.enableAutoEncrypt) }}
-      {{- include "consul.getAutoEncryptClientCA" . | nindent 6 }}
-      {{- end }}
       {{- end }}
       containers:
       - command:
@@ -98,7 +131,24 @@ spec:
             -consul-cross-namespace-acl-policy=cross-namespace-policy \
             {{- end }}
             {{- end }}
+        {{- if .Values.global.acls.manageSystemACLs }}
+        lifecycle:
+          preStop:
+            exec:
+              command:
+                - "/bin/sh"
+                - "-ec"
+                - |
+                  consul-k8s-control-plane consul-logout \
+                  {{- if .Values.global.adminPartitions.enabled }}
+                  -partition={{ .Values.global.adminPartitions.name }} \
+                  {{- end }}
+        {{- end }}
         env:
+        {{- if .Values.global.acls.manageSystemACLs }}
+        - name: CONSUL_HTTP_TOKEN_FILE
+          value: "/consul/login/acl-token"
+        {{- end }}
         - name: HOST_IP
           valueFrom:
             fieldRef:
@@ -110,13 +160,6 @@ spec:
               name: {{ .Values.controller.aclToken.secretName }}
               key: {{ .Values.controller.aclToken.secretKey }}
         {{- end }}
-        {{- if .Values.global.acls.manageSystemACLs }}
-        - name: CONSUL_HTTP_TOKEN
-          valueFrom:
-            secretKeyRef:
-              name: "{{ template "consul.fullname" . }}-controller-acl-token"
-              key: "token"
-        {{- end}}
         {{- if .Values.global.tls.enabled }}
         - name: CONSUL_CACERT
           value: /consul/tls/ca/tls.crt
@@ -138,6 +181,9 @@ spec:
           {{- toYaml . | nindent 12 }}
         {{- end }}
         volumeMounts:
+        - mountPath: /consul/login
+          name: consul-data
+          readOnly: true
         - mountPath: /tmp/controller-webhook/certs
           name: cert
           readOnly: true
@@ -175,6 +221,9 @@ spec:
           medium: "Memory"
       {{- end }}
       {{- end }}
+      - name: consul-data
+        emptyDir:
+          medium: "Memory"
       serviceAccountName: {{ template "consul.fullname" . }}-controller
       {{- if .Values.controller.nodeSelector }}
       nodeSelector:
diff --git a/charts/consul/templates/server-acl-init-role.yaml b/charts/consul/templates/server-acl-init-role.yaml
index e828ae9b3f..43d610f7a7 100644
--- a/charts/consul/templates/server-acl-init-role.yaml
+++ b/charts/consul/templates/server-acl-init-role.yaml
@@ -19,7 +19,6 @@ rules:
     verbs:
       - create
       - get
-{{- if .Values.connectInject.enabled }}
   - apiGroups: [""]
     resources:
       - serviceaccounts
@@ -27,7 +26,6 @@ rules:
       - {{ template "consul.fullname" . }}-connect-injector
     verbs:
       - get
-{{- end }}
 {{- if .Values.global.enablePodSecurityPolicies }}
   - apiGroups: ["policy"]
     resources: ["podsecuritypolicies"]
diff --git a/charts/consul/test/unit/connect-inject-clusterrole.bats b/charts/consul/test/unit/connect-inject-clusterrole.bats
index db6649b33c..b2f1a35906 100644
--- a/charts/consul/test/unit/connect-inject-clusterrole.bats
+++ b/charts/consul/test/unit/connect-inject-clusterrole.bats
@@ -2,33 +2,6 @@
 
 load _helpers
 
-@test "connectInject/ClusterRole: disabled by default" {
-  cd `chart_dir`
-  assert_empty helm template \
-      -s templates/connect-inject-clusterrole.yaml  \
-      .
-}
-
-@test "connectInject/ClusterRole: enabled with global.enabled false" {
-  cd `chart_dir`
-  local actual=$(helm template \
-      -s templates/connect-inject-clusterrole.yaml  \
-      --set 'global.enabled=false' \
-      --set 'client.enabled=true' \
-      --set 'connectInject.enabled=true' \
-      . | tee /dev/stderr |
-      yq -s 'length > 0' | tee /dev/stderr)
-  [ "${actual}" = "true" ]
-}
-
-@test "connectInject/ClusterRole: disabled with connectInject.enabled" {
-  cd `chart_dir`
-  assert_empty helm template \
-      -s templates/connect-inject-clusterrole.yaml  \
-      --set 'connectInject.enabled=false' \
-      .
-}
-
 #--------------------------------------------------------------------
 # global.enablePodSecurityPolicies
 
diff --git a/charts/consul/test/unit/connect-inject-clusterrolebinding.bats b/charts/consul/test/unit/connect-inject-clusterrolebinding.bats
index 1f6fc94a88..6900cb2bd5 100644
--- a/charts/consul/test/unit/connect-inject-clusterrolebinding.bats
+++ b/charts/consul/test/unit/connect-inject-clusterrolebinding.bats
@@ -2,29 +2,3 @@
 
 load _helpers
 
-@test "connectInject/ClusterRoleBinding: disabled by default" {
-  cd `chart_dir`
-  assert_empty helm template \
-      -s templates/connect-inject-clusterrolebinding.yaml  \
-      .
-}
-
-@test "connectInject/ClusterRoleBinding: enabled with global.enabled false" {
-  cd `chart_dir`
-  local actual=$(helm template \
-      -s templates/connect-inject-clusterrolebinding.yaml  \
-      --set 'global.enabled=false' \
-      --set 'client.enabled=true' \
-      --set 'connectInject.enabled=true' \
-      . | tee /dev/stderr |
-      yq -s 'length > 0' | tee /dev/stderr)
-  [ "${actual}" = "true" ]
-}
-
-@test "connectInject/ClusterRoleBinding: disabled with connectInject.enabled false" {
-  cd `chart_dir`
-  assert_empty helm template \
-      -s templates/connect-inject-clusterrolebinding.yaml  \
-      --set 'connectInject.enabled=false' \
-      .
-}
diff --git a/charts/consul/test/unit/connect-inject-serviceaccount.bats b/charts/consul/test/unit/connect-inject-serviceaccount.bats
index 464a838b07..4c5fb8e316 100644
--- a/charts/consul/test/unit/connect-inject-serviceaccount.bats
+++ b/charts/consul/test/unit/connect-inject-serviceaccount.bats
@@ -2,32 +2,6 @@
 
 load _helpers
 
-@test "connectInject/ServiceAccount: disabled by default" {
-  cd `chart_dir`
-  assert_empty helm template \
-      -s templates/connect-inject-serviceaccount.yaml  \
-      .
-}
-
-@test "connectInject/ServiceAccount: enabled with global.enabled false" {
-  cd `chart_dir`
-  local actual=$(helm template \
-      -s templates/connect-inject-serviceaccount.yaml  \
-      --set 'global.enabled=false' \
-      --set 'client.enabled=true' \
-      --set 'connectInject.enabled=true' \
-      . | tee /dev/stderr |
-      yq -s 'length > 0' | tee /dev/stderr)
-  [ "${actual}" = "true" ]
-}
-
-@test "connectInject/ServiceAccount: disabled with connectInject.enabled false" {
-  cd `chart_dir`
-  assert_empty helm template \
-      -s templates/connect-inject-serviceaccount.yaml  \
-      --set 'connectInject.enabled=false' \
-      .
-}
 #--------------------------------------------------------------------
 # global.imagePullSecrets
 
diff --git a/charts/consul/test/unit/controller-deployment.bats b/charts/consul/test/unit/controller-deployment.bats
index 248811867d..5161ebeb6d 100644
--- a/charts/consul/test/unit/controller-deployment.bats
+++ b/charts/consul/test/unit/controller-deployment.bats
@@ -46,18 +46,53 @@ load _helpers
 #--------------------------------------------------------------------
 # global.acls.manageSystemACLs
 
-@test "controller/Deployment: CONSUL_HTTP_TOKEN env variable created when global.acls.manageSystemACLs=true" {
+@test "controller/Deployment: consul-logout preStop hook is added when ACLs are enabled" {
   cd `chart_dir`
   local actual=$(helm template \
       -s templates/controller-deployment.yaml \
       --set 'controller.enabled=true' \
       --set 'global.acls.manageSystemACLs=true' \
       . | tee /dev/stderr |
-      yq '[.spec.template.spec.containers[0].env[].name] | any(contains("CONSUL_HTTP_TOKEN"))' | tee /dev/stderr)
+      yq '[.spec.template.spec.containers[0].lifecycle.preStop.exec.command[2]] | any(contains("consul-k8s-control-plane consul-logout"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+}
+
+@test "controller/Deployment: consul-logout preStop hook has partition when partitions are enabled" {
+  cd `chart_dir`
+  local actual=$(helm template \
+      -s templates/controller-deployment.yaml \
+      --set 'controller.enabled=true' \
+      --set 'global.acls.manageSystemACLs=true' \
+      --set 'global.enableConsulNamespaces=true' \
+      --set 'global.adminPartitions.enabled=true' \
+      --set 'global.adminPartitions.name=default' \
+      . | tee /dev/stderr |
+      yq '[.spec.template.spec.containers[0].lifecycle.preStop.exec.command[2]] | any(contains("-partition=default"))' | tee /dev/stderr)
   [ "${actual}" = "true" ]
 }
 
-@test "controller/Deployment: init container is created when global.acls.manageSystemACLs=true" {
+@test "controller/Deployment: CONSUL_HTTP_TOKEN_FILE is not set when acls are disabled" {
+  cd `chart_dir`
+  local actual=$(helm template \
+      -s templates/controller-deployment.yaml \
+      --set 'controller.enabled=true' \
+      . | tee /dev/stderr |
+      yq '[.spec.template.spec.containers[0].env[0].name] | any(contains("CONSUL_HTTP_TOKEN_FILE"))' | tee /dev/stderr)
+  [ "${actual}" = "false" ]
+}
+
+@test "controller/Deployment: CONSUL_HTTP_TOKEN_FILE is set when acls are enabled" {
+  cd `chart_dir`
+  local actual=$(helm template \
+      -s templates/controller-deployment.yaml \
+      --set 'controller.enabled=true' \
+      --set 'global.acls.manageSystemACLs=true' \
+      . | tee /dev/stderr |
+      yq '[.spec.template.spec.containers[0].env[0].name] | any(contains("CONSUL_HTTP_TOKEN_FILE"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+}
+
+@test "controller/Deployment: init container is created when global.acls.manageSystemACLs=true and has correct command and environment with tls disabled" {
   cd `chart_dir`
   local object=$(helm template \
       -s templates/controller-deployment.yaml \
@@ -73,6 +108,139 @@ load _helpers
   local actual=$(echo $object |
       yq -r '.command | any(contains("consul-k8s-control-plane acl-init"))' | tee /dev/stderr)
   [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq '[.env[1].name] | any(contains("CONSUL_HTTP_ADDR"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq '[.env[1].value] | any(contains("http://$(HOST_IP):8500"))' | tee /dev/stderr)
+      echo $actual
+  [ "${actual}" = "true" ]
+}
+
+@test "controller/Deployment: init container is created when global.acls.manageSystemACLs=true and has correct command and environment with tls enabled" {
+  cd `chart_dir`
+  local object=$(helm template \
+      -s templates/controller-deployment.yaml \
+      --set 'controller.enabled=true' \
+      --set 'global.tls.enabled=true' \
+      --set 'global.acls.manageSystemACLs=true' \
+      . | tee /dev/stderr |
+      yq '.spec.template.spec.initContainers[] | select(.name == "controller-acl-init")' | tee /dev/stderr)
+
+  local actual=$(echo $object |
+      yq -r '.command | any(contains("consul-k8s-control-plane acl-init"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq '[.env[1].name] | any(contains("CONSUL_CACERT"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq '[.env[2].name] | any(contains("CONSUL_HTTP_ADDR"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq '[.env[2].value] | any(contains("https://$(HOST_IP):8501"))' | tee /dev/stderr)
+      echo $actual
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq '.volumeMounts[1] | any(contains("consul-ca-cert"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+}
+
+@test "controller/Deployment: init container is created when global.acls.manageSystemACLs=true and has correct command with Partitions enabled" {
+  cd `chart_dir`
+  local object=$(helm template \
+      -s templates/controller-deployment.yaml \
+      --set 'controller.enabled=true' \
+      --set 'global.tls.enabled=true' \
+      --set 'global.enableConsulNamespaces=true' \
+      --set 'global.adminPartitions.enabled=true' \
+      --set 'global.adminPartitions.name=default' \
+      --set 'global.acls.manageSystemACLs=true' \
+      . | tee /dev/stderr |
+      yq '.spec.template.spec.initContainers[] | select(.name == "controller-acl-init")' | tee /dev/stderr)
+
+  local actual=$(echo $object |
+      yq -r '.command | any(contains("consul-k8s-control-plane acl-init"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq -r '.command | any(contains("-acl-auth-method=RELEASE-NAME-consul-k8s-component-auth-method"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq -r '.command | any(contains("-partition=default"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq '[.env[1].name] | any(contains("CONSUL_CACERT"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq '[.env[2].name] | any(contains("CONSUL_HTTP_ADDR"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq '[.env[2].value] | any(contains("https://$(HOST_IP):8501"))' | tee /dev/stderr)
+      echo $actual
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq '.volumeMounts[1] | any(contains("consul-ca-cert"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+}
+
+@test "controller/Deployment: init container is created when global.acls.manageSystemACLs=true and has correct command and environment with tls enabled and autoencrypt enabled" {
+  cd `chart_dir`
+  local object=$(helm template \
+      -s templates/controller-deployment.yaml \
+      --set 'controller.enabled=true' \
+      --set 'global.tls.enabled=true' \
+      --set 'global.tls.enableAutoEncrypt=true' \
+      --set 'global.acls.manageSystemACLs=true' \
+      . | tee /dev/stderr |
+      yq '.spec.template.spec.initContainers[] | select(.name == "controller-acl-init")' | tee /dev/stderr)
+
+  local actual=$(echo $object |
+      yq -r '.command | any(contains("consul-k8s-control-plane acl-init"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq '[.env[1].name] | any(contains("CONSUL_CACERT"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq '[.env[2].name] | any(contains("CONSUL_HTTP_ADDR"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq '[.env[2].value] | any(contains("https://$(HOST_IP):8501"))' | tee /dev/stderr)
+      echo $actual
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+      yq '.volumeMounts[1] | any(contains("consul-auto-encrypt-ca-cert"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+}
+
+@test "controller/Deployment: auto-encrypt init container is created and is the first init-container when global.acls.manageSystemACLs=true and has correct command and environment with tls enabled and autoencrypt enabled" {
+  cd `chart_dir`
+  local object=$(helm template \
+      -s templates/controller-deployment.yaml \
+      --set 'controller.enabled=true' \
+      --set 'global.tls.enabled=true' \
+      --set 'global.tls.enableAutoEncrypt=true' \
+      --set 'global.acls.manageSystemACLs=true' \
+      . | tee /dev/stderr |
+      yq '.spec.template.spec.initContainers[0]' | tee /dev/stderr)
+
+  local actual=$(echo $object |
+      yq -r '.name' | tee /dev/stderr)
+  [ "${actual}" = "get-auto-encrypt-client-ca" ]
 }
 
 #--------------------------------------------------------------------
@@ -486,38 +654,37 @@ load _helpers
 #--------------------------------------------------------------------
 # aclToken
 
-@test "controller/Deployment: aclToken disabled when secretName is missing" {
+@test "controller/Deployment: aclToken enabled when secretName and secretKey is provided" {
   cd `chart_dir`
   local actual=$(helm template \
       -s templates/controller-deployment.yaml  \
       --set 'controller.enabled=true' \
+      --set 'controller.aclToken.secretName=foo' \
       --set 'controller.aclToken.secretKey=bar' \
       . | tee /dev/stderr |
       yq '[.spec.template.spec.containers[0].env[].name] | any(contains("CONSUL_HTTP_TOKEN"))' | tee /dev/stderr)
-  [ "${actual}" = "false" ]
+  [ "${actual}" = "true" ]
 }
 
-@test "controller/Deployment: aclToken disabled when secretKey is missing" {
+@test "controller/Deployment: aclToken env is set when ACLs are enabled" {
   cd `chart_dir`
   local actual=$(helm template \
       -s templates/controller-deployment.yaml  \
       --set 'controller.enabled=true' \
-      --set 'controller.aclToken.secretName=foo' \
+      --set 'global.acls.manageSystemACLs=true' \
       . | tee /dev/stderr |
       yq '[.spec.template.spec.containers[0].env[].name] | any(contains("CONSUL_HTTP_TOKEN"))' | tee /dev/stderr)
-  [ "${actual}" = "false" ]
+  [ "${actual}" = "true" ]
 }
 
-@test "controller/Deployment: aclToken enabled when secretName and secretKey is provided" {
+@test "controller/Deployment: aclToken env is not set when ACLs are disabled" {
   cd `chart_dir`
   local actual=$(helm template \
       -s templates/controller-deployment.yaml  \
       --set 'controller.enabled=true' \
-      --set 'controller.aclToken.secretName=foo' \
-      --set 'controller.aclToken.secretKey=bar' \
       . | tee /dev/stderr |
       yq '[.spec.template.spec.containers[0].env[].name] | any(contains("CONSUL_HTTP_TOKEN"))' | tee /dev/stderr)
-  [ "${actual}" = "true" ]
+  [ "${actual}" = "false" ]
 }
 
 #--------------------------------------------------------------------
@@ -528,11 +695,16 @@ load _helpers
   local cmd=$(helm template \
       -s templates/controller-deployment.yaml  \
       --set 'controller.enabled=true' \
+      --set 'global.acls.manageSystemACLs=true' \
       . | tee /dev/stderr |
-      yq '.spec.template.spec.containers[0].command' | tee /dev/stderr)
+      yq '.spec.template.spec' | tee /dev/stderr)
 
   local actual=$(echo "$cmd" |
-    yq 'any(contains("-log-level=info"))' | tee /dev/stderr)
+    yq '.containers[0].command | any(contains("-log-level=info"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo "$cmd" |
+    yq '.initContainers[0].command | any(contains("-log-level=info"))' | tee /dev/stderr)
   [ "${actual}" = "true" ]
 }
 
@@ -542,11 +714,16 @@ load _helpers
       -s templates/controller-deployment.yaml  \
       --set 'controller.enabled=true' \
       --set 'controller.logLevel=error' \
+      --set 'global.acls.manageSystemACLs=true' \
       . | tee /dev/stderr |
-      yq '.spec.template.spec.containers[0].command' | tee /dev/stderr)
+      yq '.spec.template.spec' | tee /dev/stderr)
+
+  local actual=$(echo "$cmd" |
+    yq '.containers[0].command | any(contains("-log-level=error"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
 
   local actual=$(echo "$cmd" |
-    yq 'any(contains("-log-level=error"))' | tee /dev/stderr)
+    yq '.initContainers[0].command | any(contains("-log-level=error"))' | tee /dev/stderr)
   [ "${actual}" = "true" ]
 }
 
diff --git a/control-plane/commands.go b/control-plane/commands.go
index db43863642..8d5e8de23e 100644
--- a/control-plane/commands.go
+++ b/control-plane/commands.go
@@ -5,6 +5,7 @@ import (
 
 	cmdACLInit "github.com/hashicorp/consul-k8s/control-plane/subcommand/acl-init"
 	cmdConnectInit "github.com/hashicorp/consul-k8s/control-plane/subcommand/connect-init"
+	cmdConsulLogout "github.com/hashicorp/consul-k8s/control-plane/subcommand/consul-logout"
 	cmdConsulSidecar "github.com/hashicorp/consul-k8s/control-plane/subcommand/consul-sidecar"
 	cmdController "github.com/hashicorp/consul-k8s/control-plane/subcommand/controller"
 	cmdCreateFederationSecret "github.com/hashicorp/consul-k8s/control-plane/subcommand/create-federation-secret"
@@ -46,6 +47,10 @@ func init() {
 			return &cmdConsulSidecar.Command{UI: ui}, nil
 		},
 
+		"consul-logout": func() (cli.Command, error) {
+			return &cmdConsulLogout.Command{UI: ui}, nil
+		},
+
 		"server-acl-init": func() (cli.Command, error) {
 			return &cmdServerACLInit.Command{UI: ui}, nil
 		},
diff --git a/control-plane/helper/test/test_util.go b/control-plane/helper/test/test_util.go
index 365dc54363..c804efdf02 100644
--- a/control-plane/helper/test/test_util.go
+++ b/control-plane/helper/test/test_util.go
@@ -15,6 +15,10 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+const (
+	componentAuthMethod = "consul-k8s-component-auth-method"
+)
+
 // GenerateServerCerts generates Consul CA
 // and a server certificate and saves them to temp files.
 // It returns file names in this order:
@@ -57,6 +61,79 @@ func GenerateServerCerts(t *testing.T) (string, string, string) {
 	return caFile.Name(), certFile.Name(), certKeyFile.Name()
 }
 
+// SetupK8sComponentAuthMethod creates a k8s auth method, sample "acl:write" ACL policy, Role and BindingRule
+// that allows a client using serviceAccount's JWT token to call "consul login".
+func SetupK8sComponentAuthMethod(t *testing.T, consulClient *api.Client, serviceAccountName, k8sComponentNS string) {
+	t.Helper()
+	// Start the mock k8s server.
+	k8sMockServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("content-type", "application/json")
+		if r != nil && r.URL.Path == "/apis/authentication.k8s.io/v1/tokenreviews" && r.Method == "POST" {
+			w.Write([]byte(tokenReviewsResponse(serviceAccountName, k8sComponentNS)))
+		}
+		if r != nil && r.URL.Path == fmt.Sprintf("/api/v1/namespaces/%s/serviceaccounts/%s", k8sComponentNS, serviceAccountName) &&
+			r.Method == "GET" {
+			w.Write([]byte(serviceAccountGetResponse(serviceAccountName, k8sComponentNS)))
+		}
+	}))
+	t.Cleanup(k8sMockServer.Close)
+
+	// Set up Component's auth method.
+	authMethodTmpl := api.ACLAuthMethod{
+		Name:        componentAuthMethod,
+		Type:        "kubernetes",
+		Description: "Kubernetes Auth Method",
+		Config: map[string]interface{}{
+			"Host":              k8sMockServer.URL,
+			"CACert":            serviceAccountCACert,
+			"ServiceAccountJWT": ServiceAccountJWTToken,
+		},
+	}
+	// This API call will idempotently create the auth method (it won't fail if it already exists).
+	_, _, err := consulClient.ACL().AuthMethodCreate(&authMethodTmpl, nil)
+	require.NoError(t, err)
+
+	rules := `acl = "write"`
+	policyName := fmt.Sprintf("%s-token", serviceAccountName)
+	policy := api.ACLPolicy{
+		Name:        policyName,
+		Description: fmt.Sprintf("%s Token Policy", policyName),
+		Rules:       rules,
+		Datacenters: []string{"dc1"},
+	}
+	_, _, err = consulClient.ACL().PolicyCreate(&policy, &api.WriteOptions{})
+	require.NoError(t, err)
+
+	// Create the ACL Role, it requires an ACLRolePolicyLink which contains a list
+	// of ACL policies that are allowed to be fetched by an associated ACLBindingRule.
+	ap := &api.ACLRolePolicyLink{
+		Name: policyName,
+	}
+	apl := []*api.ACLRolePolicyLink{}
+	apl = append(apl, ap)
+	aclRoleName := fmt.Sprintf("%s-acl-role", serviceAccountName)
+	role := &api.ACLRole{
+		Name:        aclRoleName,
+		Description: fmt.Sprintf("ACL Role for %s", serviceAccountName),
+		Policies:    apl,
+	}
+	_, _, err = consulClient.ACL().RoleCreate(role, &api.WriteOptions{})
+	require.NoError(t, err)
+
+	// Create the ACLBindingRule, this specifies that a user using the AuthMethod
+	// is able to request an ACL Token with associated ACLRole from above via BindName
+	// as long as its serviceaccount matches the Selector.
+	abr := api.ACLBindingRule{
+		Description: fmt.Sprintf("Binding Rule for %s", serviceAccountName),
+		AuthMethod:  componentAuthMethod,
+		Selector:    fmt.Sprintf("serviceaccount.name==%q", serviceAccountName),
+		BindType:    api.BindingRuleBindTypeRole,
+		BindName:    aclRoleName,
+	}
+	_, _, err = consulClient.ACL().BindingRuleCreate(&abr, nil)
+	require.NoError(t, err)
+}
+
 // SetupK8sAuthMethod create a k8s auth method and a binding rule in Consul for the
 // given k8s service and namespace.
 func SetupK8sAuthMethod(t *testing.T, consulClient *api.Client, serviceName, k8sServiceNS string) {
diff --git a/control-plane/subcommand/acl-init/command.go b/control-plane/subcommand/acl-init/command.go
index 6017137bbf..b4e940db08 100644
--- a/control-plane/subcommand/acl-init/command.go
+++ b/control-plane/subcommand/acl-init/command.go
@@ -12,53 +12,87 @@ import (
 	"text/template"
 	"time"
 
+	"github.com/hashicorp/consul-k8s/control-plane/consul"
 	"github.com/hashicorp/consul-k8s/control-plane/subcommand"
+	"github.com/hashicorp/consul-k8s/control-plane/subcommand/common"
 	"github.com/hashicorp/consul-k8s/control-plane/subcommand/flags"
+	"github.com/hashicorp/consul/api"
+	"github.com/hashicorp/go-hclog"
 	"github.com/mitchellh/cli"
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 )
 
+const (
+	defaultBearerTokenFile = "/var/run/secrets/kubernetes.io/serviceaccount/token"
+	defaultTokenSinkFile   = "/consul/login/acl-token"
+)
+
 type Command struct {
 	UI cli.Ui
 
-	flags             *flag.FlagSet
-	k8s               *flags.K8SFlags
+	flags *flag.FlagSet
+	k8s   *flags.K8SFlags
+	http  *flags.HTTPFlags
+
 	flagSecretName    string
 	flagInitType      string
 	flagNamespace     string
 	flagACLDir        string
 	flagTokenSinkFile string
 
+	flagACLAuthMethod string // Auth Method to use for ACLs.
+	flagLogLevel      string
+	flagLogJSON       bool
+
+	bearerTokenFile   string // Location of the bearer token. Default is defaultBearerTokenFile.
+	flagComponentName string // Name of the component to be used as metadata to ACL Login.
+
 	k8sClient kubernetes.Interface
 
-	once sync.Once
-	help string
+	once   sync.Once
+	help   string
+	logger hclog.Logger
 
-	ctx context.Context
+	ctx          context.Context
+	consulClient *api.Client
 }
 
 func (c *Command) init() {
 	c.flags = flag.NewFlagSet("", flag.ContinueOnError)
+
 	c.flags.StringVar(&c.flagSecretName, "secret-name", "",
 		"Name of secret to watch for an ACL token")
 	c.flags.StringVar(&c.flagInitType, "init-type", "",
 		"ACL init type. The only supported value is 'client'. If set to 'client' will write Consul client ACL config to an acl-config.json file in -acl-dir")
-	c.flags.StringVar(&c.flagNamespace, "k8s-namespace", "",
-		"Name of Kubernetes namespace where the servers are deployed")
 	c.flags.StringVar(&c.flagACLDir, "acl-dir", "/consul/aclconfig",
 		"Directory name of shared volume where client acl config file acl-config.json will be written if -init-type=client")
 	c.flags.StringVar(&c.flagTokenSinkFile, "token-sink-file", "",
 		"Optional filepath to write acl token")
 
+	// Flags related to using consul login to fetch the ACL token.
+	c.flags.StringVar(&c.flagNamespace, "k8s-namespace", "", "Name of Kubernetes namespace where the token Kubernetes secret is stored.")
+	c.flags.StringVar(&c.flagACLAuthMethod, "acl-auth-method", "", "Name of the auth method to login with.")
+	c.flags.StringVar(&c.flagComponentName, "component-name", "",
+		"Name of the component to pass to ACL Login as metadata.")
+	c.flags.StringVar(&c.flagLogLevel, "log-level", "info",
+		"Log verbosity level. Supported values (in order of detail) are \"trace\", "+
+			"\"debug\", \"info\", \"warn\", and \"error\".")
+	c.flags.BoolVar(&c.flagLogJSON, "log-json", false,
+		"Enable or disable JSON output format for logging.")
+
 	c.k8s = &flags.K8SFlags{}
+	c.http = &flags.HTTPFlags{}
 	flags.Merge(c.flags, c.k8s.Flags())
+	flags.Merge(c.flags, c.http.Flags())
 	c.help = flags.Usage(help, c.flags)
 }
 
 func (c *Command) Run(args []string) int {
+	var err error
 	c.once.Do(c.init)
-	if err := c.flags.Parse(args); err != nil {
+	if err = c.flags.Parse(args); err != nil {
 		return 1
 	}
 	if len(c.flags.Args()) > 0 {
@@ -66,6 +100,20 @@ func (c *Command) Run(args []string) int {
 		return 1
 	}
 
+	if c.bearerTokenFile == "" {
+		c.bearerTokenFile = defaultBearerTokenFile
+	}
+	// This allows us to utilize the default path of `/consul/login/acl-token` for the ACL token
+	// but only in the case of when we're using ACL.Login. If flagACLAuthMethod is not set and
+	// the tokenSinkFile is also unset it means we do not want to write an ACL token in the case
+	// of the client token.
+	if c.flagTokenSinkFile == "" && c.flagACLAuthMethod != "" {
+		c.flagTokenSinkFile = defaultTokenSinkFile
+	}
+	if c.flagNamespace == "" {
+		c.flagNamespace = corev1.NamespaceDefault
+	}
+
 	if c.ctx == nil {
 		c.ctx = context.Background()
 	}
@@ -84,6 +132,35 @@ func (c *Command) Run(args []string) int {
 		}
 	}
 
+	// Set up logging.
+	if c.logger == nil {
+		c.logger, err = common.Logger(c.flagLogLevel, c.flagLogJSON)
+		if err != nil {
+			c.UI.Error(err.Error())
+			return 1
+		}
+	}
+
+	if c.flagACLAuthMethod != "" {
+		cfg := api.DefaultConfig()
+		c.http.MergeOntoConfig(cfg)
+		c.consulClient, err = consul.NewClient(cfg)
+		if err != nil {
+			c.logger.Error("Unable to get client connection", "error", err)
+			return 1
+		}
+
+		meta := map[string]string{
+			"component": c.flagComponentName,
+		}
+		err := common.ConsulLogin(c.consulClient, cfg, c.logger, c.bearerTokenFile, c.flagACLAuthMethod, c.flagTokenSinkFile, "", "", meta)
+		if err != nil {
+			c.logger.Error("Consul login failed", "error", err)
+			return 1
+		}
+		c.logger.Info("Consul login succeeded")
+		return 0
+	}
 	// Check if the client secret exists yet
 	// If not, wait until it does
 	var secret string
@@ -91,9 +168,10 @@ func (c *Command) Run(args []string) int {
 		var err error
 		secret, err = c.getSecret(c.flagSecretName)
 		if err != nil {
-			c.UI.Error(fmt.Sprintf("Error getting Kubernetes secret: %s", err))
+			c.logger.Error("Error getting Kubernetes secret", "error", err)
 		}
 		if err == nil {
+			c.logger.Info("Successfully read Kubernetes secret")
 			break
 		}
 		time.Sleep(1 * time.Second)
@@ -106,7 +184,7 @@ func (c *Command) Run(args []string) int {
 		tpl := template.Must(template.New("root").Parse(strings.TrimSpace(clientACLConfigTpl)))
 		err := tpl.Execute(&buf, secret)
 		if err != nil {
-			c.UI.Error(fmt.Sprintf("Error creating template: %s", err))
+			c.logger.Error("Error creating template", "error", err)
 			return 1
 		}
 
@@ -115,7 +193,7 @@ func (c *Command) Run(args []string) int {
 		// to be readable by the consul user.
 		err = ioutil.WriteFile(filepath.Join(c.flagACLDir, "acl-config.json"), buf.Bytes(), 0644)
 		if err != nil {
-			c.UI.Error(fmt.Sprintf("Error writing config file: %s", err))
+			c.logger.Error("Error writing config file", "error", err)
 			return 1
 		}
 	}
@@ -125,7 +203,7 @@ func (c *Command) Run(args []string) int {
 		// to have permissions to overwrite our file.
 		err := ioutil.WriteFile(c.flagTokenSinkFile, []byte(secret), 0600)
 		if err != nil {
-			c.UI.Error(fmt.Sprintf("Error writing token to file %q: %s", c.flagTokenSinkFile, err))
+			c.logger.Error("Error writing token to file", "file", c.flagTokenSinkFile, "error", err)
 			return 1
 		}
 	}
diff --git a/control-plane/subcommand/acl-init/command_test.go b/control-plane/subcommand/acl-init/command_test.go
index 0a3a7ab8bf..e0804b64c8 100644
--- a/control-plane/subcommand/acl-init/command_test.go
+++ b/control-plane/subcommand/acl-init/command_test.go
@@ -2,12 +2,16 @@ package aclinit
 
 import (
 	"context"
+	"fmt"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 	"testing"
 
+	"github.com/hashicorp/consul-k8s/control-plane/helper/test"
 	"github.com/hashicorp/consul-k8s/control-plane/subcommand/common"
+	"github.com/hashicorp/consul/api"
+	"github.com/hashicorp/consul/sdk/testutil"
 	"github.com/mitchellh/cli"
 	"github.com/stretchr/testify/require"
 	v1 "k8s.io/api/core/v1"
@@ -15,6 +19,10 @@ import (
 	"k8s.io/client-go/kubernetes/fake"
 )
 
+const (
+	componentAuthMethod = "consul-k8s-component-auth-method"
+)
+
 // Test that we write the secret data to a file.
 func TestRun_TokenSinkFile(t *testing.T) {
 	t.Parallel()
@@ -50,7 +58,6 @@ func TestRun_TokenSinkFile(t *testing.T) {
 		k8sClient: k8s,
 	}
 	code := cmd.Run([]string{
-		"-k8s-namespace", k8sNS,
 		"-token-sink-file", sinkFile,
 		"-secret-name", secretName,
 	})
@@ -91,15 +98,11 @@ func TestRun_TokenSinkFileErr(t *testing.T) {
 		k8sClient: k8s,
 	}
 	code := cmd.Run([]string{
-		"-k8s-namespace", k8sNS,
 		"-token-sink-file", "/this/filepath/does/not/exist",
 		"-secret-name", secretName,
 	})
 
 	require.Equal(1, code)
-	require.Contains(ui.ErrorWriter.String(),
-		`Error writing token to file "/this/filepath/does/not/exist": open /this/filepath/does/not/exist: no such file or directory`,
-	)
 }
 
 // Test that if the command is run twice it succeeds. This test is the result
@@ -142,7 +145,6 @@ func TestRun_TokenSinkFileTwice(t *testing.T) {
 	// Run twice.
 	for i := 0; i < 2; i++ {
 		code := cmd.Run([]string{
-			"-k8s-namespace", k8sNS,
 			"-token-sink-file", sinkFile,
 			"-secret-name", secretName,
 		})
@@ -153,3 +155,58 @@ func TestRun_TokenSinkFileTwice(t *testing.T) {
 		require.Equal(token, string(bytes), "exp: %s, got: %s", token, string(bytes))
 	}
 }
+
+// TestRun_PerformsConsulLogin executes the consul login path and validates the token
+// is written to disk.
+func TestRun_PerformsConsulLogin(t *testing.T) {
+	// This is the test file that we will write the token to so consul-logout can read it.
+	tokenFile := common.WriteTempFile(t, "")
+	bearerFile := common.WriteTempFile(t, test.ServiceAccountJWTToken)
+
+	k8s := fake.NewSimpleClientset()
+
+	// Start Consul server with ACLs enabled and default deny policy.
+	masterToken := "b78d37c7-0ca7-5f4d-99ee-6d9975ce4586"
+	server, err := testutil.NewTestServerConfigT(t, func(c *testutil.TestServerConfig) {
+		c.ACL.Enabled = true
+		c.ACL.DefaultPolicy = "deny"
+		c.ACL.Tokens.InitialManagement = masterToken
+	})
+	require.NoError(t, err)
+	defer server.Stop()
+	server.WaitForLeader(t)
+	cfg := &api.Config{
+		Scheme:  "http",
+		Address: server.HTTPAddr,
+		Token:   masterToken,
+	}
+	consulClient, err := api.NewClient(cfg)
+	require.NoError(t, err)
+
+	// Set up the Component Auth Method, this pre-loads Consul with bindingrule, roles and an acl:write policy so we
+	// can issue an ACL.Login().
+	test.SetupK8sComponentAuthMethod(t, consulClient, "test-sa", "default")
+
+	ui := cli.NewMockUi()
+	cmd := Command{
+		UI:              ui,
+		k8sClient:       k8s,
+		bearerTokenFile: bearerFile,
+	}
+
+	code := cmd.Run([]string{
+		"-token-sink-file", tokenFile,
+		"-acl-auth-method", componentAuthMethod,
+		"-component-name", "foo",
+		"-http-addr", fmt.Sprintf("%s://%s", cfg.Scheme, cfg.Address),
+	})
+	require.Equal(t, 0, code, ui.ErrorWriter.String())
+	// Validate the Token got written.
+	tokenBytes, err := ioutil.ReadFile(tokenFile)
+	require.NoError(t, err)
+	require.Equal(t, 36, len(tokenBytes))
+	// Validate the Token and its Description.
+	tok, _, err := consulClient.ACL().TokenReadSelf(&api.QueryOptions{Token: string(tokenBytes)})
+	require.NoError(t, err)
+	require.Equal(t, "token created via login: {\"component\":\"foo\"}", tok.Description)
+}
diff --git a/control-plane/subcommand/common/common.go b/control-plane/subcommand/common/common.go
index bd60b5822c..b7929f2dae 100644
--- a/control-plane/subcommand/common/common.go
+++ b/control-plane/subcommand/common/common.go
@@ -7,8 +7,11 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"time"
 
+	"github.com/cenkalti/backoff"
 	"github.com/go-logr/logr"
+	"github.com/hashicorp/consul-k8s/control-plane/consul"
 	godiscover "github.com/hashicorp/consul-k8s/control-plane/helper/go-discover"
 	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/go-discover"
@@ -31,6 +34,12 @@ const (
 	// which secrets to delete on an uninstall.
 	CLILabelKey   = "managed-by"
 	CLILabelValue = "consul-k8s"
+
+	// The number of times to attempt ACL Login.
+	numLoginRetries = 100
+
+	raftReplicationTimeout   = 2 * time.Second
+	tokenReadPollingInterval = 100 * time.Millisecond
 )
 
 // Logger returns an hclog instance with log level set and JSON logging enabled/disabled, or an error if level is invalid.
@@ -79,10 +88,8 @@ func ValidateUnprivilegedPort(flagName, flagValue string) error {
 
 // ConsulLogin issues an ACL().Login to Consul and writes out the token to tokenSinkFile.
 // The logic of this is taken from the `consul login` command.
-func ConsulLogin(client *api.Client, bearerTokenFile, authMethodName, tokenSinkFile, namespace string, meta map[string]string) error {
-	if meta == nil {
-		return fmt.Errorf("invalid meta")
-	}
+func ConsulLogin(client *api.Client, cfg *api.Config, log hclog.Logger, bearerTokenFile, authMethodName, tokenSinkFile, namespace string, serviceAccountName string, meta map[string]string) error {
+	// Read the bearerTokenFile.
 	data, err := ioutil.ReadFile(bearerTokenFile)
 	if err != nil {
 		return fmt.Errorf("unable to read bearerTokenFile: %v, err: %v", bearerTokenFile, err)
@@ -91,20 +98,85 @@ func ConsulLogin(client *api.Client, bearerTokenFile, authMethodName, tokenSinkF
 	if bearerToken == "" {
 		return fmt.Errorf("no bearer token found in %s", bearerTokenFile)
 	}
-	// Do the login.
-	req := &api.ACLLoginParams{
-		AuthMethod:  authMethodName,
-		BearerToken: bearerToken,
-		Meta:        meta,
+	err = backoff.Retry(func() error {
+		// Do the login.
+		req := &api.ACLLoginParams{
+			AuthMethod:  authMethodName,
+			BearerToken: bearerToken,
+			Meta:        meta,
+		}
+		tok, _, err := client.ACL().Login(req, &api.WriteOptions{Namespace: namespace})
+		if err != nil {
+			log.Error("unable to login", "error", err)
+			return fmt.Errorf("error logging in: %s", err)
+		}
+		// Write out the resultant token file.
+		if err := WriteFileWithPerms(tokenSinkFile, tok.SecretID, 0444); err != nil {
+			return fmt.Errorf("error writing token to file sink: %v", err)
+		}
+		return err
+	}, backoff.WithMaxRetries(backoff.NewConstantBackOff(1*time.Second), numLoginRetries))
+	if err != nil {
+		if serviceAccountName == "default" {
+			log.Warn("The service account name for this Pod is \"default\"." +
+				" In default installations this is not a supported service account name." +
+				" The service account name must match the name of the Kubernetes Service" +
+				" or the consul.hashicorp.com/connect-service annotation.")
+		}
+		log.Error("Hit maximum retries for consul login", "error", err)
+		return err
 	}
-	tok, _, err := client.ACL().Login(req, &api.WriteOptions{Namespace: namespace})
+	// Now update the client so that it will read the ACL token we just fetched.
+	cfg.TokenFile = tokenSinkFile
+	client, err = consul.NewClient(cfg)
 	if err != nil {
-		return fmt.Errorf("error logging in: %s", err)
+		log.Error("Unable to update client connection", "error", err)
+		return err
 	}
+	log.Info("Consul login complete")
 
-	if err := WriteFileWithPerms(tokenSinkFile, tok.SecretID, 0444); err != nil {
-		return fmt.Errorf("error writing token to file sink: %v", err)
+	// A workaround to check that the ACL token is replicated to other Consul servers.
+	//
+	// A consul client may reach out to a follower instead of a leader to resolve the token during the
+	// call to get services below. This is because clients talk to servers in the stale consistency mode
+	// to decrease the load on the servers (see https://www.consul.io/docs/architecture/consensus#stale).
+	// In that case, it's possible that the token isn't replicated
+	// to that server instance yet. The client will then get an "ACL not found" error
+	// and subsequently cache this not found response. Then our call below
+	// to get services from the agent will keep hitting the same "ACL not found" error
+	// until the cache entry expires (determined by the `acl_token_ttl` which defaults to 30 seconds).
+	// This is not great because it will delay app start up time by 30 seconds in most cases
+	// (if you are running 3 servers, then the probability of ending up on a follower is close to 2/3).
+	//
+	// To help with that, we try to first read the token in the stale consistency mode until we
+	// get a successful response. This should not take more than 100ms because raft replication
+	// should in most cases take less than that (see https://www.consul.io/docs/install/performance#read-write-tuning)
+	// but we set the timeout to 2s to be sure.
+	//
+	// Note though that this workaround does not eliminate this problem completely. It's still possible
+	// for this call and the next call to reach different servers and those servers to have different
+	// states from each other.
+	// For example, this call can reach a leader and succeed, while the call below can go to a follower
+	// that is still behind the leader and get an "ACL not found" error.
+	// However, this is a pretty unlikely case because
+	// clients have sticky connections to a server, and those connections get rebalanced only every 2-3min.
+	// And so, this workaround should work in a vast majority of cases.
+	log.Info("Checking that the ACL token exists when reading it in the stale consistency mode")
+	// Use raft timeout and polling interval to determine the number of retries.
+	numTokenReadRetries := uint64(raftReplicationTimeout.Milliseconds() / tokenReadPollingInterval.Milliseconds())
+	err = backoff.Retry(func() error {
+		_, _, err := client.ACL().TokenReadSelf(&api.QueryOptions{AllowStale: true})
+		if err != nil {
+			log.Error("Unable to read ACL token; retrying", "err", err)
+		}
+		return err
+	}, backoff.WithMaxRetries(backoff.NewConstantBackOff(tokenReadPollingInterval), numTokenReadRetries))
+	if err != nil {
+		log.Error("Unable to read ACL token from a Consul server; "+
+			"please check that your server cluster is healthy", "err", err)
+		return err
 	}
+	log.Info("Successfully read ACL token from the server")
 	return nil
 }
 
diff --git a/control-plane/subcommand/common/common_test.go b/control-plane/subcommand/common/common_test.go
index 179d10a114..f4bcfd172b 100644
--- a/control-plane/subcommand/common/common_test.go
+++ b/control-plane/subcommand/common/common_test.go
@@ -60,8 +60,11 @@ func TestConsulLogin(t *testing.T) {
 	bearerTokenFile := WriteTempFile(t, "foo")
 	tokenFile := WriteTempFile(t, "")
 
-	client := startMockServer(t, &counter)
-	err := ConsulLogin(client, bearerTokenFile, testAuthMethod, tokenFile, "", testPodMeta)
+	// This is a common.Logger.
+	log, err := Logger("INFO", false)
+	require.NoError(err)
+	client, cfg := startMockServer(t, &counter)
+	err = ConsulLogin(client, cfg, log, bearerTokenFile, testAuthMethod, tokenFile, "", "", testPodMeta)
 	require.NoError(err)
 	require.Equal(counter, 1)
 	// Validate that the token file was written to disk.
@@ -76,11 +79,12 @@ func TestConsulLogin_EmptyBearerTokenFile(t *testing.T) {
 
 	bearerTokenFile := WriteTempFile(t, "")
 	err := ConsulLogin(
-		nil,
+		nil, nil, nil,
 		bearerTokenFile,
 		testAuthMethod,
 		"",
 		"",
+		"",
 		testPodMeta,
 	)
 	require.EqualError(err, fmt.Sprintf("no bearer token found in %s", bearerTokenFile))
@@ -91,11 +95,12 @@ func TestConsulLogin_BearerTokenFileDoesNotExist(t *testing.T) {
 	require := require.New(t)
 	randFileName := fmt.Sprintf("/foo/%d/%d", rand.Int(), rand.Int())
 	err := ConsulLogin(
-		nil,
+		nil, nil, nil,
 		randFileName,
 		testAuthMethod,
 		"",
 		"",
+		"",
 		testPodMeta,
 	)
 	require.Error(err)
@@ -107,14 +112,18 @@ func TestConsulLogin_TokenFileUnwritable(t *testing.T) {
 	require := require.New(t)
 	counter := 0
 	bearerTokenFile := WriteTempFile(t, "foo")
-	client := startMockServer(t, &counter)
+	client, cfg := startMockServer(t, &counter)
+	// This is a common.Logger.
+	log, err := Logger("INFO", false)
+	require.NoError(err)
 	randFileName := fmt.Sprintf("/foo/%d/%d", rand.Int(), rand.Int())
-	err := ConsulLogin(
-		client,
+	err = ConsulLogin(
+		client, cfg, log,
 		bearerTokenFile,
 		testAuthMethod,
 		randFileName,
 		"",
+		"",
 		testPodMeta,
 	)
 	require.Error(err)
@@ -214,7 +223,7 @@ func TestGetResolvedServerAddresses(t *testing.T) {
 // startMockServer starts an httptest server used to mock a Consul server's
 // /v1/acl/login endpoint. apiCallCounter will be incremented on each call to /v1/acl/login.
 // It returns a consul client pointing at the server.
-func startMockServer(t *testing.T, apiCallCounter *int) *api.Client {
+func startMockServer(t *testing.T, apiCallCounter *int) (*api.Client, *api.Config) {
 
 	// Start the Consul server.
 	consulServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -232,7 +241,7 @@ func startMockServer(t *testing.T, apiCallCounter *int) *api.Client {
 	client, err := api.NewClient(clientConfig)
 	require.NoError(t, err)
 
-	return client
+	return client, clientConfig
 }
 
 const testAuthMethod = "consul-k8s-auth-method"
diff --git a/control-plane/subcommand/connect-init/command.go b/control-plane/subcommand/connect-init/command.go
index 01a23e9c2c..883cdc0658 100644
--- a/control-plane/subcommand/connect-init/command.go
+++ b/control-plane/subcommand/connect-init/command.go
@@ -22,13 +22,8 @@ const (
 	defaultTokenSinkFile   = "/consul/connect-inject/acl-token"
 	defaultProxyIDFile     = "/consul/connect-inject/proxyid"
 
-	// The number of times to attempt ACL Login.
-	numLoginRetries = 3
 	// The number of times to attempt to read this service (120s).
 	defaultServicePollingRetries = 120
-
-	raftReplicationTimeout   = 2 * time.Second
-	tokenReadPollingInterval = 100 * time.Millisecond
 )
 
 type Command struct {
@@ -84,7 +79,6 @@ func (c *Command) init() {
 	c.http = &flags.HTTPFlags{}
 	flags.Merge(c.flagSet, c.http.Flags())
 	c.help = flags.Usage(help, c.flagSet)
-
 }
 
 func (c *Command) Run(args []string) int {
@@ -129,74 +123,11 @@ func (c *Command) Run(args []string) int {
 	if c.flagACLAuthMethod != "" {
 		// loginMeta is the default metadata that we pass to the consul login API.
 		loginMeta := map[string]string{"pod": fmt.Sprintf("%s/%s", c.flagPodNamespace, c.flagPodName)}
-		err = backoff.Retry(func() error {
-			err := common.ConsulLogin(consulClient, c.flagBearerTokenFile, c.flagACLAuthMethod, c.flagACLTokenSink, c.flagAuthMethodNamespace, loginMeta)
-			if err != nil {
-				c.logger.Error("Consul login failed; retrying", "error", err)
-			}
-			return err
-		}, backoff.WithMaxRetries(backoff.NewConstantBackOff(1*time.Second), numLoginRetries))
+		err = common.ConsulLogin(consulClient, cfg, c.logger, c.flagBearerTokenFile, c.flagACLAuthMethod, c.flagACLTokenSink, c.flagAuthMethodNamespace, c.flagServiceAccountName, loginMeta)
 		if err != nil {
-			if c.flagServiceAccountName == "default" {
-				c.logger.Warn("The service account name for this Pod is \"default\"." +
-					" In default installations this is not a supported service account name." +
-					" The service account name must match the name of the Kubernetes Service" +
-					" or the consul.hashicorp.com/connect-service annotation.")
-			}
-			c.logger.Error("Hit maximum retries for consul login", "error", err)
+			c.logger.Error("unable to complete login", "error", err)
 			return 1
 		}
-		// Now update the client so that it will read the ACL token we just fetched.
-		cfg.TokenFile = c.flagACLTokenSink
-		consulClient, err = consul.NewClient(cfg)
-		if err != nil {
-			c.logger.Error("Unable to update client connection", "error", err)
-			return 1
-		}
-		c.logger.Info("Consul login complete")
-
-		// A workaround to check that the ACL token is replicated to other Consul servers.
-		//
-		// A consul client may reach out to a follower instead of a leader to resolve the token during the
-		// call to get services below. This is because clients talk to servers in the stale consistency mode
-		// to decrease the load on the servers (see https://www.consul.io/docs/architecture/consensus#stale).
-		// In that case, it's possible that the token isn't replicated
-		// to that server instance yet. The client will then get an "ACL not found" error
-		// and subsequently cache this not found response. Then our call below
-		// to get services from the agent will keep hitting the same "ACL not found" error
-		// until the cache entry expires (determined by the `acl_token_ttl` which defaults to 30 seconds).
-		// This is not great because it will delay app start up time by 30 seconds in most cases
-		// (if you are running 3 servers, then the probability of ending up on a follower is close to 2/3).
-		//
-		// To help with that, we try to first read the token in the stale consistency mode until we
-		// get a successful response. This should not take more than 100ms because raft replication
-		// should in most cases take less than that (see https://www.consul.io/docs/install/performance#read-write-tuning)
-		// but we set the timeout to 2s to be sure.
-		//
-		// Note though that this workaround does not eliminate this problem completely. It's still possible
-		// for this call and the next call to reach different servers and those servers to have different
-		// states from each other.
-		// For example, this call can reach a leader and succeed, while the call below can go to a follower
-		// that is still behind the leader and get an "ACL not found" error.
-		// However, this is a pretty unlikely case because
-		// clients have sticky connections to a server, and those connections get rebalanced only every 2-3min.
-		// And so, this workaround should work in a vast majority of cases.
-		c.logger.Info("Checking that the ACL token exists when reading it in the stale consistency mode")
-		// Use raft timeout and polling interval to determine the number of retries.
-		numTokenReadRetries := uint64(raftReplicationTimeout.Milliseconds() / tokenReadPollingInterval.Milliseconds())
-		err = backoff.Retry(func() error {
-			_, _, err := consulClient.ACL().TokenReadSelf(&api.QueryOptions{AllowStale: true})
-			if err != nil {
-				c.logger.Error("Unable to read ACL token; retrying", "err", err)
-			}
-			return err
-		}, backoff.WithMaxRetries(backoff.NewConstantBackOff(tokenReadPollingInterval), numTokenReadRetries))
-		if err != nil {
-			c.logger.Error("Unable to read ACL token from a Consul server; "+
-				"please check that your server cluster is healthy", "err", err)
-			return 1
-		}
-		c.logger.Info("Successfully read ACL token from the server")
 	}
 
 	// Now wait for the service to be registered. Do this by querying the Agent for a service
@@ -204,6 +135,13 @@ func (c *Command) Run(args []string) int {
 	var proxyID string
 	registrationRetryCount := 0
 	var errServiceNameMismatch error
+	// We need a new client so that we can use the ACL token that was fetched during login to do the next bit,
+	// otherwise `consulClient` will still be using the bearerToken that was passed in.
+	consulClient, err = consul.NewClient(cfg)
+	if err != nil {
+		c.logger.Error("Unable to update client connection", "error", err)
+		return 1
+	}
 	err = backoff.Retry(func() error {
 		registrationRetryCount++
 		filter := fmt.Sprintf("Meta[%q] == %q and Meta[%q] == %q ",
diff --git a/control-plane/subcommand/consul-logout/command.go b/control-plane/subcommand/consul-logout/command.go
new file mode 100644
index 0000000000..74ec6ccf57
--- /dev/null
+++ b/control-plane/subcommand/consul-logout/command.go
@@ -0,0 +1,99 @@
+package consullogout
+
+import (
+	"flag"
+	"sync"
+
+	"github.com/hashicorp/consul-k8s/control-plane/consul"
+	"github.com/hashicorp/consul-k8s/control-plane/subcommand/common"
+	"github.com/hashicorp/consul-k8s/control-plane/subcommand/flags"
+	"github.com/hashicorp/consul/api"
+	"github.com/hashicorp/go-hclog"
+	"github.com/mitchellh/cli"
+)
+
+const (
+	defaultACLTokenFile = "/consul/login/acl-token"
+)
+
+// The consul-logout command issues a Consul logout API request to delete an ACL token.
+type Command struct {
+	UI cli.Ui
+
+	flagLogLevel string
+	flagLogJSON  bool
+
+	flagSet *flag.FlagSet
+	http    *flags.HTTPFlags
+
+	once   sync.Once
+	help   string
+	logger hclog.Logger
+}
+
+func (c *Command) init() {
+	c.flagSet = flag.NewFlagSet("", flag.ContinueOnError)
+	c.flagSet.StringVar(&c.flagLogLevel, "log-level", "info",
+		"Log verbosity level. Supported values (in order of detail) are \"trace\", "+
+			"\"debug\", \"info\", \"warn\", and \"error\".")
+	c.flagSet.BoolVar(&c.flagLogJSON, "log-json", false,
+		"Enable or disable JSON output format for logging.")
+
+	c.http = &flags.HTTPFlags{}
+	flags.Merge(c.flagSet, c.http.Flags())
+	c.help = flags.Usage(help, c.flagSet)
+
+}
+
+func (c *Command) Run(args []string) int {
+	var err error
+	c.once.Do(c.init)
+
+	if err := c.flagSet.Parse(args); err != nil {
+		return 1
+	}
+	if c.logger == nil {
+		c.logger, err = common.Logger(c.flagLogLevel, c.flagLogJSON)
+		if err != nil {
+			c.UI.Error(err.Error())
+			return 1
+		}
+	}
+	// Set a default if it is not already set.
+	if c.http.TokenFile() == "" {
+		if err := c.http.SetTokenFile(defaultACLTokenFile); err != nil {
+			c.logger.Error("Unable to update client", "error", err)
+			return 1
+		}
+	}
+
+	cfg := api.DefaultConfig()
+	c.http.MergeOntoConfig(cfg)
+	consulClient, err := consul.NewClient(cfg)
+	if err != nil {
+		c.logger.Error("Unable to get client connection", "error", err)
+		return 1
+	}
+	// Issue the logout.
+	_, err = consulClient.ACL().Logout(&api.WriteOptions{})
+	if err != nil {
+		c.logger.Error("Unable to delete consul ACL token as logout failed", "error", err)
+		return 1
+	}
+	c.logger.Error("ACL token successfully deleted")
+	return 0
+}
+
+func (c *Command) Synopsis() string { return synopsis }
+func (c *Command) Help() string {
+	c.once.Do(c.init)
+	return c.help
+}
+
+const synopsis = "Issue a consul logout to delete the ACL token."
+const help = `
+Usage: consul-k8s-control-plane consul-logout [options]
+
+  Deletes the ACL token for this pod.
+  Not intended for stand-alone use.
+`
diff --git a/control-plane/subcommand/consul-logout/command_test.go b/control-plane/subcommand/consul-logout/command_test.go
new file mode 100644
index 0000000000..5ff4c216be
--- /dev/null
+++ b/control-plane/subcommand/consul-logout/command_test.go
@@ -0,0 +1,153 @@
+package consullogout
+
+import (
+	"fmt"
+	"math/rand"
+	"os"
+	"testing"
+
+	"github.com/hashicorp/consul-k8s/control-plane/consul"
+	"github.com/hashicorp/consul-k8s/control-plane/helper/test"
+	"github.com/hashicorp/consul/api"
+	"github.com/hashicorp/consul/sdk/testutil"
+	"github.com/mitchellh/cli"
+	"github.com/stretchr/testify/require"
+)
+
+// TestRun_InvalidSinkFile validates that we correctly fail in case the token sink file
+// does not exist.
+func TestRun_InvalidSinkFile(t *testing.T) {
+	t.Parallel()
+	randFileName := fmt.Sprintf("/foo/%d/%d", rand.Int(), rand.Int())
+
+	ui := cli.NewMockUi()
+	cmd := Command{
+		UI: ui,
+	}
+	code := cmd.Run([]string{
+		"-token-file", randFileName,
+	})
+	require.Equal(t, 1, code)
+}
+
+// Test_UnableToLogoutDueToInvalidToken checks the error path for when Consul is not
+// aware of an ACL token. This is a big corner case but covers the rare occurrance that
+// the preStop hook where `consul-logout` is run might be executed more than once by Kubelet.
+// This also covers obscure cases where the acl-token file is corrupted somehow.
+func Test_UnableToLogoutDueToInvalidToken(t *testing.T) {
+	tokenFile := fmt.Sprintf("/tmp/%d1", rand.Int())
+	t.Cleanup(func() {
+		os.Remove(tokenFile)
+	})
+
+	var caFile, certFile, keyFile string
+	// Start Consul server with ACLs enabled and default deny policy.
+	masterToken := "b78d37c7-0ca7-5f4d-99ee-6d9975ce4586"
+	server, err := testutil.NewTestServerConfigT(t, func(c *testutil.TestServerConfig) {
+		c.ACL.Enabled = true
+		c.ACL.DefaultPolicy = "deny"
+		c.ACL.Tokens.InitialManagement = masterToken
+		caFile, certFile, keyFile = test.GenerateServerCerts(t)
+		c.CAFile = caFile
+		c.CertFile = certFile
+		c.KeyFile = keyFile
+	})
+	require.NoError(t, err)
+	defer server.Stop()
+	server.WaitForLeader(t)
+	cfg := &api.Config{
+		Address: server.HTTPSAddr,
+		Scheme:  "https",
+		Token:   masterToken,
+		TLSConfig: api.TLSConfig{
+			CAFile: caFile,
+		},
+	}
+	require.NoError(t, err)
+
+	bogusToken := "00000000-00-0-001110aacddbderf"
+	err = os.WriteFile(tokenFile, []byte(bogusToken), 0444)
+	require.NoError(t, err)
+
+	ui := cli.NewMockUi()
+	cmd := Command{
+		UI: ui,
+	}
+
+	// Run the command.
+	code := cmd.Run([]string{
+		"-http-addr", fmt.Sprintf("%s://%s", cfg.Scheme, cfg.Address),
+		"-token-file", tokenFile,
+	})
+	require.Equal(t, 1, code, ui.ErrorWriter.String())
+	require.Contains(t, "Unexpected response code: 403 (ACL not found)", ui.ErrorWriter.String())
+}
+
+// Test_RunUsingLogin creates an AuthMethod and issues an ACL Token via ACL().Login()
+// which is the code path that is taken to provision the ACL tokens at runtime through
+// subcommand/acl-init. It then runs `consul-logout` and ensures that the ACL token
+// is properly destroyed.
+func Test_RunUsingLogin(t *testing.T) {
+	// This is the test file that we will write the token to so consul-logout can read it.
+	tokenFile := fmt.Sprintf("/tmp/%d1", rand.Int())
+	t.Cleanup(func() {
+		os.Remove(tokenFile)
+	})
+
+	// Start Consul server with ACLs enabled and default deny policy.
+	masterToken := "b78d37c7-0ca7-5f4d-99ee-6d9975ce4586"
+	server, err := testutil.NewTestServerConfigT(t, func(c *testutil.TestServerConfig) {
+		c.ACL.Enabled = true
+		c.ACL.DefaultPolicy = "deny"
+		c.ACL.Tokens.InitialManagement = masterToken
+	})
+	require.NoError(t, err)
+	defer server.Stop()
+	server.WaitForLeader(t)
+	cfg := &api.Config{
+		Address: server.HTTPAddr,
+		Scheme:  "http",
+		Token:   masterToken,
+	}
+	consulClient, err := consul.NewClient(cfg)
+	require.NoError(t, err)
+
+	// We are not setting up the Component Auth Method here because testing logout
+	// does not need to use the auth method and this auth method can still issue a login.
+	test.SetupK8sAuthMethod(t, consulClient, "test-sa", "default")
+
+	// Do the login.
+	req := &api.ACLLoginParams{
+		AuthMethod:  test.AuthMethod,
+		BearerToken: test.ServiceAccountJWTToken,
+		Meta:        map[string]string{},
+	}
+	token, _, err := consulClient.ACL().Login(req, &api.WriteOptions{})
+	require.NoError(t, err)
+
+	// Validate that the token was created.
+	tok, _, err := consulClient.ACL().TokenRead(token.AccessorID, &api.QueryOptions{})
+	require.NoError(t, err)
+
+	// Write the token's SecretID to the tokenFile which mimics loading
+	// the ACL token from subcommand/acl-init path.
+	err = os.WriteFile(tokenFile, []byte(token.SecretID), 0444)
+	require.NoError(t, err)
+
+	ui := cli.NewMockUi()
+	cmd := Command{
+		UI: ui,
+	}
+
+	// Run the command.
+	code := cmd.Run([]string{
+		"-http-addr", fmt.Sprintf("%s://%s", cfg.Scheme, cfg.Address),
+		"-token-file", tokenFile,
+	})
+	require.Equal(t, 0, code, ui.ErrorWriter.String())
+
+	// Validate the ACL token was destroyed.
+	noTok, _, err := consulClient.ACL().TokenReadSelf(&api.QueryOptions{Token: tok.SecretID})
+	require.Error(t, err)
+	require.Nil(t, noTok)
+}
diff --git a/control-plane/subcommand/server-acl-init/command.go b/control-plane/subcommand/server-acl-init/command.go
index 64906d44ad..617403022b 100644
--- a/control-plane/subcommand/server-acl-init/command.go
+++ b/control-plane/subcommand/server-acl-init/command.go
@@ -47,7 +47,7 @@ type Command struct {
 	flagInjectAuthMethodHost string
 	flagBindingRuleSelector  string
 
-	flagCreateControllerToken bool
+	flagCreateControllerPoliciesAndBindings bool
 
 	flagCreateEntLicenseToken bool
 
@@ -147,8 +147,8 @@ func (c *Command) init() {
 	c.flags.StringVar(&c.flagBindingRuleSelector, "acl-binding-rule-selector", "",
 		"Selector string for connectInject ACL Binding Rule.")
 
-	c.flags.BoolVar(&c.flagCreateControllerToken, "create-controller-token", false,
-		"Toggle for creating a token for the controller.")
+	c.flags.BoolVar(&c.flagCreateControllerPoliciesAndBindings, "create-controller-token", false,
+		"Toggle for creating acl policies and rolebindings for the controller.")
 
 	c.flags.BoolVar(&c.flagCreateEntLicenseToken, "create-enterprise-license-token", false,
 		"Toggle for creating a token for the enterprise license job.")
@@ -257,7 +257,6 @@ func (c *Command) Run(args []string) int {
 		c.UI.Error(err.Error())
 		return 1
 	}
-
 	var aclReplicationToken string
 	if c.flagACLReplicationTokenFile != "" {
 		// Load the ACL replication token from file.
@@ -365,10 +364,10 @@ func (c *Command) Run(args []string) int {
 			CAFile:  c.flagConsulCACert,
 		},
 	}
+
 	if c.flagEnablePartitions {
 		clientConfig.Partition = c.flagPartitionName
 	}
-
 	consulClient, err := consul.NewClient(clientConfig)
 	if err != nil {
 		c.log.Error(fmt.Sprintf("Error creating Consul client for addr %q: %s", serverAddr, err))
@@ -440,6 +439,15 @@ func (c *Command) Run(args []string) int {
 		}
 	}
 
+	// Create the component auth method, this is the auth method that Consul components will use
+	// to issue an `ACL().Login()` against at startup, for local tokens.
+	componentAuthMethodName := c.withPrefix("k8s-component-auth-method")
+	err = c.configureComponentAuthMethod(consulClient, componentAuthMethodName)
+	if err != nil {
+		c.log.Error(err.Error())
+		return 1
+	}
+
 	if c.flagCreateClientToken {
 		agentRules, err := c.agentRules()
 		if err != nil {
@@ -497,7 +505,8 @@ func (c *Command) Run(args []string) int {
 	}
 
 	if c.flagCreateInjectToken {
-		err := c.configureConnectInjectAuthMethod(consulClient)
+		authMethodName := c.withPrefix("k8s-auth-method")
+		err := c.configureConnectInjectAuthMethod(consulClient, authMethodName)
 		if err != nil {
 			c.log.Error(err.Error())
 			return 1
@@ -705,22 +714,25 @@ func (c *Command) Run(args []string) int {
 		}
 	}
 
-	if c.flagCreateControllerToken {
+	if c.flagCreateControllerPoliciesAndBindings {
 		rules, err := c.controllerRules()
 		if err != nil {
 			c.log.Error("Error templating controller token rules", "err", err)
 			return 1
 		}
+
+		serviceAccountName := c.withPrefix("controller")
+
+		// Create the controller ACL Policy, Role and BindingRule but do not issue any ACLTokens or create Kube Secrets.
 		// Controller token must be global because config entry writes all
 		// go to the primary datacenter. This means secondary datacenters need
 		// a token that is known by the primary datacenters.
-		err = c.createGlobalACL("controller", rules, consulDC, isPrimary, consulClient)
+		err = c.createACLPolicyRoleAndBindingRule("controller", rules, consulDC, isPrimary, componentAuthMethodName, serviceAccountName, consulClient)
 		if err != nil {
 			c.log.Error(err.Error())
 			return 1
 		}
 	}
-
 	c.log.Info("server-acl-init completed successfully")
 	return 0
 }
@@ -886,6 +898,25 @@ func (c *Command) validateFlags() error {
 	return nil
 }
 
+// configureComponentAuthMethod sets up an AuthMethod that the Consul components will use to issue ACL logins with.
+func (c *Command) configureComponentAuthMethod(consulClient *api.Client, authMethodName string) error {
+	// Create the auth method template. This requires calls to the kubernetes environment.
+	authMethodTmpl, err := c.createAuthMethodTmpl(authMethodName, false)
+	if err != nil {
+		return err
+	}
+	err = c.untilSucceeds(fmt.Sprintf("creating auth method %s", authMethodTmpl.Name),
+		func() error {
+			var err error
+			// `AuthMethodCreate` will also be able to update an existing
+			// AuthMethod based on the name provided. This means that any
+			// configuration changes will correctly update the AuthMethod.
+			_, _, err = consulClient.ACL().AuthMethodCreate(&authMethodTmpl, &api.WriteOptions{})
+			return err
+		})
+	return err
+}
+
 const consulDefaultNamespace = "default"
 const consulDefaultPartition = "default"
 const synopsis = "Initialize ACLs on Consul servers and other components."
diff --git a/control-plane/subcommand/server-acl-init/command_ent_test.go b/control-plane/subcommand/server-acl-init/command_ent_test.go
index 5824d4af9b..a539e01e56 100644
--- a/control-plane/subcommand/server-acl-init/command_ent_test.go
+++ b/control-plane/subcommand/server-acl-init/command_ent_test.go
@@ -64,7 +64,13 @@ func TestRun_ConnectInject_SingleDestinationNamespace(t *testing.T) {
 			}
 			methods, _, err := consul.ACL().AuthMethodList(namespaceQuery)
 			require.NoError(err)
-			require.Len(methods, 1)
+			if consulDestNamespace == "default" {
+				// If the destination mamespace is default then AuthMethodList
+				// will return the component-auth-method as well.
+				require.Len(methods, 2)
+			} else {
+				require.Len(methods, 1)
+			}
 
 			// Check the ACL auth method is created in the expected namespace.
 			authMethodName := resourcePrefix + "-k8s-auth-method"
@@ -324,7 +330,7 @@ func TestRun_ACLPolicyUpdates(t *testing.T) {
 				"gw-terminating-gateway-token",
 				"anothergw-terminating-gateway-token",
 				"connect-inject-token",
-				"controller-token",
+				"controller-policy",
 			}
 			policies, _, err := consul.ACL().PolicyList(nil)
 			require.NoError(err)
@@ -376,7 +382,7 @@ func TestRun_ACLPolicyUpdates(t *testing.T) {
 				"anothergw-ingress-gateway-token",
 				"gw-terminating-gateway-token",
 				"anothergw-terminating-gateway-token",
-				"controller-token",
+				"controller-policy",
 				"partitions-token",
 			}
 			policies, _, err = consul.ACL().PolicyList(nil)
@@ -747,13 +753,6 @@ func TestRun_TokensWithNamespacesEnabled(t *testing.T) {
 			SecretNames: []string{resourcePrefix + "-connect-inject-acl-token"},
 			LocalToken:  false,
 		},
-		"controller token": {
-			TokenFlags:  []string{"-create-controller-token"},
-			PolicyNames: []string{"controller-token"},
-			PolicyDCs:   nil,
-			SecretNames: []string{resourcePrefix + "-controller-acl-token"},
-			LocalToken:  false,
-		},
 		"partitions token": {
 			TokenFlags:  []string{"-enable-partitions", "-partition=default"},
 			PolicyNames: []string{"partitions-token"},
@@ -797,17 +796,16 @@ func TestRun_TokensWithNamespacesEnabled(t *testing.T) {
 			})
 			require.NoError(err)
 
+			// Check that the expected policy was created.
 			for i := range c.PolicyNames {
 				policy := policyExists(t, c.PolicyNames[i], consul)
 				require.Equal(c.PolicyDCs, policy.Datacenters)
-
 				// Test that the token was created as a Kubernetes Secret.
 				tokenSecret, err := k8s.CoreV1().Secrets(ns).Get(context.Background(), c.SecretNames[i], metav1.GetOptions{})
 				require.NoError(err)
 				require.NotNil(tokenSecret)
 				token, ok := tokenSecret.Data["token"]
 				require.True(ok)
-
 				// Test that the token has the expected policies in Consul.
 				tokenData, _, err := consul.ACL().TokenReadSelf(&api.QueryOptions{Token: string(token)})
 				require.NoError(err)
@@ -1023,6 +1021,7 @@ partition "default" {
 		t.Run(c.TestName, func(t *testing.T) {
 			k8s, testSvr := completeEnterpriseSetup(t)
 			defer testSvr.Stop()
+			setUpK8sServiceAccount(t, k8s, ns)
 			require := require.New(t)
 
 			// Run the command.
@@ -1076,6 +1075,99 @@ partition "default" {
 	}
 }
 
+// Test creating the correct ACL policies and Binding Rules for components that use the auth method.
+// The test works by running the command and then ensuring that:
+// * An ACLBindingRule exists which references the ACLRole.
+// * An ACLRole exists and has the correct PolicyName in it's ACLPolicyLinkRule list.
+// * The ACLPolicy exists.
+func TestRun_PoliciesAndBindingRulesForACLLoginNamespacesEnabled(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		TestName    string
+		TokenFlags  []string
+		PolicyNames []string
+		Roles       []string
+		Namespace   string
+	}{
+		{
+			TestName:    "Controller",
+			TokenFlags:  []string{"-create-controller-token"},
+			PolicyNames: []string{"controller-policy"},
+			Roles:       []string{resourcePrefix + "-controller-acl-role"},
+			Namespace:   ns,
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.TestName, func(t *testing.T) {
+			k8s, testSvr := completeSetup(t)
+			defer testSvr.Stop()
+			setUpK8sServiceAccount(t, k8s, ns)
+
+			// Run the command.
+			ui := cli.NewMockUi()
+			cmd := Command{
+				UI:        ui,
+				clientset: k8s,
+			}
+			cmdArgs := append([]string{
+				"-timeout=500ms",
+				"-resource-prefix=" + resourcePrefix,
+				"-k8s-namespace=" + ns,
+				"-server-address", strings.Split(testSvr.HTTPAddr, ":")[0],
+				"-server-port", strings.Split(testSvr.HTTPAddr, ":")[1],
+			}, c.TokenFlags...)
+			cmd.init()
+			responseCode := cmd.Run(cmdArgs)
+			require.Equal(t, 0, responseCode, ui.ErrorWriter.String())
+
+			bootToken := getBootToken(t, k8s, resourcePrefix, ns)
+			consul, err := api.NewClient(&api.Config{
+				Namespace: ns,
+				Address:   testSvr.HTTPAddr,
+				Token:     bootToken,
+			})
+			require.NoError(t, err)
+
+			// Check that the Role exists + has correct Policy and is associated with a BindingRule.
+			for i := range c.Roles {
+				// Check that the Policy exists.
+				policy, _, err := consul.ACL().PolicyReadByName(c.PolicyNames[i], &api.QueryOptions{})
+				require.NoError(t, err)
+				require.NotNil(t, policy)
+
+				// Check that the Role exists.
+				role, _, err := consul.ACL().RoleReadByName(c.Roles[i], &api.QueryOptions{})
+				require.NoError(t, err)
+				require.NotNil(t, role)
+
+				// Check that the Role references the Policy.
+				found := false
+				for x := range role.Policies {
+					if role.Policies[x].Name == policy.Name {
+						found = true
+						break
+					}
+				}
+				require.True(t, found)
+
+				// Check that there exists a BindingRule that references this Role.
+				rb, _, err := consul.ACL().BindingRuleList("release-name-"+componentAuthMethod, &api.QueryOptions{})
+				require.NoError(t, err)
+				require.NotNil(t, rb)
+				found = false
+				for x := range rb {
+					if rb[x].BindName == c.Roles[i] {
+						found = true
+						break
+					}
+				}
+				require.True(t, found)
+			}
+		})
+	}
+}
+
 // Set up test consul agent and kubernetes cluster.
 func completeEnterpriseSetup(t *testing.T) (*fake.Clientset, *testutil.TestServer) {
 	k8s := fake.NewSimpleClientset()
diff --git a/control-plane/subcommand/server-acl-init/command_test.go b/control-plane/subcommand/server-acl-init/command_test.go
index bbccfcc20d..6323aa1479 100644
--- a/control-plane/subcommand/server-acl-init/command_test.go
+++ b/control-plane/subcommand/server-acl-init/command_test.go
@@ -40,6 +40,10 @@ import (
 var ns = "default"
 var resourcePrefix = "release-name-consul"
 
+const (
+	componentAuthMethod = "consul-k8s-component-auth-method"
+)
+
 func TestRun_FlagValidation(t *testing.T) {
 	t.Parallel()
 
@@ -104,6 +108,7 @@ func TestRun_Defaults(t *testing.T) {
 
 	k8s, testSvr := completeSetup(t)
 	defer testSvr.Stop()
+	setUpK8sServiceAccount(t, k8s, ns)
 	require := require.New(t)
 
 	// Run the command.
@@ -244,14 +249,6 @@ func TestRun_TokensPrimaryDC(t *testing.T) {
 			SecretNames: []string{resourcePrefix + "-acl-replication-acl-token"},
 			LocalToken:  false,
 		},
-		{
-			TestName:    "Controller token",
-			TokenFlags:  []string{"-create-controller-token"},
-			PolicyNames: []string{"controller-token"},
-			PolicyDCs:   nil,
-			SecretNames: []string{resourcePrefix + "-controller-acl-token"},
-			LocalToken:  false,
-		},
 		{
 			TestName:    "Endpoints Controller ACL token",
 			TokenFlags:  []string{"-create-inject-token"},
@@ -333,6 +330,7 @@ func TestRun_ReplicationTokenPrimaryDC_WithProvidedSecretID(t *testing.T) {
 	k8s, testSvr := completeSetup(t)
 	defer testSvr.Stop()
 	require := require.New(t)
+	setUpK8sServiceAccount(t, k8s, ns)
 
 	replicationToken := "123e4567-e89b-12d3-a456-426614174000"
 	replicationTokenFile, err := ioutil.TempFile("", "replicationtoken")
@@ -487,14 +485,6 @@ func TestRun_TokensReplicatedDC(t *testing.T) {
 			SecretNames: []string{resourcePrefix + "-connect-inject-acl-token"},
 			LocalToken:  true,
 		},
-		{
-			TestName:    "Controller token",
-			TokenFlags:  []string{"-create-controller-token"},
-			PolicyNames: []string{"controller-token-dc2"},
-			PolicyDCs:   nil,
-			SecretNames: []string{resourcePrefix + "-controller-acl-token"},
-			LocalToken:  false,
-		},
 	}
 	for _, c := range cases {
 		t.Run(c.TestName, func(t *testing.T) {
@@ -631,12 +621,6 @@ func TestRun_TokensWithProvidedBootstrapToken(t *testing.T) {
 			PolicyNames: []string{"acl-replication-token"},
 			SecretNames: []string{resourcePrefix + "-acl-replication-acl-token"},
 		},
-		{
-			TestName:    "Controller token",
-			TokenFlags:  []string{"-create-controller-token"},
-			PolicyNames: []string{"controller-token"},
-			SecretNames: []string{resourcePrefix + "-controller-acl-token"},
-		},
 	}
 	for _, c := range cases {
 		t.Run(c.TestName, func(t *testing.T) {
@@ -1117,6 +1101,7 @@ func TestRun_SyncPolicyUpdates(t *testing.T) {
 	t.Parallel()
 	k8s, testSvr := completeSetup(t)
 	defer testSvr.Stop()
+	setUpK8sServiceAccount(t, k8s, ns)
 	require := require.New(t)
 
 	ui := cli.NewMockUi()
@@ -1257,7 +1242,7 @@ func TestRun_DelayedServers(t *testing.T) {
 	t.Parallel()
 	require := require.New(t)
 	k8s := fake.NewSimpleClientset()
-
+	setUpK8sServiceAccount(t, k8s, ns)
 	randomPorts := freeport.GetN(t, 6)
 
 	ui := cli.NewMockUi()
@@ -1345,6 +1330,7 @@ func TestRun_NoLeader(t *testing.T) {
 	t.Parallel()
 	require := require.New(t)
 	k8s := fake.NewSimpleClientset()
+	setUpK8sServiceAccount(t, k8s, ns)
 
 	type APICall struct {
 		Method string
@@ -1450,6 +1436,10 @@ func TestRun_NoLeader(t *testing.T) {
 			"GET",
 			"/v1/agent/self",
 		},
+		{
+			"PUT",
+			"/v1/acl/auth-method",
+		},
 		{
 			"PUT",
 			"/v1/acl/policy",
@@ -1563,6 +1553,7 @@ func TestRun_ClientTokensRetry(t *testing.T) {
 	t.Parallel()
 	require := require.New(t)
 	k8s := fake.NewSimpleClientset()
+	setUpK8sServiceAccount(t, k8s, ns)
 
 	type APICall struct {
 		Method string
@@ -1644,6 +1635,10 @@ func TestRun_ClientTokensRetry(t *testing.T) {
 			"GET",
 			"/v1/agent/self",
 		},
+		{
+			"PUT",
+			"/v1/acl/auth-method",
+		},
 		// This call should happen twice since the first will fail.
 		{
 			"PUT",
@@ -1666,6 +1661,7 @@ func TestRun_AlreadyBootstrapped(t *testing.T) {
 	t.Parallel()
 	require := require.New(t)
 	k8s := fake.NewSimpleClientset()
+	setUpK8sServiceAccount(t, k8s, ns)
 
 	type APICall struct {
 		Method string
@@ -1756,6 +1752,10 @@ func TestRun_AlreadyBootstrapped(t *testing.T) {
 			"GET",
 			"/v1/agent/self",
 		},
+		{
+			"PUT",
+			"/v1/acl/auth-method",
+		},
 		{
 			"PUT",
 			"/v1/acl/policy",
@@ -1848,6 +1848,7 @@ func TestRun_SkipBootstrapping_WhenBootstrapTokenIsProvided(t *testing.T) {
 	t.Parallel()
 	require := require.New(t)
 	k8s := fake.NewSimpleClientset()
+	setUpK8sServiceAccount(t, k8s, ns)
 
 	bootToken := "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
 	tokenFile := common.WriteTempFile(t, bootToken)
@@ -1904,6 +1905,10 @@ func TestRun_SkipBootstrapping_WhenBootstrapTokenIsProvided(t *testing.T) {
 			"GET",
 			"/v1/agent/self",
 		},
+		{
+			"PUT",
+			"/v1/acl/auth-method",
+		},
 	}, consulAPICalls)
 }
 
@@ -1933,6 +1938,7 @@ func TestRun_HTTPS(t *testing.T) {
 	t.Parallel()
 	require := require.New(t)
 	k8s := fake.NewSimpleClientset()
+	setUpK8sServiceAccount(t, k8s, ns)
 
 	caFile, certFile, keyFile := test.GenerateServerCerts(t)
 
@@ -1981,6 +1987,7 @@ func TestRun_ACLReplicationTokenValid(t *testing.T) {
 
 	secondaryK8s, secondaryConsulClient, secondaryAddr, aclReplicationToken, clean := completeReplicatedSetup(t)
 	defer clean()
+	setUpK8sServiceAccount(t, secondaryK8s, ns)
 
 	// completeReplicatedSetup ran the command in our primary dc so now we
 	// need to run the command in our secondary dc.
@@ -2076,6 +2083,7 @@ func TestRun_CloudAutoJoin(t *testing.T) {
 
 	k8s, testSvr := completeSetup(t)
 	defer testSvr.Stop()
+	setUpK8sServiceAccount(t, k8s, ns)
 	require := require.New(t)
 
 	// create a mock provider
@@ -2155,6 +2163,7 @@ func TestRun_GatewayErrors(t *testing.T) {
 
 			k8s, testSvr := completeSetup(tt)
 			defer testSvr.Stop()
+			setUpK8sServiceAccount(t, k8s, ns)
 			require := require.New(tt)
 
 			// Run the command.
@@ -2252,6 +2261,7 @@ func replicatedSetup(t *testing.T, bootToken string) (*fake.Clientset, *api.Clie
 	if bootToken == "" {
 		primaryK8s := fake.NewSimpleClientset()
 		require.NoError(t, err)
+		setUpK8sServiceAccount(t, primaryK8s, ns)
 
 		// Run the command to bootstrap ACLs
 		primaryUI := cli.NewMockUi()
@@ -2336,6 +2346,134 @@ func replicatedSetup(t *testing.T, bootToken string) (*fake.Clientset, *api.Clie
 	}
 }
 
+// Test creating the correct ACL policies and Binding Rules for components that use the auth method.
+// The test works by running the command and then ensuring that:
+// * An ACLBindingRule exists which references the ACLRole.
+// * An ACLRole exists and has the correct PolicyName in it's ACLPolicyLinkRule list.
+// * The ACLPolicy exists.
+func TestRun_PoliciesAndBindingRulesForACLLogin(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		TestName    string
+		TokenFlags  []string
+		PolicyNames []string
+		Roles       []string
+	}{
+		{
+			TestName:    "Controller",
+			TokenFlags:  []string{"-create-controller-token"},
+			PolicyNames: []string{"controller-policy"},
+			Roles:       []string{resourcePrefix + "-controller-acl-role"},
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.TestName, func(t *testing.T) {
+			k8s, testSvr := completeSetup(t)
+			defer testSvr.Stop()
+			setUpK8sServiceAccount(t, k8s, ns)
+
+			// Run the command.
+			ui := cli.NewMockUi()
+			cmd := Command{
+				UI:        ui,
+				clientset: k8s,
+			}
+			cmdArgs := append([]string{
+				"-timeout=500ms",
+				"-resource-prefix=" + resourcePrefix,
+				"-k8s-namespace=" + ns,
+				"-server-address", strings.Split(testSvr.HTTPAddr, ":")[0],
+				"-server-port", strings.Split(testSvr.HTTPAddr, ":")[1],
+			}, c.TokenFlags...)
+			cmd.init()
+			responseCode := cmd.Run(cmdArgs)
+			require.Equal(t, 0, responseCode, ui.ErrorWriter.String())
+
+			bootToken := getBootToken(t, k8s, resourcePrefix, ns)
+			consul, err := api.NewClient(&api.Config{
+				Address: testSvr.HTTPAddr,
+				Token:   bootToken,
+			})
+			require.NoError(t, err)
+
+			// Check that the Role exists + has correct Policy and is associated with a BindingRule.
+			for i := range c.Roles {
+				// Check that the Policy exists.
+				policy, _, err := consul.ACL().PolicyReadByName(c.PolicyNames[i], &api.QueryOptions{})
+				require.NoError(t, err)
+				require.NotNil(t, policy)
+
+				// Check that the Role exists.
+				role, _, err := consul.ACL().RoleReadByName(c.Roles[i], &api.QueryOptions{})
+				require.NoError(t, err)
+				require.NotNil(t, role)
+
+				// Check that the Role references the Policy.
+				found := false
+				for x := range role.Policies {
+					if role.Policies[x].Name == policy.Name {
+						found = true
+						break
+					}
+				}
+				require.True(t, found)
+
+				// Check that there exists a BindingRule that references this Role.
+				rb, _, err := consul.ACL().BindingRuleList("release-name-"+componentAuthMethod, &api.QueryOptions{})
+				require.NoError(t, err)
+				require.NotNil(t, rb)
+				found = false
+				for x := range rb {
+					if rb[x].BindName == c.Roles[i] {
+						found = true
+						break
+					}
+				}
+				require.True(t, found)
+			}
+		})
+	}
+}
+
+// Test that the component auth method gets created.
+func TestRun_ComponentAuthMethod(t *testing.T) {
+	t.Parallel()
+
+	k8s, testSvr := completeSetup(t)
+	setUpK8sServiceAccount(t, k8s, ns)
+	defer testSvr.Stop()
+	require := require.New(t)
+
+	// Run the command.
+	ui := cli.NewMockUi()
+	cmd := Command{
+		UI:        ui,
+		clientset: k8s,
+	}
+	cmd.init()
+	cmdArgs := []string{
+		"-timeout=1m",
+		"-k8s-namespace=" + ns,
+		"-server-address", strings.Split(testSvr.HTTPAddr, ":")[0],
+		"-server-port", strings.Split(testSvr.HTTPAddr, ":")[1],
+		"-resource-prefix=" + resourcePrefix}
+
+	responseCode := cmd.Run(cmdArgs)
+	require.Equal(0, responseCode, ui.ErrorWriter.String())
+
+	// Check that the expected policy was created.
+	bootToken := getBootToken(t, k8s, resourcePrefix, ns)
+	consulClient, err := api.NewClient(&api.Config{
+		Address: testSvr.HTTPAddr,
+		Token:   bootToken,
+	})
+	require.NoError(err)
+	authMethod, _, err := consulClient.ACL().AuthMethodRead(resourcePrefix+"-k8s-component-auth-method", &api.QueryOptions{})
+	require.NoError(err)
+	require.NotNil(authMethod)
+}
+
 // getBootToken gets the bootstrap token from the Kubernetes secret. It will
 // cause a test failure if the Secret doesn't exist or is malformed.
 func getBootToken(t *testing.T, k8s *fake.Clientset, prefix string, k8sNamespace string) string {
diff --git a/control-plane/subcommand/server-acl-init/connect_inject.go b/control-plane/subcommand/server-acl-init/connect_inject.go
index abd10f9f7f..f8fc11b4fc 100644
--- a/control-plane/subcommand/server-acl-init/connect_inject.go
+++ b/control-plane/subcommand/server-acl-init/connect_inject.go
@@ -1,9 +1,7 @@
 package serveraclinit
 
 import (
-	"errors"
 	"fmt"
-
 	"github.com/hashicorp/consul-k8s/control-plane/namespaces"
 	"github.com/hashicorp/consul/api"
 	apiv1 "k8s.io/api/core/v1"
@@ -18,13 +16,11 @@ const defaultKubernetesHost = "https://kubernetes.default.svc"
 
 // configureConnectInject sets up auth methods so that connect injection will
 // work.
-func (c *Command) configureConnectInjectAuthMethod(consulClient *api.Client) error {
-
-	authMethodName := c.withPrefix("k8s-auth-method")
+func (c *Command) configureConnectInjectAuthMethod(consulClient *api.Client, authMethodName string) error {
 
 	// Create the auth method template. This requires calls to the
 	// kubernetes environment.
-	authMethodTmpl, err := c.createAuthMethodTmpl(authMethodName)
+	authMethodTmpl, err := c.createAuthMethodTmpl(authMethodName, true)
 	if err != nil {
 		return err
 	}
@@ -68,6 +64,7 @@ func (c *Command) configureConnectInjectAuthMethod(consulClient *api.Client) err
 		return err
 	}
 
+	c.log.Info("creating inject binding rule")
 	// Create the binding rule.
 	abr := api.ACLBindingRule{
 		Description: "Kubernetes binding rule",
@@ -76,67 +73,13 @@ func (c *Command) configureConnectInjectAuthMethod(consulClient *api.Client) err
 		BindName:    "${serviceaccount.name}",
 		Selector:    c.flagBindingRuleSelector,
 	}
-
-	// Binding rule list api call query options
-	queryOptions := api.QueryOptions{}
-
-	// Add a namespace if appropriate
-	// If namespaces and mirroring are enabled, this is not necessary because
-	// the binding rule will fall back to being created in the Consul `default`
-	// namespace automatically, as is necessary for mirroring.
-	if c.flagEnableNamespaces && !c.flagEnableInjectK8SNSMirroring {
-		abr.Namespace = c.flagConsulInjectDestinationNamespace
-		queryOptions.Namespace = c.flagConsulInjectDestinationNamespace
-	}
-
-	var existingRules []*api.ACLBindingRule
-	err = c.untilSucceeds(fmt.Sprintf("listing binding rules for auth method %s", authMethodName),
-		func() error {
-			var err error
-			existingRules, _, err = consulClient.ACL().BindingRuleList(authMethodName, &queryOptions)
-			return err
-		})
-	if err != nil {
-		return err
-	}
-
-	// If the binding rule already exists, update it
-	// This updates the binding rule any time the acl bootstrapping
-	// command is rerun, which is a bit of extra overhead, but is
-	// necessary to pick up any potential config changes.
-	if len(existingRules) > 0 {
-		// Find the policy that matches our name and description
-		// and that's the ID we need
-		for _, existingRule := range existingRules {
-			if existingRule.BindName == abr.BindName && existingRule.Description == abr.Description {
-				abr.ID = existingRule.ID
-			}
-		}
-
-		// This will only happen if there are existing policies
-		// for this auth method, but none that match the binding
-		// rule set up here in the bootstrap method.
-		if abr.ID == "" {
-			return errors.New("unable to find a matching ACL binding rule to update")
-		}
-
-		err = c.untilSucceeds(fmt.Sprintf("updating acl binding rule for %s", authMethodName),
-			func() error {
-				_, _, err := consulClient.ACL().BindingRuleUpdate(&abr, nil)
-				return err
-			})
-	} else {
-		// Otherwise create the binding rule
-		err = c.untilSucceeds(fmt.Sprintf("creating acl binding rule for %s", authMethodName),
-			func() error {
-				_, _, err := consulClient.ACL().BindingRuleCreate(&abr, nil)
-				return err
-			})
-	}
-	return err
+	return c.updateOrCreateBindingRule(consulClient, authMethodName, &abr, false)
 }
 
-func (c *Command) createAuthMethodTmpl(authMethodName string) (api.ACLAuthMethod, error) {
+// createAuthMethodTmpl sets up the auth method template based on the connect-injector's service account
+// jwt token. It is common for both the connect inject auth method and the component auth method
+// with the option to add namespace specific configuration to the auth method template via `useNS`.
+func (c *Command) createAuthMethodTmpl(authMethodName string, useNS bool) (api.ACLAuthMethod, error) {
 	// Get the Secret name for the auth method ServiceAccount.
 	var authMethodServiceAccount *apiv1.ServiceAccount
 	saName := c.withPrefix("connect-injector")
@@ -197,8 +140,9 @@ func (c *Command) createAuthMethodTmpl(authMethodName string) (api.ACLAuthMethod
 		},
 	}
 
-	// Add options for mirroring namespaces
-	if c.flagEnableNamespaces && c.flagEnableInjectK8SNSMirroring {
+	// Add options for mirroring namespaces, this is only used by the connect inject auth method
+	// and so can be disabled for the component auth method.
+	if useNS && c.flagEnableNamespaces && c.flagEnableInjectK8SNSMirroring {
 		authMethodTmpl.Config["MapNamespaces"] = true
 		authMethodTmpl.Config["ConsulNamespacePrefix"] = c.flagInjectK8SNSMirroringPrefix
 	}
diff --git a/control-plane/subcommand/server-acl-init/connect_inject_test.go b/control-plane/subcommand/server-acl-init/connect_inject_test.go
index a17d635bc1..e9c38595a0 100644
--- a/control-plane/subcommand/server-acl-init/connect_inject_test.go
+++ b/control-plane/subcommand/server-acl-init/connect_inject_test.go
@@ -64,6 +64,6 @@ func TestCommand_createAuthMethodTmpl_SecretNotFound(t *testing.T) {
 	_, err := k8s.CoreV1().Secrets(ns).Create(ctx, secret, metav1.CreateOptions{})
 	require.NoError(t, err)
 
-	_, err = cmd.createAuthMethodTmpl("test")
+	_, err = cmd.createAuthMethodTmpl("test", true)
 	require.EqualError(t, err, "found no secret of type 'kubernetes.io/service-account-token' associated with the release-name-consul-connect-injector service account")
 }
diff --git a/control-plane/subcommand/server-acl-init/create_or_update.go b/control-plane/subcommand/server-acl-init/create_or_update.go
index 80dca054bf..d7d0cb991a 100644
--- a/control-plane/subcommand/server-acl-init/create_or_update.go
+++ b/control-plane/subcommand/server-acl-init/create_or_update.go
@@ -1,6 +1,7 @@
 package serveraclinit
 
 import (
+	"errors"
 	"fmt"
 	"strings"
 
@@ -10,6 +11,172 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+// addRoleAndBindingRule adds an ACLRole and ACLBindingRule which reference the authMethod.
+func (c *Command) addRoleAndBindingRule(client *api.Client, serviceAccountName string, authMethodName string, policies []*api.ACLRolePolicyLink) error {
+
+	// This is the ACLRole which will allow the component which uses the serviceaccount
+	// to be able to do a consul login.
+	aclRoleName := fmt.Sprintf("%s-acl-role", serviceAccountName)
+	role := &api.ACLRole{
+		Name:        aclRoleName,
+		Description: fmt.Sprintf("ACL Role for %s", serviceAccountName),
+		Policies:    policies,
+	}
+	err := c.updateOrCreateACLRole(client, role)
+	if err != nil {
+		c.log.Error("unable to update or create ACL Role", err)
+		return err
+	}
+
+	// Create the ACLBindingRule, this ties the Policies defined in the Role to the authMethod via serviceaccount.
+	abr := api.ACLBindingRule{
+		Description: fmt.Sprintf("Binding Rule for %s", serviceAccountName),
+		AuthMethod:  authMethodName,
+		Selector:    fmt.Sprintf("serviceaccount.name==%q", serviceAccountName),
+		BindType:    api.BindingRuleBindTypeRole,
+		BindName:    aclRoleName,
+	}
+	return c.updateOrCreateBindingRule(client, authMethodName, &abr, true)
+}
+
+// updateOrCreateACLRole will query to see if existing role is in place and update them
+// or create them if they do not yet exist.
+func (c *Command) updateOrCreateACLRole(client *api.Client, role *api.ACLRole) error {
+	err := c.untilSucceeds(fmt.Sprintf("update or create acl role for %s", role.Name),
+		func() error {
+			var err error
+			aclRole, _, err := client.ACL().RoleReadByName(role.Name, &api.QueryOptions{})
+			if err != nil {
+				c.log.Error("unable to read ACL Roles", err)
+				return err
+			}
+			if aclRole != nil {
+				_, _, err := client.ACL().RoleUpdate(aclRole, &api.WriteOptions{})
+				if err != nil {
+					c.log.Error("unable to update role", err)
+					return err
+				}
+				return nil
+			}
+			_, _, err = client.ACL().RoleCreate(role, &api.WriteOptions{})
+			if err != nil {
+				c.log.Error("unable to create role", err)
+				return err
+			}
+			return err
+		})
+	return err
+}
+
+// updateOrCreateBindingRule will query to see if existing binding rules are in place and update them
+// or create them if they do not yet exist.
+func (c *Command) updateOrCreateBindingRule(client *api.Client, authMethodName string, abr *api.ACLBindingRule, skipNamespacing bool) error {
+	// Binding rule list api call query options.
+	queryOptions := api.QueryOptions{}
+
+	// If namespaces and mirroring are enabled, this is not necessary because
+	// the binding rule will fall back to being created in the Consul `default`
+	// namespace automatically, as is necessary for mirroring.
+	if !skipNamespacing && c.flagEnableNamespaces && !c.flagEnableInjectK8SNSMirroring {
+		abr.Namespace = c.flagConsulInjectDestinationNamespace
+		queryOptions.Namespace = c.flagConsulInjectDestinationNamespace
+	}
+
+	var existingRules []*api.ACLBindingRule
+	err := c.untilSucceeds(fmt.Sprintf("listing binding rules for auth method %s", authMethodName),
+		func() error {
+			var err error
+			existingRules, _, err = client.ACL().BindingRuleList(authMethodName, &queryOptions)
+			return err
+		})
+	if err != nil {
+		return err
+	}
+
+	// If the binding rule already exists, update it
+	// This updates the binding rule any time the acl bootstrapping
+	// command is rerun, which is a bit of extra overhead, but is
+	// necessary to pick up any potential config changes.
+	if len(existingRules) > 0 {
+		// Find the policy that matches our name and description
+		// and that's the ID we need
+		for _, existingRule := range existingRules {
+			if existingRule.BindName == abr.BindName && existingRule.Description == abr.Description {
+				abr.ID = existingRule.ID
+			}
+		}
+
+		// This will only happen if there are existing policies
+		// for this auth method, but none that match the binding
+		// rule set up here in the bootstrap method.
+		if abr.ID == "" {
+			return errors.New("unable to find a matching ACL binding rule to update")
+		}
+
+		err = c.untilSucceeds(fmt.Sprintf("updating acl binding rule for %s", authMethodName),
+			func() error {
+				_, _, err := client.ACL().BindingRuleUpdate(abr, nil)
+				return err
+			})
+	} else {
+		// Otherwise create the binding rule
+		err = c.untilSucceeds(fmt.Sprintf("creating acl binding rule for %s", authMethodName),
+			func() error {
+				_, _, err := client.ACL().BindingRuleCreate(abr, nil)
+				return err
+			})
+
+	}
+	return err
+}
+
+// createACLPolicyRoleAndBindingRule will create the ACL Policy for the component
+// then create a set of ACLRole and ACLBindingRule which tie the component's serviceaccount
+// to the authMethod, allowing the serviceaccount to later be allowed to issue a Consul Login.
+func (c *Command) createACLPolicyRoleAndBindingRule(componentName string, rules string, dc string, isPrimary bool,
+	authMethodName string, serviceAccountName string, client *api.Client) error {
+	// Create policy with the given rules.
+	policyName := fmt.Sprintf("%s-policy", componentName)
+	if c.flagFederation && !isPrimary {
+		// If performing ACL replication, we must ensure policy names are
+		// globally unique so we append the datacenter name but only in secondary datacenters..
+		policyName += fmt.Sprintf("-%s", dc)
+	}
+	var datacenters []string
+	// TODO: when we support global auth method logins we will need to append the list of dcs.
+	//if !globalToken && dc != "" {
+	if dc != "" {
+		datacenters = append(datacenters, dc)
+	}
+	policyTmpl := api.ACLPolicy{
+		Name:        policyName,
+		Description: fmt.Sprintf("%s Token Policy", policyName),
+		Rules:       rules,
+		Datacenters: datacenters,
+	}
+	err := c.untilSucceeds(fmt.Sprintf("creating %s policy", policyTmpl.Name),
+		func() error {
+			return c.createOrUpdateACLPolicy(policyTmpl, client)
+		})
+	if err != nil {
+		return err
+	}
+
+	// Create an ACLRolePolicyLink list to attach to the ACLRole.
+	ap := &api.ACLRolePolicyLink{
+		Name: policyName,
+	}
+	apl := []*api.ACLRolePolicyLink{}
+	apl = append(apl, ap)
+
+	// Add the ACLRole and ACLBindingRule.
+	err = c.addRoleAndBindingRule(client, serviceAccountName, authMethodName, apl)
+	if err != nil {
+		return err
+	}
+	return err
+}
+
 // createLocalACL creates a policy and acl token for this dc (datacenter), i.e.
 // the policy is only valid for this datacenter and the token is a local token.
 func (c *Command) createLocalACL(name, rules, dc string, isPrimary bool, consulClient *api.Client) error {
@@ -31,7 +198,6 @@ func (c *Command) createGlobalACLWithSecretID(name, rules, dc string, isPrimary
 // createACL creates a policy with rules and name. If localToken is true then
 // the token will be a local token and the policy will be scoped to only dc.
 // If localToken is false, the policy will be global.
-// The token will be written to a Kubernetes secret.
 // When secretID is provided, we will use that value for the created token and
 // will skip writing it to a Kubernetes secret (because in this case we assume that
 // this value already exists in some secrets storage).
@@ -69,9 +235,9 @@ func (c *Command) createACL(name, rules string, localToken bool, dc string, isPr
 	}
 
 	// Check if the replication token already exists in some form.
-	secretName := c.withPrefix(name + "-acl-token")
 	// When secretID is not provided, we assume that replication token should exist
 	// as a Kubernetes secret.
+	secretName := c.withPrefix(name + "-acl-token")
 	if secretID == "" {
 		// Check if the secret already exists, if so, we assume the ACL has already been
 		// created and return.