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.command.coord;
020
021import java.io.StringReader;
022import java.net.URI;
023import java.text.ParseException;
024import java.util.TimeZone;
025import java.util.Map;
026import java.util.HashMap;
027import java.util.List;
028import java.util.Date;
029import java.util.Calendar;
030
031import org.apache.hadoop.conf.Configuration;
032import org.apache.oozie.CoordinatorActionBean;
033import org.apache.oozie.ErrorCode;
034import org.apache.oozie.client.CoordinatorAction;
035import org.apache.oozie.command.CommandException;
036import org.apache.oozie.coord.CoordELEvaluator;
037import org.apache.oozie.coord.CoordELFunctions;
038import org.apache.oozie.coord.CoordUtils;
039import org.apache.oozie.coord.CoordinatorJobException;
040import org.apache.oozie.coord.SyncCoordAction;
041import org.apache.oozie.coord.TimeUnit;
042import org.apache.oozie.dependency.ActionDependency;
043import org.apache.oozie.dependency.DependencyChecker;
044import org.apache.oozie.dependency.URIHandler;
045import org.apache.oozie.dependency.URIHandler.DependencyType;
046import org.apache.oozie.service.Services;
047import org.apache.oozie.service.URIHandlerService;
048import org.apache.oozie.service.UUIDService;
049import org.apache.oozie.util.DateUtils;
050import org.apache.oozie.util.ELEvaluator;
051import org.apache.oozie.util.XConfiguration;
052import org.apache.oozie.util.XmlUtils;
053import org.jdom.Element;
054import org.jdom.JDOMException;
055import org.quartz.CronExpression;
056import org.apache.commons.lang.StringUtils;
057import org.apache.oozie.CoordinatorJobBean;
058
059public class CoordCommandUtils {
060    public static int CURRENT = 0;
061    public static int LATEST = 1;
062    public static int FUTURE = 2;
063    public static int OFFSET = 3;
064    public static int ABSOLUTE = 4;
065    public static int UNEXPECTED = -1;
066    public static final String RESOLVED_UNRESOLVED_SEPARATOR = "!!";
067    public static final String UNRESOLVED_INST_TAG = "unresolved-instances";
068
069    /**
070     * parse a function like coord:latest(n)/future() and return the 'n'.
071     * <p/>
072     *
073     * @param function
074     * @param restArg
075     * @return int instanceNumber
076     * @throws Exception
077     */
078    public static int getInstanceNumber(String function, StringBuilder restArg) throws Exception {
079        int funcType = getFuncType(function);
080        if (funcType == ABSOLUTE) {
081            return ABSOLUTE;
082        }
083        if (funcType == CURRENT || funcType == LATEST) {
084            return parseOneArg(function);
085        }
086        else {
087            return parseMoreArgs(function, restArg);
088        }
089    }
090
091    /**
092     * Evaluates function for coord-action-create-inst tag
093     * @param event
094     * @param appInst
095     * @param conf
096     * @param function
097     * @return evaluation result
098     * @throws Exception
099     */
100    private static String evaluateInstanceFunction(Element event, SyncCoordAction appInst, Configuration conf,
101            String function) throws Exception {
102        ELEvaluator eval = CoordELEvaluator.createInstancesELEvaluator("coord-action-create-inst", event, appInst, conf);
103        return CoordELFunctions.evalAndWrap(eval, function);
104    }
105
106    public static int parseOneArg(String funcName) throws Exception {
107        int firstPos = funcName.indexOf("(");
108        int lastPos = funcName.lastIndexOf(")");
109        if (firstPos >= 0 && lastPos > firstPos) {
110            String tmp = funcName.substring(firstPos + 1, lastPos).trim();
111            if (tmp.length() > 0) {
112                return (int) Double.parseDouble(tmp);
113            }
114        }
115        throw new RuntimeException("Unformatted function :" + funcName);
116    }
117
118    public static String parseOneStringArg(String funcName) throws Exception {
119        int firstPos = funcName.indexOf("(");
120        int lastPos = funcName.lastIndexOf(")");
121        if (firstPos >= 0 && lastPos > firstPos) {
122            return funcName.substring(firstPos + 1, lastPos).trim();
123        }
124        throw new RuntimeException("Unformatted function :" + funcName);
125    }
126
127    private static int parseMoreArgs(String funcName, StringBuilder restArg) throws Exception {
128        int firstPos = funcName.indexOf("(");
129        int secondPos = funcName.lastIndexOf(",");
130        int lastPos = funcName.lastIndexOf(")");
131        if (firstPos >= 0 && secondPos > firstPos) {
132            String tmp = funcName.substring(firstPos + 1, secondPos).trim();
133            if (tmp.length() > 0) {
134                restArg.append(funcName.substring(secondPos + 1, lastPos).trim());
135                return (int) Double.parseDouble(tmp);
136            }
137        }
138        throw new RuntimeException("Unformatted function :" + funcName);
139    }
140
141    /**
142     * @param EL function name
143     * @return type of EL function
144     */
145    public static int getFuncType(String function) {
146        if (function.indexOf("current") >= 0) {
147            return CURRENT;
148        }
149        else if (function.indexOf("latest") >= 0) {
150            return LATEST;
151        }
152        else if (function.indexOf("future") >= 0) {
153            return FUTURE;
154        }
155        else if (function.indexOf("offset") >= 0) {
156            return OFFSET;
157        }
158        else if (function.indexOf("absolute") >= 0) {
159            return ABSOLUTE;
160        }
161        return UNEXPECTED;
162        // throw new RuntimeException("Unexpected instance name "+ function);
163    }
164
165    /**
166     * @param startInst: EL function name
167     * @param endInst: EL function name
168     * @throws CommandException if both are not the same function
169     */
170    public static void checkIfBothSameType(String startInst, String endInst) throws CommandException {
171        if (getFuncType(startInst) != getFuncType(endInst)) {
172            if (getFuncType(startInst) == ABSOLUTE) {
173                if (getFuncType(endInst) != CURRENT) {
174                    throw new CommandException(ErrorCode.E1010,
175                            "Only start-instance as absolute and end-instance as current is supported." + " start = "
176                                    + startInst + "  end = " + endInst);
177                }
178            }
179            else {
180                throw new CommandException(ErrorCode.E1010,
181                        " start-instance and end-instance both should be either latest or current or future or offset\n"
182                                + " start " + startInst + " and end " + endInst);
183            }
184        }
185    }
186
187
188    /**
189     * Resolve list of <instance> </instance> tags.
190     *
191     * @param event
192     * @param instances
193     * @param actionInst
194     * @param conf
195     * @param eval: ELEvalautor
196     * @throws Exception
197     */
198    public static void resolveInstances(Element event, StringBuilder instances, SyncCoordAction actionInst,
199            Configuration conf, ELEvaluator eval) throws Exception {
200        for (Element eInstance : (List<Element>) event.getChildren("instance", event.getNamespace())) {
201
202            if (instances.length() > 0) {
203                instances.append(CoordELFunctions.INSTANCE_SEPARATOR);
204            }
205            instances.append(materializeInstance(event, eInstance.getTextTrim(), actionInst, conf, eval));
206        }
207        event.removeChildren("instance", event.getNamespace());
208    }
209
210    /**
211     * Resolve <start-instance> <end-insatnce> tag. Don't resolve any
212     * latest()/future()
213     *
214     * @param event
215     * @param instances
216     * @param appInst
217     * @param conf
218     * @param eval: ELEvalautor
219     * @throws Exception
220     */
221    public static void resolveInstanceRange(Element event, StringBuilder instances, SyncCoordAction appInst,
222            Configuration conf, ELEvaluator eval) throws Exception {
223        Element eStartInst = event.getChild("start-instance", event.getNamespace());
224        Element eEndInst = event.getChild("end-instance", event.getNamespace());
225        if (eStartInst != null && eEndInst != null) {
226            String strStart = evaluateInstanceFunction(event, appInst, conf, eStartInst.getTextTrim());
227            String strEnd = evaluateInstanceFunction(event, appInst, conf, eEndInst.getTextTrim());
228            checkIfBothSameType(strStart, strEnd);
229            StringBuilder restArg = new StringBuilder(); // To store rest
230                                                         // arguments for
231                                                         // future
232                                                         // function
233            int startIndex = getInstanceNumber(strStart, restArg);
234            String startRestArg = restArg.toString();
235            restArg.delete(0, restArg.length());
236            int endIndex = getInstanceNumber(strEnd, restArg);
237            String endRestArg = restArg.toString();
238            int funcType = getFuncType(strStart);
239
240            if (funcType == ABSOLUTE) {
241                StringBuffer bf = new StringBuffer();
242                bf.append("${coord:absoluteRange(\"").append(parseOneStringArg(strStart))
243                        .append("\",").append(endIndex).append(")}");
244                String matInstance = materializeInstance(event, bf.toString(), appInst, conf, eval);
245                if (matInstance != null && !matInstance.isEmpty()) {
246                    if (instances.length() > 0) {
247                        instances.append(CoordELFunctions.INSTANCE_SEPARATOR);
248                    }
249                    instances.append(matInstance);
250                }
251            }
252            else {
253                if (funcType == OFFSET) {
254                    TimeUnit startU = TimeUnit.valueOf(startRestArg);
255                    TimeUnit endU = TimeUnit.valueOf(endRestArg);
256                    if (startU.getCalendarUnit() * startIndex > endU.getCalendarUnit() * endIndex) {
257                        throw new CommandException(ErrorCode.E1010,
258                                " start-instance should be equal or earlier than the end-instance \n"
259                                        + XmlUtils.prettyPrint(event));
260                    }
261                    Calendar startCal = CoordELFunctions.resolveOffsetRawTime(startIndex, startU, eval);
262                    Calendar endCal = CoordELFunctions.resolveOffsetRawTime(endIndex, endU, eval);
263                    if (startCal != null && endCal != null) {
264                        List<Integer> expandedFreqs = CoordELFunctions.expandOffsetTimes(startCal, endCal, eval);
265                        for (int i = expandedFreqs.size() - 1; i >= 0; i--) {
266                            //we need to use DS timeout, bcz expandOffsetTimes will expand offset in Freqs in DS timeunit
267                            String matInstance = materializeInstance(event, "${coord:offset(" + expandedFreqs.get(i)
268                                    + ", \"" + CoordELFunctions.getDSTimeUnit(eval) + "\")}", appInst, conf, eval);
269                            if (matInstance == null || matInstance.length() == 0) {
270                                // Earlier than dataset's initial instance
271                                break;
272                            }
273                            if (instances.length() > 0) {
274                                instances.append(CoordELFunctions.INSTANCE_SEPARATOR);
275                            }
276                            instances.append(matInstance);
277                        }
278                    }
279                }
280                else {
281                    if (startIndex > endIndex) {
282                        throw new CommandException(ErrorCode.E1010,
283                                " start-instance should be equal or earlier than the end-instance \n"
284                                        + XmlUtils.prettyPrint(event));
285                    }
286                    if (funcType == CURRENT) {
287                        // Everything could be resolved NOW. no latest() ELs
288                        String matInstance = materializeInstance(event, "${coord:currentRange(" + startIndex + ","
289                                + endIndex + ")}", appInst, conf, eval);
290                        if (matInstance != null && !matInstance.isEmpty()) {
291                            if (instances.length() > 0) {
292                                instances.append(CoordELFunctions.INSTANCE_SEPARATOR);
293                            }
294                            instances.append(matInstance);
295                        }
296                    }
297
298                    else { // latest(n)/future() EL is present
299                        if (funcType == LATEST) {
300                            instances.append("${coord:latestRange(").append(startIndex).append(",").append(endIndex)
301                            .append(")}");
302                        }
303                        else if (funcType == FUTURE) {
304                            instances.append("${coord:futureRange(").append(startIndex).append(",").append(endIndex)
305                            .append(",'").append(endRestArg).append("')}");
306                        }
307                    }
308                }
309            }
310            // Remove start-instance and end-instances
311            event.removeChild("start-instance", event.getNamespace());
312            event.removeChild("end-instance", event.getNamespace());
313        }
314    }
315
316    /**
317     * Materialize one instance like current(-2)
318     *
319     * @param event : <data-in>
320     * @param expr : instance like current(-1)
321     * @param appInst : application specific info
322     * @param conf
323     * @param evalInst :ELEvaluator
324     * @return materialized date string
325     * @throws Exception
326     */
327    public static String materializeInstance(Element event, String expr, SyncCoordAction appInst, Configuration conf,
328            ELEvaluator evalInst) throws Exception {
329        if (event == null) {
330            return null;
331        }
332        // ELEvaluator eval = CoordELEvaluator.createInstancesELEvaluator(event,
333        // appInst, conf);
334        return CoordELFunctions.evalAndWrap(evalInst, expr);
335    }
336
337    /**
338     * Create two new tags with <uris> and <unresolved-instances>.
339     *
340     * @param event
341     * @param instances
342     * @throws Exception
343     */
344    private static String separateResolvedAndUnresolved(Element event, StringBuilder instances)
345            throws Exception {
346        StringBuilder unresolvedInstances = new StringBuilder();
347        StringBuilder urisWithDoneFlag = new StringBuilder();
348        StringBuilder depList = new StringBuilder();
349        String uris = createEarlyURIs(event, instances.toString(), unresolvedInstances, urisWithDoneFlag);
350        if (uris.length() > 0) {
351            Element uriInstance = new Element("uris", event.getNamespace());
352            uriInstance.addContent(uris);
353            event.getContent().add(1, uriInstance);
354            if (depList.length() > 0) {
355                depList.append(CoordELFunctions.INSTANCE_SEPARATOR);
356            }
357            depList.append(urisWithDoneFlag);
358        }
359        if (unresolvedInstances.length() > 0) {
360            Element elemInstance = new Element(UNRESOLVED_INST_TAG, event.getNamespace());
361            elemInstance.addContent(unresolvedInstances.toString());
362            event.getContent().add(1, elemInstance);
363        }
364        return depList.toString();
365    }
366
367    /**
368     * The function create a list of URIs separated by "," using the instances
369     * time stamp and URI-template
370     *
371     * @param event : <data-in> event
372     * @param instances : List of time stamp separated by ","
373     * @param unresolvedInstances : list of instance with latest function
374     * @param urisWithDoneFlag : list of URIs with the done flag appended
375     * @return : list of URIs separated by ";" as a string.
376     * @throws Exception
377     */
378    public static String createEarlyURIs(Element event, String instances, StringBuilder unresolvedInstances,
379            StringBuilder urisWithDoneFlag) throws Exception {
380        if (instances == null || instances.length() == 0) {
381            return "";
382        }
383        String[] instanceList = instances.split(CoordELFunctions.INSTANCE_SEPARATOR);
384        StringBuilder uris = new StringBuilder();
385
386        Element doneFlagElement = event.getChild("dataset", event.getNamespace()).getChild("done-flag",
387                event.getNamespace());
388        URIHandlerService uriService = Services.get().get(URIHandlerService.class);
389
390        for (int i = 0; i < instanceList.length; i++) {
391            if (instanceList[i].trim().length() == 0) {
392                continue;
393            }
394            int funcType = getFuncType(instanceList[i]);
395            if (funcType == LATEST || funcType == FUTURE) {
396                if (unresolvedInstances.length() > 0) {
397                    unresolvedInstances.append(CoordELFunctions.INSTANCE_SEPARATOR);
398                }
399                unresolvedInstances.append(instanceList[i]);
400                continue;
401            }
402            ELEvaluator eval = CoordELEvaluator.createURIELEvaluator(instanceList[i]);
403            if (uris.length() > 0) {
404                uris.append(CoordELFunctions.INSTANCE_SEPARATOR);
405                urisWithDoneFlag.append(CoordELFunctions.INSTANCE_SEPARATOR);
406            }
407
408            String uriPath = CoordELFunctions.evalAndWrap(eval, event.getChild("dataset", event.getNamespace())
409                    .getChild("uri-template", event.getNamespace()).getTextTrim());
410            URIHandler uriHandler = uriService.getURIHandler(uriPath);
411            uriHandler.validate(uriPath);
412            uris.append(uriPath);
413            urisWithDoneFlag.append(uriHandler.getURIWithDoneFlag(uriPath, CoordUtils.getDoneFlag(doneFlagElement)));
414        }
415        return uris.toString();
416    }
417
418    /**
419     * @param eAction
420     * @param coordAction
421     * @param conf
422     * @return boolean to determine whether the SLA element is present or not
423     * @throws CoordinatorJobException
424     */
425    public static boolean materializeSLA(Element eAction, CoordinatorActionBean coordAction, Configuration conf)
426            throws CoordinatorJobException {
427        Element eSla = eAction.getChild("action", eAction.getNamespace()).getChild("info", eAction.getNamespace("sla"));
428        if (eSla == null) {
429            // eAppXml.getNamespace("sla"));
430            return false;
431        }
432        try {
433            ELEvaluator evalSla = CoordELEvaluator.createSLAEvaluator(eAction, coordAction, conf);
434            List<Element> elemList = eSla.getChildren();
435            for (Element elem : elemList) {
436                String updated;
437                try {
438                    updated = CoordELFunctions.evalAndWrap(evalSla, elem.getText().trim());
439                }
440                catch (Exception e) {
441                    throw new CoordinatorJobException(ErrorCode.E1004, e.getMessage(), e);
442                }
443                elem.removeContent();
444                elem.addContent(updated);
445            }
446        }
447        catch (Exception e) {
448            throw new CoordinatorJobException(ErrorCode.E1004, e.getMessage(), e);
449        }
450        return true;
451    }
452
453    /**
454     * Materialize one instance for specific nominal time. It includes: 1.
455     * Materialize data events (i.e. <data-in> and <data-out>) 2. Materialize
456     * data properties (i.e dataIn(<DS>) and dataOut(<DS>) 3. remove 'start' and
457     * 'end' tag 4. Add 'instance_number' and 'nominal-time' tag
458     *
459     * @param jobId coordinator job id
460     * @param dryrun true if it is dryrun
461     * @param eAction frequency unexploded-job
462     * @param nominalTime materialization time
463     * @param actualTime action actual time
464     * @param instanceCount instance numbers
465     * @param conf job configuration
466     * @param actionBean CoordinatorActionBean to materialize
467     * @return one materialized action for specific nominal time
468     * @throws Exception
469     */
470    @SuppressWarnings("unchecked")
471    public static String materializeOneInstance(String jobId, boolean dryrun, Element eAction, Date nominalTime,
472            Date actualTime, int instanceCount, Configuration conf, CoordinatorActionBean actionBean) throws Exception {
473        String actionId = Services.get().get(UUIDService.class).generateChildId(jobId, instanceCount + "");
474        SyncCoordAction appInst = new SyncCoordAction();
475        appInst.setActionId(actionId);
476        appInst.setName(eAction.getAttributeValue("name"));
477        appInst.setNominalTime(nominalTime);
478        appInst.setActualTime(actualTime);
479        String frequency = eAction.getAttributeValue("frequency");
480        appInst.setFrequency(frequency);
481        appInst.setTimeUnit(TimeUnit.valueOf(eAction.getAttributeValue("freq_timeunit")));
482        appInst.setTimeZone(DateUtils.getTimeZone(eAction.getAttributeValue("timezone")));
483        appInst.setEndOfDuration(TimeUnit.valueOf(eAction.getAttributeValue("end_of_duration")));
484
485        Map<String, StringBuilder> dependencyMap = null;
486
487        Element inputList = eAction.getChild("input-events", eAction.getNamespace());
488        List<Element> dataInList = null;
489        if (inputList != null) {
490            dataInList = inputList.getChildren("data-in", eAction.getNamespace());
491            dependencyMap = materializeDataEvents(dataInList, appInst, conf);
492        }
493
494        Element outputList = eAction.getChild("output-events", eAction.getNamespace());
495        List<Element> dataOutList = null;
496        if (outputList != null) {
497            dataOutList = outputList.getChildren("data-out", eAction.getNamespace());
498            materializeDataEvents(dataOutList, appInst, conf);
499        }
500
501        eAction.removeAttribute("start");
502        eAction.removeAttribute("end");
503        eAction.setAttribute("instance-number", Integer.toString(instanceCount));
504        eAction.setAttribute("action-nominal-time", DateUtils.formatDateOozieTZ(nominalTime));
505        eAction.setAttribute("action-actual-time", DateUtils.formatDateOozieTZ(actualTime));
506
507        // Setting up action bean
508        actionBean.setCreatedConf(XmlUtils.prettyPrint(conf).toString());
509        actionBean.setRunConf(XmlUtils.prettyPrint(conf).toString());
510        actionBean.setCreatedTime(actualTime);
511        actionBean.setJobId(jobId);
512        actionBean.setId(actionId);
513        actionBean.setLastModifiedTime(new Date());
514        actionBean.setStatus(CoordinatorAction.Status.WAITING);
515        actionBean.setActionNumber(instanceCount);
516        if (dependencyMap != null) {
517            StringBuilder sbPull = dependencyMap.get(DependencyType.PULL.name());
518            if (sbPull != null) {
519                actionBean.setMissingDependencies(sbPull.toString());
520            }
521            StringBuilder sbPush = dependencyMap.get(DependencyType.PUSH.name());
522            if (sbPush != null) {
523                actionBean.setPushMissingDependencies(sbPush.toString());
524            }
525        }
526        actionBean.setNominalTime(nominalTime);
527        boolean isSla = CoordCommandUtils.materializeSLA(eAction, actionBean, conf);
528        if (isSla == true) {
529            actionBean.setSlaXml(XmlUtils.prettyPrint(
530                    eAction.getChild("action", eAction.getNamespace()).getChild("info", eAction.getNamespace("sla")))
531                    .toString());
532        }
533
534        // actionBean.setTrackerUri(trackerUri);//TOOD:
535        // actionBean.setConsoleUrl(consoleUrl); //TODO:
536        // actionBean.setType(type);//TODO:
537        // actionBean.setErrorInfo(errorCode, errorMessage); //TODO:
538        // actionBean.setExternalStatus(externalStatus);//TODO
539        if (!dryrun) {
540            return XmlUtils.prettyPrint(eAction).toString();
541        }
542        else {
543            return dryRunCoord(eAction, actionBean);
544        }
545    }
546
547    /**
548     * @param eAction the actionXml related element
549     * @param actionBean the coordinator action bean
550     * @return
551     * @throws Exception
552     */
553    static String dryRunCoord(Element eAction, CoordinatorActionBean actionBean) throws Exception {
554        String action = XmlUtils.prettyPrint(eAction).toString();
555        StringBuilder actionXml = new StringBuilder(action);
556        Configuration actionConf = new XConfiguration(new StringReader(actionBean.getRunConf()));
557
558        boolean isPushDepAvailable = true;
559        if (actionBean.getPushMissingDependencies() != null) {
560            ActionDependency actionDep = DependencyChecker.checkForAvailability(
561                    actionBean.getPushMissingDependencies(), actionConf, true);
562            if (actionDep.getMissingDependencies().size() != 0) {
563                isPushDepAvailable = false;
564            }
565
566        }
567        boolean isPullDepAvailable = true;
568        CoordActionInputCheckXCommand coordActionInput = new CoordActionInputCheckXCommand(actionBean.getId(),
569                actionBean.getJobId());
570        if (actionBean.getMissingDependencies() != null) {
571            StringBuilder existList = new StringBuilder();
572            StringBuilder nonExistList = new StringBuilder();
573            StringBuilder nonResolvedList = new StringBuilder();
574            getResolvedList(actionBean.getMissingDependencies(), nonExistList, nonResolvedList);
575            isPullDepAvailable = coordActionInput.checkInput(actionXml, existList, nonExistList, actionConf);
576        }
577
578        if (isPullDepAvailable && isPushDepAvailable) {
579            // Check for latest/future
580            boolean isLatestFutureDepAvailable = coordActionInput.checkUnResolvedInput(actionXml, actionConf);
581            if (isLatestFutureDepAvailable) {
582                String newActionXml = CoordActionInputCheckXCommand.resolveCoordConfiguration(actionXml, actionConf,
583                        actionBean.getId());
584                actionXml.replace(0, actionXml.length(), newActionXml);
585            }
586        }
587
588        return actionXml.toString();
589    }
590
591    /**
592     * Materialize all <input-events>/<data-in> or <output-events>/<data-out>
593     * tags Create uris for resolved instances. Create unresolved instance for
594     * latest()/future().
595     *
596     * @param events
597     * @param appInst
598     * @param conf
599     * @throws Exception
600     */
601    public static Map<String, StringBuilder> materializeDataEvents(List<Element> events, SyncCoordAction appInst, Configuration conf
602            ) throws Exception {
603
604        if (events == null) {
605            return null;
606        }
607        StringBuilder unresolvedList = new StringBuilder();
608        Map<String, StringBuilder> dependencyMap = new HashMap<String, StringBuilder>();
609        URIHandlerService uriService = Services.get().get(URIHandlerService.class);
610        StringBuilder pullMissingDep = null;
611        StringBuilder pushMissingDep = null;
612
613        for (Element event : events) {
614            StringBuilder instances = new StringBuilder();
615            ELEvaluator eval = CoordELEvaluator.createInstancesELEvaluator(event, appInst, conf);
616            // Handle list of instance tag
617            resolveInstances(event, instances, appInst, conf, eval);
618            // Handle start-instance and end-instance
619            resolveInstanceRange(event, instances, appInst, conf, eval);
620            // Separate out the unresolved instances
621            String resolvedList = separateResolvedAndUnresolved(event, instances);
622            if (!resolvedList.isEmpty()) {
623                Element uri = event.getChild("dataset", event.getNamespace()).getChild("uri-template",
624                        event.getNamespace());
625                String uriTemplate = uri.getText();
626                URI baseURI = uriService.getAuthorityWithScheme(uriTemplate);
627                URIHandler handler = uriService.getURIHandler(baseURI);
628                if (handler.getDependencyType(baseURI).equals(DependencyType.PULL)) {
629                    pullMissingDep = (pullMissingDep == null) ? new StringBuilder(resolvedList) : pullMissingDep.append(
630                            CoordELFunctions.INSTANCE_SEPARATOR).append(resolvedList);
631                }
632                else {
633                    pushMissingDep = (pushMissingDep == null) ? new StringBuilder(resolvedList) : pushMissingDep.append(
634                            CoordELFunctions.INSTANCE_SEPARATOR).append(resolvedList);
635                }
636            }
637
638            String tmpUnresolved = event.getChildTextTrim(UNRESOLVED_INST_TAG, event.getNamespace());
639            if (tmpUnresolved != null) {
640                if (unresolvedList.length() > 0) {
641                    unresolvedList.append(CoordELFunctions.INSTANCE_SEPARATOR);
642                }
643                unresolvedList.append(tmpUnresolved);
644            }
645        }
646        if (unresolvedList.length() > 0) {
647            if (pullMissingDep == null) {
648                pullMissingDep = new StringBuilder();
649            }
650            pullMissingDep.append(RESOLVED_UNRESOLVED_SEPARATOR).append(unresolvedList);
651        }
652        dependencyMap.put(DependencyType.PULL.name(), pullMissingDep);
653        dependencyMap.put(DependencyType.PUSH.name(), pushMissingDep);
654        return dependencyMap;
655    }
656
657    /**
658     * Get resolved string from missDepList
659     *
660     * @param missDepList
661     * @param resolved
662     * @param unresolved
663     * @return resolved string
664     */
665    public static String getResolvedList(String missDepList, StringBuilder resolved, StringBuilder unresolved) {
666        if (missDepList != null) {
667            int index = missDepList.indexOf(RESOLVED_UNRESOLVED_SEPARATOR);
668            if (index < 0) {
669                resolved.append(missDepList);
670            }
671            else {
672                resolved.append(missDepList.substring(0, index));
673                unresolved.append(missDepList.substring(index + RESOLVED_UNRESOLVED_SEPARATOR.length()));
674            }
675        }
676        return resolved.toString();
677    }
678
679    /**
680     * Get the next action time after a given time
681     *
682     * @param targetDate
683     * @param coordJob
684     * @return the next valid action time
685     */
686    public static Date getNextValidActionTimeForCronFrequency(Date targetDate, CoordinatorJobBean coordJob) throws ParseException {
687
688        String freq = coordJob.getFrequency();
689        TimeZone tz = DateUtils.getOozieProcessingTimeZone();
690        String[] cronArray = freq.split(" ");
691        Date nextTime = null;
692
693        // Current CronExpression doesn't support operations
694        // where both date of months and day of weeks are specified.
695        // As a result, we need to split this scenario into two cases
696        // and return the earlier time
697        if (!cronArray[2].trim().equals("?") && !cronArray[4].trim().equals("?")) {
698
699            // When any one of day of month or day of week fields is a wildcard
700            // we need to replace the wildcard with "?"
701            if (cronArray[2].trim().equals("*") || cronArray[4].trim().equals("*")) {
702                if (cronArray[2].trim().equals("*")) {
703                    cronArray[2] = "?";
704                }
705                else {
706                    cronArray[4] = "?";
707                }
708                freq= StringUtils.join(cronArray, " ");
709
710                // The cronExpression class takes second
711                // as the first field where oozie is operating on
712                // minute basis
713                CronExpression expr = new CronExpression("0 " + freq);
714                expr.setTimeZone(tz);
715                nextTime = expr.getNextValidTimeAfter(targetDate);
716            }
717            // If both fields are specified by non-wildcards,
718            // we need to split it into two expressions
719            else {
720                String[] cronArray1 = freq.split(" ");
721                String[] cronArray2 = freq.split(" ");
722
723                cronArray1[2] = "?";
724                cronArray2[4] = "?";
725
726                String freq1 = StringUtils.join(cronArray1, " ");
727                String freq2 = StringUtils.join(cronArray2, " ");
728
729                // The cronExpression class takes second
730                // as the first field where oozie is operating on
731                // minute basis
732                CronExpression expr1 = new CronExpression("0 " + freq1);
733                expr1.setTimeZone(tz);
734                CronExpression expr2 = new CronExpression("0 " + freq2);
735                expr2.setTimeZone(tz);
736                nextTime = expr1.getNextValidTimeAfter(targetDate);
737                Date nextTime2 = expr2.getNextValidTimeAfter(targetDate);
738                nextTime = nextTime.compareTo(nextTime2) < 0 ? nextTime: nextTime2;
739            }
740        }
741        else {
742            // The cronExpression class takes second
743            // as the first field where oozie is operating on
744            // minute basis
745            CronExpression expr  = new CronExpression("0 " + freq);
746            expr.setTimeZone(tz);
747            nextTime = expr.getNextValidTimeAfter(targetDate);
748        }
749
750        return nextTime;
751    }
752
753    /**
754     * Computes the nominal time of the next action.
755     * Based on CoordMaterializeTransitionXCommand#materializeActions
756     *
757     * The Coordinator Job needs to have the frequency, time unit, time zone, start time, end time, and job xml.
758     * The Coordinator Action needs to have the nominal time and action number.
759     *
760     * @param coordJob The Coordinator Job
761     * @param coordAction The Coordinator Action
762     * @return the nominal time of the next action
763     * @throws ParseException
764     * @throws JDOMException
765     */
766    public static Date computeNextNominalTime(CoordinatorJobBean coordJob, CoordinatorActionBean coordAction)
767            throws ParseException, JDOMException {
768        Date nextNominalTime;
769        boolean isCronFrequency = false;
770        int freq = -1;
771        try {
772            freq = Integer.parseInt(coordJob.getFrequency());
773        } catch (NumberFormatException e) {
774            isCronFrequency = true;
775        }
776
777        if (isCronFrequency) {
778            nextNominalTime = CoordCommandUtils.getNextValidActionTimeForCronFrequency(coordAction.getNominalTime(), coordJob);
779        } else {
780            TimeZone appTz = DateUtils.getTimeZone(coordJob.getTimeZone());
781            Calendar nextNominalTimeCal = Calendar.getInstance(appTz);
782            nextNominalTimeCal.setTime(coordJob.getStartTimestamp());
783            TimeUnit freqTU = TimeUnit.valueOf(coordJob.getTimeUnitStr());
784            // Action Number is indexed by 1, so no need to +1 here
785            nextNominalTimeCal.add(freqTU.getCalendarUnit(), coordAction.getActionNumber() * freq);
786            String jobXml = coordJob.getJobXml();
787            Element eJob = XmlUtils.parseXml(jobXml);
788            TimeUnit endOfFlag = TimeUnit.valueOf(eJob.getAttributeValue("end_of_duration"));
789            // Move to the End of duration, if needed.
790            DateUtils.moveToEnd(nextNominalTimeCal, endOfFlag);
791            nextNominalTime = nextNominalTimeCal.getTime();
792        }
793
794        // If the next nominal time is after the job's end time, then this is the last action, so return null
795        if (nextNominalTime.after(coordJob.getEndTime())) {
796            nextNominalTime = null;
797        }
798        return nextNominalTime;
799    }
800}