/**
 * 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;

import java.io.IOException;
import java.io.StringReader;
import java.io.Writer;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.StringTokenizer;
import com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.conf.Configuration;
import org.apache.oozie.client.CoordinatorJob;
import org.apache.oozie.client.OozieClient;
import org.apache.oozie.client.WorkflowJob;
import org.apache.oozie.command.CommandException;
import org.apache.oozie.command.OperationType;
import org.apache.oozie.command.wf.BulkWorkflowXCommand;
import org.apache.oozie.command.wf.CompletedActionXCommand;
import org.apache.oozie.command.wf.DefinitionXCommand;
import org.apache.oozie.command.wf.ExternalIdXCommand;
import org.apache.oozie.command.wf.JobXCommand;
import org.apache.oozie.command.wf.JobsXCommand;
import org.apache.oozie.command.wf.KillXCommand;
import org.apache.oozie.command.wf.ReRunXCommand;
import org.apache.oozie.command.wf.ResumeXCommand;
import org.apache.oozie.command.wf.StartXCommand;
import org.apache.oozie.command.wf.SubmitHiveXCommand;
import org.apache.oozie.command.wf.SubmitHttpXCommand;
import org.apache.oozie.command.wf.SubmitMRXCommand;
import org.apache.oozie.command.wf.SubmitPigXCommand;
import org.apache.oozie.command.wf.SubmitSqoopXCommand;
import org.apache.oozie.command.wf.SubmitXCommand;
import org.apache.oozie.command.wf.SuspendXCommand;
import org.apache.oozie.command.wf.WorkflowActionInfoXCommand;
import org.apache.oozie.command.wf.WorkflowActionRetryInfoXCommand;
import org.apache.oozie.executor.jpa.JPAExecutorException;
import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor;
import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor.WorkflowJobQuery;
import org.apache.oozie.service.CallableQueueService;
import org.apache.oozie.service.ConfigurationService;
import org.apache.oozie.service.DagXLogInfoService;
import org.apache.oozie.service.Services;
import org.apache.oozie.service.XLogService;
import org.apache.oozie.service.XLogStreamingService;
import org.apache.oozie.util.ParamChecker;
import org.apache.oozie.util.XCallable;
import org.apache.oozie.util.XConfiguration;
import org.apache.oozie.util.XLog;
import org.apache.oozie.util.XLogStreamer;

/**
 * The DagEngine provides all the DAG engine functionality for WS calls.
 */
public class DagEngine extends BaseEngine {

    private static final int HIGH_PRIORITY = 2;
    private static XLog LOG = XLog.getLog(DagEngine.class);

    /**
     * Create a system Dag engine, with no user and no group.
     */
    public DagEngine() {

    }

    /**
     * Create a Dag engine to perform operations on behave of a user.
     *
     * @param user user name.
     */
    public DagEngine(String user) {
        this();

        this.user = ParamChecker.notEmpty(user, "user");
    }

    /**
     * Submit a workflow job. <p> It validates configuration properties.
     *
     * @param conf job configuration.
     * @param startJob indicates if the job should be started or not.
     * @return the job Id.
     * @throws DagEngineException thrown if the job could not be created.
     */
    @Override
    public String submitJob(Configuration conf, boolean startJob) throws DagEngineException {
        validateSubmitConfiguration(conf);

        try {
            String jobId;
            SubmitXCommand submit = new SubmitXCommand(conf);
            jobId = submit.call();
            if (startJob) {
                start(jobId);
            }
            return jobId;
        }
        catch (CommandException ex) {
            throw new DagEngineException(ex);
        }
    }

    /**
     * Submit a workflow through a coordinator. It validates configuration properties.
     * @param conf job conf
     * @param parentId parent of workflow
     * @return jobId returns jobId of the launched workflow
     * @throws DagEngineException if the job can't be submitted
     */
    public String submitJobFromCoordinator(Configuration conf, String parentId) throws DagEngineException {
        validateSubmitConfiguration(conf);
        try {
            String jobId;
            SubmitXCommand submit = new SubmitXCommand(conf, parentId);
            jobId = submit.call();
            start(jobId);
            return jobId;
        }
        catch (CommandException ex) {
            throw new DagEngineException(ex);
        }
    }

    /**
     * Submit a pig/hive/mapreduce job through HTTP.
     * <p>
     * It validates configuration properties.
     *
     * @param conf job configuration.
     * @param jobType job type - can be "pig", "hive", "sqoop" or "mapreduce".
     * @return the job Id.
     * @throws DagEngineException thrown if the job could not be created.
     */
    public String submitHttpJob(Configuration conf, String jobType) throws DagEngineException {
        validateSubmitConfiguration(conf);

        try {
            String jobId;
            SubmitHttpXCommand submit = null;
            if (jobType.equals("pig")) {
                submit = new SubmitPigXCommand(conf);
            }
            else if (jobType.equals("mapreduce")) {
                submit = new SubmitMRXCommand(conf);
            }
            else if (jobType.equals("hive")) {
                submit = new SubmitHiveXCommand(conf);
            }
            else if (jobType.equals("sqoop")) {
                submit = new SubmitSqoopXCommand(conf);
            }

            jobId = submit.call();
            start(jobId);
            return jobId;
        }
        catch (CommandException ex) {
            throw new DagEngineException(ex);
        }
    }

    private void validateSubmitConfiguration(Configuration conf) throws DagEngineException {
        if (conf.get(OozieClient.APP_PATH) == null) {
            throw new DagEngineException(ErrorCode.E0401, OozieClient.APP_PATH);
        }
    }

    /**
     * Start a job.
     *
     * @param jobId job Id.
     * @throws DagEngineException thrown if the job could not be started.
     */
    @Override
    public void start(String jobId) throws DagEngineException {
        // Changing to synchronous call from asynchronous queuing to prevent the
        // loss of command if the queue is full or the queue is lost in case of
        // failure.
        try {
                new StartXCommand(jobId).call();
        }
        catch (CommandException e) {
            throw new DagEngineException(e);
        }
    }

    /**
     * Resume a job.
     *
     * @param jobId job Id.
     * @throws DagEngineException thrown if the job could not be resumed.
     */
    @Override
    public void resume(String jobId) throws DagEngineException {
        // Changing to synchronous call from asynchronous queuing to prevent the
        // loss of command if the queue is full or the queue is lost in case of
        // failure.
        try {
                new ResumeXCommand(jobId).call();
        }
        catch (CommandException e) {
            throw new DagEngineException(e);
        }
    }

    /**
     * Suspend a job.
     *
     * @param jobId job Id.
     * @throws DagEngineException thrown if the job could not be suspended.
     */
    @Override
    public void suspend(String jobId) throws DagEngineException {
        // Changing to synchronous call from asynchronous queuing to prevent the
        // loss of command if the queue is full or the queue is lost in case of
        // failure.
        try {
			new SuspendXCommand(jobId).call();
        }
        catch (CommandException e) {
            throw new DagEngineException(e);
        }
    }

    /**
     * Kill a job.
     *
     * @param jobId job Id.
     * @throws DagEngineException thrown if the job could not be killed.
     */
    @Override
    public void kill(String jobId) throws DagEngineException {
        // Changing to synchronous call from asynchronous queuing to prevent the
        // loss of command if the queue is full or the queue is lost in case of
        // failure.
        try {
			new KillXCommand(jobId).call();
			LOG.info("User " + user + " killed the WF job " + jobId);
        }
        catch (CommandException e) {
            throw new DagEngineException(e);
        }
    }

    @Override
    public void change(String jobId, String changeValue) throws DagEngineException {
        // This code should not be reached.
        throw new DagEngineException(ErrorCode.E1017);
    }

    /**
     * Rerun a job.
     *
     * @param jobId job Id to rerun.
     * @param conf configuration information for the rerun.
     * @throws DagEngineException thrown if the job could not be rerun.
     */
    @Override
    public void reRun(String jobId, Configuration conf) throws DagEngineException {
        try {
            WorkflowJobBean wfBean = WorkflowJobQueryExecutor.getInstance().get(WorkflowJobQuery.GET_WORKFLOW, jobId);
            Configuration wfConf = new XConfiguration(new StringReader(wfBean.getConf()));
            XConfiguration.copy(conf, wfConf);
            validateReRunConfiguration(wfConf);
            new ReRunXCommand(jobId, wfConf).call();
        }
        catch (CommandException ex) {
            throw new DagEngineException(ex);
        }
        catch (JPAExecutorException ex) {
            throw new DagEngineException(ex);
        }
        catch (IOException ex) {
            throw new DagEngineException(ErrorCode.E0803, ex.getMessage());
        }
    }

    @VisibleForTesting
    protected void validateReRunConfiguration(Configuration conf) throws DagEngineException {
        if (conf.get(OozieClient.APP_PATH) == null) {
            throw new DagEngineException(ErrorCode.E0401, OozieClient.APP_PATH);
        }
        boolean rerunFailNodes = ConfigurationService.getBoolean(conf, OozieClient.RERUN_FAIL_NODES);
        String skipNodes = conf.get(OozieClient.RERUN_SKIP_NODES);

        if (rerunFailNodes && skipNodes != null) {
            throw new DagEngineException(ErrorCode.E0404, OozieClient.RERUN_SKIP_NODES + " OR "
                    + OozieClient.RERUN_FAIL_NODES);
        }
    }

    /**
     * Process an action callback.
     *
     * @param actionId the action Id.
     * @param externalStatus the action external status.
     * @param actionData the action output data, <code>null</code> if none.
     * @throws DagEngineException thrown if the callback could not be processed.
     */
    public void processCallback(String actionId, String externalStatus, Properties actionData)
            throws DagEngineException {
        XLog.Info.get().clearParameter(XLogService.GROUP);
        XLog.Info.get().clearParameter(XLogService.USER);
        XCallable<Void> command = null;

		command = new CompletedActionXCommand(actionId, externalStatus,
				actionData, HIGH_PRIORITY);
        if (!Services.get().get(CallableQueueService.class).queue(command)) {
            LOG.warn(XLog.OPS, "queue is full or system is in SAFEMODE, ignoring callback");
        }
    }

    /**
     * Return the info about a job.
     *
     * @param jobId job Id.
     * @return the workflow job info.
     * @throws DagEngineException thrown if the job info could not be obtained.
     */
    @Override
    public WorkflowJob getJob(String jobId) throws DagEngineException {
        try {
                return new JobXCommand(jobId).call();
        }
        catch (CommandException ex) {
            throw new DagEngineException(ex);
        }
    }

    /**
     * Return the info about a job with actions subset.
     *
     * @param jobId job Id
     * @param start starting from this index in the list of actions belonging to the job
     * @param length number of actions to be returned
     * @return the workflow job info.
     * @throws DagEngineException thrown if the job info could not be obtained.
     */
    @Override
    public WorkflowJob getJob(String jobId, int start, int length) throws DagEngineException {
        try {
			return new JobXCommand(jobId, start, length).call();
        }
        catch (CommandException ex) {
            throw new DagEngineException(ex);
        }
    }

    /**
     * Return the a job definition.
     *
     * @param jobId job Id.
     * @return the job definition.
     * @throws DagEngineException thrown if the job definition could no be obtained.
     */
    @Override
    public String getDefinition(String jobId) throws DagEngineException {
        try {
            return new DefinitionXCommand(jobId).call();
        }
        catch (CommandException ex) {
            throw new DagEngineException(ex);
        }
    }

    @Override
    protected void streamJobLog(XLogStreamer logStreamer, String jobId, Writer writer)
            throws IOException, DagEngineException {
        logStreamer.getXLogFilter().setParameter(DagXLogInfoService.JOB, jobId);
        WorkflowJob job = getJob(jobId);
        Date lastTime = job.getEndTime();
        if (lastTime == null) {
            lastTime = job.getLastModifiedTime();
        }
        Services.get().get(XLogStreamingService.class).streamLog(logStreamer, job.getCreatedTime(), lastTime, writer);
    }

    private static final Set<String> FILTER_NAMES = new HashSet<String>();

    static {
        FILTER_NAMES.add(OozieClient.FILTER_TEXT);
        FILTER_NAMES.add(OozieClient.FILTER_USER);
        FILTER_NAMES.add(OozieClient.FILTER_NAME);
        FILTER_NAMES.add(OozieClient.FILTER_GROUP);
        FILTER_NAMES.add(OozieClient.FILTER_STATUS);
        FILTER_NAMES.add(OozieClient.FILTER_ID);
        FILTER_NAMES.add(OozieClient.FILTER_CREATED_TIME_START);
        FILTER_NAMES.add(OozieClient.FILTER_CREATED_TIME_END);
        FILTER_NAMES.add(OozieClient.FILTER_SORT_BY);
    }

    /**
     * Validate a jobs filter.
     *
     * @param filter filter to validate.
     * @return the parsed filter.
     * @throws DagEngineException thrown if the filter is invalid.
     */
    protected Map<String, List<String>> parseFilter(String filter) throws DagEngineException {
        Map<String, List<String>> map = new HashMap<String, List<String>>();
        if (filter != null) {
            StringTokenizer st = new StringTokenizer(filter, ";");
            while (st.hasMoreTokens()) {
                String token = st.nextToken();
                if (token.contains("=")) {
                    String[] pair = token.split("=");
                    if (pair.length != 2) {
                        throw new DagEngineException(ErrorCode.E0420, filter,
                                "elements must be semicolon-separated name=value pairs");
                    }
                    pair[0] = pair[0].toLowerCase();
                    if (!FILTER_NAMES.contains(pair[0])) {
                        throw new DagEngineException(ErrorCode.E0420, filter, XLog
                                .format("invalid name [{0}]", pair[0]));
                    }
                    if (pair[0].equals("status")) {
                        try {
                            WorkflowJob.Status.valueOf(pair[1]);
                        }
                        catch (IllegalArgumentException ex) {
                            throw new DagEngineException(ErrorCode.E0420, filter, XLog.format("invalid status [{0}]",
                                                                                              pair[1]));
                        }
                    }
                    List<String> list = map.get(pair[0]);
                    if (list == null) {
                        list = new ArrayList<String>();
                        map.put(pair[0], list);
                    }
                    list.add(pair[1]);
                }
                else {
                    throw new DagEngineException(ErrorCode.E0420, filter,
                            "elements must be semicolon-separated name=value pairs");
                }
            }
        }
        return map;
    }

    /**
     * Return the info about a set of jobs.
     *
     * @param filter job filter. Refer to the {@link org.apache.oozie.client.OozieClient} for the filter syntax.
     * @param start offset, base 1.
     * @param len number of jobs to return.
     * @return job info for all matching jobs, the jobs don't contain node action information.
     * @throws DagEngineException thrown if the jobs info could not be obtained.
     */
    public WorkflowsInfo getJobs(String filter, int start, int len) throws DagEngineException {
        Map<String, List<String>> filterList = parseFilter(filter);
        try {
			return new JobsXCommand(filterList, start, len).call();
        }
        catch (CommandException dce) {
            throw new DagEngineException(dce);
        }
    }

    /**
     * Return the workflow Job ID for an external ID. <p> This is reverse lookup for recovery purposes.
     *
     * @param externalId external ID provided at job submission time.
     * @return the associated workflow job ID if any, <code>null</code> if none.
     * @throws DagEngineException thrown if the lookup could not be done.
     */
    @Override
    public String getJobIdForExternalId(String externalId) throws DagEngineException {
        try {
			return new ExternalIdXCommand(externalId).call();
        }
        catch (CommandException dce) {
            throw new DagEngineException(dce);
        }
    }

    @Override
    public CoordinatorJob getCoordJob(String jobId) throws BaseEngineException {
        throw new BaseEngineException(new XException(ErrorCode.E0301, "cannot get a coordinator job from DagEngine"));
    }

    @Override
    public CoordinatorJob getCoordJob(String jobId, String filter, int start, int length, boolean desc)
            throws BaseEngineException {
        throw new BaseEngineException(new XException(ErrorCode.E0301, "cannot get a coordinator job from DagEngine"));
    }

    public WorkflowActionBean getWorkflowAction(String actionId) throws BaseEngineException {
        try {
			return new WorkflowActionInfoXCommand(actionId).call();
        }
        catch (CommandException ex) {
            throw new BaseEngineException(ex);
        }
    }

    /**
     * Gets the workflow action retries.
     *
     * @param actionId the action id
     * @return the workflow action retries
     * @throws BaseEngineException the base engine exception
     */
    public List<Map<String, String>> getWorkflowActionRetries(String actionId) throws BaseEngineException {
        try {
            return new WorkflowActionRetryInfoXCommand(actionId).call();
        }
        catch (CommandException ex) {
            throw new BaseEngineException(ex);
        }
    }

    @Override
    public String dryRunSubmit(Configuration conf) throws BaseEngineException {
        try {
            SubmitXCommand submit = new SubmitXCommand(true, conf);
            return submit.call();
        } catch (CommandException ex) {
            throw new DagEngineException(ex);
        }
    }

    /**
     * Return the status for a Job ID
     *
     * @param jobId job Id.
     * @return the job's status
     * @throws DagEngineException thrown if the job's status could not be obtained
     */
    @Override
    public String getJobStatus(String jobId) throws DagEngineException {
        try {
            WorkflowJobBean wfJob = WorkflowJobQueryExecutor.getInstance().get(WorkflowJobQuery.GET_WORKFLOW_STATUS, jobId);
            return wfJob.getStatusStr();
        }
        catch (JPAExecutorException ex) {
            throw new DagEngineException(ex);
        }
    }

    @Override
    public void enableSLAAlert(String id, String actions, String dates, String childIds) throws BaseEngineException {
        throw new BaseEngineException(new XException(ErrorCode.E0301, "Not supported for workflow"));
    }

    @Override
    public void disableSLAAlert(String id, String actions, String dates, String childIds) throws BaseEngineException {
        throw new BaseEngineException(new XException(ErrorCode.E0301, "Not supported for workflow"));
    }

    @Override
    public void changeSLA(String id, String actions, String dates, String childIds, String newParams) throws BaseEngineException {
        throw new BaseEngineException(new XException(ErrorCode.E0301, "Not supported for workflow"));
    }

    /**
     * return the jobs that've been killed
     * @param filter Jobs that satisfy the filter will be killed
     * @param start start index in the database of jobs
     * @param len maximum number of jobs that will be killed
     * @return workflowsInfo return the jobs that've been killed
     * @throws DagEngineException if the jobs could not be killed
     */
    public WorkflowsInfo killJobs(String filter, int start, int len) throws DagEngineException {
        try {
            Map<String, List<String>> filterList = parseFilter(filter);
            WorkflowsInfo workflowsInfo = new BulkWorkflowXCommand(filterList, start, len, OperationType.Kill).call();
            if (workflowsInfo == null) {
                return new WorkflowsInfo(new ArrayList<WorkflowJobBean>(), 0, 0, 0);
            }
            return workflowsInfo;
        }
        catch (CommandException ex) {
            throw new DagEngineException(ex);
        }
    }

    /**
     * return the jobs that've been suspended
     * @param filter Filter for jobs that will be suspended, can be name, user, group, status, id or combination of any
     * @param start Offset for the jobs that will be suspended
     * @param len maximum number of jobs that will be suspended
     * @return workflowsInfo return the jobs that've been suspended
     * @throws DagEngineException if the jobs cloud not be suspended
     */
    public WorkflowsInfo suspendJobs(String filter, int start, int len) throws DagEngineException {
        try {
            Map<String, List<String>> filterList = parseFilter(filter);
            WorkflowsInfo workflowsInfo = new BulkWorkflowXCommand(filterList, start, len, OperationType.Suspend).call();
            if (workflowsInfo == null) {
                return new WorkflowsInfo(new ArrayList<WorkflowJobBean>(), 0, 0, 0);
            }
            return workflowsInfo;
        }
        catch (CommandException ex) {
            throw new DagEngineException(ex);
        }
    }

    /**
     * return the jobs that've been resumed
     * @param filter Filter for jobs that will be resumed, can be name, user, group, status, id or combination of any
     * @param start Offset for the jobs that will be resumed
     * @param len maximum number of jobs that will be resumed
     * @return workflowsInfo returns the jobs that've been resumed
     * @throws DagEngineException if the jobs cloud not be resumed
     */
    public WorkflowsInfo resumeJobs(String filter, int start, int len) throws DagEngineException {
        try {
            Map<String, List<String>> filterList = parseFilter(filter);
            WorkflowsInfo workflowsInfo = new BulkWorkflowXCommand(filterList, start, len, OperationType.Resume).call();
            if (workflowsInfo == null) {
                return new WorkflowsInfo(new ArrayList<WorkflowJobBean>(), 0, 0, 0);
            }
            return workflowsInfo;
        }
        catch (CommandException ex) {
            throw new DagEngineException(ex);
        }
    }
}