diff --git a/clouddriver-docker/src/main/groovy/com/netflix/spinnaker/clouddriver/docker/registry/api/v2/auth/DockerBearerTokenService.groovy b/clouddriver-docker/src/main/groovy/com/netflix/spinnaker/clouddriver/docker/registry/api/v2/auth/DockerBearerTokenService.groovy
index 835d63f409e..656e06506ce 100644
--- a/clouddriver-docker/src/main/groovy/com/netflix/spinnaker/clouddriver/docker/registry/api/v2/auth/DockerBearerTokenService.groovy
+++ b/clouddriver-docker/src/main/groovy/com/netflix/spinnaker/clouddriver/docker/registry/api/v2/auth/DockerBearerTokenService.groovy
@@ -27,16 +27,16 @@ import retrofit.http.Query
 class DockerBearerTokenService {
   private Map<String, TokenService> realmToService
   private Map<String, DockerBearerToken> cachedTokens
-  private String basicAuthenticate
+  public String basicAuth
 
   DockerBearerTokenService(String username, String password) {
     realmToService = new HashMap<String, TokenService>()
     cachedTokens = new HashMap<String, DockerBearerToken>()
     if (username) {
-      basicAuthenticate = new String(Base64.encoder.encode(("${username}:${password}").bytes))
-      basicAuthenticate = "Basic $basicAuthenticate"
+      basicAuth = new String(Base64.encoder.encode(("${username}:${password}").bytes))
+      basicAuth = "Basic $basicAuth"
     } else {
-      basicAuthenticate = null
+      basicAuth = null
     }
   }
 
@@ -168,8 +168,8 @@ class DockerBearerTokenService {
 
     def tokenService = getTokenService(authenticateDetails.realm)
     def token
-    if (basicAuthenticate) {
-      token = tokenService.getToken(authenticateDetails.path, authenticateDetails.service, authenticateDetails.scope, basicAuthenticate)
+    if (basicAuth) {
+      token = tokenService.getToken(authenticateDetails.path, authenticateDetails.service, authenticateDetails.scope, basicAuth)
     }
     else {
       token = tokenService.getToken(authenticateDetails.path, authenticateDetails.service, authenticateDetails.scope)
diff --git a/clouddriver-docker/src/main/groovy/com/netflix/spinnaker/clouddriver/docker/registry/api/v2/client/DockerRegistryClient.groovy b/clouddriver-docker/src/main/groovy/com/netflix/spinnaker/clouddriver/docker/registry/api/v2/client/DockerRegistryClient.groovy
index 119ba31e2da..c665ae85987 100644
--- a/clouddriver-docker/src/main/groovy/com/netflix/spinnaker/clouddriver/docker/registry/api/v2/client/DockerRegistryClient.groovy
+++ b/clouddriver-docker/src/main/groovy/com/netflix/spinnaker/clouddriver/docker/registry/api/v2/client/DockerRegistryClient.groovy
@@ -35,10 +35,16 @@ class DockerRegistryClient {
   public String address
   private DockerRegistryService registryService
   private GsonConverter converter
+  private String basicAuth
+
+  public getBasicAuth() {
+    return basicAuth
+  }
 
   DockerRegistryClient(String address, String email, String username, String password) {
     this.tokenService = new DockerBearerTokenService(username, password)
-    this.registryService = new RestAdapter.Builder().setEndpoint(address).setLogLevel(RestAdapter.LogLevel.NONE).build().create(DockerRegistryService)
+    this.basicAuth = this.tokenService.basicAuth;
+    this.registryService = new RestAdapter.Builder().setEndpoint(address).setLogLevel(RestAdapter.LogLevel.BASIC).build().create(DockerRegistryService)
     this.converter = new GsonConverter(new GsonBuilder().create())
     this.address = address
   }
diff --git a/clouddriver-docker/src/main/groovy/com/netflix/spinnaker/clouddriver/docker/registry/security/DockerRegistryNamedAccountCredentials.groovy b/clouddriver-docker/src/main/groovy/com/netflix/spinnaker/clouddriver/docker/registry/security/DockerRegistryNamedAccountCredentials.groovy
index 2612b8f587d..be44d66c89b 100644
--- a/clouddriver-docker/src/main/groovy/com/netflix/spinnaker/clouddriver/docker/registry/security/DockerRegistryNamedAccountCredentials.groovy
+++ b/clouddriver-docker/src/main/groovy/com/netflix/spinnaker/clouddriver/docker/registry/security/DockerRegistryNamedAccountCredentials.groovy
@@ -16,88 +16,118 @@
 
 package com.netflix.spinnaker.clouddriver.docker.registry.security
 
-import com.netflix.spinnaker.clouddriver.docker.registry.api.v2.client.DockerRegistryClient;
-import com.netflix.spinnaker.clouddriver.security.AccountCredentials;
-
-import java.util.*;
+import com.netflix.spinnaker.clouddriver.docker.registry.api.v2.client.DockerRegistryClient
+import com.netflix.spinnaker.clouddriver.security.AccountCredentials
 
 public class DockerRegistryNamedAccountCredentials implements AccountCredentials<DockerRegistryCredentials> {
   public DockerRegistryNamedAccountCredentials(String accountName, String environment, String accountType,
                                                String address, String username, String password, String email,
                                                List<String> repositories) {
-    this(accountName, environment, accountType, address, username, password, email, repositories, null);
+    this(accountName, environment, accountType, address, username, password, email, repositories, null)
   }
 
   public DockerRegistryNamedAccountCredentials(String accountName, String environment, String accountType,
                                                String address, String username, String password, String email,
                                                List<String> repositories, List<String> requiredGroupMembership) {
-    this.accountName = accountName;
-    this.environment = environment;
-    this.accountType = accountType;
-    this.address = address;
-    this.username = username;
-    this.password = password;
-    this.email = email;
+    if (!accountName == 0) {
+      throw new IllegalArgumentException("Docker Registry account must be provided with a name.")
+    }
+    this.accountName = accountName
+    this.environment = environment
+    this.accountType = accountType
+
+    if (!address == 0) {
+      throw new IllegalArgumentException("Docker Registry account $accountName must provide an endpoint address.");
+    } else {
+      int addressLen = address.length();
+      if (address[addressLen - 1] == '/') {
+        address = address.substring(0, addressLen - 1)
+        addressLen -= 1
+      }
+      // Strip the v2 endpoint, as the Docker API assumes it's not present.
+      if (address.endsWith('/v2')) {
+        address = address.substring(0, addressLen - 3)
+      }
+    }
+
+    this.address = address
+    this.username = username
+    this.password = password
+    this.email = email
     this.repositories = (repositories == null) ? [] : repositories
-    this.requiredGroupMembership = requiredGroupMembership == null ? Collections.emptyList() : Collections.unmodifiableList(requiredGroupMembership);
-    this.credentials = buildCredentials();
+    this.requiredGroupMembership = requiredGroupMembership == null ? Collections.emptyList() : Collections.unmodifiableList(requiredGroupMembership)
+    this.credentials = buildCredentials()
   }
 
   @Override
   public String getName() {
-    return accountName;
+    return accountName
+  }
+
+  public String getBasicAuth() {
+    return this.credentials ?
+      this.credentials.client ?
+        this.credentials.client.basicAuth ?
+          this.credentials.client.basicAuth :
+          "" :
+        "" :
+      ""
+  }
+
+  public String getEmail() {
+    return email
+  }
+
+  public String getV2Endpoint() {
+    return "$address/v2"
   }
 
   @Override
   public String getEnvironment() {
-    return environment;
+    return environment
   }
 
   @Override
   public String getAccountType() {
-    return accountType;
+    return accountType
   }
 
   @Override
   public String getCloudProvider() {
-    return CLOUD_PROVIDER;
+    return CLOUD_PROVIDER
   }
 
   public DockerRegistryCredentials getCredentials() {
-    return credentials;
+    return credentials
   }
 
   private DockerRegistryCredentials buildCredentials() {
-    DockerRegistryClient client = new DockerRegistryClient(this.address, this.email, this.username, this.password);
-    return new DockerRegistryCredentials(client, this.repositories);
-  }
-
-  private static String getLocalName(String fullUrl) {
-    return fullUrl.substring(fullUrl.lastIndexOf('/') + 1);
+    DockerRegistryClient client = new DockerRegistryClient(this.address, this.email, this.username, this.password)
+    return new DockerRegistryCredentials(client, this.repositories)
   }
 
   @Override
   public String getProvider() {
-    return getCloudProvider();
+    return getCloudProvider()
   }
 
   public String getAccountName() {
-    return accountName;
+    return accountName
   }
 
   public List<String> getRequiredGroupMembership() {
-    return requiredGroupMembership;
+    return requiredGroupMembership
   }
 
-  private static final String CLOUD_PROVIDER = "dockerRegistry";
-  private final String accountName;
-  private final String environment;
-  private final String accountType;
-  private final String address;
-  private final String username;
-  private final String password;
-  private final String email;
-  private final List<String> repositories;
-  private final DockerRegistryCredentials credentials;
-  private final List<String> requiredGroupMembership;
+  private static final String CLOUD_PROVIDER = "dockerRegistry"
+  private final String accountName
+  private final String environment
+  private final String accountType
+  private final String address
+  private final String username
+  private final String password
+  private final String email
+  private final List<String> repositories
+  private final DockerRegistryCredentials credentials
+  private final List<String> requiredGroupMembership
 }
diff --git a/clouddriver-kubernetes/clouddriver-kubernetes.gradle b/clouddriver-kubernetes/clouddriver-kubernetes.gradle
index a113de71e7a..07f3a8bc076 100644
--- a/clouddriver-kubernetes/clouddriver-kubernetes.gradle
+++ b/clouddriver-kubernetes/clouddriver-kubernetes.gradle
@@ -1,5 +1,6 @@
 dependencies {
   compile project(":clouddriver-core")
+  compile project(":clouddriver-docker")
   compile spinnaker.dependency('frigga')
   compile spinnaker.dependency('bootActuator')
   compile spinnaker.dependency('bootWeb')
diff --git a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/api/KubernetesApiAdaptor.groovy b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/api/KubernetesApiAdaptor.groovy
new file mode 100644
index 00000000000..9c6c9b178ec
--- /dev/null
+++ b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/api/KubernetesApiAdaptor.groovy
@@ -0,0 +1,84 @@
+/*
+ * Copyright 2016 Google, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License")
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netflix.spinnaker.clouddriver.kubernetes.api
+
+import com.netflix.spinnaker.clouddriver.kubernetes.deploy.KubernetesUtil
+import io.fabric8.kubernetes.api.model.Namespace
+import io.fabric8.kubernetes.api.model.Pod
+import io.fabric8.kubernetes.api.model.ReplicationController
+import io.fabric8.kubernetes.api.model.Secret
+import io.fabric8.kubernetes.api.model.Service
+import io.fabric8.kubernetes.client.KubernetesClient
+
+class KubernetesApiAdaptor {
+  KubernetesClient client
+
+  KubernetesApiAdaptor(KubernetesClient client) {
+    if (!client) {
+      throw new IllegalArgumentException("Client may not be null.")
+    }
+    this.client = client
+  }
+
+  List<ReplicationController> getReplicationControllers(String namespace) {
+    client.replicationControllers().inNamespace(namespace).list().items
+  }
+
+  List<Pod> getPods(String namespace, String replicationControllerName) {
+    client.pods().inNamespace(namespace).withLabel(KubernetesUtil.REPLICATION_CONTROLLER_LABEL, replicationControllerName).list().items
+  }
+
+  ReplicationController getReplicationController(String namespace, String serverGroupName) {
+    client.replicationControllers().inNamespace(namespace).withName(serverGroupName).get()
+  }
+
+  ReplicationController createReplicationController(String namespace, ReplicationController replicationController) {
+    client.replicationControllers().inNamespace(namespace).create(replicationController)
+  }
+
+  Service getService(String namespace, String service) {
+    client.services().inNamespace(namespace).withName(service).get()
+  }
+
+  Service getSecurityGroup(String namespace, String securityGroup) {
+    getService(namespace, securityGroup)
+  }
+
+  Service getLoadBalancer(String namespace, String loadBalancer) {
+    getService(namespace, loadBalancer)
+  }
+
+  Secret getSecret(String namespace, String secret) {
+    client.secrets().inNamespace(namespace).withName(secret).get()
+  }
+
+  Boolean deleteSecret(String namespace, String secret) {
+    client.secrets().inNamespace(namespace).withName(secret).delete()
+  }
+
+  Secret createSecret(String namespace, Secret secret) {
+    client.secrets().inNamespace(namespace).create(secret)
+  }
+
+  Namespace getNamespace(String namespace) {
+    client.namespaces().withName(namespace).get()
+  }
+
+  Namespace createNamespace(Namespace namespace) {
+    client.namespaces().create(namespace)
+  }
+}
diff --git a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/config/KubernetesConfigurationProperties.groovy b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/config/KubernetesConfigurationProperties.groovy
index 60fda8612da..08aaa219eea 100644
--- a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/config/KubernetesConfigurationProperties.groovy
+++ b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/config/KubernetesConfigurationProperties.groovy
@@ -29,7 +29,15 @@ class KubernetesConfigurationProperties {
     String username
     String password
     List<String> namespaces
+    List<LinkedDockerRegistryConfiguration> dockerRegistries
+
   }
 
   List<ManagedAccount> accounts = []
 }
+
+@ToString(includeNames = true)
+class LinkedDockerRegistryConfiguration {
+  String accountName
+  List<String> namespaces
+}
diff --git a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/KubernetesUtil.groovy b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/KubernetesUtil.groovy
index 44638b668eb..70aacd25f57 100644
--- a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/KubernetesUtil.groovy
+++ b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/KubernetesUtil.groovy
@@ -22,51 +22,27 @@ import com.netflix.spinnaker.clouddriver.kubernetes.security.KubernetesCredentia
 import io.fabric8.kubernetes.api.model.PodList
 import io.fabric8.kubernetes.api.model.ReplicationController
 import io.fabric8.kubernetes.api.model.ReplicationControllerList
+import io.fabric8.kubernetes.api.model.Secret
 import io.fabric8.kubernetes.api.model.Service
 
 class KubernetesUtil {
-
   static String SECURITY_GROUP_LABEL_PREFIX = "security-group-"
   static String LOAD_BALANCER_LABEL_PREFIX = "load-balancer-"
   static String REPLICATION_CONTROLLER_LABEL = "replication-controller"
   private static int SECURITY_GROUP_LABEL_PREFIX_LENGTH = SECURITY_GROUP_LABEL_PREFIX.length()
   private static int LOAD_BALANCER_LABEL_PREFIX_LENGTH = LOAD_BALANCER_LABEL_PREFIX.length()
 
-  ReplicationControllerList getReplicationControllers(KubernetesCredentials credentials, String namespace) {
-    credentials.client.replicationControllers().inNamespace(namespace).list()
-  }
-
-  PodList getPods(KubernetesCredentials credentials, String namespace, String replicationControllerName) {
-    credentials.client.pods().inNamespace(namespace).withLabel(REPLICATION_CONTROLLER_LABEL, replicationControllerName).list()
-  }
-
-  ReplicationController getReplicationController(KubernetesCredentials credentials, String namespace, String serverGroupName) {
-    credentials.client.replicationControllers().inNamespace(namespace).withName(serverGroupName).get()
-  }
-
-  Service getService(KubernetesCredentials credentials, String namespace, String service) {
-    credentials.client.services().inNamespace(namespace).withName(service).get()
-  }
-
-  Service getSecurityGroup(KubernetesCredentials credentials, String namespace, String securityGroup) {
-    getService(credentials, namespace, securityGroup)
-  }
-
-  Service getLoadBalancer(KubernetesCredentials credentials, String namespace, String loadBalancer) {
-    getService(credentials, namespace, loadBalancer)
-  }
-
-  String getNextSequence(String clusterName, String namespace, KubernetesCredentials credentials) {
+  static String getNextSequence(String clusterName, String namespace, KubernetesCredentials credentials) {
     def maxSeqNumber = -1
-    def replicationControllers = getReplicationControllers(credentials, namespace)
+    def replicationControllers = credentials.apiAdaptor.getReplicationControllers(namespace)
 
-    for (def replicationController : replicationControllers.getItems()) {
+    replicationControllers.forEach( { replicationController ->
       def names = Names.parseName(replicationController.getMetadata().getName())
 
       if (names.cluster == clusterName) {
         maxSeqNumber = Math.max(maxSeqNumber, names.sequence)
       }
-    }
+    })
 
     String.format("%03d", ++maxSeqNumber)
   }
diff --git a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/description/DeployKubernetesAtomicOperationDescription.groovy b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/description/DeployKubernetesAtomicOperationDescription.groovy
index d6f0f510121..a3a7065d809 100644
--- a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/description/DeployKubernetesAtomicOperationDescription.groovy
+++ b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/description/DeployKubernetesAtomicOperationDescription.groovy
@@ -30,6 +30,7 @@ class DeployKubernetesAtomicOperationDescription extends KubernetesAtomicOperati
   Integer targetSize
   List<String> loadBalancers
   List<String> securityGroups
+  List<String> imagePullSecrets
   List<KubernetesContainerDescription> containers
 }
 
diff --git a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/CloneKubernetesAtomicOperation.groovy b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/CloneKubernetesAtomicOperation.groovy
index 28bdee36340..8773ca576bf 100644
--- a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/CloneKubernetesAtomicOperation.groovy
+++ b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/CloneKubernetesAtomicOperation.groovy
@@ -32,9 +32,6 @@ import org.springframework.beans.factory.annotation.Autowired
 class CloneKubernetesAtomicOperation implements AtomicOperation<DeploymentResult> {
   private static final String BASE_PHASE = "CLONE_SERVER_GROUP"
 
-  @Autowired
-  KubernetesUtil kubernetesUtil
-
   CloneKubernetesAtomicOperation(CloneKubernetesAtomicOperationDescription description) {
     this.description = description
   }
@@ -46,8 +43,8 @@ class CloneKubernetesAtomicOperation implements AtomicOperation<DeploymentResult
   CloneKubernetesAtomicOperationDescription description
 
   /*
-   * curl -X POST -H "Content-Type: application/json" -d  '[ { "cloneServerGroup": { "source": { "serverGroupName": "k8s-test-v000" }, "credentials":  "my-k8s-account" } } ]' localhost:7002/kubernetes/ops
-   * curl -X POST -H "Content-Type: application/json" -d  '[ { "cloneServerGroup": { "stack": "prod", "freeFormDetails": "mdservice", "targetSize": "4", "source": { "serverGroupName": "k8s-test-v000" }, "credentials":  "my-k8s-account" } } ]' localhost:7002/kubernetes/ops
+   * curl -X POST -H "Content-Type: application/json" -d  '[ { "cloneServerGroup": { "source": { "serverGroupName": "kub-test-v000" }, "credentials":  "my-kubernetes-account" } } ]' localhost:7002/kubernetes/ops
+   * curl -X POST -H "Content-Type: application/json" -d  '[ { "cloneServerGroup": { "stack": "prod", "freeFormDetails": "mdservice", "targetSize": "4", "source": { "serverGroupName": "kub-test-v000" }, "credentials":  "my-kubernetes-account" } } ]' localhost:7002/kubernetes/ops
   */
   @Override
   DeploymentResult operate(List priorOutputs) {
@@ -57,7 +54,6 @@ class CloneKubernetesAtomicOperation implements AtomicOperation<DeploymentResult
       "${description.source.serverGroupName}..."
 
     DeployKubernetesAtomicOperation deployer = new DeployKubernetesAtomicOperation(newDescription)
-    deployer.kubernetesUtil = kubernetesUtil
     DeploymentResult deploymentResult = deployer.operate(priorOutputs)
 
     task.updateStatus BASE_PHASE, "Finished copying server group for " +
@@ -75,7 +71,9 @@ class CloneKubernetesAtomicOperation implements AtomicOperation<DeploymentResult
 
     task.updateStatus BASE_PHASE, "Reading ancestor server group ${description.source.serverGroupName}..."
 
-    ReplicationController ancestorServerGroup = kubernetesUtil.getReplicationController(description.kubernetesCredentials, description.source.namespace, description.source.serverGroupName)
+    def credentials = description.kubernetesCredentials
+
+    ReplicationController ancestorServerGroup = credentials.apiAdaptor.getReplicationController(description.source.namespace, description.source.serverGroupName)
 
     if (!ancestorServerGroup) {
       throw new KubernetesResourceNotFoundException("Source server group $description.source.serverGroupName does not exist.")
@@ -91,6 +89,7 @@ class CloneKubernetesAtomicOperation implements AtomicOperation<DeploymentResult
     newDescription.namespace = description.namespace ?: description.source.namespace
     newDescription.loadBalancers = description.loadBalancers != null ? description.loadBalancers : KubernetesUtil.getDescriptionLoadBalancers(ancestorServerGroup)
     newDescription.securityGroups = description.securityGroups != null ? description.securityGroups : KubernetesUtil.getDescriptionSecurityGroups(ancestorServerGroup)
+    newDescription.imagePullSecrets = description.imagePullSecrets != null ? description.imagePullSecrets : ancestorServerGroup.spec?.template?.spec?.imagePullSecrets?.collect({ it.name })
     if (!description.containers) {
       newDescription.containers = []
       ancestorServerGroup.spec?.template?.spec?.containers?.each { it ->
diff --git a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/DeployKubernetesAtomicOperation.groovy b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/DeployKubernetesAtomicOperation.groovy
index e09c07b7308..184b84f7dd9 100644
--- a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/DeployKubernetesAtomicOperation.groovy
+++ b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/DeployKubernetesAtomicOperation.groovy
@@ -25,14 +25,10 @@ import com.netflix.spinnaker.clouddriver.kubernetes.deploy.exception.KubernetesI
 import com.netflix.spinnaker.clouddriver.orchestration.AtomicOperation
 import io.fabric8.kubernetes.api.model.ReplicationController
 import io.fabric8.kubernetes.api.model.ReplicationControllerBuilder
-import org.springframework.beans.factory.annotation.Autowired
 
 class DeployKubernetesAtomicOperation implements AtomicOperation<DeploymentResult> {
   private static final String BASE_PHASE = "DEPLOY"
 
-  @Autowired
-  KubernetesUtil kubernetesUtil
-
   DeployKubernetesAtomicOperation(DeployKubernetesAtomicOperationDescription description) {
     this.description = description
   }
@@ -44,17 +40,17 @@ class DeployKubernetesAtomicOperation implements AtomicOperation<DeploymentResul
   DeployKubernetesAtomicOperationDescription description
 
   /*
-   * curl -X POST -H "Content-Type: application/json" -d  '[ {  "createServerGroup": { "application": "k8s", "stack": "test",  "targetSize": "3", "securityGroups": [], "loadBalancers":  [],  "containers": [ { "name": "nginx", "image": "nginx" } ], "credentials":  "my-k8s-account" } } ]' localhost:7002/kubernetes/ops
+   * curl -X POST -H "Content-Type: application/json" -d  '[ {  "createServerGroup": { "application": "kub", "stack": "test",  "targetSize": "3", "securityGroups": [], "loadBalancers":  [],  "containers": [ { "name": "nginx", "image": "nginx" } ], "credentials":  "my-kubernetes-account" } } ]' localhost:7002/kubernetes/ops
    *
-   * curl -X POST -H "Content-Type: application/json" -d  '[ {  "createServerGroup": { "application": "k8s", "stack": "test",  "targetSize": "3", "securityGroups": ["frontend-sg"], "loadBalancers":  ["frontend-lb"],  "containers": [ { "name": "nginx", "image": "nginx" } ], "credentials":  "my-k8s-account" } } ]' localhost:7002/kubernetes/ops
+   * curl -X POST -H "Content-Type: application/json" -d  '[ {  "createServerGroup": { "application": "kub", "stack": "test",  "targetSize": "3", "securityGroups": ["frontend-sg"], "loadBalancers":  ["frontend-lb"],  "containers": [ { "name": "nginx", "image": "nginx" } ], "imagePullSecrets": ["my-docker-account"], "credentials":  "my-kubernetes-account" } } ]' localhost:7002/kubernetes/ops
   */
   @Override
   DeploymentResult operate(List priorOutputs) {
 
-    ReplicationController rc = deployDescription()
+    ReplicationController replicationController = deployDescription()
     DeploymentResult deploymentResult = new DeploymentResult()
-    deploymentResult.serverGroupNames = Arrays.asList("${rc.metadata.namespace}:${rc.metadata.name}".toString())
-    deploymentResult.serverGroupNameByRegion[rc.metadata.namespace] = rc.metadata.name
+    deploymentResult.serverGroupNames = Arrays.asList("${replicationController.metadata.namespace}:${replicationController.metadata.name}".toString())
+    deploymentResult.serverGroupNameByRegion[replicationController.metadata.namespace] = replicationController.metadata.name
     return deploymentResult
   }
 
@@ -79,21 +75,11 @@ class DeployKubernetesAtomicOperation implements AtomicOperation<DeploymentResul
 
     def clusterName = KubernetesUtil.combineAppStackDetail(description.application, description.stack, description.freeFormDetails)
     task.updateStatus BASE_PHASE, "Looking up next sequence index..."
-    def sequenceIndex = kubernetesUtil.getNextSequence(clusterName, namespace, credentials)
+    def sequenceIndex = KubernetesUtil.getNextSequence(clusterName, namespace, credentials)
     task.updateStatus BASE_PHASE, "Sequence index chosen to be ${sequenceIndex}."
     def replicationControllerName = String.format("%s-v%s", clusterName, sequenceIndex)
     task.updateStatus BASE_PHASE, "Replication controller name chosen to be ${replicationControllerName}."
 
-    task.updateStatus BASE_PHASE, "Collecting ports for associated security groups..."
-    def ports = []
-    for (def securityGroupName : description.securityGroups) {
-      def securityGroup = kubernetesUtil.getSecurityGroup(credentials, namespace, securityGroupName)
-
-      for (def port : securityGroup.getSpec().getPorts()) {
-        ports.add(port.getTargetPort().intVal)
-      }
-    }
-
     def replicationControllerBuilder = new ReplicationControllerBuilder()
                         .withNewMetadata().withName(replicationControllerName)
 
@@ -131,13 +117,16 @@ class DeployKubernetesAtomicOperation implements AtomicOperation<DeploymentResul
 
     replicationControllerBuilder = replicationControllerBuilder.endMetadata().withNewSpec()
 
+    task.updateStatus BASE_PHASE, "Adding image pull secrets... "
+    replicationControllerBuilder = replicationControllerBuilder.withImagePullSecrets()
+
+    for (def imagePullSecret : description.imagePullSecrets) {
+      replicationControllerBuilder = replicationControllerBuilder.addNewImagePullSecret(imagePullSecret)
+    }
+
     for (def container : description.containers) {
       task.updateStatus BASE_PHASE, "Adding container ${container.name} with image ${container.image}..."
       replicationControllerBuilder = replicationControllerBuilder.addNewContainer().withName(container.name).withImage(container.image)
-      task.updateStatus BASE_PHASE, "Opening container ports..."
-      for (def port : ports) {
-        replicationControllerBuilder = replicationControllerBuilder.addNewPort().withContainerPort(port).endPort()
-      }
 
       replicationControllerBuilder = replicationControllerBuilder.withNewResources()
       if (container.requests) {
@@ -171,13 +160,13 @@ class DeployKubernetesAtomicOperation implements AtomicOperation<DeploymentResul
       task.updateStatus BASE_PHASE, "Finished adding container ${container.name}."
     }
 
-    replicationControllerBuilder = replicationControllerBuilder.endSpec().endTemplate().endSpec().build()
+    replicationControllerBuilder = replicationControllerBuilder.endSpec().endTemplate().endSpec()
 
     task.updateStatus BASE_PHASE, "Sending replication controller spec to the Kubernetes master."
-	  ReplicationController rc = credentials.client.replicationControllers().inNamespace(namespace).create(replicationControllerBuilder)
+	  ReplicationController replicationController = credentials.apiAdaptor.createReplicationController(namespace, replicationControllerBuilder.build())
 
-    task.updateStatus BASE_PHASE, "Finished creating replication controller ${rc.metadata.name}."
+    task.updateStatus BASE_PHASE, "Finished creating replication controller ${replicationController.metadata.name}."
 
-    return rc
+    return replicationController
   }
 }
diff --git a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/CloneKubernetesAtomicOperationValidator.groovy b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/CloneKubernetesAtomicOperationValidator.groovy
index e9dc525a70c..7eedd53d7d4 100644
--- a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/CloneKubernetesAtomicOperationValidator.groovy
+++ b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/CloneKubernetesAtomicOperationValidator.groovy
@@ -17,8 +17,9 @@
 package com.netflix.spinnaker.clouddriver.kubernetes.deploy.validators
 
 import com.netflix.spinnaker.clouddriver.deploy.DescriptionValidator
-import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.CloneKubernetesAtomicOperationDescription
 import com.netflix.spinnaker.clouddriver.kubernetes.KubernetesOperation
+import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.CloneKubernetesAtomicOperationDescription
+import com.netflix.spinnaker.clouddriver.kubernetes.security.KubernetesCredentials
 import com.netflix.spinnaker.clouddriver.orchestration.AtomicOperations
 import com.netflix.spinnaker.clouddriver.security.AccountCredentialsProvider
 import org.springframework.beans.factory.annotation.Autowired
@@ -34,8 +35,13 @@ class CloneKubernetesAtomicOperationValidator extends DescriptionValidator<Clone
   @Override
   void validate(List priorDescriptions, CloneKubernetesAtomicOperationDescription description, Errors errors) {
     def helper = new StandardKubernetesAttributeValidator("cloneKubernetesAtomicOperationDescription", errors)
-    helper.validateCredentials(description.credentials, accountCredentialsProvider)
-    helper.validateSource(description.source, "source")
+    if (!helper.validateCredentials(description.credentials, accountCredentialsProvider)) {
+      return
+    }
+
+    KubernetesCredentials credentials = (KubernetesCredentials) accountCredentialsProvider.getCredentials(description.credentials).credentials
+
+    helper.validateCloneSource(description.source, "source")
     if (description.application) {
       helper.validateApplication(description.application, "application")
     }
@@ -53,20 +59,31 @@ class CloneKubernetesAtomicOperationValidator extends DescriptionValidator<Clone
     }
 
     if (description.namespace) {
-      helper.validateNamespace(description.namespace, "namespace")
+      helper.validateNamespace(credentials, description.namespace, "namespace")
     }
 
-    description.loadBalancers.eachWithIndex { name, idx ->
-      helper.validateName(name, "loadBalancers[${idx}]")
+    if (description.imagePullSecrets) {
+      description.imagePullSecrets.eachWithIndex { name, idx ->
+        helper.validateImagePullSecret(credentials, name, description.namespace ?: 'default', "imagePullSecrets[${idx}]")
+      }
     }
 
-    description.securityGroups.eachWithIndex { name, idx ->
-      helper.validateName(name, "securityGroups[${idx}]")
+    if (description.loadBalancers) {
+      description.loadBalancers.eachWithIndex { name, idx ->
+        helper.validateName(name, "loadBalancers[${idx}]")
+      }
     }
 
-    description.containers.eachWithIndex { container, idx ->
-      KubernetesContainerValidator.validate(container, helper, "container[${idx}]")
+    if (description.securityGroups) {
+      description.securityGroups.eachWithIndex { name, idx ->
+        helper.validateName(name, "securityGroups[${idx}]")
+      }
     }
 
+    if (description.containers) {
+      description.containers.eachWithIndex { container, idx ->
+        KubernetesContainerValidator.validate(container, helper, "container[${idx}]")
+      }
+    }
   }
 }
diff --git a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/DeployKubernetesAtomicOperationValidator.groovy b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/DeployKubernetesAtomicOperationValidator.groovy
index 13b04ea9e01..928e2559be4 100644
--- a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/DeployKubernetesAtomicOperationValidator.groovy
+++ b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/DeployKubernetesAtomicOperationValidator.groovy
@@ -19,6 +19,7 @@ package com.netflix.spinnaker.clouddriver.kubernetes.deploy.validators
 import com.netflix.spinnaker.clouddriver.deploy.DescriptionValidator
 import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.DeployKubernetesAtomicOperationDescription
 import com.netflix.spinnaker.clouddriver.kubernetes.KubernetesOperation
+import com.netflix.spinnaker.clouddriver.kubernetes.security.KubernetesCredentials
 import com.netflix.spinnaker.clouddriver.orchestration.AtomicOperations
 import com.netflix.spinnaker.clouddriver.security.AccountCredentialsProvider
 import org.springframework.beans.factory.annotation.Autowired
@@ -35,12 +36,21 @@ class DeployKubernetesAtomicOperationValidator extends DescriptionValidator<Depl
   void validate(List priorDescriptions, DeployKubernetesAtomicOperationDescription description, Errors errors) {
     def helper = new StandardKubernetesAttributeValidator("deployKubernetesAtomicOperationDescription", errors)
 
-    helper.validateCredentials(description.credentials, accountCredentialsProvider)
+    if (!helper.validateCredentials(description.credentials, accountCredentialsProvider)) {
+      return
+    }
+
+    KubernetesCredentials credentials = (KubernetesCredentials) accountCredentialsProvider.getCredentials(description.credentials).credentials
+
     helper.validateApplication(description.application, "application")
     helper.validateStack(description.stack, "stack")
     helper.validateDetails(description.freeFormDetails, "details")
     helper.validateNonNegative(description.targetSize, "targetSize")
-    helper.validateNamespace(description.namespace, "namespace")
+    helper.validateNamespace(credentials, description.namespace, "namespace")
+
+    description.imagePullSecrets.eachWithIndex { name, idx ->
+      helper.validateImagePullSecret(credentials, name, description.namespace ?: 'default', "imagePullSecrets[${idx}]")
+    }
 
     description.loadBalancers.eachWithIndex { name, idx ->
       helper.validateName(name, "loadBalancers[${idx}]")
diff --git a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/StandardKubernetesAttributeValidator.groovy b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/StandardKubernetesAttributeValidator.groovy
index 9c63cb87e78..6ff62adad6d 100644
--- a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/StandardKubernetesAttributeValidator.groovy
+++ b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/StandardKubernetesAttributeValidator.groovy
@@ -97,11 +97,24 @@ class StandardKubernetesAttributeValidator {
     }
   }
 
-  def validateNamespace(String value, String attribute) {
+  def validateImagePullSecret(KubernetesCredentials credentials, String value, String namespace, String attribute) {
+    if (!credentials.isRegisteredImagePullSecret(value, namespace)) {
+      errors.rejectValue("${context}.${attribute}", "${context}.${attribute}.notRegistered")
+      return false
+    }
+    return validateByRegex(value, attribute, namePattern)
+  }
+
+
+  def validateNamespace(KubernetesCredentials credentials, String value, String attribute) {
     // Namespace is optional, empty taken to mean 'default'.
     if (!value) {
       return true
     } else {
+      if (!credentials.isRegisteredNamespace(value)) {
+        errors.rejectValue("${context}.${attribute}", "${context}.${attribute}.notRegistered")
+        return false
+      }
       return validateByRegex(value, attribute, namePattern)
     }
   }
@@ -139,7 +152,7 @@ class StandardKubernetesAttributeValidator {
     result
   }
 
-  def validateSource(Object value, String attribute) {
+  def validateCloneSource(Object value, String attribute) {
     if (!value) {
       errors.rejectValue("${context}.${attribute}",  "${context}.${attribute}.empty")
       return false
diff --git a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/health/KubernetesHealthIndicator.groovy b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/health/KubernetesHealthIndicator.groovy
index f3b4c26e552..be07e7d52e8 100644
--- a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/health/KubernetesHealthIndicator.groovy
+++ b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/health/KubernetesHealthIndicator.groovy
@@ -69,11 +69,11 @@ class KubernetesHealthIndicator implements HealthIndicator {
         // This verifies that the specified credentials are sufficient to
         // access the referenced Kubernetes master endpoint.
         kubernetesCredentials.getNamespaces().each { namespace ->
-          Namespace res = kubernetesCredentials.client.namespaces().withName(namespace).get();
+          Namespace res = kubernetesCredentials.apiAdaptor.getNamespace(namespace)
           if (res == null) {
             NamespaceBuilder namespaceBuilder = new NamespaceBuilder();
             EditableNamespace newNamespace = namespaceBuilder.withNewMetadata().withName(namespace).endMetadata().build()
-            kubernetesCredentials.client.namespaces().create(newNamespace)
+            kubernetesCredentials.apiAdaptor.createNamespace(newNamespace)
             LOG.info "Created missing namespace $namespace"
           }
         }
diff --git a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/provider/agent/KubernetesServerGroupCachingAgent.groovy b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/provider/agent/KubernetesServerGroupCachingAgent.groovy
index 087482f090e..0c201bf6c0a 100644
--- a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/provider/agent/KubernetesServerGroupCachingAgent.groovy
+++ b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/provider/agent/KubernetesServerGroupCachingAgent.groovy
@@ -48,7 +48,6 @@ class KubernetesServerGroupCachingAgent implements CachingAgent, OnDemandAgent,
   final String accountName
   final String namespace
   final KubernetesCredentials credentials
-  final KubernetesUtil util
   final ObjectMapper objectMapper
 
   final OnDemandMetricsSupport metricsSupport
@@ -65,14 +64,12 @@ class KubernetesServerGroupCachingAgent implements CachingAgent, OnDemandAgent,
                                     KubernetesCredentials credentials,
                                     String namespace,
                                     ObjectMapper objectMapper,
-                                    Registry registry,
-                                    KubernetesUtil util) {
+                                    Registry registry) {
     this.kubernetesCloudProvider = kubernetesCloudProvider
     this.accountName = accountName
     this.credentials = credentials
     this.objectMapper = objectMapper
     this.namespace = namespace
-    this.util = util
     this.metricsSupport = new OnDemandMetricsSupport(registry, this, kubernetesCloudProvider.id + ":" + ON_DEMAND_TYPE)
   }
 
@@ -142,11 +139,11 @@ class KubernetesServerGroupCachingAgent implements CachingAgent, OnDemandAgent,
   }
 
   List<ReplicationController> loadReplicationControllers() {
-    util.getReplicationControllers(credentials, namespace).items
+    credentials.apiAdaptor.getReplicationControllers(namespace)
   }
 
   List<Pod> loadPods(String replicationControllerName) {
-    util.getPods(credentials, namespace, replicationControllerName).items
+    credentials.apiAdaptor.getPods(namespace, replicationControllerName)
   }
 
   @Override
@@ -175,7 +172,7 @@ class KubernetesServerGroupCachingAgent implements CachingAgent, OnDemandAgent,
       def applicationKey = Keys.getApplicationKey(applicationName)
       def clusterKey = Keys.getClusterKey(accountName, applicationName, clusterName)
       def instanceKeys = []
-      def loadBalancerKeys = util.getDescriptionLoadBalancers(replicationController).collect({
+      def loadBalancerKeys = KubernetesUtil.getDescriptionLoadBalancers(replicationController).collect({
         Keys.getLoadBalancerKey(accountName, namespace, it)
       })
 
diff --git a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/provider/config/KubernetesProviderConfig.groovy b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/provider/config/KubernetesProviderConfig.groovy
index aac685e1f39..26ba9f39ac0 100644
--- a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/provider/config/KubernetesProviderConfig.groovy
+++ b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/provider/config/KubernetesProviderConfig.groovy
@@ -74,13 +74,12 @@ class KubernetesProviderConfig {
     def scheduledAccounts = ProviderUtils.getScheduledAccounts(kubernetesProvider)
     def allAccounts = ProviderUtils.buildThreadSafeSetOfAccounts(accountCredentialsRepository, KubernetesNamedAccountCredentials)
 
-    def util = new KubernetesUtil()
     allAccounts.each { KubernetesNamedAccountCredentials credentials ->
       if (!scheduledAccounts.contains(credentials.accountName)) {
         def newlyAddedAgents = []
 
         credentials.credentials.namespaces.forEach({ namespace ->
-          newlyAddedAgents << new KubernetesServerGroupCachingAgent(kubernetesCloudProvider, credentials.accountName, credentials.credentials, namespace, objectMapper, registry, util)
+          newlyAddedAgents << new KubernetesServerGroupCachingAgent(kubernetesCloudProvider, credentials.accountName, credentials.credentials, namespace, objectMapper, registry)
         })
 
         // If there is an agent scheduler, then this provider has been through the AgentController in the past.
diff --git a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/security/KubernetesCredentials.java b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/security/KubernetesCredentials.java
index fdcbc8312a9..3d88a2128a4 100644
--- a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/security/KubernetesCredentials.java
+++ b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/security/KubernetesCredentials.java
@@ -16,28 +16,100 @@
 
 package com.netflix.spinnaker.clouddriver.kubernetes.security;
 
-import io.fabric8.kubernetes.client.KubernetesClient;
+import com.netflix.spinnaker.clouddriver.docker.registry.security.DockerRegistryNamedAccountCredentials;
+import com.netflix.spinnaker.clouddriver.kubernetes.api.KubernetesApiAdaptor;
+import com.netflix.spinnaker.clouddriver.kubernetes.config.LinkedDockerRegistryConfiguration;
+import com.netflix.spinnaker.clouddriver.security.AccountCredentialsRepository;
+import io.fabric8.kubernetes.api.model.Secret;
+import io.fabric8.kubernetes.api.model.SecretBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-import java.util.List;
+import java.util.*;
 
 public class KubernetesCredentials {
-  private final KubernetesClient client;
+  private final KubernetesApiAdaptor apiAdaptor;
   private final List<String> namespaces;
+  private final List<LinkedDockerRegistryConfiguration> dockerRegistries;
+  private final HashMap<String, List<String>> imagePullSecrets;
+  private final Logger LOG;
 
-  public KubernetesCredentials(List<String> namespaces, KubernetesClient client) {
-    this.client = client;
-    this.namespaces = namespaces;
+  public KubernetesCredentials(KubernetesApiAdaptor apiAdaptor,
+                               List<String> namespaces,
+                               List<LinkedDockerRegistryConfiguration> dockerRegistries,
+                               AccountCredentialsRepository accountCredentialsRepository) {
+    this.apiAdaptor = apiAdaptor;
+    this.namespaces = namespaces != null ? namespaces : new ArrayList<>();
+    this.dockerRegistries = dockerRegistries != null ? dockerRegistries : new ArrayList<>();
+    this.imagePullSecrets = new HashMap<>();
+    this.LOG = LoggerFactory.getLogger(KubernetesCredentials.class);
+
+    for (int i = 0; i < this.dockerRegistries.size(); i++) {
+      LinkedDockerRegistryConfiguration registry = this.dockerRegistries.get(i);
+      this.LOG.info("Adding secrets for docker registry " + registry.getAccountName() + ".");
+      List<String> registryNamespaces = registry.getNamespaces();
+      List<String> affectedNamespaces = registryNamespaces != null && registryNamespaces.size() > 0 ? registryNamespaces : this.namespaces;
+
+      DockerRegistryNamedAccountCredentials account = (DockerRegistryNamedAccountCredentials) accountCredentialsRepository.getOne(registry.getAccountName());
+
+      if (account == null) {
+        throw new IllegalArgumentException("The account " + registry.getAccountName() + " was not configured inside Clouddriver.");
+      }
+
+      for (int j = 0; j < affectedNamespaces.size(); j++) {
+        String inNamespace = affectedNamespaces.get(j);
+        SecretBuilder secretBuilder = new SecretBuilder();
+        String secretName = registry.getAccountName();
+
+        Secret exists = this.apiAdaptor.getSecret(inNamespace, secretName);
+        if (exists != null) {
+          this.LOG.info("Secret for docker registry " + registry.getAccountName() + " in namespace " + inNamespace + " is being repopulated.");
+          this.apiAdaptor.deleteSecret(inNamespace, secretName);
+        }
+
+        secretBuilder = secretBuilder.withNewMetadata().withName(secretName).endMetadata();
+
+        HashMap<String, String> secretData = new HashMap<>(1);
+        String dockerCfg = String.format("{ \"%s\": { \"auth\": \"%s\", \"email\": \"%s\" } }", account.getV2Endpoint(), account.getBasicAuth(), account.getEmail());
+        dockerCfg = new String(Base64.getEncoder().encode(dockerCfg.getBytes()));
+        secretData.put(".dockercfg", dockerCfg);
+
+        secretBuilder = secretBuilder.withData(secretData).withType("kubernetes.io/dockercfg");
+        this.apiAdaptor.createSecret(inNamespace, secretBuilder.build());
+
+        List<String> existingSecrets = imagePullSecrets.get(inNamespace);
+        existingSecrets = existingSecrets != null ? existingSecrets : new ArrayList<>();
+        existingSecrets.add(secretName);
+        imagePullSecrets.put(inNamespace, existingSecrets);
+      }
+    }
   }
 
-  public KubernetesClient getClient() {
-    return client;
+  public KubernetesApiAdaptor getApiAdaptor() {
+    return apiAdaptor;
   }
 
   public List<String> getNamespaces() {
     return namespaces;
   }
 
+  public List<LinkedDockerRegistryConfiguration> getDockerRegistries() {
+    return dockerRegistries;
+  }
+
+  public Map<String, List<String>> getImagePullSecrets() {
+    return imagePullSecrets;
+  }
+
   public Boolean isRegisteredNamespace(String namespace) {
-    return namespaces != null && namespaces.contains(namespace);
+    return namespaces.contains(namespace);
+  }
+
+  public Boolean isRegisteredImagePullSecret(String secret, String namespace) {
+    List<String> secrets = imagePullSecrets.get(namespace);
+    if (secrets == null) {
+      return false;
+    }
+    return secrets.contains(secret);
   }
 }
diff --git a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/security/KubernetesCredentialsInitializer.groovy b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/security/KubernetesCredentialsInitializer.groovy
index 4b44dab416b..1cd4424e58d 100644
--- a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/security/KubernetesCredentialsInitializer.groovy
+++ b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/security/KubernetesCredentialsInitializer.groovy
@@ -28,6 +28,7 @@ import org.springframework.beans.factory.config.ConfigurableBeanFactory
 import org.springframework.context.ApplicationContext
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
+import org.springframework.context.annotation.DependsOn
 import org.springframework.context.annotation.Scope
 import org.springframework.stereotype.Component
 
@@ -58,21 +59,25 @@ class KubernetesCredentialsInitializer implements CredentialsInitializerSynchron
 
   @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
   @Bean
+  @DependsOn("dockerRegistryNamedAccountCredentials")
   List<?> synchronizeKubernetesAccounts(KubernetesConfigurationProperties kubernetesConfigurationProperties, CatsModule catsModule) {
     def (ArrayList<KubernetesConfigurationProperties.ManagedAccount> accountsToAdd, List<String> namesOfDeletedAccounts) =
     ProviderUtils.calculateAccountDeltas(accountCredentialsRepository,
                                          KubernetesNamedAccountCredentials,
                                          kubernetesConfigurationProperties.accounts)
 
+    // TODO(lwander): Modify accounts when their dockerRegistries attribute is updated as well -- need to ask @duftler.
     accountsToAdd.each { KubernetesConfigurationProperties.ManagedAccount managedAccount ->
       try {
-        def kubernetesAccount = new KubernetesNamedAccountCredentials(managedAccount.name,
+        def kubernetesAccount = new KubernetesNamedAccountCredentials(accountCredentialsRepository,
+                                                                      managedAccount.name,
                                                                       managedAccount.environment ?: managedAccount.name,
                                                                       managedAccount.accountType ?: managedAccount.name,
                                                                       managedAccount.master,
                                                                       managedAccount.username,
                                                                       managedAccount.password,
-                                                                      managedAccount.namespaces)
+                                                                      managedAccount.namespaces,
+                                                                      managedAccount.dockerRegistries)
 
         accountCredentialsRepository.save(managedAccount.name, kubernetesAccount)
       } catch (e) {
diff --git a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/security/KubernetesNamedAccountCredentials.java b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/security/KubernetesNamedAccountCredentials.java
index adcc72006ad..349e8b48f19 100644
--- a/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/security/KubernetesNamedAccountCredentials.java
+++ b/clouddriver-kubernetes/src/main/groovy/com/netflix/spinnaker/clouddriver/kubernetes/security/KubernetesNamedAccountCredentials.java
@@ -16,94 +16,123 @@
 
 package com.netflix.spinnaker.clouddriver.kubernetes.security;
 
+import com.netflix.spinnaker.clouddriver.kubernetes.api.KubernetesApiAdaptor;
+import com.netflix.spinnaker.clouddriver.kubernetes.config.LinkedDockerRegistryConfiguration;
 import com.netflix.spinnaker.clouddriver.security.AccountCredentials;
+import com.netflix.spinnaker.clouddriver.security.AccountCredentialsRepository;
 import io.fabric8.kubernetes.client.Config;
 import io.fabric8.kubernetes.client.ConfigBuilder;
 import io.fabric8.kubernetes.client.DefaultKubernetesClient;
 import io.fabric8.kubernetes.client.KubernetesClient;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
 public class KubernetesNamedAccountCredentials implements AccountCredentials<KubernetesCredentials> {
-    public KubernetesNamedAccountCredentials(String accountName, String environment, String accountType, String master, String username, String password, List<String> namespaces) {
-      this(accountName, environment, accountType, master, username, password, namespaces, null);
+  public KubernetesNamedAccountCredentials(AccountCredentialsRepository accountCredentialsRepository,
+                                           String accountName,
+                                           String environment,
+                                           String accountType,
+                                           String master,
+                                           String username,
+                                           String password,
+                                           List<String> namespaces,
+                                           List<LinkedDockerRegistryConfiguration> dockerRegistries) {
+    this(accountCredentialsRepository, accountName, environment, accountType, master, username, password, namespaces, dockerRegistries, null);
+  }
+
+  public KubernetesNamedAccountCredentials(AccountCredentialsRepository accountCredentialsRepository,
+                                           String accountName,
+                                           String environment,
+                                           String accountType,
+                                           String master,
+                                           String username,
+                                           String password,
+                                           List<String> namespaces,
+                                           List<LinkedDockerRegistryConfiguration> dockerRegistries,
+                                           List<String> requiredGroupMembership) {
+    this.accountName = accountName;
+    this.environment = environment;
+    this.accountType = accountType;
+    this.master = master;
+    this.username = username;
+    this.password = password;
+    this.namespaces = (namespaces == null || namespaces.size() == 0) ? Arrays.asList("default") : namespaces;
+    // TODO(lwander): what is this?
+    this.requiredGroupMembership = requiredGroupMembership == null ? Collections.emptyList() : Collections.unmodifiableList(requiredGroupMembership);
+    this.dockerRegistries = dockerRegistries != null ? dockerRegistries : new ArrayList<>();
+    this.accountCredentialsRepository = accountCredentialsRepository;
+    this.credentials = buildCredentials();
+  }
+
+  @Override
+  public String getName() {
+    return accountName;
+  }
+
+  @Override
+  public String getEnvironment() {
+    return environment;
+  }
+
+  @Override
+  public String getAccountType() {
+    return accountType;
+  }
+
+  @Override
+  public String getCloudProvider() {
+    return CLOUD_PROVIDER;
+  }
+
+  public List<LinkedDockerRegistryConfiguration> getDockerRegistries() {
+    return dockerRegistries;
+  }
+
+  public KubernetesCredentials getCredentials() {
+    return credentials;
+  }
+
+  private KubernetesCredentials buildCredentials() {
+    Config config = new ConfigBuilder().withMasterUrl(master).withUsername(username).withPassword(password).withTrustCerts(true).build();
+    KubernetesClient client;
+    try {
+      client = new DefaultKubernetesClient(config);
+    } catch (Exception e) {
+      throw new RuntimeException("failed to create credentials", e);
     }
-
-    public KubernetesNamedAccountCredentials(String accountName, String environment, String accountType, String master, String username, String password, List<String> namespaces, List<String> requiredGroupMembership) {
-      this.accountName = accountName;
-      this.environment = environment;
-      this.accountType = accountType;
-      this.master = master;
-      this.username = username;
-      this.password = password;
-      this.namespaces = (namespaces == null || namespaces.size() == 0) ? Arrays.asList("default") : namespaces;
-      // TODO(lwander): what is this?
-      this.requiredGroupMembership = requiredGroupMembership == null ? Collections.emptyList() : Collections.unmodifiableList(requiredGroupMembership);
-      this.credentials = buildCredentials();
-    }
-
-    @Override
-    public String getName() {
-      return accountName;
-    }
-
-    @Override
-    public String getEnvironment() {
-      return environment;
-    }
-
-    @Override
-    public String getAccountType() {
-      return accountType;
-    }
-
-    @Override
-    public String getCloudProvider() {
-      return CLOUD_PROVIDER;
-    }
-
-    public KubernetesCredentials getCredentials() {
-      return credentials;
-    }
-
-    private KubernetesCredentials buildCredentials() {
-      Config config = new ConfigBuilder().withMasterUrl(master).withUsername(username).withPassword(password).withTrustCerts(true).build();
-      KubernetesClient client;
-      try {
-        client = new DefaultKubernetesClient(config);
-      } catch (Exception e) {
-        throw new RuntimeException("failed to create credentials", e);
-      }
-      return new KubernetesCredentials(this.namespaces, client);
-    }
-
-    private static String getLocalName(String fullUrl) {
-        return fullUrl.substring(fullUrl.lastIndexOf('/') + 1);
-    }
-
-    @Override
-    public String getProvider() {
-      return getCloudProvider();
-    }
-
-    public String getAccountName() {
-      return accountName;
-    }
-
-    public List<String> getRequiredGroupMembership() {
-      return requiredGroupMembership;
-    }
-
-    private static final String CLOUD_PROVIDER = "kubernetes";
-    private final String accountName;
-    private final String environment;
-    private final String accountType;
-    private final String master;
-    private final String username;
-    private final String password;
-    private final List<String> namespaces;
-    private final KubernetesCredentials credentials;
-    private final List<String> requiredGroupMembership;
+    return new KubernetesCredentials(new KubernetesApiAdaptor(client), this.namespaces, this.dockerRegistries, this.accountCredentialsRepository);
+  }
+
+  private static String getLocalName(String fullUrl) {
+    return fullUrl.substring(fullUrl.lastIndexOf('/') + 1);
+  }
+
+  @Override
+  public String getProvider() {
+    return getCloudProvider();
+  }
+
+  public String getAccountName() {
+    return accountName;
+  }
+
+  public List<String> getRequiredGroupMembership() {
+    return requiredGroupMembership;
+  }
+
+  private static final String CLOUD_PROVIDER = "kubernetes";
+  private final String accountName;
+  private final String environment;
+  private final String accountType;
+  private final String master;
+  private final String username;
+  private final String password;
+  private final List<String> namespaces;
+  private final KubernetesCredentials credentials;
+  private final List<String> requiredGroupMembership;
+  private final List<LinkedDockerRegistryConfiguration> dockerRegistries;
+  private final AccountCredentialsRepository accountCredentialsRepository;
 }
diff --git a/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/KubernetesUtilSpec.groovy b/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/KubernetesUtilSpec.groovy
deleted file mode 100644
index 0c87cc3e388..00000000000
--- a/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/KubernetesUtilSpec.groovy
+++ /dev/null
@@ -1,110 +0,0 @@
-/*
- * Copyright 2015 Google, Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.netflix.spinnaker.clouddriver.kubernetes.deploy
-
-import com.netflix.spinnaker.clouddriver.data.task.Task
-import com.netflix.spinnaker.clouddriver.data.task.TaskRepository
-import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.DeployKubernetesAtomicOperationDescription
-import com.netflix.spinnaker.clouddriver.kubernetes.security.KubernetesCredentials;
-import io.fabric8.kubernetes.api.model.ReplicationController
-import io.fabric8.kubernetes.api.model.ReplicationControllerList
-import io.fabric8.kubernetes.api.model.Service
-import io.fabric8.kubernetes.api.model.ServiceList
-import io.fabric8.kubernetes.api.model.ServiceSpec
-import io.fabric8.kubernetes.client.dsl.internal.ServiceOperationsImpl
-import io.fabric8.kubernetes.client.dsl.internal.ReplicationControllerOperationsImpl
-import io.fabric8.kubernetes.client.KubernetesClient;
-import spock.lang.Subject
-import spock.lang.Specification
-
-class KubernetesUtilSpec extends Specification {
-  private static final NAMESPACE = "default"
-  private static final SERVICE = "service"
-
-  def kubernetesClientMock
-  def kubernetesUtil
-  def credentials
-  def replicationControllerOperationsMock
-  def replicationControllerListMock
-  def replicationControllerMock
-
-  def serviceOperationsMock
-  def serviceListMock
-  def serviceMock
-  def serviceSpecMock
-
-  def setupSpec() {
-    TaskRepository.threadLocalTask.set(Mock(Task))
-  }
-
-  def setup() {
-    kubernetesUtil = new KubernetesUtil()
-    kubernetesClientMock = Mock(KubernetesClient)
-    credentials = new KubernetesCredentials([NAMESPACE], kubernetesClientMock)
-    replicationControllerOperationsMock = Mock(ReplicationControllerOperationsImpl)
-    replicationControllerListMock = Mock(ReplicationControllerList)
-    replicationControllerMock = Mock(ReplicationController)
-
-    serviceOperationsMock = Mock(ServiceOperationsImpl)
-    serviceListMock = Mock(ServiceList)
-    serviceMock = Mock(Service)
-    serviceSpecMock = Mock(ServiceSpec)
-  }
-
-  void "list replication controllers"() {
-    when:
-      kubernetesUtil.getReplicationControllers(credentials, NAMESPACE)
-
-    then:
-      1 * kubernetesClientMock.replicationControllers() >> replicationControllerOperationsMock
-      1 * replicationControllerOperationsMock.inNamespace(NAMESPACE) >> replicationControllerOperationsMock
-      1 * replicationControllerOperationsMock.list() >> replicationControllerListMock
-  }
-
-  void "get service"() {
-    when:
-      kubernetesUtil.getService(credentials, NAMESPACE, SERVICE)
-
-    then:
-      1 * kubernetesClientMock.services() >> serviceOperationsMock
-      1 * serviceOperationsMock.inNamespace(NAMESPACE) >> serviceOperationsMock
-      1 * serviceOperationsMock.withName(SERVICE) >> serviceOperationsMock
-      1 * serviceOperationsMock.get() >> serviceMock
-  }
-
-  void "get security group"() {
-    when:
-      kubernetesUtil.getSecurityGroup(credentials, NAMESPACE, SERVICE)
-
-    then:
-      1 * kubernetesClientMock.services() >> serviceOperationsMock
-      1 * serviceOperationsMock.inNamespace(NAMESPACE) >> serviceOperationsMock
-      1 * serviceOperationsMock.withName(SERVICE) >> serviceOperationsMock
-      1 * serviceOperationsMock.get() >> serviceMock
-  }
-
-  void "get load balancer"() {
-    when:
-      kubernetesUtil.getLoadBalancer(credentials, NAMESPACE, SERVICE)
-
-    then:
-      1 * kubernetesClientMock.services() >> serviceOperationsMock
-      1 * serviceOperationsMock.inNamespace(NAMESPACE) >> serviceOperationsMock
-      1 * serviceOperationsMock.withName(SERVICE) >> serviceOperationsMock
-      1 * serviceOperationsMock.get() >> serviceMock
-  }
-}
diff --git a/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/CloneKubernetesAtomicOperationSpec.groovy b/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/CloneKubernetesAtomicOperationSpec.groovy
index cc73dafba4f..9b06a9fc161 100644
--- a/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/CloneKubernetesAtomicOperationSpec.groovy
+++ b/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/CloneKubernetesAtomicOperationSpec.groovy
@@ -18,19 +18,13 @@ package com.netflix.spinnaker.clouddriver.kubernetes.deploy.ops
 
 import com.netflix.spinnaker.clouddriver.data.task.Task
 import com.netflix.spinnaker.clouddriver.data.task.TaskRepository
-import com.netflix.spinnaker.clouddriver.kubernetes.deploy.KubernetesUtil
+import com.netflix.spinnaker.clouddriver.kubernetes.api.KubernetesApiAdaptor
 import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.CloneKubernetesAtomicOperationDescription
 import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.KubernetesContainerDescription
 import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.KubernetesResourceDescription
-import io.fabric8.kubernetes.api.model.Container
-import io.fabric8.kubernetes.api.model.ObjectMeta
-import io.fabric8.kubernetes.api.model.PodSpec
-import io.fabric8.kubernetes.api.model.PodTemplateSpec
-import io.fabric8.kubernetes.api.model.Quantity
-import io.fabric8.kubernetes.api.model.ReplicationController
-import io.fabric8.kubernetes.api.model.ReplicationControllerSpec
-import io.fabric8.kubernetes.api.model.ResourceRequirements
-import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder
+import com.netflix.spinnaker.clouddriver.kubernetes.security.KubernetesCredentials
+import com.netflix.spinnaker.clouddriver.security.AccountCredentialsRepository
+import io.fabric8.kubernetes.api.model.*
 import spock.lang.Specification
 import spock.lang.Subject
 
@@ -60,15 +54,17 @@ class CloneKubernetesAtomicOperationSpec extends Specification {
   def podTemplateSpec
   def objectMetadata
   def podSpec
-  def kubernetesUtilMock
   def replicationControllerContainers
+  def apiMock
+  def credentials
+  def accountCredentialsRepositoryMock
 
   def setupSpec() {
     TaskRepository.threadLocalTask.set(Mock(Task))
   }
 
   def setup() {
-    kubernetesUtilMock = Mock(KubernetesUtil)
+    apiMock = Mock(KubernetesApiAdaptor)
 
     containers = []
     CONTAINER_NAMES.eachWithIndex { name, idx ->
@@ -99,6 +95,8 @@ class CloneKubernetesAtomicOperationSpec extends Specification {
     podTemplateSpec= new PodTemplateSpec()
     objectMetadata = new ObjectMeta()
     podSpec = new PodSpec()
+    accountCredentialsRepositoryMock = Mock(AccountCredentialsRepository)
+    credentials = new KubernetesCredentials(apiMock, [], [], accountCredentialsRepositoryMock)
 
     objectMetadata.setLabels(LABELS)
     podTemplateSpec.setMetadata(objectMetadata)
@@ -133,13 +131,13 @@ class CloneKubernetesAtomicOperationSpec extends Specification {
   void "builds a description based on ancestor server group, overrides nothing"() {
     setup:
       def inputDescription = new CloneKubernetesAtomicOperationDescription(
-        source: [serverGroupName: ANCESTOR_SERVER_GROUP_NAME, namespace: NAMESPACE1]
+        source: [serverGroupName: ANCESTOR_SERVER_GROUP_NAME, namespace: NAMESPACE1],
+        kubernetesCredentials: credentials
       )
 
       @Subject def operation = new CloneKubernetesAtomicOperation(inputDescription)
 
-      kubernetesUtilMock.getReplicationController(inputDescription.kubernetesCredentials, NAMESPACE1, inputDescription.source.serverGroupName) >> replicationController
-      operation.kubernetesUtil = kubernetesUtilMock
+      apiMock.getReplicationController(NAMESPACE1, inputDescription.source.serverGroupName) >> replicationController
 
     when:
       def resultDescription = operation.cloneAndOverrideDescription()
@@ -173,13 +171,13 @@ class CloneKubernetesAtomicOperationSpec extends Specification {
         loadBalancers: LOAD_BALANCER_NAMES,
         securityGroups: SECURITY_GROUP_NAMES,
         containers: containers,
+        kubernetesCredentials: credentials,
         source: [serverGroupName: ANCESTOR_SERVER_GROUP_NAME, namespace: NAMESPACE2]
       )
 
       @Subject def operation = new CloneKubernetesAtomicOperation(inputDescription)
 
-      kubernetesUtilMock.getReplicationController(inputDescription.kubernetesCredentials, NAMESPACE2, inputDescription.source.serverGroupName) >> replicationController
-      operation.kubernetesUtil = kubernetesUtilMock
+      apiMock.getReplicationController(NAMESPACE2, inputDescription.source.serverGroupName) >> replicationController
 
     when:
       def resultDescription = operation.cloneAndOverrideDescription()
diff --git a/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/DeployKubernetesAtomicOperationSpec.groovy b/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/DeployKubernetesAtomicOperationSpec.groovy
index 8b25f755540..ac0e0bef477 100644
--- a/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/DeployKubernetesAtomicOperationSpec.groovy
+++ b/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/ops/DeployKubernetesAtomicOperationSpec.groovy
@@ -18,24 +18,20 @@ package com.netflix.spinnaker.clouddriver.kubernetes.deploy.ops
 
 import com.netflix.spinnaker.clouddriver.data.task.Task
 import com.netflix.spinnaker.clouddriver.data.task.TaskRepository
-import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.DeployKubernetesAtomicOperationDescription
+import com.netflix.spinnaker.clouddriver.docker.registry.security.DockerRegistryNamedAccountCredentials
+import com.netflix.spinnaker.clouddriver.kubernetes.api.KubernetesApiAdaptor
+import com.netflix.spinnaker.clouddriver.kubernetes.config.LinkedDockerRegistryConfiguration
 import com.netflix.spinnaker.clouddriver.kubernetes.deploy.KubernetesUtil
+import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.DeployKubernetesAtomicOperationDescription
 import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.KubernetesContainerDescription
 import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.KubernetesResourceDescription
 import com.netflix.spinnaker.clouddriver.kubernetes.security.KubernetesCredentials
-import io.fabric8.kubernetes.api.model.ObjectMeta;
-import io.fabric8.kubernetes.api.model.IntOrString
-import io.fabric8.kubernetes.api.model.ReplicationController
-import io.fabric8.kubernetes.api.model.ReplicationControllerList
-import io.fabric8.kubernetes.api.model.Service
-import io.fabric8.kubernetes.api.model.ServiceList
-import io.fabric8.kubernetes.api.model.ServicePort
-import io.fabric8.kubernetes.api.model.ServiceSpec
-import io.fabric8.kubernetes.client.KubernetesClient;
+import com.netflix.spinnaker.clouddriver.security.AccountCredentialsRepository
+import io.fabric8.kubernetes.api.model.*
 import io.fabric8.kubernetes.client.dsl.internal.ReplicationControllerOperationsImpl
 import io.fabric8.kubernetes.client.dsl.internal.ServiceOperationsImpl
-import spock.lang.Subject
 import spock.lang.Specification
+import spock.lang.Subject
 
 class DeployKubernetesAtomicOperationSpec extends Specification {
   private static final NAMESPACE = "default"
@@ -51,10 +47,9 @@ class DeployKubernetesAtomicOperationSpec extends Specification {
   private static final REQUEST_MEMORY = ["100Mi", "200Mi"]
   private static final LIMIT_CPU = ["120m", "200m"]
   private static final LIMIT_MEMORY = ["200Mi", "300Mi"]
-  private static final TARGET_PORT = 80
+  private static final DOCKER_REGISTRY_ACCOUNTS = [new LinkedDockerRegistryConfiguration(accountName: "my-docker-account")]
 
-  def kubernetesClientMock
-  def kubernetesUtilMock
+  def apiMock
   def credentials
   def containers
   def description
@@ -74,13 +69,14 @@ class DeployKubernetesAtomicOperationSpec extends Specification {
   def clusterName
   def replicationControllerName
 
+  def accountCredentialsRepositoryMock
+
   def setupSpec() {
     TaskRepository.threadLocalTask.set(Mock(Task))
   }
 
   def setup() {
-    kubernetesClientMock = Mock(KubernetesClient)
-    kubernetesUtilMock = Mock(KubernetesUtil)
+    apiMock = Mock(KubernetesApiAdaptor)
     replicationControllerOperationsMock = Mock(ReplicationControllerOperationsImpl)
     replicationControllerListMock = Mock(ReplicationControllerList)
     replicationControllerMock = Mock(ReplicationController)
@@ -91,8 +87,17 @@ class DeployKubernetesAtomicOperationSpec extends Specification {
     servicePortMock = Mock(ServicePort)
     metadataMock = Mock(ObjectMeta)
     intOrStringMock = Mock(IntOrString)
+    accountCredentialsRepositoryMock = Mock(AccountCredentialsRepository)
+
+    DOCKER_REGISTRY_ACCOUNTS.forEach({ account ->
+      def dockerRegistryAccountMock = Mock(DockerRegistryNamedAccountCredentials)
+      accountCredentialsRepositoryMock.getOne(account.accountName) >> dockerRegistryAccountMock
+      dockerRegistryAccountMock.getAccountName() >> account
+      apiMock.getSecret(NAMESPACE, account.accountName) >> null
+      apiMock.createSecret(NAMESPACE, _) >> null
+    })
 
-    credentials = new KubernetesCredentials([NAMESPACE], kubernetesClientMock)
+    credentials = new KubernetesCredentials(apiMock, [NAMESPACE], DOCKER_REGISTRY_ACCOUNTS, accountCredentialsRepositoryMock)
     clusterName = KubernetesUtil.combineAppStackDetail(APPLICATION, STACK, DETAILS)
     replicationControllerName = String.format("%s-v%s", clusterName, SEQUENCE)
 
@@ -118,25 +123,16 @@ class DeployKubernetesAtomicOperationSpec extends Specification {
       )
 
       @Subject def operation = new DeployKubernetesAtomicOperation(description)
-      operation.kubernetesUtil = kubernetesUtilMock
 
     when:
       operation.operate([])
 
     then:
-      1 * kubernetesUtilMock.getNextSequence(_, NAMESPACE, _) >> SEQUENCE
-      SECURITY_GROUP_NAMES.each { name ->
-        1 * kubernetesUtilMock.getSecurityGroup(credentials, NAMESPACE, name) >> serviceMock
-        1 * serviceMock.getSpec() >> serviceSpecMock
-        1 * serviceSpecMock.getPorts() >> [servicePortMock]
-        1 * servicePortMock.getTargetPort() >> intOrStringMock
-        1 * intOrStringMock.getIntVal() >> TARGET_PORT
-      }
 
-    then:
-      1 * kubernetesClientMock.replicationControllers() >> replicationControllerOperationsMock
-      1 * replicationControllerOperationsMock.inNamespace(NAMESPACE) >> replicationControllerOperationsMock
-      1 * replicationControllerOperationsMock.create({ rc ->
+      1 * apiMock.getReplicationControllers(NAMESPACE) >> []
+      5 * replicationControllerMock.getMetadata() >> metadataMock
+      3 * metadataMock.getName() >> replicationControllerName
+      1 * apiMock.createReplicationController(NAMESPACE, { ReplicationController rc ->
         LOAD_BALANCER_NAMES.each { name ->
           assert(rc.metadata.labels[KubernetesUtil.loadBalancerKey(name)])
         }
@@ -153,19 +149,16 @@ class DeployKubernetesAtomicOperationSpec extends Specification {
           assert(rc.spec.template.metadata.labels[KubernetesUtil.securityGroupKey(name)])
         }
 
-        assert(rc.spec[0].replicas == TARGET_SIZE)
+        assert(rc.spec.replicas == TARGET_SIZE)
 
         CONTAINER_NAMES.eachWithIndex { name, idx ->
-          assert(rc.spec.template.spec.containers[0][idx].name == name)
-          assert(rc.spec.template.spec.containers[0][idx].image == name)
-          assert(rc.spec.template.spec.containers[0][idx].ports[0].containerPort == TARGET_PORT)
-          assert(rc.spec.template.spec.containers[0][idx].resources.requests.cpu == REQUEST_CPU[idx])
-          assert(rc.spec.template.spec.containers[0][idx].resources.requests.memory == REQUEST_MEMORY[idx])
-          assert(rc.spec.template.spec.containers[0][idx].resources.limits.cpu == LIMIT_CPU[idx])
-          assert(rc.spec.template.spec.containers[0][idx].resources.limits.memory == LIMIT_MEMORY[idx])
+          assert(rc.spec.template.spec.containers[idx].name == name)
+          assert(rc.spec.template.spec.containers[idx].image == name)
+          assert(rc.spec.template.spec.containers[idx].resources.requests.cpu == REQUEST_CPU[idx])
+          assert(rc.spec.template.spec.containers[idx].resources.requests.memory == REQUEST_MEMORY[idx])
+          assert(rc.spec.template.spec.containers[idx].resources.limits.cpu == LIMIT_CPU[idx])
+          assert(rc.spec.template.spec.containers[idx].resources.limits.memory == LIMIT_MEMORY[idx])
         }
       }) >> replicationControllerMock
-      5 * replicationControllerMock.getMetadata() >> metadataMock
-      3 * metadataMock.getName() >> replicationControllerName
   }
 }
diff --git a/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/CloneKubernetesAtomicOperationValidatorSpec.groovy b/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/CloneKubernetesAtomicOperationValidatorSpec.groovy
index 0ba791616d2..a69786f3b92 100644
--- a/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/CloneKubernetesAtomicOperationValidatorSpec.groovy
+++ b/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/CloneKubernetesAtomicOperationValidatorSpec.groovy
@@ -16,11 +16,15 @@
 
 package com.netflix.spinnaker.clouddriver.kubernetes.deploy.validators
 
+import com.netflix.spinnaker.clouddriver.docker.registry.security.DockerRegistryNamedAccountCredentials
+import com.netflix.spinnaker.clouddriver.kubernetes.api.KubernetesApiAdaptor
+import com.netflix.spinnaker.clouddriver.kubernetes.config.LinkedDockerRegistryConfiguration
 import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.CloneKubernetesAtomicOperationDescription
 import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.KubernetesContainerDescription
 import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.KubernetesResourceDescription
 import com.netflix.spinnaker.clouddriver.kubernetes.security.KubernetesCredentials
 import com.netflix.spinnaker.clouddriver.kubernetes.security.KubernetesNamedAccountCredentials
+import com.netflix.spinnaker.clouddriver.security.AccountCredentialsRepository
 import com.netflix.spinnaker.clouddriver.security.DefaultAccountCredentialsProvider
 import com.netflix.spinnaker.clouddriver.security.MapBackedAccountCredentialsRepository
 import org.springframework.validation.Errors
@@ -29,6 +33,10 @@ import spock.lang.Specification
 
 class CloneKubernetesAtomicOperationValidatorSpec extends Specification {
   private static final DESCRIPTION = "cloneKubernetesAtomicOperationDescription"
+  private static final List<String> NAMESPACES = ["default", "prod"]
+  private static final List<LinkedDockerRegistryConfiguration> DOCKER_REGISTRY_ACCOUNTS = [
+    new LinkedDockerRegistryConfiguration(accountName: "my-docker-account"),
+    new LinkedDockerRegistryConfiguration(accountName: "restricted-docker-account", namespaces: ["prod"])]
 
   private static final VALID_APPLICATION = "app"
   private static final VALID_STACK = "stack"
@@ -44,6 +52,7 @@ class CloneKubernetesAtomicOperationValidatorSpec extends Specification {
   private static final VALID_LOAD_BALANCERS = ["x", "y"]
   private static final VALID_SECURITY_GROUPS = ["a-1", "b-2"]
   private static final VALID_SOURCE_SERVER_GROUP_NAME = "myapp-test-v000"
+  private static final VALID_SECRET = DOCKER_REGISTRY_ACCOUNTS[0].accountName
 
   private static final INVALID_APPLICATION = "-app-"
   private static final INVALID_STACK = " stack"
@@ -65,8 +74,23 @@ class CloneKubernetesAtomicOperationValidatorSpec extends Specification {
     def credentialsRepo = new MapBackedAccountCredentialsRepository()
     def credentialsProvider = new DefaultAccountCredentialsProvider(credentialsRepo)
     def namedCredentialsMock = Mock(KubernetesNamedAccountCredentials)
+
+    def apiMock = Mock(KubernetesApiAdaptor)
+    def accountCredentialsRepositoryMock = Mock(AccountCredentialsRepository)
+
+    DOCKER_REGISTRY_ACCOUNTS.forEach({ account ->
+      def dockerRegistryAccountMock = Mock(DockerRegistryNamedAccountCredentials)
+      accountCredentialsRepositoryMock.getOne(account.accountName) >> dockerRegistryAccountMock
+      dockerRegistryAccountMock.getAccountName() >> account
+      NAMESPACES.forEach({ namespace ->
+        apiMock.getSecret(namespace, account.accountName) >> null
+        apiMock.createSecret(namespace, _) >> null
+      })
+    })
+
+    def credentials = new KubernetesCredentials(apiMock, NAMESPACES, DOCKER_REGISTRY_ACCOUNTS, accountCredentialsRepositoryMock)
     namedCredentialsMock.getName() >> VALID_CREDENTIALS
-    namedCredentialsMock.getCredentials() >> new KubernetesCredentials(null, null)
+    namedCredentialsMock.getCredentials() >> credentials
     credentialsRepo.save(VALID_CREDENTIALS, namedCredentialsMock)
     validator.accountCredentialsProvider = credentialsProvider
   }
@@ -108,6 +132,7 @@ class CloneKubernetesAtomicOperationValidatorSpec extends Specification {
         loadBalancers: VALID_LOAD_BALANCERS,
         securityGroups: VALID_SECURITY_GROUPS,
         credentials: VALID_CREDENTIALS,
+        imagePullSecrets: [VALID_SECRET],
         source: [
           serverGroupName: VALID_SOURCE_SERVER_GROUP_NAME
         ])
diff --git a/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/DeployKubernetesAtomicOperationValidatorSpec.groovy b/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/DeployKubernetesAtomicOperationValidatorSpec.groovy
index 4cba5d3d356..1fea1e80da5 100644
--- a/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/DeployKubernetesAtomicOperationValidatorSpec.groovy
+++ b/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/DeployKubernetesAtomicOperationValidatorSpec.groovy
@@ -17,11 +17,15 @@
 
 package com.netflix.spinnaker.clouddriver.kubernetes.deploy.validators
 
+import com.netflix.spinnaker.clouddriver.docker.registry.security.DockerRegistryNamedAccountCredentials
+import com.netflix.spinnaker.clouddriver.kubernetes.api.KubernetesApiAdaptor
+import com.netflix.spinnaker.clouddriver.kubernetes.config.LinkedDockerRegistryConfiguration
 import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.DeployKubernetesAtomicOperationDescription
 import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.KubernetesContainerDescription
 import com.netflix.spinnaker.clouddriver.kubernetes.deploy.description.KubernetesResourceDescription
 import com.netflix.spinnaker.clouddriver.kubernetes.security.KubernetesCredentials
 import com.netflix.spinnaker.clouddriver.kubernetes.security.KubernetesNamedAccountCredentials
+import com.netflix.spinnaker.clouddriver.security.AccountCredentialsRepository
 import com.netflix.spinnaker.clouddriver.security.DefaultAccountCredentialsProvider
 import com.netflix.spinnaker.clouddriver.security.MapBackedAccountCredentialsRepository
 import org.springframework.validation.Errors
@@ -30,6 +34,10 @@ import spock.lang.Specification
 
 class DeployKubernetesAtomicOperationValidatorSpec extends Specification {
   private static final DESCRIPTION = "deployKubernetesAtomicOperationDescription"
+  private static final List<String> NAMESPACES = ["default", "prod"]
+  private static final List<LinkedDockerRegistryConfiguration> DOCKER_REGISTRY_ACCOUNTS = [
+    new LinkedDockerRegistryConfiguration(accountName: "my-docker-account"),
+    new LinkedDockerRegistryConfiguration(accountName: "restricted-docker-account", namespaces: ["prod"])]
 
   private static final VALID_APPLICATION = "app"
   private static final VALID_STACK = "stack"
@@ -44,7 +52,8 @@ class DeployKubernetesAtomicOperationValidatorSpec extends Specification {
   private static final VALID_CREDENTIALS = "auto"
   private static final VALID_LOAD_BALANCERS = ["x", "y"]
   private static final VALID_SECURITY_GROUPS = ["a-1", "b-2"]
-  private static final VALID_NAMESPACE = "default"
+  private static final VALID_NAMESPACE = NAMESPACES[0]
+  private static final VALID_SECRET = DOCKER_REGISTRY_ACCOUNTS[0].accountName
 
   private static final INVALID_APPLICATION = "-app-"
   private static final INVALID_STACK = " stack"
@@ -67,8 +76,23 @@ class DeployKubernetesAtomicOperationValidatorSpec extends Specification {
     def credentialsRepo = new MapBackedAccountCredentialsRepository()
     def credentialsProvider = new DefaultAccountCredentialsProvider(credentialsRepo)
     def namedCredentialsMock = Mock(KubernetesNamedAccountCredentials)
+
+    def apiMock = Mock(KubernetesApiAdaptor)
+    def accountCredentialsRepositoryMock = Mock(AccountCredentialsRepository)
+
+    DOCKER_REGISTRY_ACCOUNTS.forEach({ account ->
+      def dockerRegistryAccountMock = Mock(DockerRegistryNamedAccountCredentials)
+      accountCredentialsRepositoryMock.getOne(account.accountName) >> dockerRegistryAccountMock
+      dockerRegistryAccountMock.getAccountName() >> account
+      NAMESPACES.forEach({ namespace ->
+        apiMock.getSecret(namespace, account.accountName) >> null
+        apiMock.createSecret(namespace, _) >> null
+      })
+    })
+
+    def credentials = new KubernetesCredentials(apiMock, NAMESPACES, DOCKER_REGISTRY_ACCOUNTS, accountCredentialsRepositoryMock)
     namedCredentialsMock.getName() >> VALID_CREDENTIALS
-    namedCredentialsMock.getCredentials() >> new KubernetesCredentials(null, null)
+    namedCredentialsMock.getCredentials() >> credentials
     credentialsRepo.save(VALID_CREDENTIALS, namedCredentialsMock)
     validator.accountCredentialsProvider = credentialsProvider
   }
@@ -104,6 +128,7 @@ class DeployKubernetesAtomicOperationValidatorSpec extends Specification {
         namespace: VALID_NAMESPACE,
         freeFormDetails: VALID_STACK,
         targetSize: VALID_TARGET_SIZE,
+        imagePullSecrets: [VALID_SECRET],
         containers: [
           fullValidContainerDescription1,
           fullValidContainerDescription2
@@ -238,7 +263,7 @@ class DeployKubernetesAtomicOperationValidatorSpec extends Specification {
     when:
       validator.validate([], description, errorsMock)
     then:
-      1 * errorsMock.rejectValue("${DESCRIPTION}.namespace", "${DESCRIPTION}.namespace.invalid (Must match ${StandardKubernetesAttributeValidator.namePattern})")
+      1 * errorsMock.rejectValue("${DESCRIPTION}.namespace", "${DESCRIPTION}.namespace.notRegistered")
       0 * errorsMock._
   }
 
diff --git a/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/StandardKubernetesAttributeValidatorSpec.groovy b/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/StandardKubernetesAttributeValidatorSpec.groovy
index cd5cc3cd730..204ee6b9185 100644
--- a/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/StandardKubernetesAttributeValidatorSpec.groovy
+++ b/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/deploy/validators/StandardKubernetesAttributeValidatorSpec.groovy
@@ -16,10 +16,15 @@
 
 package com.netflix.spinnaker.clouddriver.kubernetes.deploy.validators
 
+import com.netflix.spinnaker.clouddriver.docker.registry.security.DockerRegistryNamedAccountCredentials
+import com.netflix.spinnaker.clouddriver.kubernetes.api.KubernetesApiAdaptor
+import com.netflix.spinnaker.clouddriver.kubernetes.config.LinkedDockerRegistryConfiguration
 import com.netflix.spinnaker.clouddriver.kubernetes.security.KubernetesCredentials
 import com.netflix.spinnaker.clouddriver.kubernetes.security.KubernetesNamedAccountCredentials
+import com.netflix.spinnaker.clouddriver.security.AccountCredentialsRepository
 import com.netflix.spinnaker.clouddriver.security.DefaultAccountCredentialsProvider
 import com.netflix.spinnaker.clouddriver.security.MapBackedAccountCredentialsRepository
+import io.fabric8.kubernetes.api.model.Secret
 import org.springframework.validation.Errors
 import spock.lang.Shared
 import spock.lang.Specification
@@ -28,6 +33,13 @@ import spock.lang.Unroll
 class StandardKubernetesAttributeValidatorSpec extends Specification {
   private static final ACCOUNT_NAME = "auto"
   private static final DECORATOR = "decorator"
+  private static final List<String> NAMESPACES = ["default", "prod"]
+  private static final List<LinkedDockerRegistryConfiguration> DOCKER_REGISTRY_ACCOUNTS = [
+    new LinkedDockerRegistryConfiguration(accountName: "my-docker-account"),
+    new LinkedDockerRegistryConfiguration(accountName: "restricted-docker-account", namespaces: ["prod"])]
+
+  @Shared
+  KubernetesCredentials credentials
 
   @Shared
   DefaultAccountCredentialsProvider accountCredentialsProvider
@@ -37,7 +49,21 @@ class StandardKubernetesAttributeValidatorSpec extends Specification {
     accountCredentialsProvider = new DefaultAccountCredentialsProvider(credentialsRepo)
     def namedAccountCredentialsMock = Mock(KubernetesNamedAccountCredentials)
     namedAccountCredentialsMock.getName() >> ACCOUNT_NAME
-    namedAccountCredentialsMock.getCredentials() >> new KubernetesCredentials(null, null)
+    def apiMock = Mock(KubernetesApiAdaptor)
+    def accountCredentialsRepositoryMock = Mock(AccountCredentialsRepository)
+
+    DOCKER_REGISTRY_ACCOUNTS.forEach({ account ->
+      def dockerRegistryAccountMock = Mock(DockerRegistryNamedAccountCredentials)
+      accountCredentialsRepositoryMock.getOne(account.accountName) >> dockerRegistryAccountMock
+      dockerRegistryAccountMock.getAccountName() >> account
+      NAMESPACES.forEach({ namespace ->
+        apiMock.getSecret(namespace, account.accountName) >> null
+        apiMock.createSecret(namespace, _) >> null
+      })
+    })
+
+    credentials = new KubernetesCredentials(apiMock, NAMESPACES, DOCKER_REGISTRY_ACCOUNTS, accountCredentialsRepositoryMock)
+    namedAccountCredentialsMock.getCredentials() >> credentials
     credentialsRepo.save(ACCOUNT_NAME, namedAccountCredentialsMock)
   }
 
@@ -525,17 +551,17 @@ class StandardKubernetesAttributeValidatorSpec extends Specification {
       def label = "label"
 
     when:
-      validator.validateNamespace("", label)
+      validator.validateNamespace(credentials, "", label)
     then:
       0 * errorsMock._
 
     when:
-      validator.validateNamespace("default", label)
+      validator.validateNamespace(credentials, NAMESPACES[0], label)
     then:
       0 * errorsMock._
 
     when:
-      validator.validateNamespace("prod-staging", label)
+      validator.validateNamespace(credentials, NAMESPACES[1], label)
     then:
       0 * errorsMock._
   }
@@ -547,21 +573,63 @@ class StandardKubernetesAttributeValidatorSpec extends Specification {
       def label = "label"
 
     when:
-      validator.validateNamespace(" .-100z", label)
+      validator.validateNamespace(credentials, " .-100z", label)
     then:
-      1 * errorsMock.rejectValue("${DECORATOR}.${label}", "${DECORATOR}.${label}.invalid (Must match ${StandardKubernetesAttributeValidator.namePattern})")
+      1 * errorsMock.rejectValue("${DECORATOR}.${label}", "${DECORATOR}.${label}.notRegistered")
       0 * errorsMock._
 
     when:
-      validator.validateNamespace("?", label)
+      validator.validateNamespace(credentials, "?", label)
     then:
-      1 * errorsMock.rejectValue("${DECORATOR}.${label}", "${DECORATOR}.${label}.invalid (Must match ${StandardKubernetesAttributeValidator.namePattern})")
+      1 * errorsMock.rejectValue("${DECORATOR}.${label}", "${DECORATOR}.${label}.notRegistered")
       0 * errorsMock._
 
     when:
-      validator.validateNamespace("- ", label)
+      validator.validateNamespace(credentials, "- ", label)
     then:
-      1 * errorsMock.rejectValue("${DECORATOR}.${label}", "${DECORATOR}.${label}.invalid (Must match ${StandardKubernetesAttributeValidator.namePattern})")
+      1 * errorsMock.rejectValue("${DECORATOR}.${label}", "${DECORATOR}.${label}.notRegistered")
+      0 * errorsMock._
+  }
+
+  void "image pull secret accept"() {
+    setup:
+      def errorsMock = Mock(Errors)
+      def validator = new StandardKubernetesAttributeValidator(DECORATOR, errorsMock)
+      def label = "label"
+
+    when:
+      validator.validateImagePullSecret(credentials, DOCKER_REGISTRY_ACCOUNTS[0].accountName, NAMESPACES[0], label)
+    then:
+      0 * errorsMock._
+
+    when:
+      validator.validateImagePullSecret(credentials, DOCKER_REGISTRY_ACCOUNTS[1].accountName, NAMESPACES[1], label)
+    then:
+      0 * errorsMock._
+  }
+
+  void "image pull secret reject"() {
+    setup:
+    def errorsMock = Mock(Errors)
+    def validator = new StandardKubernetesAttributeValidator(DECORATOR, errorsMock)
+    def label = "label"
+
+    when:
+      validator.validateImagePullSecret(credentials, DOCKER_REGISTRY_ACCOUNTS[1].accountName, NAMESPACES[0], label)
+    then:
+      1 * errorsMock.rejectValue("${DECORATOR}.${label}", "${DECORATOR}.${label}.notRegistered")
+      0 * errorsMock._
+
+    when:
+      validator.validateImagePullSecret(credentials, "?", NAMESPACES[0], label)
+    then:
+      1 * errorsMock.rejectValue("${DECORATOR}.${label}", "${DECORATOR}.${label}.notRegistered")
+      0 * errorsMock._
+
+    when:
+      validator.validateImagePullSecret(credentials, DOCKER_REGISTRY_ACCOUNTS[0].accountName, "not a namespace", label)
+    then:
+      1 * errorsMock.rejectValue("${DECORATOR}.${label}", "${DECORATOR}.${label}.notRegistered")
       0 * errorsMock._
   }
 }
diff --git a/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/provider/agent/KubernetesServerGroupCachingAgentSpec.groovy b/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/provider/agent/KubernetesServerGroupCachingAgentSpec.groovy
index b8eb828cc77..db8e0c74cac 100644
--- a/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/provider/agent/KubernetesServerGroupCachingAgentSpec.groovy
+++ b/clouddriver-kubernetes/src/test/groovy/com/netflix/spinnaker/clouddriver/kubernetes/provider/agent/KubernetesServerGroupCachingAgentSpec.groovy
@@ -20,8 +20,12 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import com.netflix.spectator.api.Registry
 import com.netflix.spinnaker.cats.agent.CacheResult
 import com.netflix.spinnaker.clouddriver.kubernetes.KubernetesCloudProvider
+import com.netflix.spinnaker.clouddriver.kubernetes.api.KubernetesApiAdaptor
 import com.netflix.spinnaker.clouddriver.kubernetes.cache.Keys
 import com.netflix.spinnaker.clouddriver.kubernetes.deploy.KubernetesUtil
+import com.netflix.spinnaker.clouddriver.kubernetes.security.KubernetesCredentials
+import com.netflix.spinnaker.clouddriver.kubernetes.security.KubernetesCredentialsInitializer
+import com.netflix.spinnaker.clouddriver.security.AccountCredentialsRepository
 import io.fabric8.kubernetes.api.model.ObjectMeta
 import io.fabric8.kubernetes.api.model.PodList
 import io.fabric8.kubernetes.api.model.ReplicationController
@@ -39,8 +43,9 @@ class KubernetesServerGroupCachingAgentSpec extends Specification {
   KubernetesServerGroupCachingAgent cachingAgent
   ReplicationControllerList replicationControllerList
   PodList podList
-  KubernetesUtil utilMock
+  KubernetesApiAdaptor apiMock
   Registry registryMock
+  KubernetesCredentials kubernetesCredentials
 
   String applicationKey
   String clusterKey
@@ -55,16 +60,18 @@ class KubernetesServerGroupCachingAgentSpec extends Specification {
     replicationControllerList = Mock(ReplicationControllerList)
     podList = Mock(PodList)
 
-    utilMock = Mock(KubernetesUtil)
-    utilMock.getReplicationControllers(_, NAMESPACE) >> replicationControllerList
-    utilMock.getPods(_, NAMESPACE, _) >> podList
+    apiMock = Mock(KubernetesApiAdaptor)
+
+    def accountCredentialsRepositoryMock = Mock(AccountCredentialsRepository)
+
+    kubernetesCredentials = new KubernetesCredentials(apiMock, [], [], accountCredentialsRepositoryMock)
 
     applicationKey = Keys.getApplicationKey(APP)
     clusterKey = Keys.getClusterKey(ACCOUNT_NAME, APP, CLUSTER)
     serverGroupKey = Keys.getServerGroupKey(ACCOUNT_NAME, NAMESPACE, REPLICATION_CONTROLLER)
     instanceKey = Keys.getInstanceKey(ACCOUNT_NAME, NAMESPACE, REPLICATION_CONTROLLER, POD)
 
-    cachingAgent = new KubernetesServerGroupCachingAgent(new KubernetesCloudProvider(), ACCOUNT_NAME, null, NAMESPACE, new ObjectMapper(), registryMock, utilMock)
+    cachingAgent = new KubernetesServerGroupCachingAgent(new KubernetesCloudProvider(), ACCOUNT_NAME, kubernetesCredentials, NAMESPACE, new ObjectMapper(), registryMock)
   }
 
   void "Should store a single replication controller object and relationships"() {
@@ -79,9 +86,10 @@ class KubernetesServerGroupCachingAgentSpec extends Specification {
       podMetadataMock.getName() >> POD
       podMock.getMetadata() >> podMetadataMock
 
+      apiMock.getReplicationControllers(NAMESPACE) >> [replicationControllerMock]
+      apiMock.getPods(NAMESPACE, _) >> [podMock]
+
     when:
-      replicationControllerList.getItems() >> [replicationControllerMock]
-      podList.getItems() >> [podMock]
       def result = cachingAgent.loadData(null)
 
     then: