sharelib/distcp/src/main/java/org/apache/oozie/action/hadoop/DistcpMain.java

/**
 * 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.oozie.action.hadoop;

import java.io.File;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.PropertyConfigurator;

import com.google.common.annotations.VisibleForTesting;

public class DistcpMain extends JavaMain {

    private Constructor<?> construct;
    private Object[] constArgs;
    private static final String DISTCP_LOG4J_PROPS = "distcp-log4j.properties";

    @VisibleForTesting
    static final Pattern[] DISTCP_JOB_IDS_PATTERNS = {
            Pattern.compile("Job complete: (job_\\S*)"),
            Pattern.compile("Job (job_\\S*) completed successfully"),
            Pattern.compile("Submitted application (application[0-9_]*)")
    };
    public static void main(String[] args) throws Exception {
        run(DistcpMain.class, args);
    }

    @Override
    protected void run(String[] args) throws Exception {

        Configuration actionConf = loadActionConf();
        LauncherMain.killChildYarnJobs(actionConf);
        String logFile = setUpDistcpLog4J(actionConf);
        Class<?> klass = actionConf.getClass(LauncherAMUtils.CONF_OOZIE_ACTION_MAIN_CLASS,
                org.apache.hadoop.tools.DistCp.class);
        System.out.println("Main class        : " + klass.getName());
        System.out.println("Arguments         :");
        for (String arg : args) {
            System.out.println("                    " + arg);
        }

        // propagate delegation related props from launcher job to MR job
        if (getFilePathFromEnv("HADOOP_TOKEN_FILE_LOCATION") != null) {
            actionConf.set("mapreduce.job.credentials.binary", getFilePathFromEnv("HADOOP_TOKEN_FILE_LOCATION"));
        }

        getConstructorAndArgs(klass, actionConf);
        if (construct == null) {
            throw new RuntimeException("Distcp constructor was not found, unable to instantiate");
        }
        if (constArgs == null) {
            throw new RuntimeException("Arguments for distcp constructor is null, unable to instantiate");
        }
        try {
            Tool distcp = (Tool) construct.newInstance(constArgs);
            int i = ToolRunner.run(distcp, args);
            if (i != 0) {
                throw new RuntimeException("Returned value from distcp is non-zero (" + i + ")");
            }
        }
        catch (InvocationTargetException ex) {
            throw new JavaMainException(ex.getCause());
        }
        finally {
            System.out.println("\n<<< Invocation of DistCp command completed <<<\n");
            writeExternalChildIDs(logFile, DISTCP_JOB_IDS_PATTERNS, "Distcp");
        }
    }

    protected void getConstructorAndArgs(Class<?> klass, Configuration actionConf) throws Exception {
        Constructor<?>[] allConstructors = klass.getConstructors();
        for (Constructor<?> cstruct : allConstructors) {
            Class<?>[] pType = cstruct.getParameterTypes();
            construct = cstruct;
            if (pType.length == 1 && pType[0].equals(Class.forName("org.apache.hadoop.conf.Configuration"))) {
                System.out.println("found Distcp v1 Constructor");
                System.out.println("                    " + cstruct.toString());
                constArgs = new Object[1];
                constArgs[0] = actionConf;
                break;
            }
            else if (pType.length == 2 && pType[0].equals(Class.forName("org.apache.hadoop.conf.Configuration"))) {
                // 2nd argument is org.apache.hadoop.tools.DistCpOptions
                System.out.println("found Distcp v2 Constructor");
                System.out.println("                    " + cstruct.toString());
                constArgs = new Object[2];
                constArgs[0] = actionConf;
                constArgs[1] = null;
                break;
            }
        }
    }

    private String setUpDistcpLog4J(Configuration distcpConf) throws IOException {
        // Logfile to capture job IDs
        String hadoopJobId = System.getProperty("oozie.launcher.job.id");
        if (hadoopJobId == null) {
            throw new RuntimeException("Launcher Hadoop Job ID system,property not set");
        }

        String logFile = new File("distcp-oozie-" + hadoopJobId + ".log").getAbsolutePath();

        String logLevel = distcpConf.get("oozie.distcp.log.level", "INFO");
        String rootLogLevel = distcpConf.get("oozie.action." + LauncherAMUtils.ROOT_LOGGER_LEVEL, "INFO");

        log4jProperties.setProperty("log4j.rootLogger", rootLogLevel + ", A");
        log4jProperties.setProperty("log4j.logger.org.apache.hadoop.tools", logLevel + ", A, jobid");
        log4jProperties.setProperty("log4j.additivity.org.apache.hadoop.tools", "false");
        log4jProperties.setProperty("log4j.appender.A", "org.apache.log4j.ConsoleAppender");
        log4jProperties.setProperty("log4j.appender.A.layout", "org.apache.log4j.PatternLayout");
        log4jProperties.setProperty("log4j.appender.A.layout.ConversionPattern", "%-4r [%t] %-5p %c %x - %m%n");

        log4jProperties.setProperty("log4j.appender.jobid", "org.apache.log4j.FileAppender");
        log4jProperties.setProperty("log4j.appender.jobid.file", logFile);
        log4jProperties.setProperty("log4j.appender.jobid.layout", "org.apache.log4j.PatternLayout");
        log4jProperties.setProperty("log4j.appender.jobid.layout.ConversionPattern", "%-4r [%t] %-5p %c %x - %m%n");
        log4jProperties.setProperty("log4j.logger.org.apache.hadoop.mapred", "INFO, jobid");
        log4jProperties.setProperty("log4j.logger.org.apache.hadoop.mapreduce.Job", "INFO, jobid");
        log4jProperties.setProperty("log4j.logger.org.apache.hadoop.yarn.client.api.impl.YarnClientImpl", "INFO, jobid");

        String localProps = new File(DISTCP_LOG4J_PROPS).getAbsolutePath();
        createFileWithContentIfNotExists(localProps, log4jProperties);
        PropertyConfigurator.configure(DISTCP_LOG4J_PROPS);

        return logFile;
    }
}