Source code

001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *   http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing,
013 * software distributed under the License is distributed on an
014 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
015 * KIND, either express or implied.  See the License for the
016 * specific language governing permissions and limitations
017 * under the License.
018 */
019package org.apache.reef.runtime.common.files;
020
021import org.apache.reef.annotations.audience.ClientSide;
022import org.apache.reef.annotations.audience.Private;
023import org.apache.reef.annotations.audience.RuntimeAuthor;
024import org.apache.reef.runtime.common.client.api.JobSubmissionEvent;
025import org.apache.reef.runtime.common.parameters.DeleteTempFiles;
026import org.apache.reef.tang.Configuration;
027import org.apache.reef.tang.annotations.Parameter;
028import org.apache.reef.tang.formats.ConfigurationSerializer;
029import org.apache.reef.util.JARFileMaker;
030
031import javax.inject.Inject;
032import java.io.File;
033import java.io.IOException;
034import java.nio.file.Files;
035import java.util.logging.Level;
036import java.util.logging.Logger;
037
038/**
039 * Utility that takes a JobSubmissionProto and turns it into a Job Submission Jar.
040 */
041@Private
042@RuntimeAuthor
043@ClientSide
044public final class JobJarMaker {
045
046  private static final Logger LOG = Logger.getLogger(JobJarMaker.class.getName());
047
048  private final ConfigurationSerializer configurationSerializer;
049  private final REEFFileNames fileNames;
050  private final boolean deleteTempFilesOnExit;
051
052  @Inject
053  JobJarMaker(final ConfigurationSerializer configurationSerializer,
054              final REEFFileNames fileNames,
055              @Parameter(DeleteTempFiles.class) final boolean deleteTempFilesOnExit) {
056    this.configurationSerializer = configurationSerializer;
057    this.fileNames = fileNames;
058    this.deleteTempFilesOnExit = deleteTempFilesOnExit;
059  }
060
061  public static void copy(final Iterable<FileResource> files, final File destinationFolder) {
062
063    if (!destinationFolder.exists() && !destinationFolder.mkdirs()) {
064      LOG.log(Level.WARNING, "Failed to create [{0}]", destinationFolder.getAbsolutePath());
065    }
066
067    for (final FileResource fileProto : files) {
068      final File sourceFile = toFile(fileProto);
069      final File destinationFile = new File(destinationFolder, fileProto.getName());
070      if (destinationFile.exists()) {
071        LOG.log(Level.FINEST,
072            "Will not add {0} to the job jar because another file with the same name was already added.",
073            sourceFile.getAbsolutePath()
074        );
075      } else {
076        try {
077          Files.copy(sourceFile.toPath(), destinationFile.toPath());
078        } catch (final IOException e) {
079          final String message = new StringBuilder("Copy of file [")
080              .append(sourceFile.getAbsolutePath())
081              .append("] to [")
082              .append(destinationFile.getAbsolutePath())
083              .append("] failed.")
084              .toString();
085          throw new RuntimeException(message, e);
086        }
087      }
088    }
089  }
090
091  private static File toFile(final FileResource fileProto) {
092    return new File(fileProto.getPath());
093  }
094
095  public File createJobSubmissionJAR(
096      final JobSubmissionEvent jobSubmissionEvent,
097      final Configuration driverConfiguration) throws IOException {
098
099    // Copy all files to a local job submission folder
100    final File jobSubmissionFolder = makejobSubmissionFolder();
101    LOG.log(Level.FINE, "Staging submission in {0}", jobSubmissionFolder);
102
103    final File localFolder = new File(jobSubmissionFolder, this.fileNames.getLocalFolderName());
104    final File globalFolder = new File(jobSubmissionFolder, this.fileNames.getGlobalFolderName());
105
106    copy(jobSubmissionEvent.getGlobalFileSet(), globalFolder);
107    copy(jobSubmissionEvent.getLocalFileSet(), localFolder);
108
109    // Store the Driver Configuration in the JAR file.
110    this.configurationSerializer.toFile(
111        driverConfiguration, new File(localFolder, this.fileNames.getDriverConfigurationName()));
112
113    // Create a JAR File for the submission
114    final File jarFile = File.createTempFile(this.fileNames.getJobFolderPrefix(), this.fileNames.getJarFileSuffix());
115
116    LOG.log(Level.FINE, "Creating job submission jar file: {0}", jarFile);
117    new JARFileMaker(jarFile).addChildren(jobSubmissionFolder).close();
118
119    if (this.deleteTempFilesOnExit) {
120      LOG.log(Level.FINE,
121          "Deleting the temporary job folder [{0}] and marking the jar file [{1}] for deletion after the JVM exits.",
122          new Object[]{jobSubmissionFolder.getAbsolutePath(), jarFile.getAbsolutePath()});
123      if (!jobSubmissionFolder.delete()) {
124        LOG.log(Level.WARNING, "Failed to delete [{0}]", jobSubmissionFolder.getAbsolutePath());
125      }
126      jarFile.deleteOnExit();
127    } else {
128      LOG.log(Level.FINE, "Keeping the temporary job folder [{0}] and jar file [{1}] available after job submission.",
129          new Object[]{jobSubmissionFolder.getAbsolutePath(), jarFile.getAbsolutePath()});
130    }
131    return jarFile;
132  }
133
134  private File makejobSubmissionFolder() throws IOException {
135    return Files.createTempDirectory(this.fileNames.getJobFolderPrefix()).toFile();
136  }
137}