Source code

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.coord;
020
021import java.util.Calendar;
022import java.util.Date;
023import java.util.HashMap;
024import java.util.Iterator;
025import java.util.List;
026import java.util.Map;
027
028import org.apache.hadoop.conf.Configuration;
029import org.apache.oozie.CoordinatorActionBean;
030import org.apache.oozie.command.coord.CoordCommandUtils;
031import org.apache.oozie.coord.input.dependency.CoordInputDependency;
032import org.apache.oozie.coord.input.logic.CoordInputLogicEvaluator;
033import org.apache.oozie.service.ELService;
034import org.apache.oozie.service.Services;
035import org.apache.oozie.util.DateUtils;
036import org.apache.oozie.util.ELEvaluator;
037import org.apache.oozie.util.XmlUtils;
038import org.jdom.Element;
039
040/**
041 * This class provide different evaluators required at different stages
042 */
043public class CoordELEvaluator {
044    public static final Integer MINUTE = 1;
045    public static final Integer HOUR = 60 * MINUTE;
046
047    /**
048     * Create an evaluator to be used in resolving configuration vars and frequency constant/functions (used in Stage
049     * 1)
050     *
051     * @param conf : Configuration containing property variables
052     * @return configured ELEvaluator
053     */
054    public static ELEvaluator createELEvaluatorForGroup(Configuration conf, String group) {
055        ELEvaluator eval = Services.get().get(ELService.class).createEvaluator(group);
056        setConfigToEval(eval, conf);
057        return eval;
058    }
059
060    /**
061     * Create a new Evaluator to resolve the EL functions and variables using action creation time (Phase 2)
062     *
063     * @param event : Xml element for data-in element usually enclosed by &lt;data-in(out)&gt; tag
064     * @param appInst : Application Instance related information such as Action creation Time
065     * @param conf :Configuration to substitute any variables
066     * @return configured ELEvaluator
067     * @throws Exception : If there is any date-time string in wrong format, the exception is thrown
068     */
069    public static ELEvaluator createInstancesELEvaluator(Element event, SyncCoordAction appInst, Configuration conf)
070            throws Exception {
071        return createInstancesELEvaluator("coord-action-create", event, appInst, conf);
072    }
073
074    public static ELEvaluator createInstancesELEvaluator(String tag, Element event, SyncCoordAction appInst,
075                                                         Configuration conf) throws Exception {
076        ELEvaluator eval = Services.get().get(ELService.class).createEvaluator(tag);
077        setConfigToEval(eval, conf);
078        SyncCoordDataset ds = getDSObject(event);
079        CoordELFunctions.configureEvaluator(eval, ds, appInst);
080        return eval;
081    }
082
083    public static ELEvaluator createELEvaluatorForDataEcho(Configuration conf, String group,
084                                                           HashMap<String, String> dataNameList) throws Exception {
085        ELEvaluator eval = createELEvaluatorForGroup(conf, group);
086        for (Iterator<String> it = dataNameList.keySet().iterator(); it.hasNext();) {
087            String key = it.next();
088            String value = dataNameList.get(key);
089            eval.setVariable("oozie.dataname." + key, value);
090        }
091        return eval;
092    }
093
094    /**
095     * Create a new evaluator for Lazy resolve (phase 3). For example, coord_latest(n) and coord_actualTime()function
096     * should be resolved when all other data dependencies are met.
097     *
098     * @param actualTime : Action start time
099     * @param nominalTime : Action creation time
100     * @param dEvent :XML element for data-in element usually enclosed by &lt;data-in(out)&gt; tag
101     * @param conf :Configuration to substitute any variables
102     * @return configured ELEvaluator
103     * @throws Exception : If there is any date-time string in wrong format, the exception is thrown
104     */
105    public static ELEvaluator createLazyEvaluator(Date actualTime, Date nominalTime, Element dEvent, Configuration conf)
106            throws Exception {
107        ELEvaluator eval = Services.get().get(ELService.class).createEvaluator("coord-action-start");
108        setConfigToEval(eval, conf);
109        SyncCoordDataset ds = getDSObject(dEvent);
110        SyncCoordAction appInst = new SyncCoordAction();
111        appInst.setNominalTime(nominalTime);
112        appInst.setActualTime(actualTime);
113        CoordELFunctions.configureEvaluator(eval, ds, appInst);
114        eval.setVariable(CoordELFunctions.CONFIGURATION, conf);
115        return eval;
116    }
117
118    /**
119     * Create a SLA evaluator to be used during Materialization
120     * @param eAction
121     * @param coordAction
122     * @param conf
123     * @return
124     * @throws Exception
125     */
126    public static ELEvaluator createSLAEvaluator(Element eAction, CoordinatorActionBean coordAction, Configuration conf)
127            throws Exception {
128        ELEvaluator eval = Services.get().get(ELService.class).createEvaluator("coord-sla-create");
129        setConfigToEval(eval, conf);
130        SyncCoordAction appInst = new SyncCoordAction();// TODO:
131        appInst.setNominalTime(coordAction.getNominalTime());
132        appInst.setActualTime(coordAction.getCreatedTime());
133        appInst.setActionId(coordAction.getId());
134        appInst.setName(eAction.getAttributeValue("name"));
135        CoordELFunctions.configureEvaluator(eval, null, appInst);
136
137        Element events = eAction.getChild("output-events", eAction.getNamespace());
138        if (events != null) {
139            for (Object obj : events.getChildren("data-out", eAction.getNamespace())) {
140                Element data = (Element) obj;
141                if (data.getChild("uris", data.getNamespace()) != null) {
142                    String uris = data.getChild("uris", data.getNamespace()).getTextTrim();
143                    uris = uris.replaceAll(CoordELFunctions.INSTANCE_SEPARATOR, CoordELFunctions.DIR_SEPARATOR);
144                    eval.setVariable(".dataout." + data.getAttributeValue("name"), uris);
145                }
146                if (data.getChild(CoordCommandUtils.UNRESOLVED_INSTANCES_TAG, data.getNamespace()) != null) {
147                    eval.setVariable(".dataout." + data.getAttributeValue("name") + ".unresolved", "true");
148                }
149            }
150        }
151        return eval;
152    }
153
154    /**
155     * Create an Evaluator using conf and input/output-data (used for sla)
156     * @param conf
157     * @param group
158     * @param dataNameList
159     * @return
160     * @throws Exception
161     */
162    public static ELEvaluator createELEvaluatorForDataAndConf(Configuration conf, String group,
163            HashMap<String, String> dataNameList) throws Exception {
164        ELEvaluator eval = createELEvaluatorForDataEcho(conf, group, dataNameList);
165        setConfigToEval(eval, conf);
166        return eval;
167    }
168
169    /**
170     * Create an Evaluator to resolve dataIns and dataOuts of an application instance (used in stage 3)
171     *
172     * @param eJob : XML element for the application instance
173     * @param conf :Configuration to substitute any variables
174     * @return configured ELEvaluator
175     * @throws Exception : If there is any date-time string in wrong format, the exception is thrown
176     */
177
178    public static ELEvaluator createDataEvaluator(Element eJob, Configuration conf, String actionId) throws Exception {
179        return createDataEvaluator(eJob, conf, actionId, null, null);
180    }
181
182    public static ELEvaluator createDataEvaluator(Element eJob, Configuration conf, String actionId,
183            CoordInputDependency pullDependencies, CoordInputDependency pushDependencies) throws Exception {
184        ELEvaluator e = Services.get().get(ELService.class).createEvaluator("coord-action-start");
185        setConfigToEval(e, conf);
186        SyncCoordAction appInst = new SyncCoordAction();
187        String strNominalTime = eJob.getAttributeValue("action-nominal-time");
188        if (strNominalTime != null) {
189            appInst.setNominalTime(DateUtils.parseDateOozieTZ(strNominalTime));
190            appInst.setTimeZone(DateUtils.getTimeZone(eJob.getAttributeValue("timezone")));
191            appInst.setFrequency(eJob.getAttributeValue("frequency"));
192            appInst.setTimeUnit(TimeUnit.valueOf(eJob.getAttributeValue("freq_timeunit")));
193            appInst.setActionId(actionId);
194            appInst.setName(eJob.getAttributeValue("name"));
195            appInst.setPullDependencies(pullDependencies);
196            appInst.setPushDependencies(pushDependencies);
197            if (CoordUtils.isInputLogicSpecified(eJob)) {
198                e.setVariable(".actionInputLogic",
199                        XmlUtils.prettyPrint(eJob.getChild(CoordInputLogicEvaluator.INPUT_LOGIC, eJob.getNamespace())).toString());
200            }
201        }
202        String strActualTime = eJob.getAttributeValue("action-actual-time");
203        if (strActualTime != null) {
204            appInst.setActualTime(DateUtils.parseDateOozieTZ(strActualTime));
205        }
206        CoordELFunctions.configureEvaluator(e, null, appInst);
207        Element events = eJob.getChild("input-events", eJob.getNamespace());
208        if (events != null) {
209            for (Element data : (List<Element>) events.getChildren("data-in", eJob.getNamespace())) {
210                if (data.getChild("uris", data.getNamespace()) != null) {
211                    String uris = data.getChild("uris", data.getNamespace()).getTextTrim();
212                    uris = uris.replaceAll(CoordELFunctions.INSTANCE_SEPARATOR, CoordELFunctions.DIR_SEPARATOR);
213                    e.setVariable(".datain." + data.getAttributeValue("name"), uris);
214                }
215                else {
216                }
217                if (data.getChild(CoordCommandUtils.UNRESOLVED_INSTANCES_TAG, data.getNamespace()) != null) {
218                    e.setVariable(".datain." + data.getAttributeValue("name") + ".unresolved", "true"); // TODO:
219                    // check
220                    // null
221                }
222                Element doneFlagElement = data.getChild("dataset", data.getNamespace()).getChild("done-flag",
223                        data.getNamespace());
224                String doneFlag = CoordUtils.getDoneFlag(doneFlagElement);
225                e.setVariable(".datain." + data.getAttributeValue("name") + ".doneFlag", doneFlag);
226            }
227        }
228        events = eJob.getChild("output-events", eJob.getNamespace());
229        if (events != null) {
230            for (Element data : (List<Element>) events.getChildren("data-out", eJob.getNamespace())) {
231                if (data.getChild("uris", data.getNamespace()) != null) {
232                    String uris = data.getChild("uris", data.getNamespace()).getTextTrim();
233                    uris = uris.replaceAll(CoordELFunctions.INSTANCE_SEPARATOR, CoordELFunctions.DIR_SEPARATOR);
234                    e.setVariable(".dataout." + data.getAttributeValue("name"), uris);
235                }
236                else {
237                }// TODO
238                if (data.getChild(CoordCommandUtils.UNRESOLVED_INSTANCES_TAG, data.getNamespace()) != null) {
239                    e.setVariable(".dataout." + data.getAttributeValue("name") + ".unresolved", "true"); // TODO:
240                    // check
241                    // null
242                }
243            }
244        }
245        return e;
246    }
247
248    /**
249     * Create a new Evaluator to resolve URI temple with time specific constant
250     *
251     * @param strDate : Date-time
252     * @return configured ELEvaluator
253     * @throws Exception If there is any date-time string in wrong format, the exception is thrown
254     */
255    public static ELEvaluator createURIELEvaluator(String strDate) throws Exception {
256        ELEvaluator eval = new ELEvaluator();
257        Calendar date = Calendar.getInstance(DateUtils.getOozieProcessingTimeZone());
258        // always???
259        date.setTime(DateUtils.parseDateOozieTZ(strDate));
260        eval.setVariable("YEAR", date.get(Calendar.YEAR));
261        eval.setVariable("MONTH", make2Digits(date.get(Calendar.MONTH) + 1));
262        eval.setVariable("DAY", make2Digits(date.get(Calendar.DAY_OF_MONTH)));
263        eval.setVariable("HOUR", make2Digits(date.get(Calendar.HOUR_OF_DAY)));
264        eval.setVariable("MINUTE", make2Digits(date.get(Calendar.MINUTE)));
265        return eval;
266    }
267
268    /**
269     * Create Dataset object using the Dataset XML information
270     *
271     * @param eData
272     * @return
273     * @throws Exception
274     */
275    private static SyncCoordDataset getDSObject(Element eData) throws Exception {
276        SyncCoordDataset ds = new SyncCoordDataset();
277        Element eDataset = eData.getChild("dataset", eData.getNamespace());
278        // System.out.println("eDATA :"+ XmlUtils.prettyPrint(eData));
279        Date initInstance = DateUtils.parseDateOozieTZ(eDataset.getAttributeValue("initial-instance"));
280        ds.setInitInstance(initInstance);
281        if (eDataset.getAttributeValue("frequency") != null) {
282            int frequency = Integer.parseInt(eDataset.getAttributeValue("frequency"));
283            ds.setFrequency(frequency);
284            ds.setType("SYNC");
285            if (eDataset.getAttributeValue("freq_timeunit") == null) {
286                throw new RuntimeException("No freq_timeunit defined in data set definition\n"
287                        + XmlUtils.prettyPrint(eDataset));
288            }
289            ds.setTimeUnit(TimeUnit.valueOf(eDataset.getAttributeValue("freq_timeunit")));
290            if (eDataset.getAttributeValue("timezone") == null) {
291                throw new RuntimeException("No timezone defined in data set definition\n"
292                        + XmlUtils.prettyPrint(eDataset));
293            }
294            ds.setTimeZone(DateUtils.getTimeZone(eDataset.getAttributeValue("timezone")));
295            if (eDataset.getAttributeValue("end_of_duration") == null) {
296                throw new RuntimeException("No end_of_duration defined in data set definition\n"
297                        + XmlUtils.prettyPrint(eDataset));
298            }
299            ds.setEndOfDuration(TimeUnit.valueOf(eDataset.getAttributeValue("end_of_duration")));
300
301            Element doneFlagElement = eDataset.getChild("done-flag", eData.getNamespace());
302            String doneFlag = CoordUtils.getDoneFlag(doneFlagElement);
303            ds.setDoneFlag(doneFlag);
304        }
305        else {
306            ds.setType("ASYNC");
307        }
308        String name = eDataset.getAttributeValue("name");
309        ds.setName(name);
310        // System.out.println(name + " VAL "+ eDataset.getChild("uri-template",
311        // eData.getNamespace()));
312        String uriTemplate = eDataset.getChild("uri-template", eData.getNamespace()).getTextTrim();
313        ds.setUriTemplate(uriTemplate);
314        // ds.setTimeUnit(TimeUnit.MINUTES);
315        return ds;
316    }
317
318    /**
319     * Set all job configurations properties into evaluator.
320     *
321     * @param eval : Evaluator to set variables
322     * @param conf : configurations to set Evaluator
323     */
324    private static void setConfigToEval(ELEvaluator eval, Configuration conf) {
325        for (Map.Entry<String, String> entry : conf) {
326            eval.setVariable(entry.getKey(), entry.getValue().trim());
327        }
328    }
329
330    /**
331     * make any one digit number to two digit string pre-appending a"0"
332     *
333     * @param num : number to make sting
334     * @return :String of length at least two digit.
335     */
336    private static String make2Digits(int num) {
337        String ret = "" + num;
338        if (num <= 9) {
339            ret = "0" + ret;
340        }
341        return ret;
342    }
343}