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.workflow.lite; 020 021import org.apache.commons.codec.binary.Base64; 022import org.apache.hadoop.io.Writable; 023import org.apache.oozie.action.hadoop.FsActionExecutor; 024import org.apache.oozie.action.oozie.SubWorkflowActionExecutor; 025import org.apache.oozie.service.ConfigurationService; 026import org.apache.oozie.util.ELUtils; 027import org.apache.oozie.util.IOUtils; 028import org.apache.oozie.util.XConfiguration; 029import org.apache.oozie.util.XmlUtils; 030import org.apache.oozie.util.ParamChecker; 031import org.apache.oozie.util.ParameterVerifier; 032import org.apache.oozie.util.ParameterVerifierException; 033import org.apache.oozie.util.WritableUtils; 034import org.apache.oozie.ErrorCode; 035import org.apache.oozie.workflow.WorkflowException; 036import org.apache.oozie.action.ActionExecutor; 037import org.apache.oozie.service.Services; 038import org.apache.oozie.service.ActionService; 039import org.apache.commons.lang.StringUtils; 040import org.apache.hadoop.conf.Configuration; 041import org.jdom.Element; 042import org.jdom.JDOMException; 043import org.jdom.Namespace; 044import org.xml.sax.SAXException; 045 046import javax.xml.transform.stream.StreamSource; 047import javax.xml.validation.Schema; 048import javax.xml.validation.Validator; 049 050import java.io.IOException; 051import java.io.Reader; 052import java.io.StringReader; 053import java.io.StringWriter; 054import java.io.ByteArrayOutputStream; 055import java.io.ByteArrayInputStream; 056import java.io.DataInputStream; 057import java.io.DataInput; 058import java.io.DataOutput; 059import java.io.DataOutputStream; 060import java.util.ArrayList; 061import java.util.Arrays; 062import java.util.Deque; 063import java.util.HashMap; 064import java.util.HashSet; 065import java.util.LinkedList; 066import java.util.List; 067import java.util.Map; 068import java.util.zip.*; 069 070/** 071 * Class to parse and validate workflow xml 072 */ 073public class LiteWorkflowAppParser { 074 075 private static final String DECISION_E = "decision"; 076 private static final String ACTION_E = "action"; 077 private static final String END_E = "end"; 078 private static final String START_E = "start"; 079 private static final String JOIN_E = "join"; 080 private static final String FORK_E = "fork"; 081 private static final Object KILL_E = "kill"; 082 083 private static final String SLA_INFO = "info"; 084 private static final String CREDENTIALS = "credentials"; 085 private static final String GLOBAL = "global"; 086 private static final String PARAMETERS = "parameters"; 087 088 private static final String NAME_A = "name"; 089 private static final String CRED_A = "cred"; 090 private static final String USER_RETRY_MAX_A = "retry-max"; 091 private static final String USER_RETRY_INTERVAL_A = "retry-interval"; 092 private static final String TO_A = "to"; 093 094 private static final String FORK_PATH_E = "path"; 095 private static final String FORK_START_A = "start"; 096 097 private static final String ACTION_OK_E = "ok"; 098 private static final String ACTION_ERROR_E = "error"; 099 100 private static final String DECISION_SWITCH_E = "switch"; 101 private static final String DECISION_CASE_E = "case"; 102 private static final String DECISION_DEFAULT_E = "default"; 103 104 private static final String KILL_MESSAGE_E = "message"; 105 public static final String VALIDATE_FORK_JOIN = "oozie.validate.ForkJoin"; 106 public static final String WF_VALIDATE_FORK_JOIN = "oozie.wf.validate.ForkJoin"; 107 108 public static final String DEFAULT_NAME_NODE = "oozie.actions.default.name-node"; 109 public static final String DEFAULT_JOB_TRACKER = "oozie.actions.default.job-tracker"; 110 public static final String OOZIE_GLOBAL = "oozie.wf.globalconf"; 111 112 private static final String JOB_TRACKER = "job-tracker"; 113 private static final String NAME_NODE = "name-node"; 114 private static final String JOB_XML = "job-xml"; 115 private static final String CONFIGURATION = "configuration"; 116 117 private Schema schema; 118 private Class<? extends ControlNodeHandler> controlNodeHandler; 119 private Class<? extends DecisionNodeHandler> decisionHandlerClass; 120 private Class<? extends ActionNodeHandler> actionHandlerClass; 121 122 private static enum VisitStatus { 123 VISITING, VISITED 124 } 125 126 /** 127 * We use this to store a node name and its top (eldest) decision parent node name for the forkjoin validation 128 */ 129 class NodeAndTopDecisionParent { 130 String node; 131 String topDecisionParent; 132 133 public NodeAndTopDecisionParent(String node, String topDecisionParent) { 134 this.node = node; 135 this.topDecisionParent = topDecisionParent; 136 } 137 } 138 139 private List<String> forkList = new ArrayList<String>(); 140 private List<String> joinList = new ArrayList<String>(); 141 private StartNodeDef startNode; 142 private List<NodeAndTopDecisionParent> visitedOkNodes = new ArrayList<NodeAndTopDecisionParent>(); 143 private List<String> visitedJoinNodes = new ArrayList<String>(); 144 145 private String defaultNameNode; 146 private String defaultJobTracker; 147 148 public LiteWorkflowAppParser(Schema schema, 149 Class<? extends ControlNodeHandler> controlNodeHandler, 150 Class<? extends DecisionNodeHandler> decisionHandlerClass, 151 Class<? extends ActionNodeHandler> actionHandlerClass) throws WorkflowException { 152 this.schema = schema; 153 this.controlNodeHandler = controlNodeHandler; 154 this.decisionHandlerClass = decisionHandlerClass; 155 this.actionHandlerClass = actionHandlerClass; 156 157 defaultNameNode = ConfigurationService.get(DEFAULT_NAME_NODE); 158 if (defaultNameNode != null) { 159 defaultNameNode = defaultNameNode.trim(); 160 if (defaultNameNode.isEmpty()) { 161 defaultNameNode = null; 162 } 163 } 164 defaultJobTracker = ConfigurationService.get(DEFAULT_JOB_TRACKER); 165 if (defaultJobTracker != null) { 166 defaultJobTracker = defaultJobTracker.trim(); 167 if (defaultJobTracker.isEmpty()) { 168 defaultJobTracker = null; 169 } 170 } 171 } 172 173 public LiteWorkflowApp validateAndParse(Reader reader, Configuration jobConf) throws WorkflowException { 174 return validateAndParse(reader, jobConf, null); 175 } 176 177 /** 178 * Parse and validate xml to {@link LiteWorkflowApp} 179 * 180 * @param reader 181 * @return LiteWorkflowApp 182 * @throws WorkflowException 183 */ 184 public LiteWorkflowApp validateAndParse(Reader reader, Configuration jobConf, Configuration configDefault) 185 throws WorkflowException { 186 try { 187 StringWriter writer = new StringWriter(); 188 IOUtils.copyCharStream(reader, writer); 189 String strDef = writer.toString(); 190 191 if (schema != null) { 192 Validator validator = schema.newValidator(); 193 validator.validate(new StreamSource(new StringReader(strDef))); 194 } 195 196 Element wfDefElement = XmlUtils.parseXml(strDef); 197 ParameterVerifier.verifyParameters(jobConf, wfDefElement); 198 LiteWorkflowApp app = parse(strDef, wfDefElement, configDefault, jobConf); 199 Map<String, VisitStatus> traversed = new HashMap<String, VisitStatus>(); 200 traversed.put(app.getNode(StartNodeDef.START).getName(), VisitStatus.VISITING); 201 validate(app, app.getNode(StartNodeDef.START), traversed); 202 //Validate whether fork/join are in pair or not 203 if (jobConf.getBoolean(WF_VALIDATE_FORK_JOIN, true) 204 && ConfigurationService.getBoolean(VALIDATE_FORK_JOIN)) { 205 validateForkJoin(app); 206 } 207 return app; 208 } 209 catch (ParameterVerifierException ex) { 210 throw new WorkflowException(ex); 211 } 212 catch (JDOMException ex) { 213 throw new WorkflowException(ErrorCode.E0700, ex.getMessage(), ex); 214 } 215 catch (SAXException ex) { 216 throw new WorkflowException(ErrorCode.E0701, ex.getMessage(), ex); 217 } 218 catch (IOException ex) { 219 throw new WorkflowException(ErrorCode.E0702, ex.getMessage(), ex); 220 } 221 } 222 223 /** 224 * Validate whether fork/join are in pair or not 225 * @param app LiteWorkflowApp 226 * @throws WorkflowException 227 */ 228 private void validateForkJoin(LiteWorkflowApp app) throws WorkflowException { 229 // Make sure the number of forks and joins in wf are equal 230 if (forkList.size() != joinList.size()) { 231 throw new WorkflowException(ErrorCode.E0730); 232 } 233 234 // No need to bother going through all of this if there are no fork/join nodes 235 if (!forkList.isEmpty()) { 236 visitedOkNodes.clear(); 237 visitedJoinNodes.clear(); 238 validateForkJoin(startNode, app, new LinkedList<String>(), new LinkedList<String>(), new LinkedList<String>(), true, 239 null); 240 } 241 } 242 243 /* 244 * Recursively walk through the DAG and make sure that all fork paths are valid. 245 * This should be called from validateForkJoin(LiteWorkflowApp app). It assumes that visitedOkNodes and visitedJoinNodes are 246 * both empty ArrayLists on the first call. 247 * 248 * @param node the current node; use the startNode on the first call 249 * @param app the WorkflowApp 250 * @param forkNodes a stack of the current fork nodes 251 * @param joinNodes a stack of the current join nodes 252 * @param path a stack of the current path 253 * @param okTo false if node (or an ancestor of node) was gotten to via an "error to" transition or via a join node that has 254 * already been visited at least once before 255 * @param topDecisionParent The top (eldest) decision node along the path to this node, or null if there isn't one 256 * @throws WorkflowException 257 */ 258 private void validateForkJoin(NodeDef node, LiteWorkflowApp app, Deque<String> forkNodes, Deque<String> joinNodes, 259 Deque<String> path, boolean okTo, String topDecisionParent) throws WorkflowException { 260 if (path.contains(node.getName())) { 261 // cycle 262 throw new WorkflowException(ErrorCode.E0741, node.getName(), Arrays.toString(path.toArray())); 263 } 264 path.push(node.getName()); 265 266 // Make sure that we're not revisiting a node (that's not a Kill, Join, or End type) that's been visited before from an 267 // "ok to" transition; if its from an "error to" transition, then its okay to visit it multiple times. Also, because we 268 // traverse through join nodes multiple times, we have to make sure not to throw an exception here when we're really just 269 // re-walking the same execution path (this is why we need the visitedJoinNodes list used later) 270 if (okTo && !(node instanceof KillNodeDef) && !(node instanceof JoinNodeDef) && !(node instanceof EndNodeDef)) { 271 NodeAndTopDecisionParent natdp = findInVisitedOkNodes(node.getName()); 272 if (natdp != null) { 273 // However, if we've visited the node and it's under a decision node, we may be seeing it again and it's only 274 // illegal if that decision node is not the same as what we're seeing now (because during execution we only go 275 // down one path of the decision node, so while we're seeing the node multiple times here, during runtime it will 276 // only be executed once). Also, this decision node should be the top (eldest) decision node. As null indicates 277 // that there isn't a decision node, when this happens they must both be null to be valid. Here is a good example 278 // to visualize a node ("actionX") that has three "ok to" paths to it, but should still be a valid workflow (it may 279 // be easier to see if you draw it): 280 // decisionA --> {actionX, decisionB} 281 // decisionB --> {actionX, actionY} 282 // actionY --> {actionX} 283 // And, if we visit this node twice under the same decision node in an invalid way, the path cycle checking code 284 // will catch it, so we don't have to worry about that here. 285 if ((natdp.topDecisionParent == null && topDecisionParent == null) 286 || (natdp.topDecisionParent == null && topDecisionParent != null) 287 || (natdp.topDecisionParent != null && topDecisionParent == null) 288 || !natdp.topDecisionParent.equals(topDecisionParent)) { 289 // If we get here, then we've seen this node before from an "ok to" transition but they don't have the same 290 // decision node top parent, which means that this node will be executed twice, which is illegal 291 throw new WorkflowException(ErrorCode.E0743, node.getName()); 292 } 293 } 294 else { 295 // If we haven't transitioned to this node before, add it and its top decision parent node 296 visitedOkNodes.add(new NodeAndTopDecisionParent(node.getName(), topDecisionParent)); 297 } 298 } 299 300 if (node instanceof StartNodeDef) { 301 String transition = node.getTransitions().get(0); // start always has only 1 transition 302 NodeDef tranNode = app.getNode(transition); 303 validateForkJoin(tranNode, app, forkNodes, joinNodes, path, okTo, topDecisionParent); 304 } 305 else if (node instanceof ActionNodeDef) { 306 String transition = node.getTransitions().get(0); // "ok to" transition 307 NodeDef tranNode = app.getNode(transition); 308 validateForkJoin(tranNode, app, forkNodes, joinNodes, path, okTo, topDecisionParent); // propogate okTo 309 transition = node.getTransitions().get(1); // "error to" transition 310 tranNode = app.getNode(transition); 311 validateForkJoin(tranNode, app, forkNodes, joinNodes, path, false, topDecisionParent); // use false 312 } 313 else if (node instanceof DecisionNodeDef) { 314 for(String transition : (new HashSet<String>(node.getTransitions()))) { 315 NodeDef tranNode = app.getNode(transition); 316 // if there currently isn't a topDecisionParent (i.e. null), then use this node instead of propagating null 317 String parentDecisionNode = topDecisionParent; 318 if (parentDecisionNode == null) { 319 parentDecisionNode = node.getName(); 320 } 321 validateForkJoin(tranNode, app, forkNodes, joinNodes, path, okTo, parentDecisionNode); 322 } 323 } 324 else if (node instanceof ForkNodeDef) { 325 forkNodes.push(node.getName()); 326 List<String> transitionsList = node.getTransitions(); 327 HashSet<String> transitionsSet = new HashSet<String>(transitionsList); 328 // Check that a fork doesn't go to the same node more than once 329 if (!transitionsList.isEmpty() && transitionsList.size() != transitionsSet.size()) { 330 // Now we have to figure out which node is the problem and what type of node they are (join and kill are ok) 331 for (int i = 0; i < transitionsList.size(); i++) { 332 String a = transitionsList.get(i); 333 NodeDef aNode = app.getNode(a); 334 if (!(aNode instanceof JoinNodeDef) && !(aNode instanceof KillNodeDef)) { 335 for (int k = i+1; k < transitionsList.size(); k++) { 336 String b = transitionsList.get(k); 337 if (a.equals(b)) { 338 throw new WorkflowException(ErrorCode.E0744, node.getName(), a); 339 } 340 } 341 } 342 } 343 } 344 for(String transition : transitionsSet) { 345 NodeDef tranNode = app.getNode(transition); 346 validateForkJoin(tranNode, app, forkNodes, joinNodes, path, okTo, topDecisionParent); 347 } 348 forkNodes.pop(); 349 if (!joinNodes.isEmpty()) { 350 joinNodes.pop(); 351 } 352 } 353 else if (node instanceof JoinNodeDef) { 354 if (forkNodes.isEmpty()) { 355 // no fork for join to match with 356 throw new WorkflowException(ErrorCode.E0742, node.getName()); 357 } 358 if (forkNodes.size() > joinNodes.size() && (joinNodes.isEmpty() || !joinNodes.peek().equals(node.getName()))) { 359 joinNodes.push(node.getName()); 360 } 361 if (!joinNodes.peek().equals(node.getName())) { 362 // join doesn't match fork 363 throw new WorkflowException(ErrorCode.E0732, forkNodes.peek(), node.getName(), joinNodes.peek()); 364 } 365 joinNodes.pop(); 366 String currentForkNode = forkNodes.pop(); 367 String transition = node.getTransitions().get(0); // join always has only 1 transition 368 NodeDef tranNode = app.getNode(transition); 369 // If we're already under a situation where okTo is false, use false (propogate it) 370 // Or if we've already visited this join node, use false (because we've already traversed this path before and we don't 371 // want to throw an exception from the check against visitedOkNodes) 372 if (!okTo || visitedJoinNodes.contains(node.getName())) { 373 validateForkJoin(tranNode, app, forkNodes, joinNodes, path, false, topDecisionParent); 374 // Else, use true because this is either the first time we've gone through this join node or okTo was already false 375 } else { 376 visitedJoinNodes.add(node.getName()); 377 validateForkJoin(tranNode, app, forkNodes, joinNodes, path, true, topDecisionParent); 378 } 379 forkNodes.push(currentForkNode); 380 joinNodes.push(node.getName()); 381 } 382 else if (node instanceof KillNodeDef) { 383 // do nothing 384 } 385 else if (node instanceof EndNodeDef) { 386 if (!forkNodes.isEmpty()) { 387 path.pop(); // = node 388 String parent = path.peek(); 389 // can't go to an end node in a fork 390 throw new WorkflowException(ErrorCode.E0737, parent, node.getName()); 391 } 392 } 393 else { 394 // invalid node type (shouldn't happen) 395 throw new WorkflowException(ErrorCode.E0740, node.getName()); 396 } 397 path.pop(); 398 } 399 400 /** 401 * Return a {@link NodeAndTopDecisionParent} whose {@link NodeAndTopDecisionParent#node} is equal to the passed in name, or null 402 * if it isn't in the {@link LiteWorkflowAppParser#visitedOkNodes} list. 403 * 404 * @param name The name to search for 405 * @return a NodeAndTopDecisionParent or null 406 */ 407 private NodeAndTopDecisionParent findInVisitedOkNodes(String name) { 408 NodeAndTopDecisionParent natdp = null; 409 for (NodeAndTopDecisionParent v : visitedOkNodes) { 410 if (v.node.equals(name)) { 411 natdp = v; 412 break; 413 } 414 } 415 return natdp; 416 } 417 418 /** 419 * Parse xml to {@link LiteWorkflowApp} 420 * 421 * @param strDef 422 * @param root 423 * @param configDefault 424 * @param jobConf 425 * @return LiteWorkflowApp 426 * @throws WorkflowException 427 */ 428 @SuppressWarnings({"unchecked"}) 429 private LiteWorkflowApp parse(String strDef, Element root, Configuration configDefault, Configuration jobConf) 430 throws WorkflowException { 431 Namespace ns = root.getNamespace(); 432 LiteWorkflowApp def = null; 433 GlobalSectionData gData = jobConf.get(OOZIE_GLOBAL) == null ? 434 null : getGlobalFromString(jobConf.get(OOZIE_GLOBAL)); 435 boolean serializedGlobalConf = false; 436 for (Element eNode : (List<Element>) root.getChildren()) { 437 if (eNode.getName().equals(START_E)) { 438 def = new LiteWorkflowApp(root.getAttributeValue(NAME_A), strDef, 439 new StartNodeDef(controlNodeHandler, eNode.getAttributeValue(TO_A))); 440 } else if (eNode.getName().equals(END_E)) { 441 def.addNode(new EndNodeDef(eNode.getAttributeValue(NAME_A), controlNodeHandler)); 442 } else if (eNode.getName().equals(KILL_E)) { 443 def.addNode(new KillNodeDef(eNode.getAttributeValue(NAME_A), 444 eNode.getChildText(KILL_MESSAGE_E, ns), controlNodeHandler)); 445 } else if (eNode.getName().equals(FORK_E)) { 446 List<String> paths = new ArrayList<String>(); 447 for (Element tran : (List<Element>) eNode.getChildren(FORK_PATH_E, ns)) { 448 paths.add(tran.getAttributeValue(FORK_START_A)); 449 } 450 def.addNode(new ForkNodeDef(eNode.getAttributeValue(NAME_A), controlNodeHandler, paths)); 451 } else if (eNode.getName().equals(JOIN_E)) { 452 def.addNode(new JoinNodeDef(eNode.getAttributeValue(NAME_A), controlNodeHandler, eNode.getAttributeValue(TO_A))); 453 } else if (eNode.getName().equals(DECISION_E)) { 454 Element eSwitch = eNode.getChild(DECISION_SWITCH_E, ns); 455 List<String> transitions = new ArrayList<String>(); 456 for (Element e : (List<Element>) eSwitch.getChildren(DECISION_CASE_E, ns)) { 457 transitions.add(e.getAttributeValue(TO_A)); 458 } 459 transitions.add(eSwitch.getChild(DECISION_DEFAULT_E, ns).getAttributeValue(TO_A)); 460 461 String switchStatement = XmlUtils.prettyPrint(eSwitch).toString(); 462 def.addNode(new DecisionNodeDef(eNode.getAttributeValue(NAME_A), switchStatement, decisionHandlerClass, 463 transitions)); 464 } else if (ACTION_E.equals(eNode.getName())) { 465 String[] transitions = new String[2]; 466 Element eActionConf = null; 467 for (Element elem : (List<Element>) eNode.getChildren()) { 468 if (ACTION_OK_E.equals(elem.getName())) { 469 transitions[0] = elem.getAttributeValue(TO_A); 470 } else if (ACTION_ERROR_E.equals(elem.getName())) { 471 transitions[1] = elem.getAttributeValue(TO_A); 472 } else if (SLA_INFO.equals(elem.getName()) || CREDENTIALS.equals(elem.getName())) { 473 continue; 474 } else { 475 if (!serializedGlobalConf && elem.getName().equals(SubWorkflowActionExecutor.ACTION_TYPE) && 476 elem.getChild(("propagate-configuration"), ns) != null && gData != null) { 477 serializedGlobalConf = true; 478 jobConf.set(OOZIE_GLOBAL, getGlobalString(gData)); 479 } 480 eActionConf = elem; 481 handleDefaultsAndGlobal(gData, configDefault, elem); 482 } 483 } 484 485 String credStr = eNode.getAttributeValue(CRED_A); 486 String userRetryMaxStr = eNode.getAttributeValue(USER_RETRY_MAX_A); 487 String userRetryIntervalStr = eNode.getAttributeValue(USER_RETRY_INTERVAL_A); 488 try { 489 if (!StringUtils.isEmpty(userRetryMaxStr)) { 490 userRetryMaxStr = ELUtils.resolveAppName(userRetryMaxStr, jobConf); 491 } 492 if (!StringUtils.isEmpty(userRetryIntervalStr)) { 493 userRetryIntervalStr = ELUtils.resolveAppName(userRetryIntervalStr, jobConf); 494 } 495 } 496 catch (Exception e) { 497 throw new WorkflowException(ErrorCode.E0703, e.getMessage()); 498 } 499 500 String actionConf = XmlUtils.prettyPrint(eActionConf).toString(); 501 def.addNode(new ActionNodeDef(eNode.getAttributeValue(NAME_A), actionConf, actionHandlerClass, 502 transitions[0], transitions[1], credStr, 503 userRetryMaxStr, userRetryIntervalStr)); 504 } else if (SLA_INFO.equals(eNode.getName()) || CREDENTIALS.equals(eNode.getName())) { 505 // No operation is required 506 } else if (eNode.getName().equals(GLOBAL)) { 507 if(jobConf.get(OOZIE_GLOBAL) != null) { 508 gData = getGlobalFromString(jobConf.get(OOZIE_GLOBAL)); 509 handleDefaultsAndGlobal(gData, null, eNode); 510 } 511 gData = parseGlobalSection(ns, eNode); 512 } else if (eNode.getName().equals(PARAMETERS)) { 513 // No operation is required 514 } else { 515 throw new WorkflowException(ErrorCode.E0703, eNode.getName()); 516 } 517 } 518 return def; 519 } 520 521 /** 522 * Read the GlobalSectionData from Base64 string. 523 * @param globalStr 524 * @return GlobalSectionData 525 * @throws WorkflowException 526 */ 527 private GlobalSectionData getGlobalFromString(String globalStr) throws WorkflowException { 528 GlobalSectionData globalSectionData = new GlobalSectionData(); 529 try { 530 byte[] data = Base64.decodeBase64(globalStr); 531 Inflater inflater = new Inflater(); 532 DataInputStream ois = new DataInputStream(new InflaterInputStream(new ByteArrayInputStream(data), inflater)); 533 globalSectionData.readFields(ois); 534 ois.close(); 535 } catch (Exception ex) { 536 throw new WorkflowException(ErrorCode.E0700, "Error while processing global section conf"); 537 } 538 return globalSectionData; 539 } 540 541 542 /** 543 * Write the GlobalSectionData to a Base64 string. 544 * @param globalSectionData 545 * @return String 546 * @throws WorkflowException 547 */ 548 private String getGlobalString(GlobalSectionData globalSectionData) throws WorkflowException { 549 ByteArrayOutputStream baos = new ByteArrayOutputStream(); 550 DataOutputStream oos = null; 551 try { 552 Deflater def = new Deflater(); 553 oos = new DataOutputStream(new DeflaterOutputStream(baos, def)); 554 globalSectionData.write(oos); 555 oos.close(); 556 } catch (IOException e) { 557 throw new WorkflowException(ErrorCode.E0700, "Error while processing global section conf"); 558 } 559 return Base64.encodeBase64String(baos.toByteArray()); 560 } 561 562 /** 563 * Validate workflow xml 564 * 565 * @param app 566 * @param node 567 * @param traversed 568 * @throws WorkflowException 569 */ 570 private void validate(LiteWorkflowApp app, NodeDef node, Map<String, VisitStatus> traversed) throws WorkflowException { 571 if (node instanceof StartNodeDef) { 572 startNode = (StartNodeDef) node; 573 } 574 else { 575 try { 576 ParamChecker.validateActionName(node.getName()); 577 } 578 catch (IllegalArgumentException ex) { 579 throw new WorkflowException(ErrorCode.E0724, ex.getMessage()); 580 } 581 } 582 if (node instanceof ActionNodeDef) { 583 try { 584 Element action = XmlUtils.parseXml(node.getConf()); 585 boolean supportedAction = Services.get().get(ActionService.class).getExecutor(action.getName()) != null; 586 if (!supportedAction) { 587 throw new WorkflowException(ErrorCode.E0723, node.getName(), action.getName()); 588 } 589 } 590 catch (JDOMException ex) { 591 throw new RuntimeException("It should never happen, " + ex.getMessage(), ex); 592 } 593 } 594 595 if(node instanceof ForkNodeDef){ 596 forkList.add(node.getName()); 597 } 598 599 if(node instanceof JoinNodeDef){ 600 joinList.add(node.getName()); 601 } 602 603 if (node instanceof EndNodeDef) { 604 traversed.put(node.getName(), VisitStatus.VISITED); 605 return; 606 } 607 if (node instanceof KillNodeDef) { 608 traversed.put(node.getName(), VisitStatus.VISITED); 609 return; 610 } 611 for (String transition : node.getTransitions()) { 612 613 if (app.getNode(transition) == null) { 614 throw new WorkflowException(ErrorCode.E0708, node.getName(), transition); 615 } 616 617 //check if it is a cycle 618 if (traversed.get(app.getNode(transition).getName()) == VisitStatus.VISITING) { 619 throw new WorkflowException(ErrorCode.E0707, app.getNode(transition).getName()); 620 } 621 //ignore validated one 622 if (traversed.get(app.getNode(transition).getName()) == VisitStatus.VISITED) { 623 continue; 624 } 625 626 traversed.put(app.getNode(transition).getName(), VisitStatus.VISITING); 627 validate(app, app.getNode(transition), traversed); 628 } 629 traversed.put(node.getName(), VisitStatus.VISITED); 630 } 631 632 private void addChildElement(Element parent, Namespace ns, String childName, String childValue) { 633 Element child = new Element(childName, ns); 634 child.setText(childValue); 635 parent.addContent(child); 636 } 637 638 private class GlobalSectionData implements Writable { 639 String jobTracker; 640 String nameNode; 641 List<String> jobXmls; 642 Configuration conf; 643 644 public GlobalSectionData() { 645 } 646 647 public GlobalSectionData(String jobTracker, String nameNode, List<String> jobXmls, Configuration conf) { 648 this.jobTracker = jobTracker; 649 this.nameNode = nameNode; 650 this.jobXmls = jobXmls; 651 this.conf = conf; 652 } 653 654 @Override 655 public void write(DataOutput dataOutput) throws IOException { 656 WritableUtils.writeStr(dataOutput, jobTracker); 657 WritableUtils.writeStr(dataOutput, nameNode); 658 659 if(jobXmls != null && !jobXmls.isEmpty()) { 660 dataOutput.writeInt(jobXmls.size()); 661 for (String content : jobXmls) { 662 WritableUtils.writeStr(dataOutput, content); 663 } 664 } else { 665 dataOutput.writeInt(0); 666 } 667 if(conf != null) { 668 WritableUtils.writeStr(dataOutput, XmlUtils.prettyPrint(conf).toString()); 669 } else { 670 WritableUtils.writeStr(dataOutput, null); 671 } 672 } 673 674 @Override 675 public void readFields(DataInput dataInput) throws IOException { 676 jobTracker = WritableUtils.readStr(dataInput); 677 nameNode = WritableUtils.readStr(dataInput); 678 int length = dataInput.readInt(); 679 if (length > 0) { 680 jobXmls = new ArrayList<String>(); 681 for (int i = 0; i < length; i++) { 682 jobXmls.add(WritableUtils.readStr(dataInput)); 683 } 684 } 685 String confString = WritableUtils.readStr(dataInput); 686 if(confString != null) { 687 conf = new XConfiguration(new StringReader(confString)); 688 } 689 } 690 } 691 692 private GlobalSectionData parseGlobalSection(Namespace ns, Element global) throws WorkflowException { 693 GlobalSectionData gData = null; 694 if (global != null) { 695 String globalJobTracker = null; 696 Element globalJobTrackerElement = global.getChild(JOB_TRACKER, ns); 697 if (globalJobTrackerElement != null) { 698 globalJobTracker = globalJobTrackerElement.getValue(); 699 } 700 701 String globalNameNode = null; 702 Element globalNameNodeElement = global.getChild(NAME_NODE, ns); 703 if (globalNameNodeElement != null) { 704 globalNameNode = globalNameNodeElement.getValue(); 705 } 706 707 List<String> globalJobXmls = null; 708 @SuppressWarnings("unchecked") 709 List<Element> globalJobXmlElements = global.getChildren(JOB_XML, ns); 710 if (!globalJobXmlElements.isEmpty()) { 711 globalJobXmls = new ArrayList<String>(globalJobXmlElements.size()); 712 for(Element jobXmlElement: globalJobXmlElements) { 713 globalJobXmls.add(jobXmlElement.getText()); 714 } 715 } 716 717 Configuration globalConf = null; 718 Element globalConfigurationElement = global.getChild(CONFIGURATION, ns); 719 if (globalConfigurationElement != null) { 720 try { 721 globalConf = new XConfiguration(new StringReader(XmlUtils.prettyPrint(globalConfigurationElement).toString())); 722 } catch (IOException ioe) { 723 throw new WorkflowException(ErrorCode.E0700, "Error while processing global section conf"); 724 } 725 } 726 gData = new GlobalSectionData(globalJobTracker, globalNameNode, globalJobXmls, globalConf); 727 } 728 return gData; 729 } 730 731 private void handleDefaultsAndGlobal(GlobalSectionData gData, Configuration configDefault, Element actionElement) 732 throws WorkflowException { 733 734 ActionExecutor ae = Services.get().get(ActionService.class).getExecutor(actionElement.getName()); 735 if (ae == null && !GLOBAL.equals(actionElement.getName())) { 736 throw new WorkflowException(ErrorCode.E0723, actionElement.getName(), ActionService.class.getName()); 737 } 738 739 Namespace actionNs = actionElement.getNamespace(); 740 741 // If this is the global section or ActionExecutor.requiresNameNodeJobTracker() returns true, we parse the action's 742 // <name-node> and <job-tracker> fields. If those aren't defined, we take them from the <global> section. If those 743 // aren't defined, we take them from the oozie-site defaults. If those aren't defined, we throw a WorkflowException. 744 // However, for the SubWorkflow and FS Actions, as well as the <global> section, we don't throw the WorkflowException. 745 // Also, we only parse the NN (not the JT) for the FS Action. 746 if (SubWorkflowActionExecutor.ACTION_TYPE.equals(actionElement.getName()) || 747 FsActionExecutor.ACTION_TYPE.equals(actionElement.getName()) || 748 GLOBAL.equals(actionElement.getName()) || ae.requiresNameNodeJobTracker()) { 749 if (actionElement.getChild(NAME_NODE, actionNs) == null) { 750 if (gData != null && gData.nameNode != null) { 751 addChildElement(actionElement, actionNs, NAME_NODE, gData.nameNode); 752 } else if (defaultNameNode != null) { 753 addChildElement(actionElement, actionNs, NAME_NODE, defaultNameNode); 754 } else if (!(SubWorkflowActionExecutor.ACTION_TYPE.equals(actionElement.getName()) || 755 FsActionExecutor.ACTION_TYPE.equals(actionElement.getName()) || 756 GLOBAL.equals(actionElement.getName()))) { 757 throw new WorkflowException(ErrorCode.E0701, "No " + NAME_NODE + " defined"); 758 } 759 } 760 if (actionElement.getChild(JOB_TRACKER, actionNs) == null && 761 !FsActionExecutor.ACTION_TYPE.equals(actionElement.getName())) { 762 if (gData != null && gData.jobTracker != null) { 763 addChildElement(actionElement, actionNs, JOB_TRACKER, gData.jobTracker); 764 } else if (defaultJobTracker != null) { 765 addChildElement(actionElement, actionNs, JOB_TRACKER, defaultJobTracker); 766 } else if (!(SubWorkflowActionExecutor.ACTION_TYPE.equals(actionElement.getName()) || 767 GLOBAL.equals(actionElement.getName()))) { 768 throw new WorkflowException(ErrorCode.E0701, "No " + JOB_TRACKER + " defined"); 769 } 770 } 771 } 772 773 // If this is the global section or ActionExecutor.supportsConfigurationJobXML() returns true, we parse the action's 774 // <configuration> and <job-xml> fields. We also merge this with those from the <global> section, if given. If none are 775 // defined, empty values are placed. Exceptions are thrown if there's an error parsing, but not if they're not given. 776 if ( GLOBAL.equals(actionElement.getName()) || ae.supportsConfigurationJobXML()) { 777 @SuppressWarnings("unchecked") 778 List<Element> actionJobXmls = actionElement.getChildren(JOB_XML, actionNs); 779 if (gData != null && gData.jobXmls != null) { 780 for(String gJobXml : gData.jobXmls) { 781 boolean alreadyExists = false; 782 for (Element actionXml : actionJobXmls) { 783 if (gJobXml.equals(actionXml.getText())) { 784 alreadyExists = true; 785 break; 786 } 787 } 788 if (!alreadyExists) { 789 Element ejobXml = new Element(JOB_XML, actionNs); 790 ejobXml.setText(gJobXml); 791 actionElement.addContent(ejobXml); 792 } 793 } 794 } 795 796 try { 797 XConfiguration actionConf = new XConfiguration(); 798 if (configDefault != null) 799 XConfiguration.copy(configDefault, actionConf); 800 if (gData != null && gData.conf != null) { 801 XConfiguration.copy(gData.conf, actionConf); 802 } 803 Element actionConfiguration = actionElement.getChild(CONFIGURATION, actionNs); 804 if (actionConfiguration != null) { 805 //copy and override 806 XConfiguration.copy(new XConfiguration(new StringReader(XmlUtils.prettyPrint(actionConfiguration).toString())), 807 actionConf); 808 } 809 int position = actionElement.indexOf(actionConfiguration); 810 actionElement.removeContent(actionConfiguration); //replace with enhanced one 811 Element eConfXml = XmlUtils.parseXml(actionConf.toXmlString(false)); 812 eConfXml.detach(); 813 eConfXml.setNamespace(actionNs); 814 if (position > 0) { 815 actionElement.addContent(position, eConfXml); 816 } 817 else { 818 actionElement.addContent(eConfXml); 819 } 820 } 821 catch (IOException e) { 822 throw new WorkflowException(ErrorCode.E0700, "Error while processing action conf"); 823 } 824 catch (JDOMException e) { 825 throw new WorkflowException(ErrorCode.E0700, "Error while processing action conf"); 826 } 827 } 828 } 829}