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.wf; 020 021import org.apache.hadoop.conf.Configuration; 022import org.apache.hadoop.fs.Path; 023import org.apache.hadoop.fs.FileSystem; 024import org.apache.oozie.AppType; 025import org.apache.oozie.SLAEventBean; 026import org.apache.oozie.command.wf.ActionXCommand.ActionExecutorContext; 027import org.apache.oozie.WorkflowJobBean; 028import org.apache.oozie.ErrorCode; 029import org.apache.oozie.action.oozie.SubWorkflowActionExecutor; 030import org.apache.oozie.service.HadoopAccessorException; 031import org.apache.oozie.service.JPAService; 032import org.apache.oozie.service.UUIDService; 033import org.apache.oozie.service.WorkflowStoreService; 034import org.apache.oozie.service.WorkflowAppService; 035import org.apache.oozie.service.HadoopAccessorService; 036import org.apache.oozie.service.Services; 037import org.apache.oozie.service.DagXLogInfoService; 038import org.apache.oozie.util.ELUtils; 039import org.apache.oozie.util.LogUtils; 040import org.apache.oozie.sla.SLAOperations; 041import org.apache.oozie.util.XLog; 042import org.apache.oozie.util.ParamChecker; 043import org.apache.oozie.util.XConfiguration; 044import org.apache.oozie.util.XmlUtils; 045import org.apache.oozie.command.CommandException; 046import org.apache.oozie.executor.jpa.BatchQueryExecutor; 047import org.apache.oozie.executor.jpa.JPAExecutorException; 048import org.apache.oozie.service.ELService; 049import org.apache.oozie.store.StoreException; 050import org.apache.oozie.workflow.WorkflowApp; 051import org.apache.oozie.workflow.WorkflowException; 052import org.apache.oozie.workflow.WorkflowInstance; 053import org.apache.oozie.workflow.WorkflowLib; 054import org.apache.oozie.util.ELEvaluator; 055import org.apache.oozie.util.InstrumentUtils; 056import org.apache.oozie.util.PropertiesUtils; 057import org.apache.oozie.util.db.SLADbOperations; 058import org.apache.oozie.service.SchemaService.SchemaName; 059import org.apache.oozie.client.OozieClient; 060import org.apache.oozie.client.WorkflowJob; 061import org.apache.oozie.client.SLAEvent.SlaAppType; 062import org.apache.oozie.client.rest.JsonBean; 063import org.jdom.Element; 064import org.jdom.filter.ElementFilter; 065 066import java.util.ArrayList; 067import java.util.Date; 068import java.util.Iterator; 069import java.util.List; 070import java.util.Map; 071import java.util.Set; 072import java.util.HashSet; 073import java.io.IOException; 074import java.net.URI; 075 076@SuppressWarnings("deprecation") 077public class SubmitXCommand extends WorkflowXCommand<String> { 078 public static final String CONFIG_DEFAULT = "config-default.xml"; 079 080 private Configuration conf; 081 private List<JsonBean> insertList = new ArrayList<JsonBean>(); 082 private String parentId; 083 084 /** 085 * Constructor to create the workflow Submit Command. 086 * 087 * @param conf : Configuration for workflow job 088 */ 089 public SubmitXCommand(Configuration conf) { 090 super("submit", "submit", 1); 091 this.conf = ParamChecker.notNull(conf, "conf"); 092 } 093 094 /** 095 * Constructor for submitting wf through coordinator 096 * 097 * @param conf : Configuration for workflow job 098 * @param parentId: the coord action id 099 */ 100 public SubmitXCommand(Configuration conf, String parentId) { 101 this(conf); 102 this.parentId = parentId; 103 } 104 105 /** 106 * Constructor to create the workflow Submit Command. 107 * 108 * @param dryrun : if dryrun 109 * @param conf : Configuration for workflow job 110 * @param authToken : To be used for authentication 111 */ 112 public SubmitXCommand(boolean dryrun, Configuration conf) { 113 this(conf); 114 this.dryrun = dryrun; 115 } 116 117 private static final Set<String> DISALLOWED_DEFAULT_PROPERTIES = new HashSet<String>(); 118 private static final Set<String> DISALLOWED_USER_PROPERTIES = new HashSet<String>(); 119 120 static { 121 String[] badUserProps = {PropertiesUtils.DAYS, PropertiesUtils.HOURS, PropertiesUtils.MINUTES, 122 PropertiesUtils.KB, PropertiesUtils.MB, PropertiesUtils.GB, PropertiesUtils.TB, PropertiesUtils.PB, 123 PropertiesUtils.RECORDS, PropertiesUtils.MAP_IN, PropertiesUtils.MAP_OUT, PropertiesUtils.REDUCE_IN, 124 PropertiesUtils.REDUCE_OUT, PropertiesUtils.GROUPS}; 125 PropertiesUtils.createPropertySet(badUserProps, DISALLOWED_USER_PROPERTIES); 126 127 String[] badDefaultProps = {PropertiesUtils.HADOOP_USER}; 128 PropertiesUtils.createPropertySet(badUserProps, DISALLOWED_DEFAULT_PROPERTIES); 129 PropertiesUtils.createPropertySet(badDefaultProps, DISALLOWED_DEFAULT_PROPERTIES); 130 } 131 132 @Override 133 protected String execute() throws CommandException { 134 InstrumentUtils.incrJobCounter(getName(), 1, getInstrumentation()); 135 WorkflowAppService wps = Services.get().get(WorkflowAppService.class); 136 try { 137 XLog.Info.get().setParameter(DagXLogInfoService.TOKEN, conf.get(OozieClient.LOG_TOKEN)); 138 String user = conf.get(OozieClient.USER_NAME); 139 URI uri = new URI(conf.get(OozieClient.APP_PATH)); 140 HadoopAccessorService has = Services.get().get(HadoopAccessorService.class); 141 Configuration fsConf = has.createJobConf(uri.getAuthority()); 142 FileSystem fs = has.createFileSystem(user, uri, fsConf); 143 144 Path configDefault = null; 145 Configuration defaultConf = null; 146 // app path could be a directory 147 Path path = new Path(uri.getPath()); 148 if (!fs.isFile(path)) { 149 configDefault = new Path(path, CONFIG_DEFAULT); 150 } else { 151 configDefault = new Path(path.getParent(), CONFIG_DEFAULT); 152 } 153 154 if (fs.exists(configDefault)) { 155 try { 156 defaultConf = new XConfiguration(fs.open(configDefault)); 157 PropertiesUtils.checkDisallowedProperties(defaultConf, DISALLOWED_DEFAULT_PROPERTIES); 158 XConfiguration.injectDefaults(defaultConf, conf); 159 } 160 catch (IOException ex) { 161 throw new IOException("default configuration file, " + ex.getMessage(), ex); 162 } 163 } 164 165 WorkflowApp app = wps.parseDef(conf, defaultConf); 166 XConfiguration protoActionConf = wps.createProtoActionConf(conf, true); 167 WorkflowLib workflowLib = Services.get().get(WorkflowStoreService.class).getWorkflowLibWithNoDB(); 168 169 PropertiesUtils.checkDisallowedProperties(conf, DISALLOWED_USER_PROPERTIES); 170 171 // Resolving all variables in the job properties. 172 // This ensures the Hadoop Configuration semantics is preserved. 173 XConfiguration resolvedVarsConf = new XConfiguration(); 174 for (Map.Entry<String, String> entry : conf) { 175 resolvedVarsConf.set(entry.getKey(), conf.get(entry.getKey())); 176 } 177 conf = resolvedVarsConf; 178 179 WorkflowInstance wfInstance; 180 try { 181 wfInstance = workflowLib.createInstance(app, conf); 182 } 183 catch (WorkflowException e) { 184 throw new StoreException(e); 185 } 186 187 Configuration conf = wfInstance.getConf(); 188 // System.out.println("WF INSTANCE CONF:"); 189 // System.out.println(XmlUtils.prettyPrint(conf).toString()); 190 191 WorkflowJobBean workflow = new WorkflowJobBean(); 192 workflow.setId(wfInstance.getId()); 193 workflow.setAppName(ELUtils.resolveAppName(app.getName(), conf)); 194 workflow.setAppPath(conf.get(OozieClient.APP_PATH)); 195 workflow.setConf(XmlUtils.prettyPrint(conf).toString()); 196 workflow.setProtoActionConf(protoActionConf.toXmlString()); 197 workflow.setCreatedTime(new Date()); 198 workflow.setLastModifiedTime(new Date()); 199 workflow.setLogToken(conf.get(OozieClient.LOG_TOKEN, "")); 200 workflow.setStatus(WorkflowJob.Status.PREP); 201 workflow.setRun(0); 202 workflow.setUser(conf.get(OozieClient.USER_NAME)); 203 workflow.setGroup(conf.get(OozieClient.GROUP_NAME)); 204 workflow.setWorkflowInstance(wfInstance); 205 workflow.setExternalId(conf.get(OozieClient.EXTERNAL_ID)); 206 // Set parent id if it doesn't already have one (for subworkflows) 207 if (workflow.getParentId() == null) { 208 workflow.setParentId(conf.get(SubWorkflowActionExecutor.PARENT_ID)); 209 } 210 // Set to coord action Id if workflow submitted through coordinator 211 if (workflow.getParentId() == null) { 212 workflow.setParentId(parentId); 213 } 214 215 LogUtils.setLogInfo(workflow); 216 LOG.debug("Workflow record created, Status [{0}]", workflow.getStatus()); 217 Element wfElem = XmlUtils.parseXml(app.getDefinition()); 218 ELEvaluator evalSla = createELEvaluatorForGroup(conf, "wf-sla-submit"); 219 String jobSlaXml = verifySlaElements(wfElem, evalSla); 220 if (!dryrun) { 221 writeSLARegistration(wfElem, jobSlaXml, workflow.getId(), workflow.getParentId(), workflow.getUser(), 222 workflow.getGroup(), workflow.getAppName(), LOG, evalSla); 223 workflow.setSlaXml(jobSlaXml); 224 // System.out.println("SlaXml :"+ slaXml); 225 226 //store.insertWorkflow(workflow); 227 insertList.add(workflow); 228 JPAService jpaService = Services.get().get(JPAService.class); 229 if (jpaService != null) { 230 try { 231 BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(insertList, null, null); 232 } 233 catch (JPAExecutorException je) { 234 throw new CommandException(je); 235 } 236 } 237 else { 238 LOG.error(ErrorCode.E0610); 239 return null; 240 } 241 242 return workflow.getId(); 243 } 244 else { 245 // Checking variable substitution for dryrun 246 ActionExecutorContext context = new ActionXCommand.ActionExecutorContext(workflow, null, false, false); 247 Element workflowXml = XmlUtils.parseXml(app.getDefinition()); 248 removeSlaElements(workflowXml); 249 String workflowXmlString = XmlUtils.removeComments(XmlUtils.prettyPrint(workflowXml).toString()); 250 workflowXmlString = context.getELEvaluator().evaluate(workflowXmlString, String.class); 251 workflowXml = XmlUtils.parseXml(workflowXmlString); 252 253 Iterator<Element> it = workflowXml.getDescendants(new ElementFilter("job-xml")); 254 255 // Checking all variable substitutions in job-xml files 256 while (it.hasNext()) { 257 Element e = it.next(); 258 String jobXml = e.getTextTrim(); 259 Path xmlPath = new Path(workflow.getAppPath(), jobXml); 260 Configuration jobXmlConf = new XConfiguration(fs.open(xmlPath)); 261 262 263 String jobXmlConfString = XmlUtils.prettyPrint(jobXmlConf).toString(); 264 jobXmlConfString = XmlUtils.removeComments(jobXmlConfString); 265 context.getELEvaluator().evaluate(jobXmlConfString, String.class); 266 } 267 268 return "OK"; 269 } 270 } 271 catch (WorkflowException ex) { 272 throw new CommandException(ex); 273 } 274 catch (HadoopAccessorException ex) { 275 throw new CommandException(ex); 276 } 277 catch (Exception ex) { 278 throw new CommandException(ErrorCode.E0803, ex.getMessage(), ex); 279 } 280 } 281 282 private void removeSlaElements(Element eWfJob) { 283 Element sla = XmlUtils.getSLAElement(eWfJob); 284 if (sla != null) { 285 eWfJob.removeChildren(sla.getName(), sla.getNamespace()); 286 } 287 288 for (Element action : (List<Element>) eWfJob.getChildren("action", eWfJob.getNamespace())) { 289 sla = XmlUtils.getSLAElement(action); 290 if (sla != null) { 291 action.removeChildren(sla.getName(), sla.getNamespace()); 292 } 293 } 294 } 295 private String verifySlaElements(Element eWfJob, ELEvaluator evalSla) throws CommandException { 296 String jobSlaXml = ""; 297 // Validate WF job 298 Element eSla = XmlUtils.getSLAElement(eWfJob); 299 if (eSla != null) { 300 jobSlaXml = resolveSla(eSla, evalSla); 301 } 302 303 // Validate all actions 304 for (Element action : (List<Element>) eWfJob.getChildren("action", eWfJob.getNamespace())) { 305 eSla = XmlUtils.getSLAElement(action); 306 if (eSla != null) { 307 resolveSla(eSla, evalSla); 308 } 309 } 310 return jobSlaXml; 311 } 312 313 private void writeSLARegistration(Element eWfJob, String slaXml, String jobId, String parentId, String user, 314 String group, String appName, XLog log, ELEvaluator evalSla) throws CommandException { 315 try { 316 if (slaXml != null && slaXml.length() > 0) { 317 Element eSla = XmlUtils.parseXml(slaXml); 318 SLAEventBean slaEvent = SLADbOperations.createSlaRegistrationEvent(eSla, jobId, 319 SlaAppType.WORKFLOW_JOB, user, group, log); 320 if(slaEvent != null) { 321 insertList.add(slaEvent); 322 } 323 // insert into new table 324 SLAOperations.createSlaRegistrationEvent(eSla, jobId, parentId, AppType.WORKFLOW_JOB, user, appName, 325 log, false); 326 } 327 // Add sla for wf actions 328 for (Element action : (List<Element>) eWfJob.getChildren("action", eWfJob.getNamespace())) { 329 Element actionSla = XmlUtils.getSLAElement(action); 330 if (actionSla != null) { 331 String actionSlaXml = SubmitXCommand.resolveSla(actionSla, evalSla); 332 actionSla = XmlUtils.parseXml(actionSlaXml); 333 String actionId = Services.get().get(UUIDService.class) 334 .generateChildId(jobId, action.getAttributeValue("name") + ""); 335 SLAOperations.createSlaRegistrationEvent(actionSla, actionId, jobId, AppType.WORKFLOW_ACTION, 336 user, appName, log, false); 337 } 338 } 339 } 340 catch (Exception e) { 341 e.printStackTrace(); 342 throw new CommandException(ErrorCode.E1007, "workflow " + jobId, e.getMessage(), e); 343 } 344 } 345 346 /** 347 * Resolve variables in sla xml element. 348 * 349 * @param eSla sla xml element 350 * @param evalSla sla evaluator 351 * @return sla xml string after evaluation 352 * @throws CommandException 353 */ 354 public static String resolveSla(Element eSla, ELEvaluator evalSla) throws CommandException { 355 // EL evaluation 356 String slaXml = XmlUtils.prettyPrint(eSla).toString(); 357 try { 358 slaXml = XmlUtils.removeComments(slaXml); 359 slaXml = evalSla.evaluate(slaXml, String.class); 360 XmlUtils.validateData(slaXml, SchemaName.SLA_ORIGINAL); 361 return slaXml; 362 } 363 catch (Exception e) { 364 throw new CommandException(ErrorCode.E1004, "Validation error :" + e.getMessage(), e); 365 } 366 } 367 368 /** 369 * Create an EL evaluator for a given group. 370 * 371 * @param conf configuration variable 372 * @param group group variable 373 * @return the evaluator created for the group 374 */ 375 public static ELEvaluator createELEvaluatorForGroup(Configuration conf, String group) { 376 ELEvaluator eval = Services.get().get(ELService.class).createEvaluator(group); 377 for (Map.Entry<String, String> entry : conf) { 378 eval.setVariable(entry.getKey(), entry.getValue()); 379 } 380 return eval; 381 } 382 383 @Override 384 public String getEntityKey() { 385 return null; 386 } 387 388 @Override 389 protected boolean isLockRequired() { 390 return false; 391 } 392 393 @Override 394 protected void loadState() { 395 396 } 397 398 @Override 399 protected void verifyPrecondition() throws CommandException { 400 401 } 402 403}