/*
 * 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.pig.backend.hadoop.executionengine;

import java.io.IOException;
import java.io.PrintStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapred.TaskReport;
import org.apache.hadoop.mapred.jobcontrol.Job;
import org.apache.hadoop.mapred.jobcontrol.JobControl;
import org.apache.pig.FuncSpec;
import org.apache.pig.PigException;
import org.apache.pig.backend.BackendException;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
import org.apache.pig.backend.hadoop.executionengine.shims.HadoopShims;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.io.FileSpec;
import org.apache.pig.impl.plan.PlanException;
import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.impl.util.LogUtils;
import org.apache.pig.impl.util.Utils;
import org.apache.pig.tools.pigstats.PigStats;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

/**
 *
 * Provides core processing implementation for the backend of Pig
 * if ExecutionEngine chosen decides to delegate it's work to this class.
 * Also contains set of utility methods, including ones centered around
 * Hadoop.
 *
 */
public abstract class Launcher {
    private static final Log log = LogFactory.getLog(Launcher.class);
    private static final String OOM_ERR = "OutOfMemoryError";
    private boolean pigException = false;
    private boolean outOfMemory = false;
    private String newLine = "\n";

    // Used to track the exception thrown by the job control which is run in a
    // separate thread
    protected String jobControlExceptionStackTrace = null;
    protected Exception jobControlException = null;
    protected long totalHadoopTimeSpent;

    protected Map<FileSpec, Exception> failureMap;
    protected JobControl jc = null;

    protected class HangingJobKiller extends Thread {
        public HangingJobKiller() {}

        @Override
        public void run() {
            try {
                kill();
            } catch (Exception e) {
                log.warn("Error in killing Execution Engine: " + e);
            }
        }
    }

    protected Launcher() {
        // handle the windows portion of \r
        if (System.getProperty("os.name").toUpperCase().startsWith("WINDOWS")) {
            newLine = "\r\n";
        }
        reset();
    }

    /**
     * Resets the state after a launch
     */
    public void reset() {
        failureMap = Maps.newHashMap();
        totalHadoopTimeSpent = 0;
    }

    /**
     * Method to launch pig for hadoop either for a cluster's job tracker or for
     * a local job runner. THe only difference between the two is the job
     * client. Depending on the pig context the job client will be initialize to
     * one of the two. Launchers for other frameworks can overide these methods.
     * Given an input PhysicalPlan, it compiles it to get a MapReduce Plan. The
     * MapReduce plan which has multiple MapReduce operators each one of which
     * has to be run as a map reduce job with dependency information stored in
     * the plan. It compiles the MROperPlan into a JobControl object. Each Map
     * Reduce operator is converted into a Job and added to the JobControl
     * object. Each Job also has a set of dependent Jobs that are created using
     * the MROperPlan. The JobControl object is obtained from the
     * JobControlCompiler Then a new thread is spawned that submits these jobs
     * while respecting the dependency information. The parent thread monitors
     * the submitted jobs' progress and after it is complete, stops the
     * JobControl thread.
     *
     * @param php
     * @param grpName
     * @param pc
     * @throws Exception
     */
    public abstract PigStats launchPig(PhysicalPlan php, String grpName,
            PigContext pc) throws Exception;

    /**
     * Explain how a pig job will be executed on the underlying infrastructure.
     *
     * @param pp
     *            PhysicalPlan to explain
     * @param pc
     *            PigContext to use for configuration
     * @param ps
     *            PrintStream to write output on.
     * @param format
     *            Format to write in
     * @param verbose
     *            Amount of information to print
     * @throws VisitorException
     * @throws IOException
     */
    public abstract void explain(PhysicalPlan pp, PigContext pc,
            PrintStream ps, String format, boolean verbose)
            throws PlanException, VisitorException, IOException;

    public abstract void kill() throws BackendException;

    public abstract void killJob(String jobID, Configuration conf)
            throws BackendException;

    protected boolean isComplete(double prog) {
        return (int) (Math.ceil(prog)) == 1;
    }

    protected long computeTimeSpent(Iterator<TaskReport> taskReports) {
        long timeSpent = 0;
        while (taskReports.hasNext()) {
            TaskReport r = taskReports.next();
            timeSpent += (r.getFinishTime() - r.getStartTime());
        }
        return timeSpent;
    }

    protected void getErrorMessages(Iterator<TaskReport> reports, String type,
            boolean errNotDbg, PigContext pigContext) throws Exception {
        while(reports.hasNext()) {
            TaskReport report = reports.next();
            String msgs[] = report.getDiagnostics();
            ArrayList<Exception> exceptions = new ArrayList<Exception>();
            String exceptionCreateFailMsg = null;
            boolean jobFailed = false;
            if (msgs.length > 0) {
                if (HadoopShims.isJobFailed(report)) {
                    jobFailed = true;
                }
                Set<String> errorMessageSet = new HashSet<String>();
                for (int j = 0; j < msgs.length; j++) {
                    if (!errorMessageSet.contains(msgs[j])) {
                        errorMessageSet.add(msgs[j]);
                        if (errNotDbg) {
                            // errNotDbg is used only for failed jobs
                            // keep track of all the unique exceptions
                            try {
                                LogUtils.writeLog("Backend error message",
                                        msgs[j], pigContext.getProperties()
                                                .getProperty("pig.logfile"),
                                        log);
                                Exception e = getExceptionFromString(msgs[j]);
                                exceptions.add(e);
                            } catch (Exception e1) {
                                exceptionCreateFailMsg = msgs[j];

                            }
                        } else {
                            log.debug("Error message from task (" + type + ") "
                                    + report.getTaskID() + msgs[j]);
                        }
                    }
                }
            }
            // if there are no valid exception that could be created, report
            if (jobFailed && (exceptions.size() == 0) && (exceptionCreateFailMsg != null)) {
                int errCode = 2997;
                String msg = "Unable to recreate exception from backed error: "
                        + exceptionCreateFailMsg;
                throw new ExecException(msg, errCode, PigException.BUG);
            }

            // if its a failed job then check if there is more than one
            // exception
            // more than one exception implies possibly different kinds of
            // failures
            // log all the different failures and throw the exception
            // corresponding
            // to the first failure
            if (jobFailed) {
                if (exceptions.size() > 1) {
                    for (int j = 0; j < exceptions.size(); ++j) {
                        String headerMessage = "Error message from task ("
                                + type + ") " + report.getTaskID();
                        LogUtils.writeLog(exceptions.get(j), pigContext
                                .getProperties().getProperty("pig.logfile"),
                                log, false, headerMessage, false, false);
                    }
                    throw exceptions.get(0);
                } else if (exceptions.size() == 1) {
                    throw exceptions.get(0);
                } else {
                    int errCode = 2115;
                    String msg = "Internal error. Expected to throw exception from the backend. Did not find any exception to throw.";
                    throw new ExecException(msg, errCode, PigException.BUG);
                }
            }
        }
    }

    /**
     * Compute the progress of the current job submitted through the JobControl
     * object jc to the JobClient jobClient
     *
     * @param jc
     *            - The JobControl object that has been submitted
     * @param jobClient
     *            - The JobClient to which it has been submitted
     * @return The progress as a precentage in double format
     * @throws IOException
     */
    protected double calculateProgress(JobControl jc)
            throws IOException {
        double prog = 0.0;
        prog += jc.getSuccessfulJobs().size();

        List<Job> runnJobs = jc.getRunningJobs();
        for (Job j : runnJobs) {
            prog += HadoopShims.progressOfRunningJob(j);
        }
        return prog;
    }

    public long getTotalHadoopTimeSpent() {
        return totalHadoopTimeSpent;
    }

    /**
     * An exception handler class to handle exceptions thrown by the job controller thread
     * Its a local class. This is the only mechanism to catch unhandled thread exceptions
     * Unhandled exceptions in threads are handled by the VM if the handler is not registered
     * explicitly or if the default handler is null
     */
    public class JobControlThreadExceptionHandler implements Thread.UncaughtExceptionHandler {
        @Override
        public void uncaughtException(Thread thread, Throwable throwable) {
            setJobException(throwable);
        }
    }

    protected void setJobException(Throwable throwable) {
        jobControlExceptionStackTrace = Utils.getStackStraceStr(throwable);
        try {
            jobControlException = getExceptionFromString(jobControlExceptionStackTrace);
        } catch (Exception e) {
            String errMsg = "Could not resolve error that occurred when launching job: "
                    + jobControlExceptionStackTrace;
            jobControlException = new RuntimeException(errMsg, throwable);
        }
    }

    /**
     *
     * @param stackTrace
     *            The string representation of
     *            {@link Throwable#printStackTrace() printStackTrace} Handles
     *            internal PigException and its subclasses that override the
     *            {@link Throwable#toString() toString} method
     * @return An exception object whose string representation of
     *         printStackTrace is the input stackTrace
     * @throws Exception
     */
    public Exception getExceptionFromString(String stackTrace) throws Exception {
        String[] lines = stackTrace.split(newLine);
        Throwable t = getExceptionFromStrings(lines, 0);

        if (!pigException) {
            int errCode = 6015;
            String msg = "During execution, encountered a Hadoop error.";
            ExecException ee = new ExecException(msg, errCode,
                    PigException.REMOTE_ENVIRONMENT, t);
            ee.setStackTrace(t.getStackTrace());
            return ee;
        } else {
            pigException = false;
            if (outOfMemory) {
                outOfMemory = false;
                int errCode = 6016;
                String msg = "Out of memory.";
                ExecException ee = new ExecException(msg, errCode,
                        PigException.REMOTE_ENVIRONMENT, t);
                ee.setStackTrace(t.getStackTrace());
                return ee;
            }
            return (Exception) t;
        }
    }

    /**
     *
     * @param stackTraceLines
     *            An array of strings that represent
     *            {@link Throwable#printStackTrace() printStackTrace} output,
     *            split by newline
     * @return An exception object whose string representation of
     *         printStackTrace is the input stackTrace
     * @throws Exception
     */
    private Throwable getExceptionFromStrings(String[] stackTraceLines,
            int startingLineNum) throws Exception {
        /*
         * parse the array of string and throw the appropriate exception first:
         * from the line startingLineNum extract the exception name extract the
         * message if any fourth: create the appropriate exception and return it
         * An example of the stack trace:
         * org.apache.pig.backend.executionengine.ExecException: ERROR 1075:
         * Received a bytearray from the UDF. Cannot determine how to convert
         * the bytearray to int. at
         * org.apache.pig.backend.hadoop.executionengine
         * .physicalLayer.expressionOperators.POCast.getNext(POCast.java:152) at
         * org.apache.pig.backend.hadoop.executionengine.physicalLayer.
         * expressionOperators.LessThanExpr.getNext(LessThanExpr.java:85) at
         * org.apache.pig.backend.hadoop.executionengine.physicalLayer.
         * relationalOperators.POFilter.getNext(POFilter.java:148) at
         * org.apache.
         * pig.backend.hadoop.executionengine.mapReduceLayer.PigMapBase
         * .runPipeline(PigMapBase.java:184) at
         * org.apache.pig.backend.hadoop.executionengine
         * .mapReduceLayer.PigMapBase.map(PigMapBase.java:174) at
         * org.apache.pig.
         * backend.hadoop.executionengine.mapReduceLayer.PigMapOnly$Map
         * .map(PigMapOnly.java:65) at
         * org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:47) at
         * org.apache.hadoop.mapred.MapTask.run(MapTask.java:227) at
         * org.apache.hadoop
         * .mapred.TaskTracker$Child.main(TaskTracker.java:2207)
         */

        if (stackTraceLines.length > 0
                && startingLineNum < (stackTraceLines.length - 1)) {

            // the regex for matching the exception class name; note the use of
            // the $ for matching nested classes
            String exceptionNameDelimiter = "(\\w+(\\$\\w+)?\\.)+\\w+";
            Pattern exceptionNamePattern = Pattern
                    .compile(exceptionNameDelimiter);

            // from the first line extract the exception name and the exception
            // message
            Matcher exceptionNameMatcher = exceptionNamePattern
                    .matcher(stackTraceLines[startingLineNum]);
            String exceptionName = null;
            String exceptionMessage = null;
            if (exceptionNameMatcher.find()) {
                exceptionName = exceptionNameMatcher.group();
                /*
                 * note that the substring is from end + 2 the regex matcher
                 * ends at one position beyond the match in this case it will
                 * end at colon (:) the exception message will have a preceding
                 * space (after the colon (:))
                 */
                if (exceptionName.contains(OOM_ERR)) {
                    outOfMemory = true;
                }

                if (stackTraceLines[startingLineNum].length() > exceptionNameMatcher
                        .end()) {
                    exceptionMessage = stackTraceLines[startingLineNum]
                            .substring(exceptionNameMatcher.end() + 2);
                }

                ++startingLineNum;
            }

            // the exceptionName should not be null
            if (exceptionName != null) {

                ArrayList<StackTraceElement> stackTraceElements = Lists.newArrayList();

                // Create stack trace elements for the remaining lines
                String stackElementRegex = "\\s+at\\s+(\\w+(\\$\\w+)?\\.)+(\\<)?\\w+(\\>)?";
                Pattern stackElementPattern = Pattern
                        .compile(stackElementRegex);
                String pigExceptionRegex = "org\\.apache\\.pig\\.";
                Pattern pigExceptionPattern = Pattern
                        .compile(pigExceptionRegex);
                String moreElementRegex = "\\s+\\.\\.\\.\\s+\\d+\\s+more";
                Pattern moreElementPattern = Pattern.compile(moreElementRegex);

                int lineNum = startingLineNum;
                for (; lineNum < (stackTraceLines.length - 1); ++lineNum) {
                    Matcher stackElementMatcher = stackElementPattern
                            .matcher(stackTraceLines[lineNum]);

                    if (stackElementMatcher.find()) {
                        StackTraceElement ste = getStackTraceElement(stackTraceLines[lineNum]);
                        stackTraceElements.add(ste);
                        String className = ste.getClassName();
                        Matcher pigExceptionMatcher = pigExceptionPattern
                                .matcher(className);
                        if (pigExceptionMatcher.find()) {
                            pigException = true;
                        }
                    } else {
                        Matcher moreElementMatcher = moreElementPattern
                                .matcher(stackTraceLines[lineNum]);
                        if (moreElementMatcher.find()) {
                            ++lineNum;
                        }
                        break;
                    }
                }

                startingLineNum = lineNum;

                // create the appropriate exception; setup the stack trace and
                // message
                Object object = PigContext
                        .instantiateFuncFromSpec(exceptionName);

                if (object instanceof PigException) {
                    // extract the error code and message the regex for matching
                    // the custom format of ERROR <ERROR CODE>:
                    String errMessageRegex = "ERROR\\s+\\d+:";
                    Pattern errMessagePattern = Pattern
                            .compile(errMessageRegex);
                    Matcher errMessageMatcher = errMessagePattern
                            .matcher(exceptionMessage);

                    if (errMessageMatcher.find()) {
                        String errMessageStub = errMessageMatcher.group();
                        /*
                         * extract the actual exception message sans the ERROR
                         * <ERROR CODE>: again note that the matcher ends at the
                         * space following the colon (:) the exception message
                         * appears after the space and hence the end + 1
                         */
                        exceptionMessage = exceptionMessage
                                .substring(errMessageMatcher.end() + 1);

                        // the regex to match the error code wich is a string of
                        // numerals
                        String errCodeRegex = "\\d+";
                        Pattern errCodePattern = Pattern.compile(errCodeRegex);
                        Matcher errCodeMatcher = errCodePattern
                                .matcher(errMessageStub);

                        String code = null;
                        if (errCodeMatcher.find()) {
                            code = errCodeMatcher.group();
                        }

                        // could receive a number format exception here but it
                        // will be propagated up the stack
                        int errCode;
                        if (code != null)
                            errCode = Integer.parseInt(code);
                        else
                            errCode = 2998;

                        // create the exception with the message and then set
                        // the error code and error source
                        FuncSpec funcSpec = new FuncSpec(exceptionName,
                                exceptionMessage);
                        object = PigContext.instantiateFuncFromSpec(funcSpec);
                        ((PigException) object).setErrorCode(errCode);
                        ((PigException) object).setErrorSource(PigException
                                .determineErrorSource(errCode));
                    } else { // else for if(errMessageMatcher.find())
                        /*
                         * did not find the error code which means that the
                         * PigException or its subclass is not returning the
                         * error code highly unlikely: should never be here
                         */
                        FuncSpec funcSpec = new FuncSpec(exceptionName,
                                exceptionMessage);
                        object = PigContext.instantiateFuncFromSpec(funcSpec);
                        ((PigException) object).setErrorCode(2997);// generic
                                                                    // error
                                                                    // code
                        ((PigException) object)
                                .setErrorSource(PigException.BUG);
                    }
                } else { // else for if(object instanceof PigException)
                    // its not PigException; create the exception with the
                    // message
                    object = PigContext.instantiateFuncFromSpec(new FuncSpec(
                            exceptionName, exceptionMessage));
                }

                StackTraceElement[] steArr = new StackTraceElement[stackTraceElements
                        .size()];
                ((Throwable) object).setStackTrace(stackTraceElements
                        .toArray(steArr));

                if (startingLineNum < (stackTraceLines.length - 1)) {
                    Throwable e = getExceptionFromStrings(stackTraceLines,
                            startingLineNum);
                    ((Throwable) object).initCause(e);
                }

                return (Throwable) object;
            } else { // else for if(exceptionName != null)
                int errCode = 2055;
                String msg = "Did not find exception name to create exception from string: "
                        + Arrays.toString(stackTraceLines);
                throw new ExecException(msg, errCode, PigException.BUG);
            }
        } else { // else for if(lines.length > 0)
            int errCode = 2056;
            String msg = "Cannot create exception from empty string.";
            throw new ExecException(msg, errCode, PigException.BUG);
        }
    }

    /**
     *
     * @param line
     *            the string representation of a stack trace returned by
     *            {@link Throwable#printStackTrace() printStackTrace}
     * @return the StackTraceElement object representing the stack trace
     * @throws Exception
     */
    public StackTraceElement getStackTraceElement(String line) throws Exception {
        /*
         * the format of the line is something like: at
         * org.apache.pig.backend.hadoop
         * .executionengine.mapReduceLayer.PigMapOnly$Map
         * .map(PigMapOnly.java:65) note the white space before the 'at'. Its
         * not of much importance but noted for posterity.
         */
        String[] items;

        /*
         * regex for matching the fully qualified method Name note the use of
         * the $ for matching nested classes and the use of < and > for
         * constructors
         */
        String qualifiedMethodNameRegex = "(\\w+(\\$\\w+)?\\.)+(<)?\\w+(>)?";
        Pattern qualifiedMethodNamePattern = Pattern
                .compile(qualifiedMethodNameRegex);
        Matcher contentMatcher = qualifiedMethodNamePattern.matcher(line);

        // org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapOnly$Map.map(PigMapOnly.java:65)
        String content = null;
        if (contentMatcher.find()) {
            content = line.substring(contentMatcher.start());
        } else {
            int errCode = 2057;
            String msg = "Did not find fully qualified method name to reconstruct stack trace: "
                    + line;
            throw new ExecException(msg, errCode, PigException.BUG);
        }

        Matcher qualifiedMethodNameMatcher = qualifiedMethodNamePattern
                .matcher(content);

        // org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapOnly$Map.map
        String qualifiedMethodName = null;
        // (PigMapOnly.java:65)
        String fileDetails = null;

        if (qualifiedMethodNameMatcher.find()) {
            qualifiedMethodName = qualifiedMethodNameMatcher.group();
            fileDetails = content
                    .substring(qualifiedMethodNameMatcher.end() + 1);
        } else {
            int errCode = 2057;
            String msg = "Did not find fully qualified method name to reconstruct stack trace: "
                    + line;
            throw new ExecException(msg, errCode, PigException.BUG);
        }

        // From the fully qualified method name, extract the declaring class and
        // method name
        items = qualifiedMethodName.split("\\.");

        // initialize the declaringClass (to org in most cases)
        String declaringClass = items[0];
        // the last member is always the method name
        String methodName = items[items.length - 1];
        StringBuilder sb = new StringBuilder();

        // concatenate the names by adding the dot (.) between the members till
        // the penultimate member
        for (int i = 1; i < items.length - 1; ++i) {
            sb.append('.');
            sb.append(items[i]);
        }

        declaringClass += sb.toString();

        // from the file details extract the file name and the line number
        // PigMapOnly.java:65
        fileDetails = fileDetails.substring(0, fileDetails.length() - 1);
        items = fileDetails.split(":");
        // PigMapOnly.java
        String fileName = null;
        int lineNumber = -1;
        if (items.length > 0) {
            fileName = items[0];
            if (items.length > 1) {
                lineNumber = Integer.parseInt(items[1]);
            }
        }
        return new StackTraceElement(declaringClass, methodName, fileName,
                lineNumber);
    }

    public void destroy() {
    }
}