YARN-5428. Allow for specifying the docker client configuration directory. Contributed by Shane Kumpf

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/Client.java

@@ -87,6 +87,7 @@
 import org.apache.hadoop.yarn.exceptions.ResourceNotFoundException;
 import org.apache.hadoop.yarn.exceptions.YARNFeatureNotEnabledException;
 import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.util.DockerClientConfigHandler;
 import org.apache.hadoop.yarn.util.UnitsConversionUtil;
 import org.apache.hadoop.yarn.util.resource.ResourceUtils;
 import org.apache.hadoop.yarn.util.resource.Resources;
@@ -225,6 +226,9 @@ public class Client {
   private String flowVersion = null;
   private long flowRunId = 0L;
 
+  // Docker client configuration
+  private String dockerClientConfig = null;
+
   // Command line options
   private Options opts;
 
@@ -368,6 +372,10 @@ public Client(Configuration conf) throws Exception  {
         "If container could retry, it specifies max retires");
     opts.addOption("container_retry_interval", true,
         "Interval between each retry, unit is milliseconds");
+    opts.addOption("docker_client_config", true,
+        "The docker client configuration path. The scheme should be supplied"
+            + " (i.e. file:// or hdfs://)."
+            + " Only used when the Docker runtime is enabled and requested.");
     opts.addOption("placement_spec", true,
         "Placement specification. Please note, if this option is specified,"
             + " The \"num_containers\" option will be ignored. All requested"
@@ -585,6 +593,9 @@ public boolean init(String[] args) throws ParseException {
             "Flow run is not a valid long value", e);
       }
     }
+    if (cliParser.hasOption("docker_client_config")) {
+      dockerClientConfig = cliParser.getOptionValue("docker_client_config");
+    }
     return true;
   }
 
@@ -884,9 +895,10 @@ public boolean run() throws IOException, YarnException {
     // amContainer.setServiceData(serviceData);
 
     // Setup security tokens
+    Credentials rmCredentials = null;
     if (UserGroupInformation.isSecurityEnabled()) {
       // Note: Credentials class is marked as LimitedPrivate for HDFS and MapReduce
-      Credentials credentials = new Credentials();
+      rmCredentials = new Credentials();
       String tokenRenewer = YarnClientUtils.getRmPrincipal(conf);
       if (tokenRenewer == null || tokenRenewer.length() == 0) {
         throw new IOException(
@@ -895,16 +907,32 @@ public boolean run() throws IOException, YarnException {
 
       // For now, only getting tokens for the default file-system.
       final Token<?> tokens[] =
-          fs.addDelegationTokens(tokenRenewer, credentials);
+          fs.addDelegationTokens(tokenRenewer, rmCredentials);
       if (tokens != null) {
         for (Token<?> token : tokens) {
           LOG.info("Got dt for " + fs.getUri() + "; " + token);
         }
       }
+    }
+
+    // Add the docker client config credentials if supplied.
+    Credentials dockerCredentials = null;
+    if (dockerClientConfig != null) {
+      dockerCredentials =
+          DockerClientConfigHandler.readCredentialsFromConfigFile(
+              new Path(dockerClientConfig), conf, appId.toString());
+    }
+
+    if (rmCredentials != null || dockerCredentials != null) {
       DataOutputBuffer dob = new DataOutputBuffer();
-      credentials.writeTokenStorageToStream(dob);
-      ByteBuffer fsTokens = ByteBuffer.wrap(dob.getData(), 0, dob.getLength());
-      amContainer.setTokens(fsTokens);
+      if (rmCredentials != null) {
+        rmCredentials.writeTokenStorageToStream(dob);
+      }
+      if (dockerCredentials != null) {
+        dockerCredentials.writeTokenStorageToStream(dob);
+      }
+      ByteBuffer tokens = ByteBuffer.wrap(dob.getData(), 0, dob.getLength());
+      amContainer.setTokens(tokens);
     }
 
     appContext.setAMContainerSpec(amContainer);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/security/DockerCredentialTokenIdentifier.java

@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.hadoop.yarn.security;
+
+import com.google.protobuf.TextFormat;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.TokenIdentifier;
+import org.apache.hadoop.yarn.proto.YarnSecurityTokenProtos.DockerCredentialTokenIdentifierProto;
+import org.slf4j.LoggerFactory;
+
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.DataOutput;
+import java.io.IOException;
+
+/**
+ * TokenIdentifier for Docker registry credentials.
+ */
+public class DockerCredentialTokenIdentifier extends TokenIdentifier {
+
+  private static final org.slf4j.Logger LOG =
+      LoggerFactory.getLogger(DockerCredentialTokenIdentifier.class);
+
+  private DockerCredentialTokenIdentifierProto proto;
+  public static final Text KIND = new Text("DOCKER_CLIENT_CREDENTIAL_TOKEN");
+
+  public DockerCredentialTokenIdentifier(String registryUrl,
+      String applicationId) {
+    DockerCredentialTokenIdentifierProto.Builder builder =
+        DockerCredentialTokenIdentifierProto.newBuilder();
+    if (registryUrl != null) {
+      builder.setRegistryUrl(registryUrl);
+    }
+    if (applicationId != null) {
+      builder.setApplicationId(applicationId);
+    }
+    proto = builder.build();
+  }
+
+  /**
+   * Default constructor needed for the Service Loader.
+   */
+  public DockerCredentialTokenIdentifier() {
+  }
+
+  /**
+   * Write the TokenIdentifier to the output stream.
+   *
+   * @param out <code>DataOutput</code> to serialize this object into.
+   * @throws IOException if the write fails.
+   */
+  @Override
+  public void write(DataOutput out) throws IOException {
+    out.write(proto.toByteArray());
+  }
+
+  /**
+   * Populate the Proto object with the input.
+   *
+   * @param in <code>DataInput</code> to deserialize this object from.
+   * @throws IOException if the read fails.
+   */
+  @Override
+  public void readFields(DataInput in) throws IOException {
+    proto = DockerCredentialTokenIdentifierProto.parseFrom((DataInputStream)in);
+  }
+
+  /**
+   * Return the ProtoBuf formatted data.
+   *
+   * @return the ProtoBuf representation of the data.
+   */
+  public DockerCredentialTokenIdentifierProto getProto() {
+    return proto;
+  }
+
+  /**
+   * Return the TokenIdentifier kind.
+   *
+   * @return the TokenIdentifier kind.
+   */
+  @Override
+  public Text getKind() {
+    return KIND;
+  }
+
+  /**
+   * Return a remote user based on the registry URL and Application ID.
+   *
+   * @return a remote user based on the registry URL and Application ID.
+   */
+  @Override
+  public UserGroupInformation getUser() {
+    return UserGroupInformation.createRemoteUser(
+        getRegistryUrl() + "-" + getApplicationId());
+  }
+
+  /**
+   * Get the registry URL.
+   *
+   * @return the registry URL.
+   */
+  public String getRegistryUrl() {
+    String registryUrl = null;
+    if (proto.hasRegistryUrl()) {
+      registryUrl = proto.getRegistryUrl();
+    }
+    return registryUrl;
+  }
+
+  /**
+   * Get the application ID.
+   *
+   * @return the application ID.
+   */
+  public String getApplicationId() {
+    String applicationId = null;
+    if (proto.hasApplicationId()) {
+      applicationId = proto.getApplicationId();
+    }
+    return applicationId;
+  }
+
+  @Override
+  public int hashCode() {
+    return getProto().hashCode();
+  }
+
+  @Override
+  public boolean equals(Object other) {
+    if (other == null) {
+      return false;
+    }
+    if (other.getClass().isAssignableFrom(this.getClass())) {
+      return this.getProto().equals(this.getClass().cast(other).getProto());
+    }
+    return false;
+  }
+
+  @Override
+  public String toString() {
+    return TextFormat.shortDebugString(getProto());
+  }
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/DockerClientConfigHandler.java

@@ -0,0 +1,183 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.hadoop.yarn.util;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.io.DataInputByteBuffer;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenIdentifier;
+import org.apache.hadoop.yarn.security.DockerCredentialTokenIdentifier;
+import org.codehaus.jackson.JsonFactory;
+import org.codehaus.jackson.JsonNode;
+import org.codehaus.jackson.JsonParser;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.node.ObjectNode;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.Iterator;
+
+/**
+ * Commonly needed actions for handling the Docker client configurations.
+ *
+ * Credentials that are used to access private Docker registries are supplied.
+ * Actions include:
+ * <ul>
+ *   <li>Read the Docker client configuration json file from a
+ *   {@link FileSystem}.</li>
+ *   <li>Extract the authentication information from the configuration into
+ *   {@link Token} and {@link Credentials} objects.</li>
+ *   <li>Tokens are commonly shipped via the
+ *   {@link org.apache.hadoop.yarn.api.records.ContainerLaunchContext} as a
+ *   {@link ByteBuffer}, extract the {@link Credentials}.</li>
+ *   <li>Write the Docker client configuration json back to the local filesystem
+ *   to be used by the Docker command line.</li>
+ * </ul>
+ */
+public final class DockerClientConfigHandler {
+  private static final org.slf4j.Logger LOG =
+      LoggerFactory.getLogger(DockerClientConfigHandler.class);
+
+  private static final String CONFIG_AUTHS_KEY = "auths";
+  private static final String CONFIG_AUTH_KEY = "auth";
+
+  private DockerClientConfigHandler() { }
+
+  /**
+   * Read the Docker client configuration and extract the auth tokens into
+   * Credentials.
+   *
+   * @param configFile the Path to the Docker client configuration.
+   * @param conf the Configuration object, needed by the FileSystem.
+   * @param applicationId the application ID to associate the Credentials with.
+   * @return the populated Credential object with the Docker Tokens.
+   * @throws IOException if the file can not be read.
+   */
+  public static Credentials readCredentialsFromConfigFile(Path configFile,
+      Configuration conf, String applicationId) throws IOException {
+    // Read the config file
+    String contents = null;
+    configFile = new Path(configFile.toUri());
+    FileSystem fs = configFile.getFileSystem(conf);
+    if (fs != null) {
+      FSDataInputStream fileHandle = fs.open(configFile);
+      if (fileHandle != null) {
+        contents = IOUtils.toString(fileHandle);
+      }
+    }
+    if (contents == null) {
+      throw new IOException("Failed to read Docker client configuration: "
+          + configFile);
+    }
+
+    // Parse the JSON and create the Tokens/Credentials.
+    ObjectMapper mapper = new ObjectMapper();
+    JsonFactory factory = mapper.getJsonFactory();
+    JsonParser parser = factory.createJsonParser(contents);
+    JsonNode rootNode = mapper.readTree(parser);
+
+    Credentials credentials = new Credentials();
+    if (rootNode.has(CONFIG_AUTHS_KEY)) {
+      Iterator<String> iter = rootNode.get(CONFIG_AUTHS_KEY).getFieldNames();
+      for (; iter.hasNext();) {
+        String registryUrl = iter.next();
+        String registryCred = rootNode.get(CONFIG_AUTHS_KEY)
+            .get(registryUrl)
+            .get(CONFIG_AUTH_KEY)
+            .asText();
+        TokenIdentifier tokenId =
+            new DockerCredentialTokenIdentifier(registryUrl, applicationId);
+        Token<DockerCredentialTokenIdentifier> token =
+            new Token<>(tokenId.getBytes(),
+                registryCred.getBytes(Charset.forName("UTF-8")),
+                tokenId.getKind(), new Text(registryUrl));
+        credentials.addToken(
+            new Text(registryUrl + "-" + applicationId), token);
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Added token: " + token.toString());
+        }
+      }
+    }
+    return credentials;
+  }
+
+  /**
+   * Convert the Token ByteBuffer to the appropriate Credentials object.
+   *
+   * @param tokens the Tokens from the ContainerLaunchContext.
+   * @return the Credentials object populated from the Tokens.
+   */
+  public static Credentials getCredentialsFromTokensByteBuffer(
+      ByteBuffer tokens) throws IOException {
+    Credentials credentials = new Credentials();
+    DataInputByteBuffer dibb = new DataInputByteBuffer();
+    tokens.rewind();
+    dibb.reset(tokens);
+    credentials.readTokenStorageStream(dibb);
+    tokens.rewind();
+    if (LOG.isDebugEnabled()) {
+      for (Token token : credentials.getAllTokens()) {
+        LOG.debug("Added token: " + token.toString());
+      }
+    }
+    return credentials;
+  }
+
+  /**
+   * Extract the Docker related tokens from the Credentials and write the Docker
+   * client configuration to the supplied File.
+   *
+   * @param outConfigFile the File to write the Docker client configuration to.
+   * @param credentials the populated Credentials object.
+   * @throws IOException if the write fails.
+   */
+  public static void writeDockerCredentialsToPath(File outConfigFile,
+      Credentials credentials) throws IOException {
+    ObjectMapper mapper = new ObjectMapper();
+    ObjectNode rootNode = mapper.createObjectNode();
+    ObjectNode registryUrlNode = mapper.createObjectNode();
+    if (credentials.numberOfTokens() > 0) {
+      for (Token<? extends TokenIdentifier> tk : credentials.getAllTokens()) {
+        if (tk.getKind().equals(DockerCredentialTokenIdentifier.KIND)) {
+          DockerCredentialTokenIdentifier ti =
+              (DockerCredentialTokenIdentifier) tk.decodeIdentifier();
+          ObjectNode registryCredNode = mapper.createObjectNode();
+          registryUrlNode.put(ti.getRegistryUrl(), registryCredNode);
+          registryCredNode.put(CONFIG_AUTH_KEY,
+              new String(tk.getPassword(), Charset.forName("UTF-8")));
+          if (LOG.isDebugEnabled()) {
+            LOG.debug("Prepared token for write: " + tk.toString());
+          }
+        }
+      }
+    }
+    rootNode.put(CONFIG_AUTHS_KEY, registryUrlNode);
+    String json =
+        mapper.writerWithDefaultPrettyPrinter().writeValueAsString(rootNode);
+    FileUtils.writeStringToFile(outConfigFile, json, Charset.defaultCharset());
+  }
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/proto/yarn_security_token.proto

@@ -72,3 +72,8 @@ message YARNDelegationTokenIdentifierProto {
   optional int32 masterKeyId = 7;
 }
 
+message DockerCredentialTokenIdentifierProto {
+  optional string registryUrl = 1;
+  optional string applicationId = 2;
+}
+

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/META-INF/services/org.apache.hadoop.security.token.TokenIdentifier

@@ -17,3 +17,4 @@ org.apache.hadoop.yarn.security.client.ClientToAMTokenIdentifier
 org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier
 org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier
 org.apache.hadoop.yarn.security.NMTokenIdentifier
+org.apache.hadoop.yarn.security.DockerCredentialTokenIdentifier

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/security/TestDockerClientConfigHandler.java

@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.hadoop.yarn.security;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenIdentifier;
+import org.apache.hadoop.yarn.util.DockerClientConfigHandler;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.nio.ByteBuffer;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Test the functionality of the DockerClientConfigHandler.
+ */
+public class TestDockerClientConfigHandler {
+
+  public static final String JSON = "{\"auths\": "
+      + "{\"https://index.docker.io/v1/\": "
+      + "{\"auth\": \"foobarbaz\"},"
+      + "\"registry.example.com\": "
+      + "{\"auth\": \"bazbarfoo\"}}}";
+  private static final String APPLICATION_ID = "application_2313_2131341";
+
+  private File file;
+  private Configuration conf = new Configuration();
+
+  @Before
+  public void setUp() throws Exception {
+    file = File.createTempFile("docker-client-config", "test");
+    file.deleteOnExit();
+    BufferedWriter bw = new BufferedWriter(new FileWriter(file));
+    bw.write(JSON);
+    bw.close();
+  }
+
+  @Test
+  public void testReadCredentialsFromConfigFile() throws Exception {
+    Credentials credentials =
+        DockerClientConfigHandler.readCredentialsFromConfigFile(
+            new Path(file.toURI()), conf, APPLICATION_ID);
+    Token token1 = credentials.getToken(
+        new Text("https://index.docker.io/v1/-" + APPLICATION_ID));
+    assertEquals(DockerCredentialTokenIdentifier.KIND, token1.getKind());
+    assertEquals("foobarbaz", new String(token1.getPassword()));
+    DockerCredentialTokenIdentifier ti1 =
+        (DockerCredentialTokenIdentifier) token1.decodeIdentifier();
+    assertEquals("https://index.docker.io/v1/", ti1.getRegistryUrl());
+    assertEquals(APPLICATION_ID, ti1.getApplicationId());
+
+    Token token2 = credentials.getToken(
+        new Text("registry.example.com-" + APPLICATION_ID));
+    assertEquals(DockerCredentialTokenIdentifier.KIND, token2.getKind());
+    assertEquals("bazbarfoo", new String(token2.getPassword()));
+    DockerCredentialTokenIdentifier ti2 =
+        (DockerCredentialTokenIdentifier) token2.decodeIdentifier();
+    assertEquals("registry.example.com", ti2.getRegistryUrl());
+    assertEquals(APPLICATION_ID, ti2.getApplicationId());
+  }
+
+  @Test
+  public void testGetCredentialsFromTokensByteBuffer() throws Exception {
+    Credentials credentials =
+        DockerClientConfigHandler.readCredentialsFromConfigFile(
+            new Path(file.toURI()), conf, APPLICATION_ID);
+    DataOutputBuffer dob = new DataOutputBuffer();
+    credentials.writeTokenStorageToStream(dob);
+    ByteBuffer tokens = ByteBuffer.wrap(dob.getData(), 0, dob.getLength());
+    Credentials credentialsOut =
+        DockerClientConfigHandler.getCredentialsFromTokensByteBuffer(tokens);
+    assertEquals(credentials.numberOfTokens(), credentialsOut.numberOfTokens());
+    for (Token<? extends TokenIdentifier> tkIn : credentials.getAllTokens()) {
+      DockerCredentialTokenIdentifier ti =
+          (DockerCredentialTokenIdentifier) tkIn.decodeIdentifier();
+      Token tkOut = credentialsOut.getToken(
+          new Text(ti.getRegistryUrl() + "-" + ti.getApplicationId()));
+      assertEquals(tkIn.getKind(), tkOut.getKind());
+      assertEquals(new String(tkIn.getIdentifier()),
+          new String(tkOut.getIdentifier()));
+      assertEquals(new String(tkIn.getPassword()),
+          new String(tkOut.getPassword()));
+      assertEquals(tkIn.getService(), tkOut.getService());
+    }
+  }
+
+  @Test
+  public void testWriteDockerCredentialsToPath() throws Exception {
+    File outFile = File.createTempFile("docker-client-config", "out");
+    outFile.deleteOnExit();
+    Credentials credentials =
+        DockerClientConfigHandler.readCredentialsFromConfigFile(
+            new Path(file.toURI()), conf, APPLICATION_ID);
+    DockerClientConfigHandler.writeDockerCredentialsToPath(outFile,
+        credentials);
+    assertTrue(outFile.exists());
+    String fileContents = FileUtils.readFileToString(outFile);
+    assertTrue(fileContents.contains("auths"));
+    assertTrue(fileContents.contains("registry.example.com"));
+    assertTrue(fileContents.contains("https://index.docker.io/v1/"));
+    assertTrue(fileContents.contains("foobarbaz"));
+    assertTrue(fileContents.contains("bazbarfoo"));
+  }
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/DockerLinuxContainerRuntime.java

@@ -21,6 +21,7 @@
 package org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime;
 
 import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.yarn.server.nodemanager.Context;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.docker.DockerCommandExecutor;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.docker.DockerKillCommand;
@@ -28,6 +29,7 @@
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.docker.DockerVolumeCommand;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.resourceplugin.DockerCommandPlugin;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.resourceplugin.ResourcePlugin;
+import org.apache.hadoop.yarn.util.DockerClientConfigHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -58,8 +60,11 @@
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.runtime.ContainerRuntimeConstants;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.runtime.ContainerRuntimeContext;
 
+import java.io.File;
+import java.io.IOException;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.ArrayList;
@@ -846,6 +851,8 @@ public void launchContainer(ContainerRuntimeContext ctx)
       runCommand.setPrivileged();
     }
 
+    addDockerClientConfigToRunCommand(ctx, runCommand);
+
     String resourcesOpts = ctx.getExecutionAttribute(RESOURCES_OPTIONS);
 
     addCGroupParentIfRequired(resourcesOpts, containerIdStr, runCommand);
@@ -1181,4 +1188,36 @@ private void handleContainerRemove(String containerId,
       }
     }
   }
+
+  private void addDockerClientConfigToRunCommand(ContainerRuntimeContext ctx,
+      DockerRunCommand dockerRunCommand) throws ContainerExecutionException {
+    ByteBuffer tokens = ctx.getContainer().getLaunchContext().getTokens();
+    Credentials credentials;
+    if (tokens != null) {
+      tokens.rewind();
+      if (tokens.hasRemaining()) {
+        try {
+          credentials = DockerClientConfigHandler
+              .getCredentialsFromTokensByteBuffer(tokens);
+        } catch (IOException e) {
+          throw new ContainerExecutionException("Unable to read tokens.");
+        }
+        if (credentials.numberOfTokens() > 0) {
+          Path nmPrivateDir =
+              ctx.getExecutionAttribute(NM_PRIVATE_CONTAINER_SCRIPT_PATH)
+                  .getParent();
+          File dockerConfigPath = new File(nmPrivateDir + "/config.json");
+          try {
+            DockerClientConfigHandler
+                .writeDockerCredentialsToPath(dockerConfigPath, credentials);
+          } catch (IOException e) {
+            throw new ContainerExecutionException(
+                "Unable to write Docker client credentials to "
+                    + dockerConfigPath);
+          }
+          dockerRunCommand.setClientConfigDir(dockerConfigPath.getParent());
+        }
+      }
+    }
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/DockerCommand.java

@@ -88,4 +88,20 @@ public String toString() {
     }
     return ret.toString();
   }
+
+  /**
+   * Add the client configuration directory to the docker command.
+   *
+   * The client configuration option proceeds any of the docker subcommands
+   * (such as run, load, pull, etc). Ordering will be handled by
+   * container-executor. Docker expects the value to be a directory containing
+   * the file config.json. This file is typically generated via docker login.
+   *
+   * @param clientConfigDir - directory containing the docker client config.
+   */
+  public void setClientConfigDir(String clientConfigDir) {
+    if (clientConfigDir != null) {
+      addCommandArguments("docker-config", clientConfigDir);
+    }
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/TestDockerContainerRuntime.java

@@ -24,11 +24,16 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.registry.client.binding.RegistryPathUtils;
+import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.util.Shell;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
+import org.apache.hadoop.yarn.util.DockerClientConfigHandler;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.security.TestDockerClientConfigHandler;
 import org.apache.hadoop.yarn.server.nodemanager.ContainerExecutor;
 import org.apache.hadoop.yarn.server.nodemanager.Context;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
@@ -56,12 +61,18 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.BufferedWriter;
 import java.io.File;
 import java.io.FileInputStream;
+import java.io.FileWriter;
 import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.nio.file.Files;
 import java.nio.file.Paths;
+import java.nio.file.attribute.FileAttribute;
+import java.nio.file.attribute.PosixFilePermission;
+import java.nio.file.attribute.PosixFilePermissions;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -69,6 +80,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
+import java.util.Set;
 
 import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.LinuxContainerRuntimeConstants.APPID;
 import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.LinuxContainerRuntimeConstants.APPLICATION_LOCAL_DIRS;
@@ -1700,6 +1712,103 @@ public void testDockerCapabilities() throws ContainerExecutionException {
     Assert.assertEquals("DAC_OVERRIDE", it.next());
   }
 
+  @Test
+  public void testLaunchContainerWithDockerTokens()
+      throws ContainerExecutionException, PrivilegedOperationException,
+      IOException {
+    // Write the JSOn to a temp file.
+    File file = File.createTempFile("docker-client-config", "runtime-test");
+    file.deleteOnExit();
+    BufferedWriter bw = new BufferedWriter(new FileWriter(file));
+    bw.write(TestDockerClientConfigHandler.JSON);
+    bw.close();
+
+    // Get the credentials object with the Tokens.
+    Credentials credentials = DockerClientConfigHandler
+        .readCredentialsFromConfigFile(new Path(file.toURI()), conf, appId);
+    DataOutputBuffer dob = new DataOutputBuffer();
+    credentials.writeTokenStorageToStream(dob);
+    ByteBuffer tokens = ByteBuffer.wrap(dob.getData(), 0, dob.getLength());
+
+    // Configure the runtime and launch the container
+    when(context.getTokens()).thenReturn(tokens);
+    DockerLinuxContainerRuntime runtime =
+        new DockerLinuxContainerRuntime(mockExecutor, mockCGroupsHandler);
+    runtime.initialize(conf, null);
+
+    Set<PosixFilePermission> perms =
+        PosixFilePermissions.fromString("rwxr-xr--");
+    FileAttribute<Set<PosixFilePermission>> attr =
+        PosixFilePermissions.asFileAttribute(perms);
+    Path outDir = new Path(
+        Files.createTempDirectory("docker-client-config-out", attr).toUri()
+            .getPath() + "/launch_container.sh");
+    builder.setExecutionAttribute(NM_PRIVATE_CONTAINER_SCRIPT_PATH, outDir);
+    runtime.launchContainer(builder.build());
+    PrivilegedOperation op = capturePrivilegedOperation();
+    Assert.assertEquals(
+        PrivilegedOperation.OperationType.LAUNCH_DOCKER_CONTAINER,
+            op.getOperationType());
+
+    List<String> args = op.getArguments();
+
+    int expectedArgs = 13;
+    int argsCounter = 0;
+    Assert.assertEquals(expectedArgs, args.size());
+    Assert.assertEquals(runAsUser, args.get(argsCounter++));
+    Assert.assertEquals(user, args.get(argsCounter++));
+    Assert.assertEquals(Integer.toString(
+        PrivilegedOperation.RunAsUserCommand.LAUNCH_DOCKER_CONTAINER
+            .getValue()), args.get(argsCounter++));
+    Assert.assertEquals(appId, args.get(argsCounter++));
+    Assert.assertEquals(containerId, args.get(argsCounter++));
+    Assert.assertEquals(containerWorkDir.toString(), args.get(argsCounter++));
+    Assert.assertEquals(outDir.toUri().getPath(), args.get(argsCounter++));
+    Assert.assertEquals(nmPrivateTokensPath.toUri().getPath(),
+        args.get(argsCounter++));
+    Assert.assertEquals(pidFilePath.toString(), args.get(argsCounter++));
+    Assert.assertEquals(localDirs.get(0), args.get(argsCounter++));
+    Assert.assertEquals(logDirs.get(0), args.get(argsCounter++));
+    String dockerCommandFile = args.get(argsCounter++);
+    Assert.assertEquals(resourcesOptions, args.get(argsCounter));
+
+    List<String> dockerCommands = Files
+        .readAllLines(Paths.get(dockerCommandFile), Charset.forName("UTF-8"));
+
+    int expected = 15;
+    int counter = 0;
+    Assert.assertEquals(expected, dockerCommands.size());
+    Assert.assertEquals("[docker-command-execution]",
+        dockerCommands.get(counter++));
+    Assert.assertEquals("  cap-add=SYS_CHROOT,NET_BIND_SERVICE",
+        dockerCommands.get(counter++));
+    Assert.assertEquals("  cap-drop=ALL", dockerCommands.get(counter++));
+    Assert.assertEquals("  detach=true", dockerCommands.get(counter++));
+    Assert.assertEquals("  docker-command=run", dockerCommands.get(counter++));
+    Assert.assertEquals("  docker-config=" + outDir.getParent(),
+        dockerCommands.get(counter++));
+    Assert.assertEquals("  group-add=" + String.join(",", groups),
+        dockerCommands.get(counter++));
+    Assert.assertEquals("  hostname=ctr-id", dockerCommands.get(counter++));
+    Assert.assertEquals("  image=busybox:latest",
+        dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  launch-command=bash,/test_container_work_dir/launch_container.sh",
+        dockerCommands.get(counter++));
+    Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
+    Assert.assertEquals("  net=host", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  rw-mounts=/test_container_local_dir:/test_container_local_dir,"
+            + "/test_filecache_dir:/test_filecache_dir,"
+            + "/test_container_work_dir:/test_container_work_dir,"
+            + "/test_container_log_dir:/test_container_log_dir,"
+            + "/test_user_local_dir:/test_user_local_dir",
+        dockerCommands.get(counter++));
+    Assert.assertEquals("  user=" + uidGidPair, dockerCommands.get(counter++));
+    Assert.assertEquals("  workdir=/test_container_work_dir",
+        dockerCommands.get(counter++));
+  }
+
   class MockRuntime extends DockerLinuxContainerRuntime {
 
     private PrivilegedOperationExecutor privilegedOperationExecutor;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/TestDockerRunCommand.java

@@ -36,6 +36,7 @@ public class TestDockerRunCommand {
   private static final String CONTAINER_NAME = "foo";
   private static final String USER_ID = "user_id";
   private static final String IMAGE_NAME = "image_name";
+  private static final String CLIENT_CONFIG_PATH = "/path/to/client.json";
 
   @Before
   public void setUp() throws Exception {
@@ -77,4 +78,11 @@ public void testCommandArguments() {
             .get("launch-command")));
     assertEquals(7, dockerRunCommand.getDockerCommandWithArguments().size());
   }
+
+  @Test
+  public void testSetClientConfigDir() {
+    dockerRunCommand.setClientConfigDir(CLIENT_CONFIG_PATH);
+    assertEquals(CLIENT_CONFIG_PATH, StringUtils.join(",",
+        dockerRunCommand.getDockerCommandWithArguments().get("docker-config")));
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/DockerContainers.md

@@ -380,11 +380,14 @@ For [YARN Service HTTPD example](./yarn-service/Examples.html), container-execut
 Connecting to a Secure Docker Repository
 ----------------------------------------
 
-Until YARN-5428 is complete, the Docker client command will draw its
-configuration from the default location, which is $HOME/.docker/config.json on
-the NodeManager host. The Docker configuration is where secure repository
-credentials are stored, so use of the LCE with secure Docker repos is
-discouraged until YARN-5428 is complete.
+The Docker client command will draw its configuration from the default location,
+which is $HOME/.docker/config.json on the NodeManager host. The Docker
+configuration is where secure repository credentials are stored, so use of the
+LCE with secure Docker repos is discouraged using this method.
+
+YARN-5428 added support to Distributed Shell for securely supplying the Docker
+client configuration. See the Distributed Shell help for usage. Support for
+additional frameworks is planned.
 
 As a work-around, you may manually log the Docker daemon on every NodeManager
 host into the secure repo using the Docker login command: