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.IOException; 022import java.io.InputStreamReader; 023import java.io.Reader; 024import java.io.StringReader; 025import java.io.StringWriter; 026import java.net.URI; 027import java.net.URISyntaxException; 028import java.util.ArrayList; 029import java.util.Calendar; 030import java.util.Date; 031import java.util.HashMap; 032import java.util.HashSet; 033import java.util.Iterator; 034import java.util.List; 035import java.util.Map; 036import java.util.Set; 037import java.util.TreeSet; 038 039import javax.xml.transform.stream.StreamSource; 040import javax.xml.validation.Validator; 041 042import org.apache.hadoop.conf.Configuration; 043import org.apache.hadoop.fs.FileSystem; 044import org.apache.hadoop.fs.Path; 045import org.apache.oozie.CoordinatorJobBean; 046import org.apache.oozie.ErrorCode; 047import org.apache.oozie.client.CoordinatorJob; 048import org.apache.oozie.client.Job; 049import org.apache.oozie.client.OozieClient; 050import org.apache.oozie.client.CoordinatorJob.Execution; 051import org.apache.oozie.command.CommandException; 052import org.apache.oozie.command.SubmitTransitionXCommand; 053import org.apache.oozie.command.bundle.BundleStatusUpdateXCommand; 054import org.apache.oozie.coord.CoordELEvaluator; 055import org.apache.oozie.coord.CoordELFunctions; 056import org.apache.oozie.coord.CoordinatorJobException; 057import org.apache.oozie.coord.TimeUnit; 058import org.apache.oozie.executor.jpa.CoordJobQueryExecutor; 059import org.apache.oozie.executor.jpa.JPAExecutorException; 060import org.apache.oozie.service.CoordMaterializeTriggerService; 061import org.apache.oozie.service.ConfigurationService; 062import org.apache.oozie.service.HadoopAccessorException; 063import org.apache.oozie.service.HadoopAccessorService; 064import org.apache.oozie.service.JPAService; 065import org.apache.oozie.service.SchemaService; 066import org.apache.oozie.service.Service; 067import org.apache.oozie.service.Services; 068import org.apache.oozie.service.UUIDService; 069import org.apache.oozie.service.SchemaService.SchemaName; 070import org.apache.oozie.service.UUIDService.ApplicationType; 071import org.apache.oozie.util.ConfigUtils; 072import org.apache.oozie.util.DateUtils; 073import org.apache.oozie.util.ELEvaluator; 074import org.apache.oozie.util.ELUtils; 075import org.apache.oozie.util.IOUtils; 076import org.apache.oozie.util.InstrumentUtils; 077import org.apache.oozie.util.LogUtils; 078import org.apache.oozie.util.ParamChecker; 079import org.apache.oozie.util.ParameterVerifier; 080import org.apache.oozie.util.ParameterVerifierException; 081import org.apache.oozie.util.PropertiesUtils; 082import org.apache.oozie.util.XConfiguration; 083import org.apache.oozie.util.XmlUtils; 084import org.jdom.Attribute; 085import org.jdom.Element; 086import org.jdom.JDOMException; 087import org.jdom.Namespace; 088import org.xml.sax.SAXException; 089 090/** 091 * This class provides the functionalities to resolve a coordinator job XML and write the job information into a DB 092 * table. 093 * <p/> 094 * Specifically it performs the following functions: 1. Resolve all the variables or properties using job 095 * configurations. 2. Insert all datasets definition as part of the <data-in> and <data-out> tags. 3. Validate the XML 096 * at runtime. 097 */ 098public class CoordSubmitXCommand extends SubmitTransitionXCommand { 099 100 protected Configuration conf; 101 private final String bundleId; 102 private final String coordName; 103 protected boolean dryrun; 104 protected JPAService jpaService = null; 105 private CoordinatorJob.Status prevStatus = CoordinatorJob.Status.PREP; 106 107 public static final String CONFIG_DEFAULT = "coord-config-default.xml"; 108 public static final String COORDINATOR_XML_FILE = "coordinator.xml"; 109 public final String COORD_INPUT_EVENTS ="input-events"; 110 public final String COORD_OUTPUT_EVENTS = "output-events"; 111 public final String COORD_INPUT_EVENTS_DATA_IN ="data-in"; 112 public final String COORD_OUTPUT_EVENTS_DATA_OUT = "data-out"; 113 114 private static final Set<String> DISALLOWED_USER_PROPERTIES = new HashSet<String>(); 115 private static final Set<String> DISALLOWED_DEFAULT_PROPERTIES = new HashSet<String>(); 116 117 protected CoordinatorJobBean coordJob = null; 118 /** 119 * Default timeout for normal jobs, in minutes, after which coordinator input check will timeout 120 */ 121 public static final String CONF_DEFAULT_TIMEOUT_NORMAL = Service.CONF_PREFIX + "coord.normal.default.timeout"; 122 123 public static final String CONF_DEFAULT_CONCURRENCY = Service.CONF_PREFIX + "coord.default.concurrency"; 124 125 public static final String CONF_DEFAULT_THROTTLE = Service.CONF_PREFIX + "coord.default.throttle"; 126 127 public static final String CONF_MAT_THROTTLING_FACTOR = Service.CONF_PREFIX 128 + "coord.materialization.throttling.factor"; 129 130 /** 131 * Default MAX timeout in minutes, after which coordinator input check will timeout 132 */ 133 public static final String CONF_DEFAULT_MAX_TIMEOUT = Service.CONF_PREFIX + "coord.default.max.timeout"; 134 135 public static final String CONF_QUEUE_SIZE = Service.CONF_PREFIX + "CallableQueueService.queue.size"; 136 137 public static final String CONF_CHECK_MAX_FREQUENCY = Service.CONF_PREFIX + "coord.check.maximum.frequency"; 138 139 private ELEvaluator evalFreq = null; 140 private ELEvaluator evalNofuncs = null; 141 private ELEvaluator evalData = null; 142 private ELEvaluator evalInst = null; 143 private ELEvaluator evalAction = null; 144 private ELEvaluator evalSla = null; 145 private ELEvaluator evalTimeout = null; 146 private ELEvaluator evalInitialInstance = null; 147 148 static { 149 String[] badUserProps = { PropertiesUtils.YEAR, PropertiesUtils.MONTH, PropertiesUtils.DAY, 150 PropertiesUtils.HOUR, PropertiesUtils.MINUTE, PropertiesUtils.DAYS, PropertiesUtils.HOURS, 151 PropertiesUtils.MINUTES, PropertiesUtils.KB, PropertiesUtils.MB, PropertiesUtils.GB, 152 PropertiesUtils.TB, PropertiesUtils.PB, PropertiesUtils.RECORDS, PropertiesUtils.MAP_IN, 153 PropertiesUtils.MAP_OUT, PropertiesUtils.REDUCE_IN, PropertiesUtils.REDUCE_OUT, PropertiesUtils.GROUPS }; 154 PropertiesUtils.createPropertySet(badUserProps, DISALLOWED_USER_PROPERTIES); 155 156 String[] badDefaultProps = { PropertiesUtils.HADOOP_USER}; 157 PropertiesUtils.createPropertySet(badUserProps, DISALLOWED_DEFAULT_PROPERTIES); 158 PropertiesUtils.createPropertySet(badDefaultProps, DISALLOWED_DEFAULT_PROPERTIES); 159 } 160 161 /** 162 * Constructor to create the Coordinator Submit Command. 163 * 164 * @param conf : Configuration for Coordinator job 165 */ 166 public CoordSubmitXCommand(Configuration conf) { 167 super("coord_submit", "coord_submit", 1); 168 this.conf = ParamChecker.notNull(conf, "conf"); 169 this.bundleId = null; 170 this.coordName = null; 171 } 172 173 /** 174 * Constructor to create the Coordinator Submit Command by bundle job. 175 * 176 * @param conf : Configuration for Coordinator job 177 * @param bundleId : bundle id 178 * @param coordName : coord name 179 */ 180 public CoordSubmitXCommand(Configuration conf, String bundleId, String coordName) { 181 super("coord_submit", "coord_submit", 1); 182 this.conf = ParamChecker.notNull(conf, "conf"); 183 this.bundleId = ParamChecker.notEmpty(bundleId, "bundleId"); 184 this.coordName = ParamChecker.notEmpty(coordName, "coordName"); 185 } 186 187 /** 188 * Constructor to create the Coordinator Submit Command. 189 * 190 * @param dryrun : if dryrun 191 * @param conf : Configuration for Coordinator job 192 */ 193 public CoordSubmitXCommand(boolean dryrun, Configuration conf) { 194 this(conf); 195 this.dryrun = dryrun; 196 } 197 198 /* (non-Javadoc) 199 * @see org.apache.oozie.command.XCommand#execute() 200 */ 201 @Override 202 protected String submit() throws CommandException { 203 LOG.info("STARTED Coordinator Submit"); 204 String jobId = submitJob(); 205 LOG.info("ENDED Coordinator Submit jobId=" + jobId); 206 return jobId; 207 } 208 209 protected String submitJob() throws CommandException { 210 String jobId = null; 211 InstrumentUtils.incrJobCounter(getName(), 1, getInstrumentation()); 212 213 boolean exceptionOccured = false; 214 try { 215 mergeDefaultConfig(); 216 217 String appXml = readAndValidateXml(); 218 coordJob.setOrigJobXml(appXml); 219 LOG.debug("jobXml after initial validation " + XmlUtils.prettyPrint(appXml).toString()); 220 221 Element eXml = XmlUtils.parseXml(appXml); 222 223 String appNamespace = readAppNamespace(eXml); 224 coordJob.setAppNamespace(appNamespace); 225 226 ParameterVerifier.verifyParameters(conf, eXml); 227 228 appXml = XmlUtils.removeComments(appXml); 229 initEvaluators(); 230 Element eJob = basicResolveAndIncludeDS(appXml, conf, coordJob); 231 232 validateCoordinatorJob(); 233 234 // checking if the coordinator application data input/output events 235 // specify multiple data instance values in erroneous manner 236 checkMultipleTimeInstances(eJob, COORD_INPUT_EVENTS, COORD_INPUT_EVENTS_DATA_IN); 237 checkMultipleTimeInstances(eJob, COORD_OUTPUT_EVENTS, COORD_OUTPUT_EVENTS_DATA_OUT); 238 239 LOG.debug("jobXml after all validation " + XmlUtils.prettyPrint(eJob).toString()); 240 241 jobId = storeToDB(appXml, eJob, coordJob); 242 // log job info for coordinator job 243 LogUtils.setLogInfo(coordJob); 244 245 if (!dryrun) { 246 queueMaterializeTransitionXCommand(jobId); 247 } 248 else { 249 return getDryRun(coordJob); 250 } 251 } 252 catch (JDOMException jex) { 253 exceptionOccured = true; 254 LOG.warn("ERROR: ", jex); 255 throw new CommandException(ErrorCode.E0700, jex.getMessage(), jex); 256 } 257 catch (CoordinatorJobException cex) { 258 exceptionOccured = true; 259 LOG.warn("ERROR: ", cex); 260 throw new CommandException(cex); 261 } 262 catch (ParameterVerifierException pex) { 263 exceptionOccured = true; 264 LOG.warn("ERROR: ", pex); 265 throw new CommandException(pex); 266 } 267 catch (IllegalArgumentException iex) { 268 exceptionOccured = true; 269 LOG.warn("ERROR: ", iex); 270 throw new CommandException(ErrorCode.E1003, iex.getMessage(), iex); 271 } 272 catch (Exception ex) { 273 exceptionOccured = true; 274 LOG.warn("ERROR: ", ex); 275 throw new CommandException(ErrorCode.E0803, ex.getMessage(), ex); 276 } 277 finally { 278 if (exceptionOccured) { 279 if (coordJob.getId() == null || coordJob.getId().equalsIgnoreCase("")) { 280 coordJob.setStatus(CoordinatorJob.Status.FAILED); 281 coordJob.resetPending(); 282 } 283 } 284 } 285 return jobId; 286 } 287 288 /** 289 * Gets the dryrun output. 290 * 291 * @param coordJob the coordinatorJobBean 292 * @return the dry run 293 * @throws Exception the exception 294 */ 295 protected String getDryRun(CoordinatorJobBean coordJob) throws Exception{ 296 int materializationWindow = ConfigurationService 297 .getInt(CoordMaterializeTriggerService.CONF_MATERIALIZATION_WINDOW); 298 Date startTime = coordJob.getStartTime(); 299 long startTimeMilli = startTime.getTime(); 300 long endTimeMilli = startTimeMilli + (materializationWindow * 1000); 301 Date jobEndTime = coordJob.getEndTime(); 302 Date endTime = new Date(endTimeMilli); 303 if (endTime.compareTo(jobEndTime) > 0) { 304 endTime = jobEndTime; 305 } 306 String jobId = coordJob.getId(); 307 LOG.info("[" + jobId + "]: Update status to RUNNING"); 308 coordJob.setStatus(Job.Status.RUNNING); 309 coordJob.setPending(); 310 Configuration jobConf = null; 311 try { 312 jobConf = new XConfiguration(new StringReader(coordJob.getConf())); 313 } 314 catch (IOException e1) { 315 LOG.warn("Configuration parse error. read from DB :" + coordJob.getConf(), e1); 316 } 317 String action = new CoordMaterializeTransitionXCommand(coordJob, materializationWindow, startTime, 318 endTime).materializeActions(true); 319 String output = coordJob.getJobXml() + System.getProperty("line.separator") 320 + "***actions for instance***" + action; 321 return output; 322 } 323 324 /** 325 * Queue MaterializeTransitionXCommand 326 */ 327 protected void queueMaterializeTransitionXCommand(String jobId) { 328 int materializationWindow = ConfigurationService 329 .getInt(CoordMaterializeTriggerService.CONF_MATERIALIZATION_WINDOW); 330 queue(new CoordMaterializeTransitionXCommand(jobId, materializationWindow), 100); 331 } 332 333 /** 334 * Method that validates values in the definition for correctness. Placeholder to add more. 335 */ 336 private void validateCoordinatorJob() throws Exception { 337 // check if startTime < endTime 338 if (!coordJob.getStartTime().before(coordJob.getEndTime())) { 339 throw new IllegalArgumentException("Coordinator Start Time must be earlier than End Time."); 340 } 341 342 try { 343 // Check if a coord job with cron frequency will materialize actions 344 int freq = Integer.parseInt(coordJob.getFrequency()); 345 346 // Check if the frequency is faster than 5 min if enabled 347 if (ConfigurationService.getBoolean(CONF_CHECK_MAX_FREQUENCY)) { 348 CoordinatorJob.Timeunit unit = coordJob.getTimeUnit(); 349 if (freq == 0 || (freq < 5 && unit == CoordinatorJob.Timeunit.MINUTE)) { 350 throw new IllegalArgumentException("Coordinator job with frequency [" + freq + 351 "] minutes is faster than allowed maximum of 5 minutes (" 352 + CONF_CHECK_MAX_FREQUENCY + " is set to true)"); 353 } 354 } 355 } catch (NumberFormatException e) { 356 Date start = coordJob.getStartTime(); 357 Calendar cal = Calendar.getInstance(); 358 cal.setTime(start); 359 cal.add(Calendar.MINUTE, -1); 360 start = cal.getTime(); 361 362 Date nextTime = CoordCommandUtils.getNextValidActionTimeForCronFrequency(start, coordJob); 363 if (nextTime == null) { 364 throw new IllegalArgumentException("Invalid coordinator cron frequency: " + coordJob.getFrequency()); 365 } 366 if (!nextTime.before(coordJob.getEndTime())) { 367 throw new IllegalArgumentException("Coordinator job with frequency '" + 368 coordJob.getFrequency() + "' materializes no actions between start and end time."); 369 } 370 } 371 } 372 373 /* 374 * Check against multiple data instance values inside a single <instance> <start-instance> or <end-instance> tag 375 * If found, the job is not submitted and user is informed to correct the error, instead of defaulting to the first instance value in the list 376 */ 377 private void checkMultipleTimeInstances(Element eCoordJob, String eventType, String dataType) throws CoordinatorJobException { 378 Element eventsSpec, dataSpec, instance; 379 List<Element> instanceSpecList; 380 Namespace ns = eCoordJob.getNamespace(); 381 String instanceValue; 382 eventsSpec = eCoordJob.getChild(eventType, ns); 383 if (eventsSpec != null) { 384 dataSpec = eventsSpec.getChild(dataType, ns); 385 if (dataSpec != null) { 386 // In case of input-events, there can be multiple child <instance> datasets. Iterating to ensure none of them have errors 387 instanceSpecList = dataSpec.getChildren("instance", ns); 388 Iterator instanceIter = instanceSpecList.iterator(); 389 while(instanceIter.hasNext()) { 390 instance = ((Element) instanceIter.next()); 391 if(instance.getContentSize() == 0) { //empty string or whitespace 392 throw new CoordinatorJobException(ErrorCode.E1021, "<instance> tag within " + eventType + " is empty!"); 393 } 394 instanceValue = instance.getContent(0).toString(); 395 boolean isInvalid = false; 396 try { 397 isInvalid = evalAction.checkForExistence(instanceValue, ","); 398 } catch (Exception e) { 399 handleELParseException(eventType, dataType, instanceValue); 400 } 401 if (isInvalid) { // reaching this block implies instance is not empty i.e. length > 0 402 handleExpresionWithMultipleInstances(eventType, dataType, instanceValue); 403 } 404 } 405 406 // In case of input-events, there can be multiple child <start-instance> datasets. Iterating to ensure none of them have errors 407 instanceSpecList = dataSpec.getChildren("start-instance", ns); 408 instanceIter = instanceSpecList.iterator(); 409 while(instanceIter.hasNext()) { 410 instance = ((Element) instanceIter.next()); 411 if(instance.getContentSize() == 0) { //empty string or whitespace 412 throw new CoordinatorJobException(ErrorCode.E1021, "<start-instance> tag within " + eventType + " is empty!"); 413 } 414 instanceValue = instance.getContent(0).toString(); 415 boolean isInvalid = false; 416 try { 417 isInvalid = evalAction.checkForExistence(instanceValue, ","); 418 } catch (Exception e) { 419 handleELParseException(eventType, dataType, instanceValue); 420 } 421 if (isInvalid) { // reaching this block implies start instance is not empty i.e. length > 0 422 handleExpresionWithStartMultipleInstances(eventType, dataType, instanceValue); 423 } 424 } 425 426 // In case of input-events, there can be multiple child <end-instance> datasets. Iterating to ensure none of them have errors 427 instanceSpecList = dataSpec.getChildren("end-instance", ns); 428 instanceIter = instanceSpecList.iterator(); 429 while(instanceIter.hasNext()) { 430 instance = ((Element) instanceIter.next()); 431 if(instance.getContentSize() == 0) { //empty string or whitespace 432 throw new CoordinatorJobException(ErrorCode.E1021, "<end-instance> tag within " + eventType + " is empty!"); 433 } 434 instanceValue = instance.getContent(0).toString(); 435 boolean isInvalid = false; 436 try { 437 isInvalid = evalAction.checkForExistence(instanceValue, ","); 438 } catch (Exception e) { 439 handleELParseException(eventType, dataType, instanceValue); 440 } 441 if (isInvalid) { // reaching this block implies instance is not empty i.e. length > 0 442 handleExpresionWithMultipleEndInstances(eventType, dataType, instanceValue); 443 } 444 } 445 446 } 447 } 448 } 449 450 private void handleELParseException(String eventType, String dataType, String instanceValue) 451 throws CoordinatorJobException { 452 String correctAction = null; 453 if(dataType.equals(COORD_INPUT_EVENTS_DATA_IN)) { 454 correctAction = "Coordinator app definition should have valid <instance> tag for data-in"; 455 } else if(dataType.equals(COORD_OUTPUT_EVENTS_DATA_OUT)) { 456 correctAction = "Coordinator app definition should have valid <instance> tag for data-out"; 457 } 458 throw new CoordinatorJobException(ErrorCode.E1021, eventType + " instance '" + instanceValue 459 + "' is not valid. Coordinator job NOT SUBMITTED. " + correctAction); 460 } 461 462 private void handleExpresionWithMultipleInstances(String eventType, String dataType, String instanceValue) 463 throws CoordinatorJobException { 464 String correctAction = null; 465 if(dataType.equals(COORD_INPUT_EVENTS_DATA_IN)) { 466 correctAction = "Coordinator app definition should have separate <instance> tag per data-in instance"; 467 } else if(dataType.equals(COORD_OUTPUT_EVENTS_DATA_OUT)) { 468 correctAction = "Coordinator app definition can have only one <instance> tag per data-out instance"; 469 } 470 throw new CoordinatorJobException(ErrorCode.E1021, eventType + " instance '" + instanceValue 471 + "' contains more than one date instance. Coordinator job NOT SUBMITTED. " + correctAction); 472 } 473 474 private void handleExpresionWithStartMultipleInstances(String eventType, String dataType, String instanceValue) 475 throws CoordinatorJobException { 476 String correctAction = "Coordinator app definition should not have multiple start-instances"; 477 throw new CoordinatorJobException(ErrorCode.E1021, eventType + " start-instance '" + instanceValue 478 + "' contains more than one date start-instance. Coordinator job NOT SUBMITTED. " + correctAction); 479 } 480 481 private void handleExpresionWithMultipleEndInstances(String eventType, String dataType, String instanceValue) 482 throws CoordinatorJobException { 483 String correctAction = "Coordinator app definition should not have multiple end-instances"; 484 throw new CoordinatorJobException(ErrorCode.E1021, eventType + " end-instance '" + instanceValue 485 + "' contains more than one date end-instance. Coordinator job NOT SUBMITTED. " + correctAction); 486 } 487 /** 488 * Read the application XML and validate against coordinator Schema 489 * 490 * @return validated coordinator XML 491 * @throws CoordinatorJobException thrown if unable to read or validate coordinator xml 492 */ 493 protected String readAndValidateXml() throws CoordinatorJobException { 494 String appPath = ParamChecker.notEmpty(conf.get(OozieClient.COORDINATOR_APP_PATH), 495 OozieClient.COORDINATOR_APP_PATH); 496 String coordXml = readDefinition(appPath); 497 validateXml(coordXml); 498 return coordXml; 499 } 500 501 /** 502 * Validate against Coordinator XSD file 503 * 504 * @param xmlContent : Input coordinator xml 505 * @throws CoordinatorJobException thrown if unable to validate coordinator xml 506 */ 507 private void validateXml(String xmlContent) throws CoordinatorJobException { 508 javax.xml.validation.Schema schema = Services.get().get(SchemaService.class).getSchema(SchemaName.COORDINATOR); 509 Validator validator = schema.newValidator(); 510 try { 511 validator.validate(new StreamSource(new StringReader(xmlContent))); 512 } 513 catch (SAXException ex) { 514 LOG.warn("SAXException :", ex); 515 throw new CoordinatorJobException(ErrorCode.E0701, ex.getMessage(), ex); 516 } 517 catch (IOException ex) { 518 LOG.warn("IOException :", ex); 519 throw new CoordinatorJobException(ErrorCode.E0702, ex.getMessage(), ex); 520 } 521 } 522 523 /** 524 * Read the application XML schema namespace 525 * 526 * @param coordXmlElement input coordinator xml Element 527 * @return app xml namespace 528 * @throws CoordinatorJobException 529 */ 530 private String readAppNamespace(Element coordXmlElement) throws CoordinatorJobException { 531 Namespace ns = coordXmlElement.getNamespace(); 532 if (ns != null && bundleId != null && ns.getURI().equals(SchemaService.COORDINATOR_NAMESPACE_URI_1)) { 533 throw new CoordinatorJobException(ErrorCode.E1319, "bundle app can not submit coordinator namespace " 534 + SchemaService.COORDINATOR_NAMESPACE_URI_1 + ", please use 0.2 or later"); 535 } 536 if (ns != null) { 537 return ns.getURI(); 538 } 539 else { 540 throw new CoordinatorJobException(ErrorCode.E0700, "the application xml namespace is not given"); 541 } 542 } 543 544 /** 545 * Merge default configuration with user-defined configuration. 546 * 547 * @throws CommandException thrown if failed to read or merge configurations 548 */ 549 protected void mergeDefaultConfig() throws CommandException { 550 Path configDefault = null; 551 try { 552 String coordAppPathStr = conf.get(OozieClient.COORDINATOR_APP_PATH); 553 Path coordAppPath = new Path(coordAppPathStr); 554 String user = ParamChecker.notEmpty(conf.get(OozieClient.USER_NAME), OozieClient.USER_NAME); 555 HadoopAccessorService has = Services.get().get(HadoopAccessorService.class); 556 Configuration fsConf = has.createJobConf(coordAppPath.toUri().getAuthority()); 557 FileSystem fs = has.createFileSystem(user, coordAppPath.toUri(), fsConf); 558 559 // app path could be a directory 560 if (!fs.isFile(coordAppPath)) { 561 configDefault = new Path(coordAppPath, CONFIG_DEFAULT); 562 } else { 563 configDefault = new Path(coordAppPath.getParent(), CONFIG_DEFAULT); 564 } 565 566 if (fs.exists(configDefault)) { 567 Configuration defaultConf = new XConfiguration(fs.open(configDefault)); 568 PropertiesUtils.checkDisallowedProperties(defaultConf, DISALLOWED_DEFAULT_PROPERTIES); 569 XConfiguration.injectDefaults(defaultConf, conf); 570 } 571 else { 572 LOG.info("configDefault Doesn't exist " + configDefault); 573 } 574 PropertiesUtils.checkDisallowedProperties(conf, DISALLOWED_USER_PROPERTIES); 575 576 // Resolving all variables in the job properties. 577 // This ensures the Hadoop Configuration semantics is preserved. 578 XConfiguration resolvedVarsConf = new XConfiguration(); 579 for (Map.Entry<String, String> entry : conf) { 580 resolvedVarsConf.set(entry.getKey(), conf.get(entry.getKey())); 581 } 582 conf = resolvedVarsConf; 583 } 584 catch (IOException e) { 585 throw new CommandException(ErrorCode.E0702, e.getMessage() + " : Problem reading default config " 586 + configDefault, e); 587 } 588 catch (HadoopAccessorException e) { 589 throw new CommandException(e); 590 } 591 LOG.debug("Merged CONF :" + XmlUtils.prettyPrint(conf).toString()); 592 } 593 594 /** 595 * The method resolve all the variables that are defined in configuration. It also include the data set definition 596 * from dataset file into XML. 597 * 598 * @param appXml : Original job XML 599 * @param conf : Configuration of the job 600 * @param coordJob : Coordinator job bean to be populated. 601 * @return Resolved and modified job XML element. 602 * @throws CoordinatorJobException thrown if failed to resolve basic entities or include referred datasets 603 * @throws Exception thrown if failed to resolve basic entities or include referred datasets 604 */ 605 public Element basicResolveAndIncludeDS(String appXml, Configuration conf, CoordinatorJobBean coordJob) 606 throws CoordinatorJobException, Exception { 607 Element basicResolvedApp = resolveInitial(conf, appXml, coordJob); 608 includeDataSets(basicResolvedApp, conf); 609 return basicResolvedApp; 610 } 611 612 /** 613 * Insert data set into data-in and data-out tags. 614 * 615 * @param eAppXml : coordinator application XML 616 * @param eDatasets : DataSet XML 617 */ 618 @SuppressWarnings("unchecked") 619 private void insertDataSet(Element eAppXml, Element eDatasets) { 620 // Adding DS definition in the coordinator XML 621 Element inputList = eAppXml.getChild("input-events", eAppXml.getNamespace()); 622 if (inputList != null) { 623 for (Element dataIn : (List<Element>) inputList.getChildren("data-in", eAppXml.getNamespace())) { 624 Element eDataset = findDataSet(eDatasets, dataIn.getAttributeValue("dataset")); 625 dataIn.getContent().add(0, eDataset); 626 } 627 } 628 Element outputList = eAppXml.getChild("output-events", eAppXml.getNamespace()); 629 if (outputList != null) { 630 for (Element dataOut : (List<Element>) outputList.getChildren("data-out", eAppXml.getNamespace())) { 631 Element eDataset = findDataSet(eDatasets, dataOut.getAttributeValue("dataset")); 632 dataOut.getContent().add(0, eDataset); 633 } 634 } 635 } 636 637 /** 638 * Find a specific dataset from a list of Datasets. 639 * 640 * @param eDatasets : List of data sets 641 * @param name : queried data set name 642 * @return one Dataset element. otherwise throw Exception 643 */ 644 @SuppressWarnings("unchecked") 645 private static Element findDataSet(Element eDatasets, String name) { 646 for (Element eDataset : (List<Element>) eDatasets.getChildren("dataset", eDatasets.getNamespace())) { 647 if (eDataset.getAttributeValue("name").equals(name)) { 648 eDataset = (Element) eDataset.clone(); 649 eDataset.detach(); 650 return eDataset; 651 } 652 } 653 throw new RuntimeException("undefined dataset: " + name); 654 } 655 656 /** 657 * Initialize all the required EL Evaluators. 658 */ 659 protected void initEvaluators() { 660 evalFreq = CoordELEvaluator.createELEvaluatorForGroup(conf, "coord-job-submit-freq"); 661 evalNofuncs = CoordELEvaluator.createELEvaluatorForGroup(conf, "coord-job-submit-nofuncs"); 662 evalInst = CoordELEvaluator.createELEvaluatorForGroup(conf, "coord-job-submit-instances"); 663 evalAction = CoordELEvaluator.createELEvaluatorForGroup(conf, "coord-action-start"); 664 evalTimeout = CoordELEvaluator.createELEvaluatorForGroup(conf, "coord-job-wait-timeout"); 665 evalInitialInstance = CoordELEvaluator.createELEvaluatorForGroup(conf, "coord-job-submit-initial-instance"); 666 667 } 668 669 /** 670 * Resolve basic entities using job Configuration. 671 * 672 * @param conf :Job configuration 673 * @param appXml : Original job XML 674 * @param coordJob : Coordinator job bean to be populated. 675 * @return Resolved job XML element. 676 * @throws CoordinatorJobException thrown if failed to resolve basic entities 677 * @throws Exception thrown if failed to resolve basic entities 678 */ 679 @SuppressWarnings("unchecked") 680 protected Element resolveInitial(Configuration conf, String appXml, CoordinatorJobBean coordJob) 681 throws CoordinatorJobException, Exception { 682 Element eAppXml = XmlUtils.parseXml(appXml); 683 // job's main attributes 684 // frequency 685 String val = resolveAttribute("frequency", eAppXml, evalFreq); 686 int ival = 0; 687 688 val = ParamChecker.checkFrequency(val); 689 coordJob.setFrequency(val); 690 TimeUnit tmp = (evalFreq.getVariable("timeunit") == null) ? TimeUnit.MINUTE : ((TimeUnit) evalFreq 691 .getVariable("timeunit")); 692 try { 693 Integer.parseInt(val); 694 } 695 catch (NumberFormatException ex) { 696 tmp=TimeUnit.CRON; 697 } 698 699 addAnAttribute("freq_timeunit", eAppXml, tmp.toString()); 700 // TimeUnit 701 coordJob.setTimeUnit(CoordinatorJob.Timeunit.valueOf(tmp.toString())); 702 // End Of Duration 703 tmp = evalFreq.getVariable("endOfDuration") == null ? TimeUnit.NONE : ((TimeUnit) evalFreq 704 .getVariable("endOfDuration")); 705 addAnAttribute("end_of_duration", eAppXml, tmp.toString()); 706 // coordJob.setEndOfDuration(tmp) // TODO: Add new attribute in Job bean 707 708 // Application name 709 if (this.coordName == null) { 710 String name = ELUtils.resolveAppName(eAppXml.getAttribute("name").getValue(), conf); 711 coordJob.setAppName(name); 712 } 713 else { 714 // this coord job is created from bundle 715 coordJob.setAppName(this.coordName); 716 } 717 718 // start time 719 val = resolveAttribute("start", eAppXml, evalNofuncs); 720 ParamChecker.checkDateOozieTZ(val, "start"); 721 coordJob.setStartTime(DateUtils.parseDateOozieTZ(val)); 722 // end time 723 val = resolveAttribute("end", eAppXml, evalNofuncs); 724 ParamChecker.checkDateOozieTZ(val, "end"); 725 coordJob.setEndTime(DateUtils.parseDateOozieTZ(val)); 726 // Time zone 727 val = resolveAttribute("timezone", eAppXml, evalNofuncs); 728 ParamChecker.checkTimeZone(val, "timezone"); 729 coordJob.setTimeZone(val); 730 731 // controls 732 val = resolveTagContents("timeout", eAppXml.getChild("controls", eAppXml.getNamespace()), evalTimeout); 733 if (val != null && val != "") { 734 int t = Integer.parseInt(val); 735 tmp = (evalTimeout.getVariable("timeunit") == null) ? TimeUnit.MINUTE : ((TimeUnit) evalTimeout 736 .getVariable("timeunit")); 737 switch (tmp) { 738 case HOUR: 739 val = String.valueOf(t * 60); 740 break; 741 case DAY: 742 val = String.valueOf(t * 60 * 24); 743 break; 744 case MONTH: 745 val = String.valueOf(t * 60 * 24 * 30); 746 break; 747 default: 748 break; 749 } 750 } 751 else { 752 val = ConfigurationService.get(CONF_DEFAULT_TIMEOUT_NORMAL); 753 } 754 755 ival = ParamChecker.checkInteger(val, "timeout"); 756 if (ival < 0 || ival > ConfigurationService.getInt(CONF_DEFAULT_MAX_TIMEOUT)) { 757 ival = ConfigurationService.getInt(CONF_DEFAULT_MAX_TIMEOUT); 758 } 759 coordJob.setTimeout(ival); 760 761 val = resolveTagContents("concurrency", eAppXml.getChild("controls", eAppXml.getNamespace()), evalNofuncs); 762 if (val == null || val.isEmpty()) { 763 val = ConfigurationService.get(CONF_DEFAULT_CONCURRENCY); 764 } 765 ival = ParamChecker.checkInteger(val, "concurrency"); 766 coordJob.setConcurrency(ival); 767 768 val = resolveTagContents("throttle", eAppXml.getChild("controls", eAppXml.getNamespace()), evalNofuncs); 769 if (val == null || val.isEmpty()) { 770 int defaultThrottle = ConfigurationService.getInt(CONF_DEFAULT_THROTTLE); 771 ival = defaultThrottle; 772 } 773 else { 774 ival = ParamChecker.checkInteger(val, "throttle"); 775 } 776 int maxQueue = ConfigurationService.getInt(CONF_QUEUE_SIZE); 777 float factor = ConfigurationService.getFloat(CONF_MAT_THROTTLING_FACTOR); 778 int maxThrottle = (int) (maxQueue * factor); 779 if (ival > maxThrottle || ival < 1) { 780 ival = maxThrottle; 781 } 782 LOG.debug("max throttle " + ival); 783 coordJob.setMatThrottling(ival); 784 785 val = resolveTagContents("execution", eAppXml.getChild("controls", eAppXml.getNamespace()), evalNofuncs); 786 if (val == "") { 787 val = Execution.FIFO.toString(); 788 } 789 coordJob.setExecutionOrder(Execution.valueOf(val)); 790 String[] acceptedVals = { Execution.LIFO.toString(), Execution.FIFO.toString(), Execution.LAST_ONLY.toString(), 791 Execution.NONE.toString()}; 792 ParamChecker.isMember(val, acceptedVals, "execution"); 793 794 // datasets 795 resolveTagContents("include", eAppXml.getChild("datasets", eAppXml.getNamespace()), evalNofuncs); 796 // for each data set 797 resolveDataSets(eAppXml); 798 HashMap<String, String> dataNameList = new HashMap<String, String>(); 799 resolveIODataset(eAppXml); 800 resolveIOEvents(eAppXml, dataNameList); 801 802 resolveTagContents("app-path", eAppXml.getChild("action", eAppXml.getNamespace()).getChild("workflow", 803 eAppXml.getNamespace()), evalNofuncs); 804 // TODO: If action or workflow tag is missing, NullPointerException will 805 // occur 806 Element configElem = eAppXml.getChild("action", eAppXml.getNamespace()).getChild("workflow", 807 eAppXml.getNamespace()).getChild("configuration", eAppXml.getNamespace()); 808 evalData = CoordELEvaluator.createELEvaluatorForDataEcho(conf, "coord-job-submit-data", dataNameList); 809 if (configElem != null) { 810 for (Element propElem : (List<Element>) configElem.getChildren("property", configElem.getNamespace())) { 811 resolveTagContents("name", propElem, evalData); 812 // Want to check the data-integrity but don't want to modify the 813 // XML 814 // for properties only 815 Element tmpProp = (Element) propElem.clone(); 816 resolveTagContents("value", tmpProp, evalData); 817 } 818 } 819 evalSla = CoordELEvaluator.createELEvaluatorForDataAndConf(conf, "coord-sla-submit", dataNameList); 820 resolveSLA(eAppXml, coordJob); 821 return eAppXml; 822 } 823 824 /** 825 * Resolve SLA events 826 * 827 * @param eAppXml job XML 828 * @param coordJob coordinator job bean 829 * @throws CommandException thrown if failed to resolve sla events 830 */ 831 private void resolveSLA(Element eAppXml, CoordinatorJobBean coordJob) throws CommandException { 832 Element eSla = XmlUtils.getSLAElement(eAppXml.getChild("action", eAppXml.getNamespace())); 833 834 if (eSla != null) { 835 String slaXml = XmlUtils.prettyPrint(eSla).toString(); 836 try { 837 // EL evaluation 838 slaXml = evalSla.evaluate(slaXml, String.class); 839 // Validate against semantic SXD 840 XmlUtils.validateData(slaXml, SchemaName.SLA_ORIGINAL); 841 } 842 catch (Exception e) { 843 throw new CommandException(ErrorCode.E1004, "Validation ERROR :" + e.getMessage(), e); 844 } 845 } 846 } 847 848 /** 849 * Resolve input-events/data-in and output-events/data-out tags. 850 * 851 * @param eJobOrg : Job element 852 * @throws CoordinatorJobException thrown if failed to resolve input and output events 853 */ 854 @SuppressWarnings("unchecked") 855 private void resolveIOEvents(Element eJobOrg, HashMap<String, String> dataNameList) throws CoordinatorJobException { 856 // Resolving input-events/data-in 857 // Clone the job and don't update anything in the original 858 Element eJob = (Element) eJobOrg.clone(); 859 Element inputList = eJob.getChild("input-events", eJob.getNamespace()); 860 if (inputList != null) { 861 TreeSet<String> eventNameSet = new TreeSet<String>(); 862 for (Element dataIn : (List<Element>) inputList.getChildren("data-in", eJob.getNamespace())) { 863 String dataInName = dataIn.getAttributeValue("name"); 864 dataNameList.put(dataInName, "data-in"); 865 // check whether there is any duplicate data-in name 866 if (eventNameSet.contains(dataInName)) { 867 throw new RuntimeException("Duplicate dataIn name " + dataInName); 868 } 869 else { 870 eventNameSet.add(dataInName); 871 } 872 resolveTagContents("instance", dataIn, evalInst); 873 resolveTagContents("start-instance", dataIn, evalInst); 874 resolveTagContents("end-instance", dataIn, evalInst); 875 876 } 877 } 878 // Resolving output-events/data-out 879 Element outputList = eJob.getChild("output-events", eJob.getNamespace()); 880 if (outputList != null) { 881 TreeSet<String> eventNameSet = new TreeSet<String>(); 882 for (Element dataOut : (List<Element>) outputList.getChildren("data-out", eJob.getNamespace())) { 883 String dataOutName = dataOut.getAttributeValue("name"); 884 dataNameList.put(dataOutName, "data-out"); 885 // check whether there is any duplicate data-out name 886 if (eventNameSet.contains(dataOutName)) { 887 throw new RuntimeException("Duplicate dataIn name " + dataOutName); 888 } 889 else { 890 eventNameSet.add(dataOutName); 891 } 892 resolveTagContents("instance", dataOut, evalInst); 893 894 } 895 } 896 897 } 898 899 /** 900 * Resolve input-events/dataset and output-events/dataset tags. 901 * 902 * @param eJob : Job element 903 * @throws CoordinatorJobException thrown if failed to resolve input and output events 904 */ 905 @SuppressWarnings("unchecked") 906 private void resolveIODataset(Element eAppXml) throws CoordinatorJobException { 907 // Resolving input-events/data-in 908 Element inputList = eAppXml.getChild("input-events", eAppXml.getNamespace()); 909 if (inputList != null) { 910 for (Element dataIn : (List<Element>) inputList.getChildren("data-in", eAppXml.getNamespace())) { 911 resolveAttribute("dataset", dataIn, evalInst); 912 913 } 914 } 915 // Resolving output-events/data-out 916 Element outputList = eAppXml.getChild("output-events", eAppXml.getNamespace()); 917 if (outputList != null) { 918 for (Element dataOut : (List<Element>) outputList.getChildren("data-out", eAppXml.getNamespace())) { 919 resolveAttribute("dataset", dataOut, evalInst); 920 921 } 922 } 923 924 } 925 926 927 /** 928 * Add an attribute into XML element. 929 * 930 * @param attrName :attribute name 931 * @param elem : Element to add attribute 932 * @param value :Value of attribute 933 */ 934 private void addAnAttribute(String attrName, Element elem, String value) { 935 elem.setAttribute(attrName, value); 936 } 937 938 /** 939 * Resolve datasets using job configuration. 940 * 941 * @param eAppXml : Job Element XML 942 * @throws Exception thrown if failed to resolve datasets 943 */ 944 @SuppressWarnings("unchecked") 945 private void resolveDataSets(Element eAppXml) throws Exception { 946 Element datasetList = eAppXml.getChild("datasets", eAppXml.getNamespace()); 947 if (datasetList != null) { 948 949 List<Element> dsElems = datasetList.getChildren("dataset", eAppXml.getNamespace()); 950 resolveDataSets(dsElems); 951 resolveTagContents("app-path", eAppXml.getChild("action", eAppXml.getNamespace()).getChild("workflow", 952 eAppXml.getNamespace()), evalNofuncs); 953 } 954 } 955 956 /** 957 * Resolve datasets using job configuration. 958 * 959 * @param dsElems : Data set XML element. 960 * @throws CoordinatorJobException thrown if failed to resolve datasets 961 */ 962 private void resolveDataSets(List<Element> dsElems) throws CoordinatorJobException { 963 for (Element dsElem : dsElems) { 964 // Setting up default TimeUnit and EndOFDuraion 965 evalFreq.setVariable("timeunit", TimeUnit.MINUTE); 966 evalFreq.setVariable("endOfDuration", TimeUnit.NONE); 967 968 String val = resolveAttribute("frequency", dsElem, evalFreq); 969 int ival = ParamChecker.checkInteger(val, "frequency"); 970 ParamChecker.checkGTZero(ival, "frequency"); 971 addAnAttribute("freq_timeunit", dsElem, evalFreq.getVariable("timeunit") == null ? TimeUnit.MINUTE 972 .toString() : ((TimeUnit) evalFreq.getVariable("timeunit")).toString()); 973 addAnAttribute("end_of_duration", dsElem, evalFreq.getVariable("endOfDuration") == null ? TimeUnit.NONE 974 .toString() : ((TimeUnit) evalFreq.getVariable("endOfDuration")).toString()); 975 val = resolveAttribute("initial-instance", dsElem, evalInitialInstance); 976 ParamChecker.checkDateOozieTZ(val, "initial-instance"); 977 checkInitialInstance(val); 978 val = resolveAttribute("timezone", dsElem, evalNofuncs); 979 ParamChecker.checkTimeZone(val, "timezone"); 980 resolveTagContents("uri-template", dsElem, evalNofuncs); 981 resolveTagContents("done-flag", dsElem, evalNofuncs); 982 } 983 } 984 985 /** 986 * Resolve the content of a tag. 987 * 988 * @param tagName : Tag name of job XML i.e. <timeout> 10 </timeout> 989 * @param elem : Element where the tag exists. 990 * @param eval : EL evealuator 991 * @return Resolved tag content. 992 * @throws CoordinatorJobException thrown if failed to resolve tag content 993 */ 994 @SuppressWarnings("unchecked") 995 private String resolveTagContents(String tagName, Element elem, ELEvaluator eval) throws CoordinatorJobException { 996 String ret = ""; 997 if (elem != null) { 998 for (Element tagElem : (List<Element>) elem.getChildren(tagName, elem.getNamespace())) { 999 if (tagElem != null) { 1000 String updated; 1001 try { 1002 updated = CoordELFunctions.evalAndWrap(eval, tagElem.getText().trim()); 1003 1004 } 1005 catch (Exception e) { 1006 throw new CoordinatorJobException(ErrorCode.E1004, e.getMessage(), e); 1007 } 1008 tagElem.removeContent(); 1009 tagElem.addContent(updated); 1010 ret += updated; 1011 } 1012 } 1013 } 1014 return ret; 1015 } 1016 1017 /** 1018 * Resolve an attribute value. 1019 * 1020 * @param attrName : Attribute name. 1021 * @param elem : XML Element where attribute is defiend 1022 * @param eval : ELEvaluator used to resolve 1023 * @return Resolved attribute value 1024 * @throws CoordinatorJobException thrown if failed to resolve an attribute value 1025 */ 1026 private String resolveAttribute(String attrName, Element elem, ELEvaluator eval) throws CoordinatorJobException { 1027 Attribute attr = elem.getAttribute(attrName); 1028 String val = null; 1029 if (attr != null) { 1030 try { 1031 val = CoordELFunctions.evalAndWrap(eval, attr.getValue().trim()); 1032 } 1033 catch (Exception e) { 1034 throw new CoordinatorJobException(ErrorCode.E1004, e.getMessage(), e); 1035 } 1036 attr.setValue(val); 1037 } 1038 return val; 1039 } 1040 1041 /** 1042 * Include referred datasets into XML. 1043 * 1044 * @param resolvedXml : Job XML element. 1045 * @param conf : Job configuration 1046 * @throws CoordinatorJobException thrown if failed to include referred datasets into XML 1047 */ 1048 @SuppressWarnings("unchecked") 1049 protected void includeDataSets(Element resolvedXml, Configuration conf) throws CoordinatorJobException { 1050 Element datasets = resolvedXml.getChild("datasets", resolvedXml.getNamespace()); 1051 Element allDataSets = new Element("all_datasets", resolvedXml.getNamespace()); 1052 List<String> dsList = new ArrayList<String>(); 1053 if (datasets != null) { 1054 for (Element includeElem : (List<Element>) datasets.getChildren("include", datasets.getNamespace())) { 1055 String incDSFile = includeElem.getTextTrim(); 1056 includeOneDSFile(incDSFile, dsList, allDataSets, datasets.getNamespace()); 1057 } 1058 for (Element e : (List<Element>) datasets.getChildren("dataset", datasets.getNamespace())) { 1059 String dsName = e.getAttributeValue("name"); 1060 if (dsList.contains(dsName)) {// Override with this DS 1061 // Remove duplicate 1062 removeDataSet(allDataSets, dsName); 1063 } 1064 else { 1065 dsList.add(dsName); 1066 } 1067 allDataSets.addContent((Element) e.clone()); 1068 } 1069 } 1070 insertDataSet(resolvedXml, allDataSets); 1071 resolvedXml.removeChild("datasets", resolvedXml.getNamespace()); 1072 } 1073 1074 /** 1075 * Include one dataset file. 1076 * 1077 * @param incDSFile : Include data set filename. 1078 * @param dsList :List of dataset names to verify the duplicate. 1079 * @param allDataSets : Element that includes all dataset definitions. 1080 * @param dsNameSpace : Data set name space 1081 * @throws CoordinatorJobException thrown if failed to include one dataset file 1082 */ 1083 @SuppressWarnings("unchecked") 1084 private void includeOneDSFile(String incDSFile, List<String> dsList, Element allDataSets, Namespace dsNameSpace) 1085 throws CoordinatorJobException { 1086 Element tmpDataSets = null; 1087 try { 1088 String dsXml = readDefinition(incDSFile); 1089 LOG.debug("DSFILE :" + incDSFile + "\n" + dsXml); 1090 tmpDataSets = XmlUtils.parseXml(dsXml); 1091 } 1092 catch (JDOMException e) { 1093 LOG.warn("Error parsing included dataset [{0}]. Message [{1}]", incDSFile, e.getMessage()); 1094 throw new CoordinatorJobException(ErrorCode.E0700, e.getMessage()); 1095 } 1096 resolveDataSets(tmpDataSets.getChildren("dataset")); 1097 for (Element e : (List<Element>) tmpDataSets.getChildren("dataset")) { 1098 String dsName = e.getAttributeValue("name"); 1099 if (dsList.contains(dsName)) { 1100 throw new RuntimeException("Duplicate Dataset " + dsName); 1101 } 1102 dsList.add(dsName); 1103 Element tmp = (Element) e.clone(); 1104 // TODO: Don't like to over-write the external/include DS's namespace 1105 tmp.setNamespace(dsNameSpace); 1106 tmp.getChild("uri-template").setNamespace(dsNameSpace); 1107 if (e.getChild("done-flag") != null) { 1108 tmp.getChild("done-flag").setNamespace(dsNameSpace); 1109 } 1110 allDataSets.addContent(tmp); 1111 } 1112 // nested include 1113 for (Element includeElem : (List<Element>) tmpDataSets.getChildren("include", tmpDataSets.getNamespace())) { 1114 String incFile = includeElem.getTextTrim(); 1115 includeOneDSFile(incFile, dsList, allDataSets, dsNameSpace); 1116 } 1117 } 1118 1119 /** 1120 * Remove a dataset from a list of dataset. 1121 * 1122 * @param eDatasets : List of dataset 1123 * @param name : Dataset name to be removed. 1124 */ 1125 @SuppressWarnings("unchecked") 1126 private static void removeDataSet(Element eDatasets, String name) { 1127 for (Element eDataset : (List<Element>) eDatasets.getChildren("dataset", eDatasets.getNamespace())) { 1128 if (eDataset.getAttributeValue("name").equals(name)) { 1129 eDataset.detach(); 1130 return; 1131 } 1132 } 1133 throw new RuntimeException("undefined dataset: " + name); 1134 } 1135 1136 /** 1137 * Read coordinator definition. 1138 * 1139 * @param appPath application path. 1140 * @return coordinator definition. 1141 * @throws CoordinatorJobException thrown if the definition could not be read. 1142 */ 1143 protected String readDefinition(String appPath) throws CoordinatorJobException { 1144 String user = ParamChecker.notEmpty(conf.get(OozieClient.USER_NAME), OozieClient.USER_NAME); 1145 // Configuration confHadoop = CoordUtils.getHadoopConf(conf); 1146 try { 1147 URI uri = new URI(appPath); 1148 LOG.debug("user =" + user); 1149 HadoopAccessorService has = Services.get().get(HadoopAccessorService.class); 1150 Configuration fsConf = has.createJobConf(uri.getAuthority()); 1151 FileSystem fs = has.createFileSystem(user, uri, fsConf); 1152 Path appDefPath = null; 1153 1154 // app path could be a directory 1155 Path path = new Path(uri.getPath()); 1156 // check file exists for dataset include file, app xml already checked 1157 if (!fs.exists(path)) { 1158 throw new URISyntaxException(path.toString(), "path not existed : " + path.toString()); 1159 } 1160 if (!fs.isFile(path)) { 1161 appDefPath = new Path(path, COORDINATOR_XML_FILE); 1162 } else { 1163 appDefPath = path; 1164 } 1165 1166 Reader reader = new InputStreamReader(fs.open(appDefPath)); 1167 StringWriter writer = new StringWriter(); 1168 IOUtils.copyCharStream(reader, writer); 1169 return writer.toString(); 1170 } 1171 catch (IOException ex) { 1172 LOG.warn("IOException :" + XmlUtils.prettyPrint(conf), ex); 1173 throw new CoordinatorJobException(ErrorCode.E1001, ex.getMessage(), ex); 1174 } 1175 catch (URISyntaxException ex) { 1176 LOG.warn("URISyException :" + ex.getMessage()); 1177 throw new CoordinatorJobException(ErrorCode.E1002, appPath, ex.getMessage(), ex); 1178 } 1179 catch (HadoopAccessorException ex) { 1180 throw new CoordinatorJobException(ex); 1181 } 1182 catch (Exception ex) { 1183 LOG.warn("Exception :", ex); 1184 throw new CoordinatorJobException(ErrorCode.E1001, ex.getMessage(), ex); 1185 } 1186 } 1187 1188 /** 1189 * Write a coordinator job into database 1190 * 1191 *@param appXML : Coordinator definition xml 1192 * @param eJob : XML element of job 1193 * @param coordJob : Coordinator job bean 1194 * @return Job id 1195 * @throws CommandException thrown if unable to save coordinator job to db 1196 */ 1197 protected String storeToDB(String appXML, Element eJob, CoordinatorJobBean coordJob) throws CommandException { 1198 String jobId = Services.get().get(UUIDService.class).generateId(ApplicationType.COORDINATOR); 1199 coordJob.setId(jobId); 1200 1201 coordJob.setAppPath(conf.get(OozieClient.COORDINATOR_APP_PATH)); 1202 coordJob.setCreatedTime(new Date()); 1203 coordJob.setUser(conf.get(OozieClient.USER_NAME)); 1204 String group = ConfigUtils.getWithDeprecatedCheck(conf, OozieClient.JOB_ACL, OozieClient.GROUP_NAME, null); 1205 coordJob.setGroup(group); 1206 coordJob.setConf(XmlUtils.prettyPrint(conf).toString()); 1207 coordJob.setJobXml(XmlUtils.prettyPrint(eJob).toString()); 1208 coordJob.setLastActionNumber(0); 1209 coordJob.setLastModifiedTime(new Date()); 1210 1211 if (!dryrun) { 1212 coordJob.setLastModifiedTime(new Date()); 1213 try { 1214 CoordJobQueryExecutor.getInstance().insert(coordJob); 1215 } 1216 catch (JPAExecutorException jpaee) { 1217 coordJob.setId(null); 1218 coordJob.setStatus(CoordinatorJob.Status.FAILED); 1219 throw new CommandException(jpaee); 1220 } 1221 } 1222 return jobId; 1223 } 1224 1225 /* 1226 * this method checks if the initial-instance specified for a particular 1227 is not a date earlier than the oozie server default Jan 01, 1970 00:00Z UTC 1228 */ 1229 private void checkInitialInstance(String val) throws CoordinatorJobException, IllegalArgumentException { 1230 Date initialInstance, givenInstance; 1231 try { 1232 initialInstance = DateUtils.parseDateUTC("1970-01-01T00:00Z"); 1233 givenInstance = DateUtils.parseDateOozieTZ(val); 1234 } 1235 catch (Exception e) { 1236 throw new IllegalArgumentException("Unable to parse dataset initial-instance string '" + val + 1237 "' to Date object. ",e); 1238 } 1239 if(givenInstance.compareTo(initialInstance) < 0) { 1240 throw new CoordinatorJobException(ErrorCode.E1021, "Dataset initial-instance " + val + 1241 " is earlier than the default initial instance " + DateUtils.formatDateOozieTZ(initialInstance)); 1242 } 1243 } 1244 1245 /* (non-Javadoc) 1246 * @see org.apache.oozie.command.XCommand#getEntityKey() 1247 */ 1248 @Override 1249 public String getEntityKey() { 1250 return null; 1251 } 1252 1253 /* (non-Javadoc) 1254 * @see org.apache.oozie.command.XCommand#isLockRequired() 1255 */ 1256 @Override 1257 protected boolean isLockRequired() { 1258 return false; 1259 } 1260 1261 /* (non-Javadoc) 1262 * @see org.apache.oozie.command.XCommand#loadState() 1263 */ 1264 @Override 1265 protected void loadState() throws CommandException { 1266 jpaService = Services.get().get(JPAService.class); 1267 if (jpaService == null) { 1268 throw new CommandException(ErrorCode.E0610); 1269 } 1270 coordJob = new CoordinatorJobBean(); 1271 if (this.bundleId != null) { 1272 // this coord job is created from bundle 1273 coordJob.setBundleId(this.bundleId); 1274 // first use bundle id if submit thru bundle 1275 LogUtils.setLogInfo(this.bundleId); 1276 } 1277 if (this.coordName != null) { 1278 // this coord job is created from bundle 1279 coordJob.setAppName(this.coordName); 1280 } 1281 setJob(coordJob); 1282 1283 } 1284 1285 /* (non-Javadoc) 1286 * @see org.apache.oozie.command.XCommand#verifyPrecondition() 1287 */ 1288 @Override 1289 protected void verifyPrecondition() throws CommandException { 1290 1291 } 1292 1293 /* (non-Javadoc) 1294 * @see org.apache.oozie.command.TransitionXCommand#notifyParent() 1295 */ 1296 @Override 1297 public void notifyParent() throws CommandException { 1298 // update bundle action 1299 if (coordJob.getBundleId() != null) { 1300 LOG.debug("Updating bundle record: " + coordJob.getBundleId() + " for coord id: " + coordJob.getId()); 1301 BundleStatusUpdateXCommand bundleStatusUpdate = new BundleStatusUpdateXCommand(coordJob, prevStatus); 1302 bundleStatusUpdate.call(); 1303 } 1304 } 1305 1306 /* (non-Javadoc) 1307 * @see org.apache.oozie.command.TransitionXCommand#updateJob() 1308 */ 1309 @Override 1310 public void updateJob() throws CommandException { 1311 } 1312 1313 /* (non-Javadoc) 1314 * @see org.apache.oozie.command.TransitionXCommand#getJob() 1315 */ 1316 @Override 1317 public Job getJob() { 1318 return coordJob; 1319 } 1320 1321 @Override 1322 public void performWrites() throws CommandException { 1323 } 1324}