diff --git a/awx/main/scheduler/kubernetes.py b/awx/main/scheduler/kubernetes.py
index 0313d463a0..d68aa5cb6e 100644
--- a/awx/main/scheduler/kubernetes.py
+++ b/awx/main/scheduler/kubernetes.py
@@ -1,4 +1,5 @@
 import collections
+import json
 import logging
 from base64 import b64encode
 
@@ -51,6 +52,72 @@ class PodManager(object):
 
         return pods
 
+    @classmethod
+    def create_secret(self, job):
+        task = collections.namedtuple('Task', 'id instance_group')(id='', instance_group=job.instance_group)
+        pm = PodManager(task)
+        registry_cred = job.execution_environment.credential
+        host = registry_cred.get_input('host').split('/')[0]
+        username = registry_cred.get_input("username")
+        password = registry_cred.get_input("password")
+
+        # Construct container auth dict and base64 encode it
+        token = b64encode("{}:{}".format(username, password).encode('ascii')).decode()
+        auth_dict = json.dumps({"auths": {host: {"auth": token}}}, indent=4)
+        auth_data = b64encode(str(auth_dict).encode('ascii')).decode()
+
+        # Construct Secret object
+        secret = client.V1Secret()
+        secret_name = "automation-{0}-image-pull-secret-{1}".format(settings.INSTALL_UUID[:5], job.execution_environment.credential.id)
+        secret.metadata = client.V1ObjectMeta(name="{}".format(secret_name))
+        secret.type = "kubernetes.io/dockerconfigjson"
+        secret.kind = "Secret"
+        secret.data = {".dockerconfigjson": auth_data}
+
+        # Check if secret already exists
+        secrets = None
+        try:
+            secrets = pm.kube_api.list_namespaced_secret(namespace=pm.namespace)
+        except client.rest.ApiException:
+            error_msg = 'Invalid openshift or k8s cluster credential'
+            logger.exception(error_msg)
+            job.cancel(job_explanation=error_msg)
+            raise
+
+        if secrets:
+            secret_exists = False
+            secrets_dict = secrets.to_dict().get('items', [])
+            for s in secrets_dict:
+                if s['metadata']['name'] == secret_name:
+                    secret_exists = True
+            if secret_exists:
+                try:
+                    # Try to replace existing secret
+                    pm.kube_api.delete_namespaced_secret(name=secret.metadata.name, namespace=pm.namespace)
+                    pm.kube_api.create_namespaced_secret(namespace=pm.namespace, body=secret)
+                except Exception:
+                    error_msg = 'Failed to create imagePullSecret for container group {}'.format(task.instance_group.name)
+                    logger.exception(error_msg)
+                    job.cancel(job_explanation=error_msg)
+                    raise
+            else:
+                # Create an image pull secret in namespace
+                try:
+                    pm.kube_api.create_namespaced_secret(namespace=pm.namespace, body=secret)
+                except client.rest.ApiException as e:
+                    if e.status == 401:
+                        error_msg = 'Failed to create imagePullSecret: {}. Check that openshift or k8s credential has permission to create a secret.'.format(
+                            e.status
+                        )
+                        logger.exception(error_msg)
+                        # let job run for the case that the secret exists but the cluster cred doesn't have permission to create a secret
+                except Exception:
+                    error_msg = 'Failed to create imagePullSecret for container group {}'.format(task.instance_group.name)
+                    logger.exception(error_msg)
+                    job.cancel(job_explanation=error_msg)
+
+        return secret.metadata.name
+
     @property
     def namespace(self):
         return self.pod_definition['metadata']['namespace']
diff --git a/awx/main/tasks.py b/awx/main/tasks.py
index 34e6ca6215..e81aef40c9 100644
--- a/awx/main/tasks.py
+++ b/awx/main/tasks.py
@@ -3077,7 +3077,20 @@ class AWXReceptorJob:
         # Enforce EE Pull Policy
         pull_options = {"always": "Always", "missing": "IfNotPresent", "never": "Never"}
         if self.task and self.task.instance.execution_environment:
-            pod_spec['spec']['containers'][0]['imagePullPolicy'] = pull_options[self.task.instance.execution_environment.pull]
+            if self.task.instance.execution_environment.pull:
+                pod_spec['spec']['containers'][0]['imagePullPolicy'] = pull_options[self.task.instance.execution_environment.pull]
+
+        if self.task and self.task.instance.is_container_group_task:
+            # If EE credential is passed, create an imagePullSecret
+            if self.task.instance.execution_environment and self.task.instance.execution_environment.credential:
+                # Create pull secret in k8s cluster based on ee cred
+                from awx.main.scheduler.kubernetes import PodManager  # prevent circular import
+
+                pm = PodManager(self.task.instance)
+                secret_name = pm.create_secret(job=self.task.instance)
+
+                # Inject secret name into podspec
+                pod_spec['spec']['imagePullSecrets'] = [{"name": secret_name}]
 
         if self.task:
             pod_spec['metadata'] = deepmerge(