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;
019
020import org.apache.oozie.util.XLogStreamer;
021import org.apache.oozie.service.XLogService;
022import org.apache.oozie.service.DagXLogInfoService;
023import org.apache.hadoop.conf.Configuration;
024import org.apache.oozie.client.CoordinatorJob;
025import org.apache.oozie.client.WorkflowJob;
026import org.apache.oozie.client.OozieClient;
027import org.apache.oozie.command.CommandException;
028import org.apache.oozie.command.wf.CompletedActionXCommand;
029import org.apache.oozie.command.wf.DefinitionXCommand;
030import org.apache.oozie.command.wf.ExternalIdXCommand;
031import org.apache.oozie.command.wf.JobXCommand;
032import org.apache.oozie.command.wf.JobsXCommand;
033import org.apache.oozie.command.wf.KillXCommand;
034import org.apache.oozie.command.wf.ReRunXCommand;
035import org.apache.oozie.command.wf.ResumeXCommand;
036import org.apache.oozie.command.wf.StartXCommand;
037import org.apache.oozie.command.wf.SubmitHiveXCommand;
038import org.apache.oozie.command.wf.SubmitHttpXCommand;
039import org.apache.oozie.command.wf.SubmitMRXCommand;
040import org.apache.oozie.command.wf.SubmitPigXCommand;
041import org.apache.oozie.command.wf.SubmitXCommand;
042import org.apache.oozie.command.wf.SuspendXCommand;
043import org.apache.oozie.command.wf.WorkflowActionInfoXCommand;
044import org.apache.oozie.service.Services;
045import org.apache.oozie.service.CallableQueueService;
046import org.apache.oozie.util.ParamChecker;
047import org.apache.oozie.util.XCallable;
048import org.apache.oozie.util.XLog;
049
050import java.io.Writer;
051import java.util.Date;
052import java.util.List;
053import java.util.Properties;
054import java.util.Set;
055import java.util.HashSet;
056import java.util.StringTokenizer;
057import java.util.Map;
058import java.util.HashMap;
059import java.util.ArrayList;
060import java.io.IOException;
061
062/**
063 * The DagEngine provides all the DAG engine functionality for WS calls.
064 */
065public class DagEngine extends BaseEngine {
066
067    private static final int HIGH_PRIORITY = 2;
068    private static XLog LOG = XLog.getLog(DagEngine.class);
069
070    /**
071     * Create a system Dag engine, with no user and no group.
072     */
073    public DagEngine() {
074        if (Services.get().getConf().getBoolean(USE_XCOMMAND, true) == false) {
075            LOG.debug("Oozie DagEngine is not using XCommands.");
076        }
077        else {
078            LOG.debug("Oozie DagEngine is using XCommands.");
079        }
080    }
081
082    /**
083     * Create a Dag engine to perform operations on behave of a user.
084     *
085     * @param user user name.
086     */
087    public DagEngine(String user) {
088        this();
089
090        this.user = ParamChecker.notEmpty(user, "user");
091    }
092
093    /**
094     * Submit a workflow job. <p/> It validates configuration properties.
095     *
096     * @param conf job configuration.
097     * @param startJob indicates if the job should be started or not.
098     * @return the job Id.
099     * @throws DagEngineException thrown if the job could not be created.
100     */
101    @Override
102    public String submitJob(Configuration conf, boolean startJob) throws DagEngineException {
103        validateSubmitConfiguration(conf);
104
105        try {
106            String jobId;
107            SubmitXCommand submit = new SubmitXCommand(conf);
108            jobId = submit.call();
109            if (startJob) {
110                start(jobId);
111            }
112            return jobId;
113        }
114        catch (CommandException ex) {
115            throw new DagEngineException(ex);
116        }
117    }
118
119    /**
120     * Submit a workflow through a coordinator. It validates configuration properties.
121     * @param conf job conf
122     * @param parentId parent of workflow
123     * @return
124     * @throws DagEngineException
125     */
126    public String submitJobFromCoordinator(Configuration conf, String parentId) throws DagEngineException {
127        validateSubmitConfiguration(conf);
128        try {
129            String jobId;
130            SubmitXCommand submit = new SubmitXCommand(conf, parentId);
131            jobId = submit.call();
132            start(jobId);
133            return jobId;
134        }
135        catch (CommandException ex) {
136            throw new DagEngineException(ex);
137        }
138    }
139
140    /**
141     * Submit a pig/hive/mapreduce job through HTTP.
142     * <p/>
143     * It validates configuration properties.
144     *
145     * @param conf job configuration.
146     * @param jobType job type - can be "pig", "hive, or "mapreduce".
147     * @return the job Id.
148     * @throws DagEngineException thrown if the job could not be created.
149     */
150    public String submitHttpJob(Configuration conf, String jobType) throws DagEngineException {
151        validateSubmitConfiguration(conf);
152
153        try {
154            String jobId;
155            SubmitHttpXCommand submit = null;
156            if (jobType.equals("pig")) {
157                submit = new SubmitPigXCommand(conf);
158            }
159            else if (jobType.equals("mapreduce")) {
160                submit = new SubmitMRXCommand(conf);
161            }
162            else if (jobType.equals("hive")) {
163                submit = new SubmitHiveXCommand(conf);
164            }
165
166            jobId = submit.call();
167            start(jobId);
168            return jobId;
169        }
170        catch (CommandException ex) {
171            throw new DagEngineException(ex);
172        }
173    }
174
175    private void validateSubmitConfiguration(Configuration conf) throws DagEngineException {
176        if (conf.get(OozieClient.APP_PATH) == null) {
177            throw new DagEngineException(ErrorCode.E0401, OozieClient.APP_PATH);
178        }
179    }
180
181    /**
182     * Start a job.
183     *
184     * @param jobId job Id.
185     * @throws DagEngineException thrown if the job could not be started.
186     */
187    @Override
188    public void start(String jobId) throws DagEngineException {
189        // Changing to synchronous call from asynchronous queuing to prevent the
190        // loss of command if the queue is full or the queue is lost in case of
191        // failure.
192        try {
193                new StartXCommand(jobId).call();
194        }
195        catch (CommandException e) {
196            throw new DagEngineException(e);
197        }
198    }
199
200    /**
201     * Resume a job.
202     *
203     * @param jobId job Id.
204     * @throws DagEngineException thrown if the job could not be resumed.
205     */
206    @Override
207    public void resume(String jobId) throws DagEngineException {
208        // Changing to synchronous call from asynchronous queuing to prevent the
209        // loss of command if the queue is full or the queue is lost in case of
210        // failure.
211        try {
212                new ResumeXCommand(jobId).call();
213        }
214        catch (CommandException e) {
215            throw new DagEngineException(e);
216        }
217    }
218
219    /**
220     * Suspend a job.
221     *
222     * @param jobId job Id.
223     * @throws DagEngineException thrown if the job could not be suspended.
224     */
225    @Override
226    public void suspend(String jobId) throws DagEngineException {
227        // Changing to synchronous call from asynchronous queuing to prevent the
228        // loss of command if the queue is full or the queue is lost in case of
229        // failure.
230        try {
231                        new SuspendXCommand(jobId).call();
232        }
233        catch (CommandException e) {
234            throw new DagEngineException(e);
235        }
236    }
237
238    /**
239     * Kill a job.
240     *
241     * @param jobId job Id.
242     * @throws DagEngineException thrown if the job could not be killed.
243     */
244    @Override
245    public void kill(String jobId) throws DagEngineException {
246        // Changing to synchronous call from asynchronous queuing to prevent the
247        // loss of command if the queue is full or the queue is lost in case of
248        // failure.
249        try {
250                        new KillXCommand(jobId).call();
251                        LOG.info("User " + user + " killed the WF job " + jobId);
252        }
253        catch (CommandException e) {
254            throw new DagEngineException(e);
255        }
256    }
257
258    /* (non-Javadoc)
259     * @see org.apache.oozie.BaseEngine#change(java.lang.String, java.lang.String)
260     */
261    @Override
262    public void change(String jobId, String changeValue) throws DagEngineException {
263        // This code should not be reached.
264        throw new DagEngineException(ErrorCode.E1017);
265    }
266
267    /**
268     * Rerun a job.
269     *
270     * @param jobId job Id to rerun.
271     * @param conf configuration information for the rerun.
272     * @throws DagEngineException thrown if the job could not be rerun.
273     */
274    @Override
275    public void reRun(String jobId, Configuration conf) throws DagEngineException {
276        try {
277            validateReRunConfiguration(conf);
278            new ReRunXCommand(jobId, conf).call();
279            start(jobId);
280        }
281        catch (CommandException ex) {
282            throw new DagEngineException(ex);
283        }
284    }
285
286    private void validateReRunConfiguration(Configuration conf) throws DagEngineException {
287        if (conf.get(OozieClient.APP_PATH) == null) {
288            throw new DagEngineException(ErrorCode.E0401, OozieClient.APP_PATH);
289        }
290        if (conf.get(OozieClient.RERUN_SKIP_NODES) == null && conf.get(OozieClient.RERUN_FAIL_NODES) == null) {
291            throw new DagEngineException(ErrorCode.E0401, OozieClient.RERUN_SKIP_NODES + " OR "
292                    + OozieClient.RERUN_FAIL_NODES);
293        }
294        if (conf.get(OozieClient.RERUN_SKIP_NODES) != null && conf.get(OozieClient.RERUN_FAIL_NODES) != null) {
295            throw new DagEngineException(ErrorCode.E0404, OozieClient.RERUN_SKIP_NODES + " OR "
296                    + OozieClient.RERUN_FAIL_NODES);
297        }
298    }
299
300    /**
301     * Process an action callback.
302     *
303     * @param actionId the action Id.
304     * @param externalStatus the action external status.
305     * @param actionData the action output data, <code>null</code> if none.
306     * @throws DagEngineException thrown if the callback could not be processed.
307     */
308    public void processCallback(String actionId, String externalStatus, Properties actionData)
309            throws DagEngineException {
310        XLog.Info.get().clearParameter(XLogService.GROUP);
311        XLog.Info.get().clearParameter(XLogService.USER);
312        XCallable<Void> command = null;
313
314                command = new CompletedActionXCommand(actionId, externalStatus,
315                                actionData, HIGH_PRIORITY);
316        if (!Services.get().get(CallableQueueService.class).queue(command)) {
317            LOG.warn(XLog.OPS, "queue is full or system is in SAFEMODE, ignoring callback");
318        }
319    }
320
321    /**
322     * Return the info about a job.
323     *
324     * @param jobId job Id.
325     * @return the workflow job info.
326     * @throws DagEngineException thrown if the job info could not be obtained.
327     */
328    @Override
329    public WorkflowJob getJob(String jobId) throws DagEngineException {
330        try {
331                return new JobXCommand(jobId).call();
332        }
333        catch (CommandException ex) {
334            throw new DagEngineException(ex);
335        }
336    }
337
338    /**
339     * Return the info about a job with actions subset.
340     *
341     * @param jobId job Id
342     * @param start starting from this index in the list of actions belonging to the job
343     * @param length number of actions to be returned
344     * @return the workflow job info.
345     * @throws DagEngineException thrown if the job info could not be obtained.
346     */
347    @Override
348    public WorkflowJob getJob(String jobId, int start, int length) throws DagEngineException {
349        try {
350                        return new JobXCommand(jobId, start, length).call();
351        }
352        catch (CommandException ex) {
353            throw new DagEngineException(ex);
354        }
355    }
356
357    /**
358     * Return the a job definition.
359     *
360     * @param jobId job Id.
361     * @return the job definition.
362     * @throws DagEngineException thrown if the job definition could no be obtained.
363     */
364    @Override
365    public String getDefinition(String jobId) throws DagEngineException {
366        try {
367                        return new DefinitionXCommand(jobId).call();
368        }
369        catch (CommandException ex) {
370            throw new DagEngineException(ex);
371        }
372    }
373
374    /**
375     * Stream the log of a job.
376     *
377     * @param jobId job Id.
378     * @param writer writer to stream the log to.
379     * @throws IOException thrown if the log cannot be streamed.
380     * @throws DagEngineException thrown if there is error in getting the Workflow Information for jobId.
381     */
382    @Override
383    public void streamLog(String jobId, Writer writer) throws IOException, DagEngineException {
384        XLogStreamer.Filter filter = new XLogStreamer.Filter();
385        filter.setParameter(DagXLogInfoService.JOB, jobId);
386        WorkflowJob job = getJob(jobId);
387        Date lastTime = job.getEndTime();
388        if (lastTime == null) {
389            lastTime = job.getLastModifiedTime();
390        }
391        Services.get().get(XLogService.class).streamLog(filter, job.getCreatedTime(), lastTime, writer);
392    }
393
394    private static final Set<String> FILTER_NAMES = new HashSet<String>();
395
396    static {
397        FILTER_NAMES.add(OozieClient.FILTER_USER);
398        FILTER_NAMES.add(OozieClient.FILTER_NAME);
399        FILTER_NAMES.add(OozieClient.FILTER_GROUP);
400        FILTER_NAMES.add(OozieClient.FILTER_STATUS);
401        FILTER_NAMES.add(OozieClient.FILTER_ID);
402    }
403
404    /**
405     * Validate a jobs filter.
406     *
407     * @param filter filter to validate.
408     * @return the parsed filter.
409     * @throws DagEngineException thrown if the filter is invalid.
410     */
411    protected Map<String, List<String>> parseFilter(String filter) throws DagEngineException {
412        Map<String, List<String>> map = new HashMap<String, List<String>>();
413        if (filter != null) {
414            StringTokenizer st = new StringTokenizer(filter, ";");
415            while (st.hasMoreTokens()) {
416                String token = st.nextToken();
417                if (token.contains("=")) {
418                    String[] pair = token.split("=");
419                    if (pair.length != 2) {
420                        throw new DagEngineException(ErrorCode.E0420, filter, "elements must be name=value pairs");
421                    }
422                    if (!FILTER_NAMES.contains(pair[0])) {
423                        throw new DagEngineException(ErrorCode.E0420, filter, XLog
424                                .format("invalid name [{0}]", pair[0]));
425                    }
426                    if (pair[0].equals("status")) {
427                        try {
428                            WorkflowJob.Status.valueOf(pair[1]);
429                        }
430                        catch (IllegalArgumentException ex) {
431                            throw new DagEngineException(ErrorCode.E0420, filter, XLog.format("invalid status [{0}]",
432                                                                                              pair[1]));
433                        }
434                    }
435                    List<String> list = map.get(pair[0]);
436                    if (list == null) {
437                        list = new ArrayList<String>();
438                        map.put(pair[0], list);
439                    }
440                    list.add(pair[1]);
441                }
442                else {
443                    throw new DagEngineException(ErrorCode.E0420, filter, "elements must be name=value pairs");
444                }
445            }
446        }
447        return map;
448    }
449
450    /**
451     * Return the info about a set of jobs.
452     *
453     * @param filter job filter. Refer to the {@link org.apache.oozie.client.OozieClient} for the filter syntax.
454     * @param start offset, base 1.
455     * @param len number of jobs to return.
456     * @return job info for all matching jobs, the jobs don't contain node action information.
457     * @throws DagEngineException thrown if the jobs info could not be obtained.
458     */
459    public WorkflowsInfo getJobs(String filter, int start, int len) throws DagEngineException {
460        Map<String, List<String>> filterList = parseFilter(filter);
461        try {
462                        return new JobsXCommand(filterList, start, len).call();
463        }
464        catch (CommandException dce) {
465            throw new DagEngineException(dce);
466        }
467    }
468
469    /**
470     * Return the workflow Job ID for an external ID. <p/> This is reverse lookup for recovery purposes.
471     *
472     * @param externalId external ID provided at job submission time.
473     * @return the associated workflow job ID if any, <code>null</code> if none.
474     * @throws DagEngineException thrown if the lookup could not be done.
475     */
476    @Override
477    public String getJobIdForExternalId(String externalId) throws DagEngineException {
478        try {
479                        return new ExternalIdXCommand(externalId).call();
480        }
481        catch (CommandException dce) {
482            throw new DagEngineException(dce);
483        }
484    }
485
486    @Override
487    public CoordinatorJob getCoordJob(String jobId) throws BaseEngineException {
488        throw new BaseEngineException(new XException(ErrorCode.E0301, "cannot get a coordinator job from DagEngine"));
489    }
490
491    @Override
492    public CoordinatorJob getCoordJob(String jobId, String filter, int start, int length, boolean desc) throws BaseEngineException {
493        throw new BaseEngineException(new XException(ErrorCode.E0301, "cannot get a coordinator job from DagEngine"));
494    }
495
496    public WorkflowActionBean getWorkflowAction(String actionId) throws BaseEngineException {
497        try {
498                        return new WorkflowActionInfoXCommand(actionId).call();
499        }
500        catch (CommandException ex) {
501            throw new BaseEngineException(ex);
502        }
503    }
504
505    /* (non-Javadoc)
506     * @see org.apache.oozie.BaseEngine#dryRunSubmit(org.apache.hadoop.conf.Configuration)
507     */
508    @Override
509    public String dryRunSubmit(Configuration conf) throws BaseEngineException {
510        try {
511            SubmitXCommand submit = new SubmitXCommand(true, conf);
512            return submit.call();
513        } catch (CommandException ex) {
514            throw new DagEngineException(ex);
515        }
516    }
517}