/**
 * 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.command.wf;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.oozie.AppType;
import org.apache.oozie.ErrorCode;
import org.apache.oozie.WorkflowActionBean;
import org.apache.oozie.WorkflowJobBean;
import org.apache.oozie.action.oozie.SubWorkflowActionExecutor;
import org.apache.oozie.client.OozieClient;
import org.apache.oozie.client.WorkflowAction;
import org.apache.oozie.client.WorkflowJob;
import org.apache.oozie.client.rest.JsonBean;
import org.apache.oozie.command.CommandException;
import org.apache.oozie.command.PreconditionException;
import org.apache.oozie.executor.jpa.JPAExecutorException;
import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor;
import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor.WorkflowActionQuery;
import org.apache.oozie.executor.jpa.BatchQueryExecutor;
import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor;
import org.apache.oozie.executor.jpa.BatchQueryExecutor.UpdateEntry;
import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor.WorkflowJobQuery;
import org.apache.oozie.service.ConfigurationService;
import org.apache.oozie.service.DagXLogInfoService;
import org.apache.oozie.service.HadoopAccessorException;
import org.apache.oozie.service.HadoopAccessorService;
import org.apache.oozie.service.Services;
import org.apache.oozie.service.UUIDService;
import org.apache.oozie.service.WorkflowAppService;
import org.apache.oozie.service.WorkflowStoreService;
import org.apache.oozie.sla.SLAOperations;
import org.apache.oozie.sla.service.SLAService;
import org.apache.oozie.util.ConfigUtils;
import org.apache.oozie.util.ELEvaluator;
import org.apache.oozie.util.ELUtils;
import org.apache.oozie.util.InstrumentUtils;
import org.apache.oozie.util.LogUtils;
import org.apache.oozie.util.ParamChecker;
import org.apache.oozie.util.PropertiesUtils;
import org.apache.oozie.util.XConfiguration;
import org.apache.oozie.util.XLog;
import org.apache.oozie.util.XmlUtils;
import org.apache.oozie.workflow.WorkflowApp;
import org.apache.oozie.workflow.WorkflowException;
import org.apache.oozie.workflow.WorkflowInstance;
import org.apache.oozie.workflow.WorkflowLib;
import org.apache.oozie.workflow.lite.NodeHandler;
import org.jdom.Element;
import org.jdom.JDOMException;

/**
 * This is a RerunXCommand which is used for rerunn.
 *
 */
public class ReRunXCommand extends WorkflowXCommand<Void> {
    private final String jobId;
    private Configuration conf;
    private final Set<String> nodesToSkip = new HashSet<String>();
    public static final String TO_SKIP = "TO_SKIP";
    private WorkflowJobBean wfBean;
    private List<WorkflowActionBean> actions;
    private List<UpdateEntry> updateList = new ArrayList<UpdateEntry>();
    private List<JsonBean> deleteList = new ArrayList<JsonBean>();

    private static final Set<String> DISALLOWED_USER_PROPERTIES = new HashSet<String>();
    public static final String DISABLE_CHILD_RERUN = "oozie.wf.rerun.disablechild";

    static {
        String[] badUserProps = { PropertiesUtils.DAYS, PropertiesUtils.HOURS, PropertiesUtils.MINUTES,
                PropertiesUtils.KB, PropertiesUtils.MB, PropertiesUtils.GB, PropertiesUtils.TB, PropertiesUtils.PB,
                PropertiesUtils.RECORDS, PropertiesUtils.MAP_IN, PropertiesUtils.MAP_OUT, PropertiesUtils.REDUCE_IN,
                PropertiesUtils.REDUCE_OUT, PropertiesUtils.GROUPS };
        PropertiesUtils.createPropertySet(badUserProps, DISALLOWED_USER_PROPERTIES);
    }

    public ReRunXCommand(String jobId, Configuration conf) {
        super("rerun", "rerun", 1);
        this.jobId = ParamChecker.notEmpty(jobId, "jobId");
        this.conf = Objects.requireNonNull(conf, "conf cannot be null");
    }

    @Override
    protected void setLogInfo() {
        LogUtils.setLogInfo(jobId);
    }

    @Override
    protected Void execute() throws CommandException {
        setupReRun();
        startWorkflow(jobId);
        return null;
    }

    private void startWorkflow(String jobId) throws CommandException {
        new StartXCommand(jobId).call();
    }

    private void setupReRun() throws CommandException {
        InstrumentUtils.incrJobCounter(getName(), 1, getInstrumentation());
        LogUtils.setLogInfo(wfBean);
        WorkflowInstance oldWfInstance = this.wfBean.getWorkflowInstance();
        WorkflowInstance newWfInstance;
        String appPath = null;

        WorkflowAppService wps = Services.get().get(WorkflowAppService.class);
        try {
            XLog.Info.get().setParameter(DagXLogInfoService.TOKEN, conf.get(OozieClient.LOG_TOKEN));
            WorkflowApp app = wps.parseDef(conf, null);
            XConfiguration protoActionConf = wps.createProtoActionConf(conf, true);
            WorkflowLib workflowLib = Services.get().get(WorkflowStoreService.class).getWorkflowLibWithNoDB();

            appPath = conf.get(OozieClient.APP_PATH);
            URI uri = new URI(appPath);
            HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
            Configuration fsConf = has.createConfiguration(uri.getAuthority());
            FileSystem fs = has.createFileSystem(wfBean.getUser(), uri, fsConf);

            Path configDefault = null;
            // app path could be a directory
            Path path = new Path(uri.getPath());
            if (!fs.isFile(path)) {
                configDefault = new Path(path, SubmitXCommand.CONFIG_DEFAULT);
            }
            else {
                configDefault = new Path(path.getParent(), SubmitXCommand.CONFIG_DEFAULT);
            }

            if (fs.exists(configDefault)) {
                Configuration defaultConf = new XConfiguration(fs.open(configDefault));
                PropertiesUtils.checkDisallowedProperties(defaultConf, DISALLOWED_USER_PROPERTIES);
                PropertiesUtils.checkDefaultDisallowedProperties(defaultConf);
                XConfiguration.injectDefaults(defaultConf, conf);
            }

            PropertiesUtils.checkDisallowedProperties(conf, DISALLOWED_USER_PROPERTIES);

            // Resolving all variables in the job properties. This ensures the Hadoop Configuration semantics are
            // preserved. The Configuration.get function within XConfiguration.resolve() works recursively to get the
            // final value corresponding to a key in the map Resetting the conf to contain all the resolved values is
            // necessary to ensure propagation of Oozie properties to Hadoop calls downstream
            conf = ((XConfiguration) conf).resolve();

            try {
                newWfInstance = workflowLib.createInstance(app, conf, jobId);
            }
            catch (WorkflowException e) {
                throw new CommandException(e);
            }
            String appName = ELUtils.resolveAppName(app.getName(), conf);
            if (SLAService.isEnabled()) {
                Element wfElem = XmlUtils.parseXml(app.getDefinition());
                ELEvaluator evalSla = SubmitXCommand.createELEvaluatorForGroup(conf, "wf-sla-submit");
                Element eSla = XmlUtils.getSLAElement(wfElem);
                String jobSlaXml = null;
                if (eSla != null) {
                    jobSlaXml = SubmitXCommand.resolveSla(eSla, evalSla);
                }
                writeSLARegistration(wfElem, jobSlaXml, newWfInstance.getId(),
                        conf.get(SubWorkflowActionExecutor.PARENT_ID), conf.get(OozieClient.USER_NAME), appName,
                        evalSla);
            }
            wfBean.setAppName(appName);
            wfBean.setProtoActionConf(protoActionConf.toXmlString());
        }
        catch (WorkflowException ex) {
            throw new CommandException(ex);
        }
        catch (IOException ex) {
            throw new CommandException(ErrorCode.E0803, ex.getMessage(), ex);
        }
        catch (HadoopAccessorException ex) {
            throw new CommandException(ex);
        }
        catch (URISyntaxException ex) {
            throw new CommandException(ErrorCode.E0711, appPath, ex.getMessage(), ex);
        }
        catch (Exception ex) {
            throw new CommandException(ErrorCode.E1007, ex.getMessage(), ex);
        }

        for (int i = 0; i < actions.size(); i++) {
            // Skipping to delete the sub workflow when rerun failed node option has been provided. As same
            // action will be used to rerun the job.
            if (!nodesToSkip.contains(actions.get(i).getName()) &&
                    !(conf.getBoolean(OozieClient.RERUN_FAIL_NODES, false) &&
                    SubWorkflowActionExecutor.ACTION_TYPE.equals(actions.get(i).getType()))) {
                deleteList.add(actions.get(i));
                LOG.info("Deleting Action[{0}] for re-run", actions.get(i).getId());
            }
            else {
                copyActionData(newWfInstance, oldWfInstance);
            }
        }

        wfBean.setAppPath(conf.get(OozieClient.APP_PATH));
        wfBean.setConf(XmlUtils.prettyPrint(conf).toString());
        wfBean.setLogToken(conf.get(OozieClient.LOG_TOKEN, ""));
        wfBean.setUser(conf.get(OozieClient.USER_NAME));
        String group = ConfigUtils.getWithDeprecatedCheck(conf, OozieClient.JOB_ACL, OozieClient.GROUP_NAME, null);
        wfBean.setGroup(group);
        wfBean.setExternalId(conf.get(OozieClient.EXTERNAL_ID));
        wfBean.setEndTime(null);
        wfBean.setRun(wfBean.getRun() + 1);
        wfBean.setStatus(WorkflowJob.Status.PREP);
        wfBean.setWorkflowInstance(newWfInstance);

        try {
            wfBean.setLastModifiedTime(new Date());
            updateList.add(new UpdateEntry<WorkflowJobQuery>(WorkflowJobQuery.UPDATE_WORKFLOW_RERUN, wfBean));
            // call JPAExecutor to do the bulk writes
            BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(null, updateList, deleteList);
        }
        catch (JPAExecutorException je) {
            throw new CommandException(je);
        }
        finally {
            updateParentIfNecessary(wfBean);
        }

    }

    @SuppressWarnings("unchecked")
	private void writeSLARegistration(Element wfElem, String jobSlaXml, String id, String parentId, String user,
            String appName, ELEvaluator evalSla) throws JDOMException, CommandException {
        if (jobSlaXml != null && jobSlaXml.length() > 0) {
            Element eSla = XmlUtils.parseXml(jobSlaXml);
            // insert into new table
            SLAOperations.createSlaRegistrationEvent(eSla, jobId, parentId, AppType.WORKFLOW_JOB, user, appName, LOG,
                    true);
        }
        // Add sla for wf actions
        for (Element action : (List<Element>) wfElem.getChildren("action", wfElem.getNamespace())) {
            Element actionSla = XmlUtils.getSLAElement(action);
            if (actionSla != null) {
                String actionSlaXml = SubmitXCommand.resolveSla(actionSla, evalSla);
                actionSla = XmlUtils.parseXml(actionSlaXml);
                if (!nodesToSkip.contains(action.getAttributeValue("name"))) {
                    String actionId = Services.get().get(UUIDService.class)
                            .generateChildId(jobId, action.getAttributeValue("name") + "");
                    SLAOperations.createSlaRegistrationEvent(actionSla, actionId, jobId, AppType.WORKFLOW_ACTION, user,
                            appName, LOG, true);
                }
            }
        }

    }

    /**
     * Loading the Wfjob and workflow actions. Parses the config and adds the nodes that are to be skipped to the
     * skipped node list
     *
     * @throws CommandException if loading state fails
     */
    @Override
    protected void eagerLoadState() throws CommandException {
        try {
            this.wfBean = WorkflowJobQueryExecutor.getInstance().get(WorkflowJobQuery.GET_WORKFLOW_STATUS, this.jobId);
            this.actions = WorkflowActionQueryExecutor.getInstance().getList(
                    WorkflowActionQuery.GET_ACTIONS_FOR_WORKFLOW_RERUN, this.jobId);

            if (conf != null) {
                if (conf.getBoolean(OozieClient.RERUN_FAIL_NODES, false) == false) { //Rerun with skipNodes
                    Collection<String> skipNodes = conf.getStringCollection(OozieClient.RERUN_SKIP_NODES);
                    for (String str : skipNodes) {
                        // trimming is required
                        nodesToSkip.add(str.trim());
                    }
                    LOG.debug("Skipnode size :" + nodesToSkip.size());
                }
                else {
                    for (WorkflowActionBean action : actions) { // Rerun from failed nodes
                        if (action.getStatus() == WorkflowAction.Status.OK) {
                            nodesToSkip.add(action.getName());
                        }
                    }
                    LOG.debug("Skipnode size are to rerun from FAIL nodes :" + nodesToSkip.size());
                }
                StringBuilder tmp = new StringBuilder();
                for (String node : nodesToSkip) {
                    tmp.append(node).append(",");
                }
                LOG.debug("SkipNode List :" + tmp);
            }
        }
        catch (Exception ex) {
            throw new CommandException(ErrorCode.E0603, ex.getMessage(), ex);
        }
    }

    /**
     * Checks the pre-conditions that are required for workflow to recover - Last run of Workflow should be completed -
     * The nodes that are to be skipped are to be completed successfully in the base run.
     *
     * @throws CommandException if verification fails
     * @throws PreconditionException On failure of pre-conditions
     */
    @Override
    protected void eagerVerifyPrecondition() throws CommandException, PreconditionException {
        // Throwing error if parent exist and same workflow trying to rerun, when running child workflow disabled
        // through conf.
        if (wfBean.getParentId() != null && !conf.getBoolean(SubWorkflowActionExecutor.SUBWORKFLOW_RERUN, false)
                && ConfigurationService.getBoolean(DISABLE_CHILD_RERUN)) {
            throw new CommandException(ErrorCode.E0755, " Rerun is not allowed through child workflow, please" +
                    " re-run through the parent " + wfBean.getParentId());
        }

        if (!(wfBean.getStatus().equals(WorkflowJob.Status.FAILED)
                || wfBean.getStatus().equals(WorkflowJob.Status.KILLED) || wfBean.getStatus().equals(
                        WorkflowJob.Status.SUCCEEDED))) {
            throw new CommandException(ErrorCode.E0805, wfBean.getStatus());
        }
        Set<String> unmachedNodes = new HashSet<String>(nodesToSkip);
        for (WorkflowActionBean action : actions) {
            if (nodesToSkip.contains(action.getName())) {
                if (!action.getStatus().equals(WorkflowAction.Status.OK)
                        && !action.getStatus().equals(WorkflowAction.Status.ERROR)) {
                    throw new CommandException(ErrorCode.E0806, action.getName());
                }
                unmachedNodes.remove(action.getName());
            }
        }
        if (unmachedNodes.size() > 0) {
            StringBuilder sb = new StringBuilder();
            String separator = "";
            for (String s : unmachedNodes) {
                sb.append(separator).append(s);
                separator = ",";
            }
            throw new CommandException(ErrorCode.E0807, sb);
        }
    }

    /**
     * Copys the variables for skipped nodes from the old wfInstance to new one.
     *
     * @param newWfInstance : Source WF instance object
     * @param oldWfInstance : Update WF instance
     */
    private void copyActionData(WorkflowInstance newWfInstance, WorkflowInstance oldWfInstance) {
        Map<String, String> oldVars = new HashMap<String, String>();
        Map<String, String> newVars = new HashMap<String, String>();
        oldVars = oldWfInstance.getAllVars();
        for (String var : oldVars.keySet()) {
            String actionName = var.split(WorkflowInstance.NODE_VAR_SEPARATOR)[0];
            if (nodesToSkip.contains(actionName)) {
                newVars.put(var, oldVars.get(var));
            }
        }
        for (String node : nodesToSkip) {
            // Setting the TO_SKIP variable to true. This will be used by
            // SignalCommand and LiteNodeHandler to skip the action.
            newVars.put(node + WorkflowInstance.NODE_VAR_SEPARATOR + TO_SKIP, "true");
            String visitedFlag = NodeHandler.getLoopFlag(node);
            // Removing the visited flag so that the action won't be considered
            // a loop.
            if (newVars.containsKey(visitedFlag)) {
                newVars.remove(visitedFlag);
            }
        }
        newWfInstance.setAllVars(newVars);
    }

    @Override
    public String getEntityKey() {
        return this.jobId;
    }

    @Override
    protected boolean isLockRequired() {
        return true;
    }

    @Override
    protected void loadState() throws CommandException {
        try {
            this.wfBean = WorkflowJobQueryExecutor.getInstance().get(WorkflowJobQuery.GET_WORKFLOW_RERUN, this.jobId);
            this.actions = WorkflowActionQueryExecutor.getInstance().getList(
                    WorkflowActionQuery.GET_ACTIONS_FOR_WORKFLOW_RERUN, this.jobId);
        }
        catch (JPAExecutorException jpe) {
            throw new CommandException(jpe);
        }
    }

    @Override
    protected void verifyPrecondition() throws CommandException, PreconditionException {
        eagerVerifyPrecondition();
    }
}