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}