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 */
018
019package org.apache.oozie;
020
021import java.io.IOException;
022import java.io.Writer;
023import java.text.ParseException;
024import java.util.ArrayList;
025import java.util.Date;
026import java.util.HashMap;
027import java.util.HashSet;
028import java.util.List;
029import java.util.Map;
030import java.util.Set;
031import java.util.StringTokenizer;
032
033import javax.servlet.ServletException;
034
035import org.apache.hadoop.conf.Configuration;
036import org.apache.oozie.client.CoordinatorAction;
037import org.apache.oozie.client.CoordinatorJob;
038import org.apache.oozie.client.Job;
039import org.apache.oozie.client.OozieClient;
040import org.apache.oozie.client.WorkflowJob;
041import org.apache.oozie.client.rest.BulkResponseImpl;
042import org.apache.oozie.command.BulkJobsXCommand;
043import org.apache.oozie.command.CommandException;
044import org.apache.oozie.command.OperationType;
045import org.apache.oozie.command.bundle.BulkBundleXCommand;
046import org.apache.oozie.command.bundle.BundleJobChangeXCommand;
047import org.apache.oozie.command.bundle.BundleJobResumeXCommand;
048import org.apache.oozie.command.bundle.BundleJobSuspendXCommand;
049import org.apache.oozie.command.bundle.BundleJobXCommand;
050import org.apache.oozie.command.bundle.BundleJobsXCommand;
051import org.apache.oozie.command.bundle.BundleKillXCommand;
052import org.apache.oozie.command.bundle.BundleRerunXCommand;
053import org.apache.oozie.command.bundle.BundleStartXCommand;
054import org.apache.oozie.command.bundle.BundleSubmitXCommand;
055import org.apache.oozie.executor.jpa.BundleJobQueryExecutor;
056import org.apache.oozie.executor.jpa.JPAExecutorException;
057import org.apache.oozie.service.DagXLogInfoService;
058import org.apache.oozie.service.Services;
059import org.apache.oozie.service.XLogStreamingService;
060import org.apache.oozie.util.DateUtils;
061import org.apache.oozie.util.JobUtils;
062import org.apache.oozie.util.JobsFilterUtils;
063import org.apache.oozie.util.ParamChecker;
064import org.apache.oozie.util.XLog;
065import org.apache.oozie.util.XLogFilter;
066import org.apache.oozie.util.XLogUserFilterParam;
067
068import com.google.common.annotations.VisibleForTesting;
069import sun.reflect.generics.tree.ReturnType;
070
071public class BundleEngine extends BaseEngine {
072    /**
073     * Create a system Bundle engine, with no user and no group.
074     */
075    public BundleEngine() {
076    }
077
078    /**
079     * Create a Bundle engine to perform operations on behave of a user.
080     *
081     * @param user user name.
082     */
083    public BundleEngine(String user) {
084        this.user = ParamChecker.notEmpty(user, "user");
085    }
086
087    /* (non-Javadoc)
088     * @see org.apache.oozie.BaseEngine#change(java.lang.String, java.lang.String)
089     */
090    @Override
091    public void change(String jobId, String changeValue) throws BundleEngineException {
092        try {
093            BundleJobChangeXCommand change = new BundleJobChangeXCommand(jobId, changeValue);
094            change.call();
095        }
096        catch (CommandException ex) {
097            throw new BundleEngineException(ex);
098        }
099    }
100
101    /* (non-Javadoc)
102     * @see org.apache.oozie.BaseEngine#dryRunSubmit(org.apache.hadoop.conf.Configuration)
103     */
104    @Override
105    public String dryRunSubmit(Configuration conf) throws BundleEngineException {
106        BundleSubmitXCommand submit = new BundleSubmitXCommand(true, conf);
107        try {
108            String jobId = submit.call();
109            return jobId;
110        }
111        catch (CommandException ex) {
112            throw new BundleEngineException(ex);
113        }
114    }
115
116    /* (non-Javadoc)
117     * @see org.apache.oozie.BaseEngine#getCoordJob(java.lang.String)
118     */
119    @Override
120    public CoordinatorJob getCoordJob(String jobId) throws BundleEngineException {
121        throw new BundleEngineException(new XException(ErrorCode.E0301, "cannot get a coordinator job from BundleEngine"));
122    }
123
124    public BundleJobBean getBundleJob(String jobId) throws BundleEngineException {
125        try {
126            return new BundleJobXCommand(jobId).call();
127        }
128        catch (CommandException ex) {
129            throw new BundleEngineException(ex);
130        }
131    }
132
133    /* (non-Javadoc)
134     * @see org.apache.oozie.BaseEngine#getCoordJob(java.lang.String, int, int)
135     */
136    @Override
137    public CoordinatorJob getCoordJob(String jobId, String filter, int start, int length, boolean desc)
138            throws BundleEngineException {
139        throw new BundleEngineException(new XException(ErrorCode.E0301,
140                "cannot get a coordinator job from BundleEngine"));
141    }
142
143    /* (non-Javadoc)
144     * @see org.apache.oozie.BaseEngine#getDefinition(java.lang.String)
145     */
146    @Override
147    public String getDefinition(String jobId) throws BundleEngineException {
148        BundleJobBean job;
149        try {
150            job = new BundleJobXCommand(jobId).call();
151        }
152        catch (CommandException ex) {
153            throw new BundleEngineException(ex);
154        }
155        return job.getOrigJobXml();
156    }
157
158    /* (non-Javadoc)
159     * @see org.apache.oozie.BaseEngine#getJob(java.lang.String)
160     */
161    @Override
162    public WorkflowJob getJob(String jobId) throws BundleEngineException {
163        throw new BundleEngineException(new XException(ErrorCode.E0301, "cannot get a workflow job from BundleEngine"));
164    }
165
166    /* (non-Javadoc)
167     * @see org.apache.oozie.BaseEngine#getJob(java.lang.String, int, int)
168     */
169    @Override
170    public WorkflowJob getJob(String jobId, int start, int length) throws BundleEngineException {
171        throw new BundleEngineException(new XException(ErrorCode.E0301, "cannot get a workflow job from BundleEngine"));
172    }
173
174    /* (non-Javadoc)
175     * @see org.apache.oozie.BaseEngine#getJobIdForExternalId(java.lang.String)
176     */
177    @Override
178    public String getJobIdForExternalId(String externalId) throws BundleEngineException {
179        return null;
180    }
181
182    /* (non-Javadoc)
183     * @see org.apache.oozie.BaseEngine#kill(java.lang.String)
184     */
185    @Override
186    public void kill(String jobId) throws BundleEngineException {
187        try {
188            new BundleKillXCommand(jobId).call();
189        }
190        catch (CommandException e) {
191            throw new BundleEngineException(e);
192        }
193    }
194
195    /* (non-Javadoc)
196     * @see org.apache.oozie.BaseEngine#reRun(java.lang.String, org.apache.hadoop.conf.Configuration)
197     */
198    @Override
199    @Deprecated
200    public void reRun(String jobId, Configuration conf) throws BundleEngineException {
201        throw new BundleEngineException(new XException(ErrorCode.E0301, "rerun"));
202    }
203
204    /**
205     * Rerun Bundle actions for given rerunType
206     *
207     * @param jobId bundle job id
208     * @param coordScope the rerun scope for coordinator job names separated by ","
209     * @param dateScope the rerun scope for coordinator nominal times separated by ","
210     * @param refresh true if user wants to refresh input/outpur dataset urls
211     * @param noCleanup false if user wants to cleanup output events for given rerun actions
212     * @throws BaseEngineException thrown if failed to rerun
213     */
214    public void reRun(String jobId, String coordScope, String dateScope, boolean refresh, boolean noCleanup)
215            throws BaseEngineException {
216        try {
217            new BundleRerunXCommand(jobId, coordScope, dateScope, refresh, noCleanup).call();
218        }
219        catch (CommandException ex) {
220            throw new BaseEngineException(ex);
221        }
222    }
223
224    /* (non-Javadoc)
225     * @see org.apache.oozie.BaseEngine#resume(java.lang.String)
226     */
227    @Override
228    public void resume(String jobId) throws BundleEngineException {
229        BundleJobResumeXCommand resume = new BundleJobResumeXCommand(jobId);
230        try {
231            resume.call();
232        }
233        catch (CommandException ex) {
234            throw new BundleEngineException(ex);
235        }
236    }
237
238    /* (non-Javadoc)
239     * @see org.apache.oozie.BaseEngine#start(java.lang.String)
240     */
241    @Override
242    public void start(String jobId) throws BundleEngineException {
243        try {
244            new BundleStartXCommand(jobId).call();
245        }
246        catch (CommandException e) {
247            throw new BundleEngineException(e);
248        }
249    }
250
251    /* (non-Javadoc)
252     * @see org.apache.oozie.BaseEngine#streamLog(java.lang.String, java.io.Writer)
253     */
254    @Override
255    public void streamLog(String jobId, Writer writer, Map<String, String[]> params) throws IOException,
256            BundleEngineException {
257
258        BundleJobBean job;
259        try {
260            XLogFilter filter = new XLogFilter(new XLogUserFilterParam(params));
261            filter.setParameter(DagXLogInfoService.JOB, jobId);
262            job = new BundleJobXCommand(jobId).call();
263            Date lastTime = null;
264            if (job.isTerminalStatus()) {
265                lastTime = job.getLastModifiedTime();
266            }
267            if (lastTime == null) {
268                lastTime = new Date();
269            }
270            Services.get().get(XLogStreamingService.class).streamLog(filter, job.getCreatedTime(), lastTime, writer, params);
271        }
272        catch (Exception ex) {
273            throw new IOException(ex);
274        }
275    }
276
277    /* (non-Javadoc)
278     * @see org.apache.oozie.BaseEngine#submitJob(org.apache.hadoop.conf.Configuration, boolean)
279     */
280    @Override
281    public String submitJob(Configuration conf, boolean startJob) throws BundleEngineException {
282        try {
283            String jobId = new BundleSubmitXCommand(conf).call();
284
285            if (startJob) {
286                start(jobId);
287            }
288            return jobId;
289        }
290        catch (CommandException ex) {
291            throw new BundleEngineException(ex);
292        }
293    }
294
295    /* (non-Javadoc)
296     * @see org.apache.oozie.BaseEngine#suspend(java.lang.String)
297     */
298    @Override
299    public void suspend(String jobId) throws BundleEngineException {
300        BundleJobSuspendXCommand suspend = new BundleJobSuspendXCommand(jobId);
301        try {
302            suspend.call();
303        }
304        catch (CommandException ex) {
305            throw new BundleEngineException(ex);
306        }
307    }
308
309    /**
310     * Get bundle jobs
311     *
312     * @param filter the filter string
313     * @param start start location for paging
314     * @param len total length to get
315     * @return bundle job info
316     * @throws BundleEngineException thrown if failed to get bundle job info
317     */
318    public BundleJobInfo getBundleJobs(String filter, int start, int len) throws BundleEngineException {
319        Map<String, List<String>> filterList = parseFilter(filter);
320
321        try {
322            return new BundleJobsXCommand(filterList, start, len).call();
323        }
324        catch (CommandException ex) {
325            throw new BundleEngineException(ex);
326        }
327    }
328
329    /**
330     * Parse filter string to a map with key = filter name and values = filter values
331     *
332     * @param filter the filter string
333     * @return filter key and value map
334     * @throws CoordinatorEngineException thrown if failed to parse filter string
335     */
336    @VisibleForTesting
337    Map<String, List<String>> parseFilter(String filter) throws BundleEngineException {
338        try {
339            return JobsFilterUtils.parseFilter(filter);
340        }
341        catch (ServletException ex) {
342            throw new BundleEngineException(ErrorCode.E0420, filter, ex.getMessage());
343        }
344    }
345
346    /**
347     * Get bulk job response
348     *
349     * @param bulkFilter the filter string
350     * @param start start location for paging
351     * @param len total length to get
352     * @return bulk job info
353     * @throws BundleEngineException thrown if failed to get bulk job info
354     */
355    public BulkResponseInfo getBulkJobs(String bulkFilter, int start, int len) throws BundleEngineException {
356        Map<String,List<String>> bulkRequestMap = parseBulkFilter(bulkFilter);
357        try {
358            return new BulkJobsXCommand(bulkRequestMap, start, len).call();
359        }
360        catch (CommandException ex) {
361            throw new BundleEngineException(ex);
362        }
363    }
364
365    /**
366     * Parse filter string to a map with key = filter name and values = filter values
367     * Allowed keys are defined as constants on top
368     *
369     * @param bulkParams the filter string
370     * @return filter key-value pair map
371     * @throws BundleEngineException thrown if failed to parse filter string
372     */
373    public static Map<String,List<String>> parseBulkFilter(String bulkParams) throws BundleEngineException {
374
375        Map<String,List<String>> bulkFilter = new HashMap<String,List<String>>();
376        // Functionality can be extended to different job levels - TODO extend filter parser and query
377        // E.g. String filterlevel = "coordinatoraction"; BulkResponseImpl.BULK_FILTER_LEVEL
378        if (bulkFilter != null) {
379            StringTokenizer st = new StringTokenizer(bulkParams, ";");
380            while (st.hasMoreTokens()) {
381                String token = st.nextToken();
382                if (token.contains("=")) {
383                    String[] pair = token.split("=");
384                    if (pair.length != 2) {
385                        throw new BundleEngineException(ErrorCode.E0420, token,
386                                "elements must be semicolon-separated name=value pairs");
387                    }
388                    pair[0] = pair[0].toLowerCase();
389                    String[] values = pair[1].split(",");
390                    if (!BulkResponseImpl.BULK_FILTER_NAMES.contains(pair[0])) {
391                        throw new BundleEngineException(ErrorCode.E0420, token, XLog.format("invalid parameter name [{0}]",
392                                pair[0]));
393                    }
394                    // special check and processing for time related params
395                    if (pair[0].contains("time")) {
396                        try {
397                            DateUtils.parseDateUTC(pair[1]);
398                        }
399                        catch (ParseException e) {
400                            throw new BundleEngineException(ErrorCode.E0420, token, XLog.format(
401                                    "invalid value [{0}] for time. A datetime value of pattern [{1}] is expected", pair[1],
402                                    DateUtils.ISO8601_UTC_MASK));
403                        }
404                    }
405                    // special check for action status param
406                    // TODO: when extended for levels other than coord action, check against corresponding level's Status values
407                    if (pair[0].equals(BulkResponseImpl.BULK_FILTER_STATUS)) {
408                        for(String value : values) {
409                            try {
410                                CoordinatorAction.Status.valueOf(value);
411                            }
412                            catch (IllegalArgumentException ex) {
413                                throw new BundleEngineException(ErrorCode.E0420, token, XLog.format(
414                                        "invalid action status [{0}]", value));
415                            }
416                        }
417                    }
418                    // eventually adding into map for all cases e.g. names, times, status
419                    List<String> list = bulkFilter.get(pair[0]);
420                    if (list == null) {
421                        list = new ArrayList<String>();
422                        bulkFilter.put(pair[0], list);
423                    }
424                    for(String value : values) {
425                        value = value.trim();
426                        if(value.isEmpty()) {
427                            throw new BundleEngineException(ErrorCode.E0420, token, "value is empty or whitespace");
428                        }
429                        list.add(value);
430                    }
431                } else {
432                    throw new BundleEngineException(ErrorCode.E0420, token,
433                            "elements must be semicolon-separated name=value pairs");
434                }
435            }
436            if (!bulkFilter.containsKey(BulkResponseImpl.BULK_FILTER_BUNDLE)) {
437                throw new BundleEngineException(ErrorCode.E0305, BulkResponseImpl.BULK_FILTER_BUNDLE);
438            }
439        }
440        return bulkFilter;
441    }
442
443    /**
444     * Return the status for a Job ID
445     *
446     * @param jobId job Id.
447     * @return the job's status
448     * @throws BundleEngineException thrown if the job's status could not be obtained
449     */
450    @Override
451    public String getJobStatus(String jobId) throws BundleEngineException {
452        try {
453            BundleJobBean bundleJob = BundleJobQueryExecutor.getInstance().get(
454                    BundleJobQueryExecutor.BundleJobQuery.GET_BUNDLE_JOB_STATUS, jobId);
455            return bundleJob.getStatusStr();
456        }
457        catch (JPAExecutorException e) {
458            throw new BundleEngineException(e);
459        }
460    }
461
462    /**
463     * return a list of killed Bundle job
464     *
465     * @param filter, the filter string for which the bundle jobs are killed
466     * @param start, the starting index for bundle jobs
467     * @param len, maximum number of jobs to be killed
468     * @return the list of jobs being killed
469     * @throws BundleEngineException thrown if one or more of the jobs cannot be killed
470     */
471    public BundleJobInfo killJobs(String filter, int start, int len) throws BundleEngineException {
472        try {
473            Map<String, List<String>> filterList = parseFilter(filter);
474            BundleJobInfo bundleJobInfo = new BulkBundleXCommand(filterList, start, len, OperationType.Kill).call();
475            if (bundleJobInfo == null) {
476                return new BundleJobInfo(new ArrayList<BundleJobBean>(), 0, 0, 0);
477            }
478            return bundleJobInfo;
479        }
480        catch (CommandException ex) {
481            throw new BundleEngineException(ex);
482        }
483    }
484
485    /**
486     * return a list of suspended Bundle job
487     *
488     * @param filter, the filter string for which the bundle jobs are suspended
489     * @param start, the starting index for bundle jobs
490     * @param len, maximum number of jobs to be suspended
491     * @return the list of jobs being suspended
492     * @throws BundleEngineException thrown if one or more of the jobs cannot be suspended
493     */
494    public BundleJobInfo suspendJobs(String filter, int start, int len) throws BundleEngineException {
495        try {
496            Map<String, List<String>> filterList = parseFilter(filter);
497            BundleJobInfo bundleJobInfo = new BulkBundleXCommand(filterList, start, len, OperationType.Suspend).call();
498            if (bundleJobInfo == null) {
499                return new BundleJobInfo(new ArrayList<BundleJobBean>(), 0, 0, 0);
500            }
501            return bundleJobInfo;
502        }
503        catch (CommandException ex) {
504            throw new BundleEngineException(ex);
505        }
506    }
507
508    /**
509     * return a list of resumed Bundle job
510     *
511     * @param filter, the filter string for which the bundle jobs are resumed
512     * @param start, the starting index for bundle jobs
513     * @param len, maximum number of jobs to be resumed
514     * @return the list of jobs being resumed
515     * @throws BundleEngineException thrown if one or more of the jobs cannot be resumed
516     */
517    public BundleJobInfo resumeJobs(String filter, int start, int len) throws BundleEngineException {
518        try {
519            Map<String, List<String>> filterList = parseFilter(filter);
520            BundleJobInfo bundleJobInfo = new BulkBundleXCommand(filterList, start, len, OperationType.Resume).call();
521            if (bundleJobInfo == null) {
522                return new BundleJobInfo(new ArrayList<BundleJobBean>(), 0, 0, 0);
523            }
524            return bundleJobInfo;
525        }
526        catch (CommandException ex) {
527            throw new BundleEngineException(ex);
528        }
529    }
530}