diff --git a/awx/main/management/commands/regenerate_secret_key.py b/awx/main/management/commands/regenerate_secret_key.py
new file mode 100644
index 0000000000..2e3d1a127d
--- /dev/null
+++ b/awx/main/management/commands/regenerate_secret_key.py
@@ -0,0 +1,129 @@
+import base64
+import json
+import os
+
+from django.core.management.base import BaseCommand
+from django.conf import settings
+from django.db import transaction
+from django.db.models.signals import post_save
+
+from awx.conf import settings_registry
+from awx.conf.models import Setting
+from awx.conf.signals import on_post_save_setting
+from awx.main.models import (
+    UnifiedJob, Credential, NotificationTemplate, Job, JobTemplate, WorkflowJob,
+    WorkflowJobTemplate, OAuth2Application
+)
+from awx.main.utils.encryption import (
+    encrypt_field, decrypt_field, encrypt_value, decrypt_value, get_encryption_key
+)
+
+
+class Command(BaseCommand):
+    """
+    Regenerate a new SECRET_KEY value and re-encrypt every secret in the
+    Tower database.
+    """
+
+    @transaction.atomic
+    def handle(self, **options):
+        self.old_key = settings.SECRET_KEY
+        self.new_key = base64.encodebytes(os.urandom(33)).decode().rstrip()
+        self._notification_templates()
+        self._credentials()
+        self._unified_jobs()
+        self._oauth2_app_secrets()
+        self._settings()
+        self._survey_passwords()
+        return self.new_key
+
+    def _notification_templates(self):
+        for nt in NotificationTemplate.objects.iterator():
+            CLASS_FOR_NOTIFICATION_TYPE = dict([(x[0], x[2]) for x in NotificationTemplate.NOTIFICATION_TYPES])
+            notification_class = CLASS_FOR_NOTIFICATION_TYPE[nt.notification_type]
+            for field in filter(lambda x: notification_class.init_parameters[x]['type'] == "password",
+                                notification_class.init_parameters):
+                nt.notification_configuration[field] = decrypt_field(nt, 'notification_configuration', subfield=field, secret_key=self.old_key)
+                nt.notification_configuration[field] = encrypt_field(nt, 'notification_configuration', subfield=field, secret_key=self.new_key)
+            nt.save()
+
+    def _credentials(self):
+        for credential in Credential.objects.iterator():
+            for field_name in credential.credential_type.secret_fields:
+                if field_name in credential.inputs:
+                    credential.inputs[field_name] = decrypt_field(
+                        credential,
+                        field_name,
+                        secret_key=self.old_key
+                    )
+                    credential.inputs[field_name] = encrypt_field(
+                        credential,
+                        field_name,
+                        secret_key=self.new_key
+                    )
+                credential.save()
+
+    def _unified_jobs(self):
+        for uj in UnifiedJob.objects.iterator():
+            if uj.start_args:
+                uj.start_args = decrypt_field(
+                    uj,
+                    'start_args',
+                    secret_key=self.old_key
+                )
+                uj.start_args = encrypt_field(uj, 'start_args', secret_key=self.new_key)
+                uj.save()
+
+    def _oauth2_app_secrets(self):
+        for app in OAuth2Application.objects.iterator():
+            raw = app.client_secret
+            app.client_secret = raw
+            encrypted = encrypt_value(raw, secret_key=self.new_key)
+            OAuth2Application.objects.filter(pk=app.pk).update(client_secret=encrypted)
+
+    def _settings(self):
+        # don't update memcached, the *actual* value isn't changing
+        post_save.disconnect(on_post_save_setting, sender=Setting)
+        for setting in Setting.objects.filter().order_by('pk'):
+            if settings_registry.is_setting_encrypted(setting.key):
+                setting.value = decrypt_field(setting, 'value', secret_key=self.old_key)
+                setting.value = encrypt_field(setting, 'value', secret_key=self.new_key)
+                setting.save()
+
+    def _survey_passwords(self):
+        for _type in (JobTemplate, WorkflowJobTemplate):
+            for jt in _type.objects.exclude(survey_spec={}):
+                changed = False
+                if jt.survey_spec.get('spec', []):
+                    for field in jt.survey_spec['spec']:
+                        if field.get('type') == 'password' and field.get('default', ''):
+                            raw = decrypt_value(
+                                get_encryption_key('value', None, secret_key=self.old_key),
+                                field['default']
+                            )
+                            field['default'] = encrypt_value(
+                                raw,
+                                pk=None,
+                                secret_key=self.new_key
+                            )
+                            changed = True
+                if changed:
+                    jt.save(update_fields=["survey_spec"])
+
+        for _type in (Job, WorkflowJob):
+            for job in _type.objects.exclude(survey_passwords={}).iterator():
+                changed = False
+                for key in job.survey_passwords:
+                    if key in job.extra_vars:
+                        extra_vars = json.loads(job.extra_vars)
+                        if not extra_vars.get(key):
+                            continue
+                        raw = decrypt_value(
+                            get_encryption_key('value', None, secret_key=self.old_key),
+                            extra_vars[key]
+                        )
+                        extra_vars[key] = encrypt_value(raw, pk=None, secret_key=self.new_key)
+                        job.extra_vars = json.dumps(extra_vars)
+                        changed = True
+                if changed:
+                    job.save(update_fields=["extra_vars"])
diff --git a/awx/main/tests/functional/commands/test_secret_key_regeneration.py b/awx/main/tests/functional/commands/test_secret_key_regeneration.py
new file mode 100644
index 0000000000..811363ee47
--- /dev/null
+++ b/awx/main/tests/functional/commands/test_secret_key_regeneration.py
@@ -0,0 +1,173 @@
+import json
+
+from cryptography.fernet import InvalidToken
+from django.test.utils import override_settings
+from django.conf import settings
+import pytest
+
+from awx.main import models
+from awx.conf.models import Setting
+from awx.main.management.commands import regenerate_secret_key
+from awx.main.utils.encryption import encrypt_field, decrypt_field, encrypt_value
+
+
+PREFIX = '$encrypted$UTF8$AESCBC$'
+
+
+@pytest.mark.django_db
+class TestKeyRegeneration:
+
+    def test_encrypted_ssh_password(self, credential):
+        # test basic decryption
+        assert credential.inputs['password'].startswith(PREFIX)
+        assert credential.get_input('password') == 'secret'
+
+        # re-key the credential
+        new_key = regenerate_secret_key.Command().handle()
+        new_cred = models.Credential.objects.get(pk=credential.pk)
+        assert credential.inputs['password'] != new_cred.inputs['password']
+
+        # verify that the old SECRET_KEY doesn't work
+        with pytest.raises(InvalidToken):
+            new_cred.get_input('password')
+
+        # verify that the new SECRET_KEY *does* work
+        with override_settings(SECRET_KEY=new_key):
+            assert new_cred.get_input('password') == 'secret'
+
+    def test_encrypted_setting_values(self):
+        # test basic decryption
+        settings.LOG_AGGREGATOR_PASSWORD = 'sensitive'
+        s = Setting.objects.filter(key='LOG_AGGREGATOR_PASSWORD').first()
+        assert s.value.startswith(PREFIX)
+        assert settings.LOG_AGGREGATOR_PASSWORD == 'sensitive'
+
+        # re-key the setting value
+        new_key = regenerate_secret_key.Command().handle()
+        new_setting = Setting.objects.filter(key='LOG_AGGREGATOR_PASSWORD').first()
+        assert s.value != new_setting.value
+
+        # wipe out the local cache so the value is pulled from the DB again
+        settings.cache.delete('LOG_AGGREGATOR_PASSWORD')
+
+        # verify that the old SECRET_KEY doesn't work
+        with pytest.raises(InvalidToken):
+            settings.LOG_AGGREGATOR_PASSWORD
+
+        # verify that the new SECRET_KEY *does* work
+        with override_settings(SECRET_KEY=new_key):
+            assert settings.LOG_AGGREGATOR_PASSWORD == 'sensitive'
+
+    def test_encrypted_notification_secrets(self, notification_template_with_encrypt):
+        # test basic decryption
+        nt = notification_template_with_encrypt
+        nc = nt.notification_configuration
+        assert nc['token'].startswith(PREFIX)
+
+        Slack = nt.CLASS_FOR_NOTIFICATION_TYPE[nt.notification_type]
+        class TestBackend(Slack):
+
+            def __init__(self, *args, **kw):
+                assert kw['token'] == 'token'
+
+            def send_messages(self, messages):
+                pass
+
+        nt.CLASS_FOR_NOTIFICATION_TYPE['test'] = TestBackend
+        nt.notification_type = 'test'
+        nt.send('Subject', 'Body')
+
+        # re-key the notification config
+        new_key = regenerate_secret_key.Command().handle()
+        new_nt = models.NotificationTemplate.objects.get(pk=nt.pk)
+        assert nt.notification_configuration['token'] != new_nt.notification_configuration['token']
+
+        # verify that the old SECRET_KEY doesn't work
+        with pytest.raises(InvalidToken):
+            new_nt.CLASS_FOR_NOTIFICATION_TYPE['test'] = TestBackend
+            new_nt.notification_type = 'test'
+            new_nt.send('Subject', 'Body')
+
+        # verify that the new SECRET_KEY *does* work
+        with override_settings(SECRET_KEY=new_key):
+            new_nt.send('Subject', 'Body')
+
+    def test_job_start_args(self, job_factory):
+        # test basic decryption
+        job = job_factory()
+        job.start_args = json.dumps({'foo': 'bar'})
+        job.start_args = encrypt_field(job, field_name='start_args')
+        job.save()
+        assert job.start_args.startswith(PREFIX)
+
+        # re-key the start_args
+        new_key = regenerate_secret_key.Command().handle()
+        new_job = models.Job.objects.get(pk=job.pk)
+        assert new_job.start_args != job.start_args
+
+        # verify that the old SECRET_KEY doesn't work
+        with pytest.raises(InvalidToken):
+            decrypt_field(new_job, field_name='start_args')
+
+        # verify that the new SECRET_KEY *does* work
+        with override_settings(SECRET_KEY=new_key):
+            assert json.loads(
+                decrypt_field(new_job, field_name='start_args')
+            ) == {'foo': 'bar'}
+
+    @pytest.mark.parametrize('cls', ('JobTemplate', 'WorkflowJobTemplate'))
+    def test_survey_spec(self, inventory, project, survey_spec_factory, cls):
+        params = {}
+        if cls == 'JobTemplate':
+            params['inventory'] = inventory
+            params['project'] = project
+        # test basic decryption
+        jt = getattr(models, cls).objects.create(
+            name='Example Template',
+            survey_spec=survey_spec_factory([{
+                'variable': 'secret_key',
+                'default': encrypt_value('donttell', pk=None),
+                'type': 'password'
+            }]),
+            survey_enabled=True,
+            **params
+        )
+        job = jt.create_unified_job()
+        assert jt.survey_spec['spec'][0]['default'].startswith(PREFIX)
+        assert job.survey_passwords == {'secret_key': '$encrypted$'}
+        assert json.loads(job.decrypted_extra_vars())['secret_key'] == 'donttell'
+
+        # re-key the extra_vars
+        new_key = regenerate_secret_key.Command().handle()
+        new_job = models.UnifiedJob.objects.get(pk=job.pk)
+        assert new_job.extra_vars != job.extra_vars
+
+        # verify that the old SECRET_KEY doesn't work
+        with pytest.raises(InvalidToken):
+            new_job.decrypted_extra_vars()
+
+        # verify that the new SECRET_KEY *does* work
+        with override_settings(SECRET_KEY=new_key):
+            assert json.loads(
+                new_job.decrypted_extra_vars()
+            )['secret_key'] == 'donttell'
+
+    def test_oauth2_application_client_secret(self, oauth_application):
+        # test basic decryption
+        secret = oauth_application.client_secret
+        assert len(secret) == 128
+
+        # re-key the client_secret
+        new_key = regenerate_secret_key.Command().handle()
+
+        # verify that the old SECRET_KEY doesn't work
+        with pytest.raises(InvalidToken):
+            models.OAuth2Application.objects.get(
+                pk=oauth_application.pk
+            ).client_secret
+
+        # verify that the new SECRET_KEY *does* work
+        with override_settings(SECRET_KEY=new_key):
+            assert models.OAuth2Application.objects.get(
+                pk=oauth_application.pk
+            ).client_secret == secret
diff --git a/awx/main/utils/encryption.py b/awx/main/utils/encryption.py
index 9ad89d7de4..94c7389f01 100644
--- a/awx/main/utils/encryption.py
+++ b/awx/main/utils/encryption.py
@@ -1,3 +1,5 @@
+# -*- coding: utf-8 -*-
+
 import base64
 import hashlib
 import logging
@@ -35,7 +37,7 @@ class Fernet256(Fernet):
         self._backend = backend
 
 
-def get_encryption_key(field_name, pk=None):
+def get_encryption_key(field_name, pk=None, secret_key=None):
     '''
     Generate key for encrypted password based on field name,
     ``settings.SECRET_KEY``, and instance pk (if available).
@@ -46,19 +48,57 @@ def get_encryption_key(field_name, pk=None):
     '''
     from django.conf import settings
     h = hashlib.sha512()
-    h.update(smart_bytes(settings.SECRET_KEY))
+    h.update(smart_bytes(secret_key or settings.SECRET_KEY))
     if pk is not None:
         h.update(smart_bytes(str(pk)))
     h.update(smart_bytes(field_name))
     return base64.urlsafe_b64encode(h.digest())
 
 
-def encrypt_value(value, pk=None):
+def encrypt_value(value, pk=None, secret_key=None):
+    #
+    # ⚠️  D-D-D-DANGER ZONE ⚠️
+    #
+    # !!! BEFORE USING THIS FUNCTION PLEASE READ encrypt_field !!!
+    #
     TransientField = namedtuple('TransientField', ['pk', 'value'])
-    return encrypt_field(TransientField(pk=pk, value=value), 'value')
+    return encrypt_field(TransientField(pk=pk, value=value), 'value', secret_key=secret_key)
 
 
-def encrypt_field(instance, field_name, ask=False, subfield=None):
+def encrypt_field(instance, field_name, ask=False, subfield=None, secret_key=None):
+    #
+    # ⚠️  D-D-D-DANGER ZONE ⚠️
+    #
+    # !!! PLEASE READ BEFORE USING THIS FUNCTION ANYWHERE !!!
+    #
+    # You should know that this function is used in various places throughout
+    # AWX for symmetric encryption - generally it's used to encrypt sensitive
+    # values that we store in the AWX database (such as SSH private keys for
+    # credentials).
+    #
+    # If you're reading this function's code because you're thinking about
+    # using it to encrypt *something new*, please remember that AWX has
+    # official support for *regenerating* the SECRET_KEY (on which the
+    # symmetric key is based):
+    #
+    # $ awx-manage regenerate_secret_key
+    #
+    # ...so you'll need to *also* add code to support the
+    # migration/re-encryption of these values (the code in question lives in
+    # `awx.main.management.commands.regenerate_secret_key`):
+    #
+    # For example, if you find that you're adding a new database column that is
+    # encrypted, in addition to calling `encrypt_field` in the appropriate
+    # places, you would also need to update the `awx-manage regenerate_secret_key`
+    # so that values are properly migrated when the SECRET_KEY changes.
+    #
+    # This process *generally* involves adding Python code to the
+    # `regenerate_secret_key` command, i.e.,
+    #
+    # 1.  Query the database for existing encrypted values on the appropriate object(s)
+    # 2.  Decrypting them using the *old* SECRET_KEY
+    # 3.  Storing newly encrypted values using the *newly generated* SECRET_KEY
+    #
     '''
     Return content of the given instance and field name encrypted.
     '''
@@ -76,7 +116,11 @@ def encrypt_field(instance, field_name, ask=False, subfield=None):
     value = smart_str(value)
     if not value or value.startswith('$encrypted$') or (ask and value == 'ASK'):
         return value
-    key = get_encryption_key(field_name, getattr(instance, 'pk', None))
+    key = get_encryption_key(
+        field_name,
+        getattr(instance, 'pk', None),
+        secret_key=secret_key
+    )
     f = Fernet256(key)
     encrypted = f.encrypt(smart_bytes(value))
     b64data = smart_str(base64.b64encode(encrypted))
@@ -99,7 +143,7 @@ def decrypt_value(encryption_key, value):
     return smart_str(value)
 
 
-def decrypt_field(instance, field_name, subfield=None):
+def decrypt_field(instance, field_name, subfield=None, secret_key=None):
     '''
     Return content of the given instance and field name decrypted.
     '''
@@ -115,7 +159,11 @@ def decrypt_field(instance, field_name, subfield=None):
     value = smart_str(value)
     if not value or not value.startswith('$encrypted$'):
         return value
-    key = get_encryption_key(field_name, getattr(instance, 'pk', None))
+    key = get_encryption_key(
+        field_name,
+        getattr(instance, 'pk', None),
+        secret_key=secret_key
+    )
 
     try:
         return smart_str(decrypt_value(key, value))
diff --git a/installer/roles/kubernetes/tasks/main.yml b/installer/roles/kubernetes/tasks/main.yml
index 112ce9fa8f..f17256a140 100644
--- a/installer/roles/kubernetes/tasks/main.yml
+++ b/installer/roles/kubernetes/tasks/main.yml
@@ -314,4 +314,4 @@
     {{ kubectl_or_oc }} -n {{ kubernetes_namespace }} \
       scale {{ deployment_object }} {{ kubernetes_deployment_name }} --replicas=0
     {{ kubectl_or_oc }} -n {{ kubernetes_namespace }} \
-      scale {{ deployment_object }} {{ kubernetes_deployment_name }} --replicas={{ kubernetes_deployment_replica_size }}
+      scale {{ deployment_object }} {{ kubernetes_deployment_name }} --replicas={{ replicas | default(kubernetes_deployment_replica_size) }}
diff --git a/installer/roles/kubernetes/tasks/rekey.yml b/installer/roles/kubernetes/tasks/rekey.yml
new file mode 100644
index 0000000000..b72dbaaa9a
--- /dev/null
+++ b/installer/roles/kubernetes/tasks/rekey.yml
@@ -0,0 +1,72 @@
+---
+- include_tasks: openshift_auth.yml
+  when: openshift_host is defined
+
+- include_tasks: kubernetes_auth.yml
+  when: kubernetes_context is defined
+
+- name: Use kubectl or oc
+  set_fact:
+    kubectl_or_oc: "{{ openshift_oc_bin if openshift_oc_bin is defined else 'kubectl' }}"
+
+- set_fact:
+    deployment_object: "sts"
+
+- name: Record deployment size
+  shell: |
+    {{ kubectl_or_oc }} -n {{ kubernetes_namespace }} \
+      get {{ deployment_object }} {{ kubernetes_deployment_name }} -o jsonpath="{.status.replicas}"
+  register: deployment_size
+
+- name: Scale deployment down
+  shell: |
+    {{ kubectl_or_oc }} -n {{ kubernetes_namespace }} \
+      scale {{ deployment_object }} {{ kubernetes_deployment_name }} --replicas=0
+
+- name: Wait for scale down
+  shell: |
+    {{ kubectl_or_oc }} -n {{ kubernetes_namespace }} get pods \
+      -o jsonpath='{.items[*].metadata.name}' \
+      | tr -s '[[:space:]]' '\n' \
+      | grep {{ kubernetes_deployment_name }} \
+      | grep -v postgres | wc -l
+  register: tower_pods
+  until: (tower_pods.stdout | trim) == '0'
+  retries: 30
+
+- name: Delete any existing management pod
+  shell: |
+    {{ kubectl_or_oc }} -n {{ kubernetes_namespace }} \
+      delete pod ansible-tower-management --grace-period=0 --ignore-not-found
+
+- name: Template management pod
+  set_fact:
+    management_pod: "{{ lookup('template', 'management-pod.yml.j2') }}"
+
+- name: Create management pod
+  shell: |
+    echo {{ management_pod | quote }} | {{ kubectl_or_oc }} apply -f -
+
+- name: Wait for management pod to start
+  shell: |
+    {{ kubectl_or_oc }} -n {{ kubernetes_namespace }} \
+      get pod ansible-tower-management -o jsonpath="{.status.phase}"
+  register: result
+  until: result.stdout == "Running"
+  retries: 60
+  delay: 10
+
+- name: generate a new SECRET_KEY
+  shell: |
+    {{ kubectl_or_oc }} -n {{ kubernetes_namespace }} \
+      exec -i ansible-tower-management -- bash -c "awx-manage regenerate_secret_key"
+  register: new_key
+
+- name: print the new SECRET_KEY
+  debug:
+    msg: "{{ new_key.stdout }}"
+
+- name: Delete management pod
+  shell: |
+    {{ kubectl_or_oc }} -n {{ kubernetes_namespace }} \
+      delete pod ansible-tower-management --grace-period=0 --ignore-not-found