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, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.oozie.command.wf;
019
020import java.io.IOException;
021import java.net.URI;
022import java.net.URISyntaxException;
023import java.util.ArrayList;
024import java.util.Collection;
025import java.util.Date;
026import java.util.HashMap;
027import java.util.HashSet;
028import java.util.List;
029import java.util.Map;
030import java.util.Set;
031
032import org.apache.hadoop.conf.Configuration;
033import org.apache.hadoop.fs.FileSystem;
034import org.apache.hadoop.fs.Path;
035import org.apache.oozie.AppType;
036import org.apache.oozie.ErrorCode;
037import org.apache.oozie.WorkflowActionBean;
038import org.apache.oozie.WorkflowJobBean;
039import org.apache.oozie.action.oozie.SubWorkflowActionExecutor;
040import org.apache.oozie.client.OozieClient;
041import org.apache.oozie.client.WorkflowAction;
042import org.apache.oozie.client.WorkflowJob;
043import org.apache.oozie.client.rest.JsonBean;
044import org.apache.oozie.command.CommandException;
045import org.apache.oozie.command.PreconditionException;
046import org.apache.oozie.executor.jpa.JPAExecutorException;
047import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor;
048import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor.WorkflowActionQuery;
049import org.apache.oozie.executor.jpa.BatchQueryExecutor;
050import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor;
051import org.apache.oozie.executor.jpa.BatchQueryExecutor.UpdateEntry;
052import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor.WorkflowJobQuery;
053import org.apache.oozie.service.DagXLogInfoService;
054import org.apache.oozie.service.HadoopAccessorException;
055import org.apache.oozie.service.HadoopAccessorService;
056import org.apache.oozie.service.Services;
057import org.apache.oozie.service.UUIDService;
058import org.apache.oozie.service.WorkflowAppService;
059import org.apache.oozie.service.WorkflowStoreService;
060import org.apache.oozie.sla.SLAOperations;
061import org.apache.oozie.sla.service.SLAService;
062import org.apache.oozie.util.ConfigUtils;
063import org.apache.oozie.util.ELEvaluator;
064import org.apache.oozie.util.ELUtils;
065import org.apache.oozie.util.InstrumentUtils;
066import org.apache.oozie.util.LogUtils;
067import org.apache.oozie.util.ParamChecker;
068import org.apache.oozie.util.PropertiesUtils;
069import org.apache.oozie.util.XConfiguration;
070import org.apache.oozie.util.XLog;
071import org.apache.oozie.util.XmlUtils;
072import org.apache.oozie.workflow.WorkflowApp;
073import org.apache.oozie.workflow.WorkflowException;
074import org.apache.oozie.workflow.WorkflowInstance;
075import org.apache.oozie.workflow.WorkflowLib;
076import org.apache.oozie.workflow.lite.NodeHandler;
077import org.jdom.Element;
078import org.jdom.JDOMException;
079
080/**
081 * This is a RerunXCommand which is used for rerunn.
082 *
083 */
084public class ReRunXCommand extends WorkflowXCommand<Void> {
085    private final String jobId;
086    private Configuration conf;
087    private final Set<String> nodesToSkip = new HashSet<String>();
088    public static final String TO_SKIP = "TO_SKIP";
089    private WorkflowJobBean wfBean;
090    private List<WorkflowActionBean> actions;
091    private List<UpdateEntry> updateList = new ArrayList<UpdateEntry>();
092    private List<JsonBean> deleteList = new ArrayList<JsonBean>();
093
094    private static final Set<String> DISALLOWED_DEFAULT_PROPERTIES = new HashSet<String>();
095    private static final Set<String> DISALLOWED_USER_PROPERTIES = new HashSet<String>();
096
097    static {
098        String[] badUserProps = { PropertiesUtils.DAYS, PropertiesUtils.HOURS, PropertiesUtils.MINUTES,
099                PropertiesUtils.KB, PropertiesUtils.MB, PropertiesUtils.GB, PropertiesUtils.TB, PropertiesUtils.PB,
100                PropertiesUtils.RECORDS, PropertiesUtils.MAP_IN, PropertiesUtils.MAP_OUT, PropertiesUtils.REDUCE_IN,
101                PropertiesUtils.REDUCE_OUT, PropertiesUtils.GROUPS };
102        PropertiesUtils.createPropertySet(badUserProps, DISALLOWED_USER_PROPERTIES);
103
104        String[] badDefaultProps = { PropertiesUtils.HADOOP_USER};
105        PropertiesUtils.createPropertySet(badUserProps, DISALLOWED_DEFAULT_PROPERTIES);
106        PropertiesUtils.createPropertySet(badDefaultProps, DISALLOWED_DEFAULT_PROPERTIES);
107    }
108
109    public ReRunXCommand(String jobId, Configuration conf) {
110        super("rerun", "rerun", 1);
111        this.jobId = ParamChecker.notEmpty(jobId, "jobId");
112        this.conf = ParamChecker.notNull(conf, "conf");
113    }
114
115    /* (non-Javadoc)
116     * @see org.apache.oozie.command.XCommand#execute()
117     */
118    @Override
119    protected Void execute() throws CommandException {
120        setupReRun();
121        startWorkflow(jobId);
122        return null;
123    }
124
125    private void startWorkflow(String jobId) throws CommandException {
126        new StartXCommand(jobId).call();
127    }
128
129    private void setupReRun() throws CommandException {
130        InstrumentUtils.incrJobCounter(getName(), 1, getInstrumentation());
131        LogUtils.setLogInfo(wfBean, logInfo);
132        WorkflowInstance oldWfInstance = this.wfBean.getWorkflowInstance();
133        WorkflowInstance newWfInstance;
134        String appPath = null;
135
136        WorkflowAppService wps = Services.get().get(WorkflowAppService.class);
137        try {
138            XLog.Info.get().setParameter(DagXLogInfoService.TOKEN, conf.get(OozieClient.LOG_TOKEN));
139            WorkflowApp app = wps.parseDef(conf, null);
140            XConfiguration protoActionConf = wps.createProtoActionConf(conf, true);
141            WorkflowLib workflowLib = Services.get().get(WorkflowStoreService.class).getWorkflowLibWithNoDB();
142
143            appPath = conf.get(OozieClient.APP_PATH);
144            URI uri = new URI(appPath);
145            HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
146            Configuration fsConf = has.createJobConf(uri.getAuthority());
147            FileSystem fs = has.createFileSystem(wfBean.getUser(), uri, fsConf);
148
149            Path configDefault = null;
150            // app path could be a directory
151            Path path = new Path(uri.getPath());
152            if (!fs.isFile(path)) {
153                configDefault = new Path(path, SubmitXCommand.CONFIG_DEFAULT);
154            }
155            else {
156                configDefault = new Path(path.getParent(), SubmitXCommand.CONFIG_DEFAULT);
157            }
158
159            if (fs.exists(configDefault)) {
160                Configuration defaultConf = new XConfiguration(fs.open(configDefault));
161                PropertiesUtils.checkDisallowedProperties(defaultConf, DISALLOWED_DEFAULT_PROPERTIES);
162                XConfiguration.injectDefaults(defaultConf, conf);
163            }
164
165            PropertiesUtils.checkDisallowedProperties(conf, DISALLOWED_USER_PROPERTIES);
166
167            // Resolving all variables in the job properties. This ensures the Hadoop Configuration semantics are
168            // preserved. The Configuration.get function within XConfiguration.resolve() works recursively to get the
169            // final value corresponding to a key in the map Resetting the conf to contain all the resolved values is
170            // necessary to ensure propagation of Oozie properties to Hadoop calls downstream
171            conf = ((XConfiguration) conf).resolve();
172
173            // Prepare the action endtimes map
174            Map<String, Date> actionEndTimes = new HashMap<String, Date>();
175            for (WorkflowActionBean action : actions) {
176                if (action.getEndTime() != null) {
177                    actionEndTimes.put(action.getName(), action.getEndTime());
178                }
179            }
180
181            try {
182                newWfInstance = workflowLib.createInstance(app, conf, jobId, actionEndTimes);
183            }
184            catch (WorkflowException e) {
185                throw new CommandException(e);
186            }
187            String appName = ELUtils.resolveAppName(app.getName(), conf);
188            if (SLAService.isEnabled()) {
189                Element wfElem = XmlUtils.parseXml(app.getDefinition());
190                ELEvaluator evalSla = SubmitXCommand.createELEvaluatorForGroup(conf, "wf-sla-submit");
191                Element eSla = XmlUtils.getSLAElement(wfElem);
192                String jobSlaXml = null;
193                if (eSla != null) {
194                    jobSlaXml = SubmitXCommand.resolveSla(eSla, evalSla);
195                }
196                writeSLARegistration(wfElem, jobSlaXml, newWfInstance.getId(),
197                        conf.get(SubWorkflowActionExecutor.PARENT_ID), conf.get(OozieClient.USER_NAME), appName,
198                        evalSla);
199            }
200            wfBean.setAppName(appName);
201            wfBean.setProtoActionConf(protoActionConf.toXmlString());
202        }
203        catch (WorkflowException ex) {
204            throw new CommandException(ex);
205        }
206        catch (IOException ex) {
207            throw new CommandException(ErrorCode.E0803, ex.getMessage(), ex);
208        }
209        catch (HadoopAccessorException ex) {
210            throw new CommandException(ex);
211        }
212        catch (URISyntaxException ex) {
213            throw new CommandException(ErrorCode.E0711, appPath, ex.getMessage(), ex);
214        }
215        catch (Exception ex) {
216            throw new CommandException(ErrorCode.E1007, ex.getMessage(), ex);
217        }
218
219        for (int i = 0; i < actions.size(); i++) {
220            if (!nodesToSkip.contains(actions.get(i).getName())) {
221                deleteList.add(actions.get(i));
222                LOG.info("Deleting Action[{0}] for re-run", actions.get(i).getId());
223            }
224            else {
225                copyActionData(newWfInstance, oldWfInstance);
226            }
227        }
228
229        wfBean.setAppPath(conf.get(OozieClient.APP_PATH));
230        wfBean.setConf(XmlUtils.prettyPrint(conf).toString());
231        wfBean.setLogToken(conf.get(OozieClient.LOG_TOKEN, ""));
232        wfBean.setUser(conf.get(OozieClient.USER_NAME));
233        String group = ConfigUtils.getWithDeprecatedCheck(conf, OozieClient.JOB_ACL, OozieClient.GROUP_NAME, null);
234        wfBean.setGroup(group);
235        wfBean.setExternalId(conf.get(OozieClient.EXTERNAL_ID));
236        wfBean.setEndTime(null);
237        wfBean.setRun(wfBean.getRun() + 1);
238        wfBean.setStatus(WorkflowJob.Status.PREP);
239        wfBean.setWorkflowInstance(newWfInstance);
240
241        try {
242            wfBean.setLastModifiedTime(new Date());
243            updateList.add(new UpdateEntry<WorkflowJobQuery>(WorkflowJobQuery.UPDATE_WORKFLOW_RERUN, wfBean));
244            // call JPAExecutor to do the bulk writes
245            BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(null, updateList, deleteList);
246        }
247        catch (JPAExecutorException je) {
248            throw new CommandException(je);
249        }
250
251    }
252
253    @SuppressWarnings("unchecked")
254        private void writeSLARegistration(Element wfElem, String jobSlaXml, String id, String parentId, String user,
255            String appName, ELEvaluator evalSla) throws JDOMException, CommandException {
256        if (jobSlaXml != null && jobSlaXml.length() > 0) {
257            Element eSla = XmlUtils.parseXml(jobSlaXml);
258            // insert into new table
259            SLAOperations.createSlaRegistrationEvent(eSla, jobId, parentId, AppType.WORKFLOW_JOB, user, appName, LOG,
260                    true);
261        }
262        // Add sla for wf actions
263        for (Element action : (List<Element>) wfElem.getChildren("action", wfElem.getNamespace())) {
264            Element actionSla = XmlUtils.getSLAElement(action);
265            if (actionSla != null) {
266                String actionSlaXml = SubmitXCommand.resolveSla(actionSla, evalSla);
267                actionSla = XmlUtils.parseXml(actionSlaXml);
268                if (!nodesToSkip.contains(action.getAttributeValue("name"))) {
269                    String actionId = Services.get().get(UUIDService.class)
270                            .generateChildId(jobId, action.getAttributeValue("name") + "");
271                    SLAOperations.createSlaRegistrationEvent(actionSla, actionId, jobId, AppType.WORKFLOW_ACTION, user,
272                            appName, LOG, true);
273                }
274            }
275        }
276
277    }
278
279    /**
280     * Loading the Wfjob and workflow actions. Parses the config and adds the nodes that are to be skipped to the
281     * skipped node list
282     *
283     * @throws CommandException
284     */
285    @Override
286    protected void eagerLoadState() throws CommandException {
287        try {
288            this.wfBean = WorkflowJobQueryExecutor.getInstance().get(WorkflowJobQuery.GET_WORKFLOW_STATUS, this.jobId);
289            this.actions = WorkflowActionQueryExecutor.getInstance().getList(
290                    WorkflowActionQuery.GET_ACTIONS_FOR_WORKFLOW_RERUN, this.jobId);
291
292            if (conf != null) {
293                if (conf.getBoolean(OozieClient.RERUN_FAIL_NODES, false) == false) { //Rerun with skipNodes
294                    Collection<String> skipNodes = conf.getStringCollection(OozieClient.RERUN_SKIP_NODES);
295                    for (String str : skipNodes) {
296                        // trimming is required
297                        nodesToSkip.add(str.trim());
298                    }
299                    LOG.debug("Skipnode size :" + nodesToSkip.size());
300                }
301                else {
302                    for (WorkflowActionBean action : actions) { // Rerun from failed nodes
303                        if (action.getStatus() == WorkflowAction.Status.OK) {
304                            nodesToSkip.add(action.getName());
305                        }
306                    }
307                    LOG.debug("Skipnode size are to rerun from FAIL nodes :" + nodesToSkip.size());
308                }
309                StringBuilder tmp = new StringBuilder();
310                for (String node : nodesToSkip) {
311                    tmp.append(node).append(",");
312                }
313                LOG.debug("SkipNode List :" + tmp);
314            }
315        }
316        catch (Exception ex) {
317            throw new CommandException(ErrorCode.E0603, ex.getMessage(), ex);
318        }
319    }
320
321    /**
322     * Checks the pre-conditions that are required for workflow to recover - Last run of Workflow should be completed -
323     * The nodes that are to be skipped are to be completed successfully in the base run.
324     *
325     * @throws org.apache.oozie.command.CommandException,PreconditionException On failure of pre-conditions
326     */
327    @Override
328    protected void eagerVerifyPrecondition() throws CommandException, PreconditionException {
329        if (!(wfBean.getStatus().equals(WorkflowJob.Status.FAILED)
330                || wfBean.getStatus().equals(WorkflowJob.Status.KILLED) || wfBean.getStatus().equals(
331                        WorkflowJob.Status.SUCCEEDED))) {
332            throw new CommandException(ErrorCode.E0805, wfBean.getStatus());
333        }
334        Set<String> unmachedNodes = new HashSet<String>(nodesToSkip);
335        for (WorkflowActionBean action : actions) {
336            if (nodesToSkip.contains(action.getName())) {
337                if (!action.getStatus().equals(WorkflowAction.Status.OK)
338                        && !action.getStatus().equals(WorkflowAction.Status.ERROR)) {
339                    throw new CommandException(ErrorCode.E0806, action.getName());
340                }
341                unmachedNodes.remove(action.getName());
342            }
343        }
344        if (unmachedNodes.size() > 0) {
345            StringBuilder sb = new StringBuilder();
346            String separator = "";
347            for (String s : unmachedNodes) {
348                sb.append(separator).append(s);
349                separator = ",";
350            }
351            throw new CommandException(ErrorCode.E0807, sb);
352        }
353    }
354
355    /**
356     * Copys the variables for skipped nodes from the old wfInstance to new one.
357     *
358     * @param newWfInstance : Source WF instance object
359     * @param oldWfInstance : Update WF instance
360     */
361    private void copyActionData(WorkflowInstance newWfInstance, WorkflowInstance oldWfInstance) {
362        Map<String, String> oldVars = new HashMap<String, String>();
363        Map<String, String> newVars = new HashMap<String, String>();
364        oldVars = oldWfInstance.getAllVars();
365        for (String var : oldVars.keySet()) {
366            String actionName = var.split(WorkflowInstance.NODE_VAR_SEPARATOR)[0];
367            if (nodesToSkip.contains(actionName)) {
368                newVars.put(var, oldVars.get(var));
369            }
370        }
371        for (String node : nodesToSkip) {
372            // Setting the TO_SKIP variable to true. This will be used by
373            // SignalCommand and LiteNodeHandler to skip the action.
374            newVars.put(node + WorkflowInstance.NODE_VAR_SEPARATOR + TO_SKIP, "true");
375            String visitedFlag = NodeHandler.getLoopFlag(node);
376            // Removing the visited flag so that the action won't be considered
377            // a loop.
378            if (newVars.containsKey(visitedFlag)) {
379                newVars.remove(visitedFlag);
380            }
381        }
382        newWfInstance.setAllVars(newVars);
383    }
384
385    /* (non-Javadoc)
386     * @see org.apache.oozie.command.XCommand#getEntityKey()
387     */
388    @Override
389    public String getEntityKey() {
390        return this.jobId;
391    }
392
393    /* (non-Javadoc)
394     * @see org.apache.oozie.command.XCommand#isLockRequired()
395     */
396    @Override
397    protected boolean isLockRequired() {
398        return true;
399    }
400
401    /* (non-Javadoc)
402     * @see org.apache.oozie.command.XCommand#loadState()
403     */
404    @Override
405    protected void loadState() throws CommandException {
406        try {
407            this.wfBean = WorkflowJobQueryExecutor.getInstance().get(WorkflowJobQuery.GET_WORKFLOW_RERUN, this.jobId);
408            this.actions = WorkflowActionQueryExecutor.getInstance().getList(
409                    WorkflowActionQuery.GET_ACTIONS_FOR_WORKFLOW_RERUN, this.jobId);
410        }
411        catch (JPAExecutorException jpe) {
412            throw new CommandException(jpe);
413        }
414    }
415
416    /* (non-Javadoc)
417     * @see org.apache.oozie.command.XCommand#verifyPrecondition()
418     */
419    @Override
420    protected void verifyPrecondition() throws CommandException, PreconditionException {
421        eagerVerifyPrecondition();
422    }
423}