SUBMARINE-54. Add test coverage for YarnServiceJobSubmitter and make it ready for extension for PyTorch. Contributed by Szilard Nemeth.

Author: tangzhankun

hadoop-submarine/hadoop-submarine-core/src/main/java/org/apache/hadoop/yarn/submarine/client/cli/param/RunJobParameters.java

@@ -293,10 +293,18 @@ public String getPsDockerImage() {
     return psDockerImage;
   }
 
+  public void setPsDockerImage(String psDockerImage) {
+    this.psDockerImage = psDockerImage;
+  }
+
   public String getWorkerDockerImage() {
     return workerDockerImage;
   }
 
+  public void setWorkerDockerImage(String workerDockerImage) {
+    this.workerDockerImage = workerDockerImage;
+  }
+
   public boolean isDistributed() {
     return distributed;
   }
@@ -313,6 +321,10 @@ public String getTensorboardDockerImage() {
     return tensorboardDockerImage;
   }
 
+  public void setTensorboardDockerImage(String tensorboardDockerImage) {
+    this.tensorboardDockerImage = tensorboardDockerImage;
+  }
+
   public List<Quicklink> getQuicklinks() {
     return quicklinks;
   }
@@ -366,6 +378,10 @@ public RunJobParameters setConfPairs(List<String> confPairs) {
     return this;
   }
 
+  public void setDistributed(boolean distributed) {
+    this.distributed = distributed;
+  }
+
   @VisibleForTesting
   public static class UnderscoreConverterPropertyUtils extends PropertyUtils {
     @Override

hadoop-submarine/hadoop-submarine-core/src/test/java/org/apache/hadoop/yarn/submarine/client/cli/TestRunJobCliParsing.java

@@ -177,6 +177,25 @@ public void testNoInputPathOptionButOnlyRunTensorboard() throws Exception {
     Assert.assertTrue(success);
   }
 
+  @Test
+  public void testJobWithoutName() throws Exception {
+    RunJobCli runJobCli = new RunJobCli(getMockClientContext());
+    String expectedErrorMessage =
+        "--" + CliConstants.NAME + " is absent";
+    String actualMessage = "";
+    try {
+      runJobCli.run(
+          new String[]{"--docker_image", "tf-docker:1.1.0",
+              "--num_workers", "0", "--tensorboard", "--verbose",
+              "--tensorboard_resources", "memory=2G,vcores=2",
+              "--tensorboard_docker_image", "tb_docker_image:001"});
+    } catch (ParseException e) {
+      actualMessage = e.getMessage();
+      e.printStackTrace();
+    }
+    assertEquals(expectedErrorMessage, actualMessage);
+  }
+
   @Test
   public void testLaunchCommandPatternReplace() throws Exception {
     RunJobCli runJobCli = new RunJobCli(getMockClientContext());

hadoop-submarine/hadoop-submarine-core/src/test/java/org/apache/hadoop/yarn/submarine/common/fs/MockRemoteDirectoryManager.java

@@ -26,6 +26,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.util.Objects;
 
 public class MockRemoteDirectoryManager implements RemoteDirectoryManager {
   private File jobsParentDir = null;
@@ -35,6 +36,7 @@ public class MockRemoteDirectoryManager implements RemoteDirectoryManager {
   @Override
   public Path getJobStagingArea(String jobName, boolean create)
       throws IOException {
+    Objects.requireNonNull(jobName, "Job name must not be null!");
     if (jobsParentDir == null && create) {
       jobsParentDir = new File(
           "target/_staging_area_" + System.currentTimeMillis());

hadoop-submarine/hadoop-submarine-yarnservice-runtime/pom.xml

@@ -115,6 +115,12 @@
       <artifactId>hadoop-yarn-services-core</artifactId>
       <version>3.3.0-SNAPSHOT</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-yarn-common</artifactId>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

hadoop-submarine/hadoop-submarine-yarnservice-runtime/src/main/java/org/apache/hadoop/yarn/submarine/runtimes/yarnservice/AbstractComponent.java

@@ -0,0 +1,99 @@
+/*
+ * 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.submarine.runtimes.yarnservice;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.yarn.service.api.records.Component;
+import org.apache.hadoop.yarn.submarine.client.cli.param.RunJobParameters;
+import org.apache.hadoop.yarn.submarine.common.api.TaskType;
+import org.apache.hadoop.yarn.submarine.common.fs.RemoteDirectoryManager;
+import org.apache.hadoop.yarn.submarine.runtimes.yarnservice.command.AbstractLaunchCommand;
+import org.apache.hadoop.yarn.submarine.runtimes.yarnservice.command.LaunchCommandFactory;
+
+import java.io.IOException;
+import java.util.Objects;
+
+import static org.apache.hadoop.yarn.submarine.runtimes.yarnservice.tensorflow.TensorFlowCommons.getScriptFileName;
+
+/**
+ * Abstract base class for Component classes.
+ * The implementations of this class are act like factories for
+ * {@link Component} instances.
+ * All dependencies are passed to the constructor so that child classes
+ * are obliged to provide matching constructors.
+ */
+public abstract class AbstractComponent {
+  private final FileSystemOperations fsOperations;
+  protected final RunJobParameters parameters;
+  protected final TaskType taskType;
+  private final RemoteDirectoryManager remoteDirectoryManager;
+  protected final Configuration yarnConfig;
+  private final LaunchCommandFactory launchCommandFactory;
+
+  /**
+   * This is only required for testing.
+   */
+  private String localScriptFile;
+
+  public AbstractComponent(FileSystemOperations fsOperations,
+      RemoteDirectoryManager remoteDirectoryManager,
+      RunJobParameters parameters, TaskType taskType,
+      Configuration yarnConfig,
+      LaunchCommandFactory launchCommandFactory) {
+    this.fsOperations = fsOperations;
+    this.remoteDirectoryManager = remoteDirectoryManager;
+    this.parameters = parameters;
+    this.taskType = taskType;
+    this.launchCommandFactory = launchCommandFactory;
+    this.yarnConfig = yarnConfig;
+  }
+
+  protected abstract Component createComponent() throws IOException;
+
+  /**
+   * Generates a command launch script on local disk,
+   * returns path to the script.
+   */
+  protected void generateLaunchCommand(Component component)
+      throws IOException {
+    AbstractLaunchCommand launchCommand =
+        launchCommandFactory.createLaunchCommand(taskType, component);
+    this.localScriptFile = launchCommand.generateLaunchScript();
+
+    String remoteLaunchCommand = uploadLaunchCommand(component);
+    component.setLaunchCommand(remoteLaunchCommand);
+  }
+
+  private String uploadLaunchCommand(Component component)
+      throws IOException {
+    Objects.requireNonNull(localScriptFile, "localScriptFile should be " +
+        "set before calling this method!");
+    Path stagingDir =
+        remoteDirectoryManager.getJobStagingArea(parameters.getName(), true);
+
+    String destScriptFileName = getScriptFileName(taskType);
+    fsOperations.uploadToRemoteFileAndLocalizeToContainerWorkDir(stagingDir,
+        localScriptFile, destScriptFileName, component);
+
+    return "./" + destScriptFileName;
+  }
+
+  String getLocalScriptFile() {
+    return localScriptFile;
+  }
+}

hadoop-submarine/hadoop-submarine-yarnservice-runtime/src/main/java/org/apache/hadoop/yarn/submarine/runtimes/yarnservice/FileSystemOperations.java

@@ -0,0 +1,201 @@
+/*
+ * 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.submarine.runtimes.yarnservice;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.yarn.service.api.records.Component;
+import org.apache.hadoop.yarn.service.api.records.ConfigFile;
+import org.apache.hadoop.yarn.submarine.common.ClientContext;
+import org.apache.hadoop.yarn.submarine.common.conf.SubmarineConfiguration;
+import org.apache.hadoop.yarn.submarine.common.conf.SubmarineLogs;
+import org.apache.hadoop.yarn.submarine.common.fs.RemoteDirectoryManager;
+import org.apache.hadoop.yarn.submarine.utils.ZipUtilities;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Contains methods to perform file system operations. Almost all of the methods
+ * are regular non-static methods as the operations are performed with the help
+ * of a {@link RemoteDirectoryManager} instance passed in as a constructor
+ * dependency. Please note that some operations require to read config settings
+ * as well, so that we have Submarine and YARN config objects as dependencies as
+ * well.
+ */
+public class FileSystemOperations {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(FileSystemOperations.class);
+  private final Configuration submarineConfig;
+  private final Configuration yarnConfig;
+
+  private Set<Path> uploadedFiles = new HashSet<>();
+  private RemoteDirectoryManager remoteDirectoryManager;
+
+  public FileSystemOperations(ClientContext clientContext) {
+    this.remoteDirectoryManager = clientContext.getRemoteDirectoryManager();
+    this.submarineConfig = clientContext.getSubmarineConfig();
+    this.yarnConfig = clientContext.getYarnConfig();
+  }
+
+  /**
+   * May download a remote uri(file/dir) and zip.
+   * Skip download if local dir
+   * Remote uri can be a local dir(won't download)
+   * or remote HDFS dir, s3 dir/file .etc
+   * */
+  public String downloadAndZip(String remoteDir, String zipFileName,
+      boolean doZip)
+      throws IOException {
+    //Append original modification time and size to zip file name
+    String suffix;
+    String srcDir = remoteDir;
+    String zipDirPath =
+        System.getProperty("java.io.tmpdir") + "/" + zipFileName;
+    boolean needDeleteTempDir = false;
+    if (remoteDirectoryManager.isRemote(remoteDir)) {
+      //Append original modification time and size to zip file name
+      FileStatus status =
+          remoteDirectoryManager.getRemoteFileStatus(new Path(remoteDir));
+      suffix = "_" + status.getModificationTime()
+          + "-" + remoteDirectoryManager.getRemoteFileSize(remoteDir);
+      // Download them to temp dir
+      boolean downloaded =
+          remoteDirectoryManager.copyRemoteToLocal(remoteDir, zipDirPath);
+      if (!downloaded) {
+        throw new IOException("Failed to download files from "
+            + remoteDir);
+      }
+      LOG.info("Downloaded remote: {} to local: {}", remoteDir, zipDirPath);
+      srcDir = zipDirPath;
+      needDeleteTempDir = true;
+    } else {
+      File localDir = new File(remoteDir);
+      suffix = "_" + localDir.lastModified()
+          + "-" + localDir.length();
+    }
+    if (!doZip) {
+      return srcDir;
+    }
+    // zip a local dir
+    String zipFileUri =
+        ZipUtilities.zipDir(srcDir, zipDirPath + suffix + ".zip");
+    // delete downloaded temp dir
+    if (needDeleteTempDir) {
+      deleteFiles(srcDir);
+    }
+    return zipFileUri;
+  }
+
+  public void deleteFiles(String localUri) {
+    boolean success = FileUtil.fullyDelete(new File(localUri));
+    if (!success) {
+      LOG.warn("Failed to delete {}", localUri);
+    }
+    LOG.info("Deleted {}", localUri);
+  }
+
+  @VisibleForTesting
+  public void uploadToRemoteFileAndLocalizeToContainerWorkDir(Path stagingDir,
+      String fileToUpload, String destFilename, Component comp)
+      throws IOException {
+    Path uploadedFilePath = uploadToRemoteFile(stagingDir, fileToUpload);
+    locateRemoteFileToContainerWorkDir(destFilename, comp, uploadedFilePath);
+  }
+
+  private void locateRemoteFileToContainerWorkDir(String destFilename,
+      Component comp, Path uploadedFilePath)
+      throws IOException {
+    FileSystem fs = FileSystem.get(yarnConfig);
+
+    FileStatus fileStatus = fs.getFileStatus(uploadedFilePath);
+    LOG.info("Uploaded file path = " + fileStatus.getPath());
+
+    // Set it to component's files list
+    comp.getConfiguration().getFiles().add(new ConfigFile().srcFile(
+        fileStatus.getPath().toUri().toString()).destFile(destFilename)
+        .type(ConfigFile.TypeEnum.STATIC));
+  }
+
+  public Path uploadToRemoteFile(Path stagingDir, String fileToUpload) throws
+      IOException {
+    FileSystem fs = remoteDirectoryManager.getDefaultFileSystem();
+
+    // Upload to remote FS under staging area
+    File localFile = new File(fileToUpload);
+    if (!localFile.exists()) {
+      throw new FileNotFoundException(
+          "Trying to upload file=" + localFile.getAbsolutePath()
+              + " to remote, but couldn't find local file.");
+    }
+    String filename = new File(fileToUpload).getName();
+
+    Path uploadedFilePath = new Path(stagingDir, filename);
+    if (!uploadedFiles.contains(uploadedFilePath)) {
+      if (SubmarineLogs.isVerbose()) {
+        LOG.info("Copying local file=" + fileToUpload + " to remote="
+            + uploadedFilePath);
+      }
+      fs.copyFromLocalFile(new Path(fileToUpload), uploadedFilePath);
+      uploadedFiles.add(uploadedFilePath);
+    }
+    return uploadedFilePath;
+  }
+
+  public void validFileSize(String uri) throws IOException {
+    long actualSizeByte;
+    String locationType = "Local";
+    if (remoteDirectoryManager.isRemote(uri)) {
+      actualSizeByte = remoteDirectoryManager.getRemoteFileSize(uri);
+      locationType = "Remote";
+    } else {
+      actualSizeByte = FileUtil.getDU(new File(uri));
+    }
+    long maxFileSizeMB = submarineConfig
+        .getLong(SubmarineConfiguration.LOCALIZATION_MAX_ALLOWED_FILE_SIZE_MB,
+            SubmarineConfiguration.DEFAULT_MAX_ALLOWED_REMOTE_URI_SIZE_MB);
+    LOG.info("{} fie/dir: {}, size(Byte):{},"
+            + " Allowed max file/dir size: {}",
+        locationType, uri, actualSizeByte, maxFileSizeMB * 1024 * 1024);
+
+    if (actualSizeByte > maxFileSizeMB * 1024 * 1024) {
+      throw new IOException(uri + " size(Byte): "
+          + actualSizeByte + " exceeds configured max size:"
+          + maxFileSizeMB * 1024 * 1024);
+    }
+  }
+
+  public void setPermission(Path destPath, FsPermission permission) throws
+      IOException {
+    FileSystem fs = FileSystem.get(yarnConfig);
+    fs.setPermission(destPath, new FsPermission(permission));
+  }
+
+  public static boolean needHdfs(String content) {
+    return content != null && content.contains("hdfs://");
+  }
+}