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 package org.apache.oozie; 019 020 import java.io.IOException; 021 import java.io.Writer; 022 import java.util.ArrayList; 023 import java.util.Date; 024 import java.util.HashMap; 025 import java.util.HashSet; 026 import java.util.Iterator; 027 import java.util.List; 028 import java.util.Map; 029 import java.util.Set; 030 import java.util.StringTokenizer; 031 import org.apache.hadoop.conf.Configuration; 032 import org.apache.oozie.client.CoordinatorJob; 033 import org.apache.oozie.client.OozieClient; 034 import org.apache.oozie.client.WorkflowJob; 035 import org.apache.oozie.client.rest.RestConstants; 036 import org.apache.oozie.command.CommandException; 037 import org.apache.oozie.command.coord.CoordActionInfoCommand; 038 import org.apache.oozie.command.coord.CoordActionInfoXCommand; 039 import org.apache.oozie.util.CoordActionsInDateRange; 040 import org.apache.oozie.command.coord.CoordChangeCommand; 041 import org.apache.oozie.command.coord.CoordChangeXCommand; 042 import org.apache.oozie.command.coord.CoordJobCommand; 043 import org.apache.oozie.command.coord.CoordJobXCommand; 044 import org.apache.oozie.command.coord.CoordJobsCommand; 045 import org.apache.oozie.command.coord.CoordJobsXCommand; 046 import org.apache.oozie.command.coord.CoordKillCommand; 047 import org.apache.oozie.command.coord.CoordKillXCommand; 048 import org.apache.oozie.command.coord.CoordRerunCommand; 049 import org.apache.oozie.command.coord.CoordRerunXCommand; 050 import org.apache.oozie.command.coord.CoordResumeCommand; 051 import org.apache.oozie.command.coord.CoordResumeXCommand; 052 import org.apache.oozie.command.coord.CoordSubmitCommand; 053 import org.apache.oozie.command.coord.CoordSubmitXCommand; 054 import org.apache.oozie.command.coord.CoordSuspendCommand; 055 import org.apache.oozie.command.coord.CoordSuspendXCommand; 056 import org.apache.oozie.service.DagXLogInfoService; 057 import org.apache.oozie.service.Services; 058 import org.apache.oozie.service.XLogService; 059 import org.apache.oozie.util.ParamChecker; 060 import org.apache.oozie.util.XLog; 061 import org.apache.oozie.util.XLogStreamer; 062 063 public class CoordinatorEngine extends BaseEngine { 064 private static boolean useXCommand = true; 065 private static XLog LOG = XLog.getLog(CoordinatorEngine.class); 066 067 /** 068 * Create a system Coordinator engine, with no user and no group. 069 */ 070 public CoordinatorEngine() { 071 if (Services.get().getConf().getBoolean(USE_XCOMMAND, true) == false) { 072 useXCommand = false; 073 LOG.debug("Oozie CoordinatorEngine is not using XCommands."); 074 } 075 else { 076 LOG.debug("Oozie CoordinatorEngine is using XCommands."); 077 } 078 } 079 080 /** 081 * Create a Coordinator engine to perform operations on behave of a user. 082 * 083 * @param user user name. 084 * @param authToken the authentication token. 085 */ 086 public CoordinatorEngine(String user, String authToken) { 087 this(); 088 this.user = ParamChecker.notEmpty(user, "user"); 089 this.authToken = ParamChecker.notEmpty(authToken, "authToken"); 090 } 091 092 /* 093 * (non-Javadoc) 094 * 095 * @see org.apache.oozie.BaseEngine#getDefinition(java.lang.String) 096 */ 097 @Override 098 public String getDefinition(String jobId) throws BaseEngineException { 099 CoordinatorJobBean job = getCoordJobWithNoActionInfo(jobId); 100 return job.getOrigJobXml(); 101 } 102 103 /** 104 * @param jobId 105 * @return CoordinatorJobBean 106 * @throws BaseEngineException 107 */ 108 private CoordinatorJobBean getCoordJobWithNoActionInfo(String jobId) throws BaseEngineException { 109 try { 110 if (useXCommand) { 111 return new CoordJobXCommand(jobId).call(); 112 } 113 else { 114 return new CoordJobCommand(jobId).call(); 115 } 116 } 117 catch (CommandException ex) { 118 throw new BaseEngineException(ex); 119 } 120 } 121 122 /** 123 * @param actionId 124 * @return CoordinatorActionBean 125 * @throws BaseEngineException 126 */ 127 public CoordinatorActionBean getCoordAction(String actionId) throws BaseEngineException { 128 try { 129 if (useXCommand) { 130 return new CoordActionInfoXCommand(actionId).call(); 131 } 132 else { 133 return new CoordActionInfoCommand(actionId).call(); 134 } 135 } 136 catch (CommandException ex) { 137 throw new BaseEngineException(ex); 138 } 139 } 140 141 /* 142 * (non-Javadoc) 143 * 144 * @see org.apache.oozie.BaseEngine#getCoordJob(java.lang.String) 145 */ 146 @Override 147 public CoordinatorJobBean getCoordJob(String jobId) throws BaseEngineException { 148 try { 149 if (useXCommand) { 150 return new CoordJobXCommand(jobId).call(); 151 } 152 else { 153 return new CoordJobCommand(jobId).call(); 154 } 155 } 156 catch (CommandException ex) { 157 throw new BaseEngineException(ex); 158 } 159 } 160 161 /* 162 * (non-Javadoc) 163 * 164 * @see org.apache.oozie.BaseEngine#getCoordJob(java.lang.String, int, int) 165 */ 166 @Override 167 public CoordinatorJobBean getCoordJob(String jobId, int start, int length) throws BaseEngineException { 168 try { 169 if (useXCommand) { 170 return new CoordJobXCommand(jobId, start, length).call(); 171 } 172 else { 173 return new CoordJobCommand(jobId, start, length).call(); 174 } 175 } 176 catch (CommandException ex) { 177 throw new BaseEngineException(ex); 178 } 179 } 180 181 /* 182 * (non-Javadoc) 183 * 184 * @see org.apache.oozie.BaseEngine#getJobIdForExternalId(java.lang.String) 185 */ 186 @Override 187 public String getJobIdForExternalId(String externalId) throws CoordinatorEngineException { 188 return null; 189 } 190 191 /* 192 * (non-Javadoc) 193 * 194 * @see org.apache.oozie.BaseEngine#kill(java.lang.String) 195 */ 196 @Override 197 public void kill(String jobId) throws CoordinatorEngineException { 198 try { 199 if (useXCommand) { 200 new CoordKillXCommand(jobId).call(); 201 } 202 else { 203 new CoordKillCommand(jobId).call(); 204 } 205 LOG.info("User " + user + " killed the Coordinator job " + jobId); 206 } 207 catch (CommandException e) { 208 throw new CoordinatorEngineException(e); 209 } 210 } 211 212 /* (non-Javadoc) 213 * @see org.apache.oozie.BaseEngine#change(java.lang.String, java.lang.String) 214 */ 215 @Override 216 public void change(String jobId, String changeValue) throws CoordinatorEngineException { 217 try { 218 if (useXCommand) { 219 new CoordChangeXCommand(jobId, changeValue).call(); 220 } 221 else { 222 new CoordChangeCommand(jobId, changeValue).call(); 223 } 224 LOG.info("User " + user + " changed the Coordinator job " + jobId + " to " + changeValue); 225 } 226 catch (CommandException e) { 227 throw new CoordinatorEngineException(e); 228 } 229 } 230 231 @Override 232 @Deprecated 233 public void reRun(String jobId, Configuration conf) throws BaseEngineException { 234 throw new BaseEngineException(new XException(ErrorCode.E0301)); 235 } 236 237 /** 238 * Rerun coordinator actions for given rerunType 239 * 240 * @param jobId 241 * @param rerunType 242 * @param scope 243 * @param refresh 244 * @param noCleanup 245 * @throws BaseEngineException 246 */ 247 public CoordinatorActionInfo reRun(String jobId, String rerunType, String scope, boolean refresh, boolean noCleanup) 248 throws BaseEngineException { 249 try { 250 if (useXCommand) { 251 return new CoordRerunXCommand(jobId, rerunType, scope, refresh, noCleanup).call(); 252 } 253 else { 254 return new CoordRerunCommand(jobId, rerunType, scope, refresh, noCleanup).call(); 255 } 256 } 257 catch (CommandException ex) { 258 throw new BaseEngineException(ex); 259 } 260 } 261 262 /* 263 * (non-Javadoc) 264 * 265 * @see org.apache.oozie.BaseEngine#resume(java.lang.String) 266 */ 267 @Override 268 public void resume(String jobId) throws CoordinatorEngineException { 269 try { 270 if (useXCommand) { 271 new CoordResumeXCommand(jobId).call(); 272 } 273 else { 274 new CoordResumeCommand(jobId).call(); 275 } 276 } 277 catch (CommandException e) { 278 throw new CoordinatorEngineException(e); 279 } 280 } 281 282 @Override 283 @Deprecated 284 public void start(String jobId) throws BaseEngineException { 285 throw new BaseEngineException(new XException(ErrorCode.E0301)); 286 } 287 288 /* 289 * (non-Javadoc) 290 * 291 * @see org.apache.oozie.BaseEngine#streamLog(java.lang.String, 292 * java.io.Writer) 293 */ 294 @Override 295 public void streamLog(String jobId, Writer writer) throws IOException, BaseEngineException { 296 XLogStreamer.Filter filter = new XLogStreamer.Filter(); 297 filter.setParameter(DagXLogInfoService.JOB, jobId); 298 299 CoordinatorJobBean job = getCoordJobWithNoActionInfo(jobId); 300 Services.get().get(XLogService.class).streamLog(filter, job.getCreatedTime(), new Date(), writer); 301 } 302 303 /** 304 * Add list of actions to the filter based on conditions 305 * 306 * @param jobId Job Id 307 * @param logRetrievalScope Value for the retrieval type 308 * @param logRetrievalType Based on which filter criteria the log is retrieved 309 * @param writer writer to stream the log to 310 * @throws IOException 311 * @throws BaseEngineException 312 * @throws CommandException 313 */ 314 public void streamLog(String jobId, String logRetrievalScope, String logRetrievalType, Writer writer) 315 throws IOException, BaseEngineException, CommandException { 316 XLogStreamer.Filter filter = new XLogStreamer.Filter(); 317 filter.setParameter(DagXLogInfoService.JOB, jobId); 318 if (logRetrievalScope != null && logRetrievalType != null) { 319 // if coordinator action logs are to be retrieved based on action id range 320 if (logRetrievalType.equals(RestConstants.JOB_LOG_ACTION)) { 321 Set<String> actions = new HashSet<String>(); 322 String[] list = logRetrievalScope.split(","); 323 for (String s : list) { 324 s = s.trim(); 325 if (s.contains("-")) { 326 String[] range = s.split("-"); 327 if (range.length != 2) { 328 throw new CommandException(ErrorCode.E0302, "format is wrong for action's range '" + s 329 + "'"); 330 } 331 int start; 332 int end; 333 try { 334 start = Integer.parseInt(range[0].trim()); 335 end = Integer.parseInt(range[1].trim()); 336 if (start > end) { 337 throw new CommandException(ErrorCode.E0302, "format is wrong for action's range '" + s 338 + "'"); 339 } 340 } 341 catch (NumberFormatException ne) { 342 throw new CommandException(ErrorCode.E0302, ne); 343 } 344 for (int i = start; i <= end; i++) { 345 actions.add(jobId + "@" + i); 346 } 347 } 348 else { 349 try { 350 Integer.parseInt(s); 351 } 352 catch (NumberFormatException ne) { 353 throw new CommandException(ErrorCode.E0302, "format is wrong for action id'" + s 354 + "'. Integer only."); 355 } 356 actions.add(jobId + "@" + s); 357 } 358 } 359 360 Iterator<String> actionsIterator = actions.iterator(); 361 StringBuilder orSeparatedActions = new StringBuilder(""); 362 boolean orRequired = false; 363 while (actionsIterator.hasNext()) { 364 if (orRequired) { 365 orSeparatedActions.append("|"); 366 } 367 orSeparatedActions.append(actionsIterator.next().toString()); 368 orRequired = true; 369 } 370 if (actions.size() > 1 && orRequired) { 371 orSeparatedActions.insert(0, "("); 372 orSeparatedActions.append(")"); 373 } 374 filter.setParameter(DagXLogInfoService.ACTION, orSeparatedActions.toString()); 375 } 376 // if coordinator action logs are to be retrieved based on date range 377 // this block gets the corresponding list of coordinator actions to be used by the log filter 378 if (logRetrievalType.equalsIgnoreCase(RestConstants.JOB_LOG_DATE)) { 379 List<CoordinatorActionBean> actionsList = null; 380 try { 381 actionsList = CoordActionsInDateRange.getCoordActionsFromDates(jobId, logRetrievalScope); 382 } 383 catch (XException xe) { 384 throw new CommandException(ErrorCode.E0302, "Error in date range for coordinator actions", xe); 385 } 386 StringBuilder orSeparatedActions = new StringBuilder(""); 387 boolean orRequired = false; 388 for (CoordinatorActionBean coordAction : actionsList) { 389 if (orRequired) { 390 orSeparatedActions.append("|"); 391 } 392 orSeparatedActions.append(coordAction.getId()); 393 orRequired = true; 394 } 395 if (actionsList.size() > 1 && orRequired) { 396 orSeparatedActions.insert(0, "("); 397 orSeparatedActions.append(")"); 398 } 399 filter.setParameter(DagXLogInfoService.ACTION, orSeparatedActions.toString()); 400 } 401 } 402 CoordinatorJobBean job = getCoordJobWithNoActionInfo(jobId); 403 Services.get().get(XLogService.class).streamLog(filter, job.getCreatedTime(), new Date(), writer); 404 } 405 406 /* 407 * (non-Javadoc) 408 * 409 * @see 410 * org.apache.oozie.BaseEngine#submitJob(org.apache.hadoop.conf.Configuration 411 * , boolean) 412 */ 413 @Override 414 public String submitJob(Configuration conf, boolean startJob) throws CoordinatorEngineException { 415 try { 416 String jobId; 417 if (useXCommand) { 418 CoordSubmitXCommand submit = new CoordSubmitXCommand(conf, getAuthToken()); 419 jobId = submit.call(); 420 } 421 else { 422 CoordSubmitCommand submit = new CoordSubmitCommand(conf, getAuthToken()); 423 jobId = submit.call(); 424 } 425 return jobId; 426 } 427 catch (CommandException ex) { 428 throw new CoordinatorEngineException(ex); 429 } 430 } 431 432 /* 433 * (non-Javadoc) 434 * 435 * @see 436 * org.apache.oozie.BaseEngine#dryrunSubmit(org.apache.hadoop.conf.Configuration 437 * , boolean) 438 */ 439 @Override 440 public String dryrunSubmit(Configuration conf, boolean startJob) throws CoordinatorEngineException { 441 try { 442 String jobId; 443 if (useXCommand) { 444 CoordSubmitXCommand submit = new CoordSubmitXCommand(true, conf, getAuthToken()); 445 jobId = submit.call(); 446 } 447 else { 448 CoordSubmitCommand submit = new CoordSubmitCommand(true, conf, getAuthToken()); 449 jobId = submit.call(); 450 } 451 return jobId; 452 } 453 catch (CommandException ex) { 454 throw new CoordinatorEngineException(ex); 455 } 456 } 457 458 /* 459 * (non-Javadoc) 460 * 461 * @see org.apache.oozie.BaseEngine#suspend(java.lang.String) 462 */ 463 @Override 464 public void suspend(String jobId) throws CoordinatorEngineException { 465 try { 466 if (useXCommand) { 467 new CoordSuspendXCommand(jobId).call(); 468 } 469 else { 470 new CoordSuspendCommand(jobId).call(); 471 } 472 } 473 catch (CommandException e) { 474 throw new CoordinatorEngineException(e); 475 } 476 477 } 478 479 /* 480 * (non-Javadoc) 481 * 482 * @see org.apache.oozie.BaseEngine#getJob(java.lang.String) 483 */ 484 @Override 485 public WorkflowJob getJob(String jobId) throws BaseEngineException { 486 throw new BaseEngineException(new XException(ErrorCode.E0301)); 487 } 488 489 /* 490 * (non-Javadoc) 491 * 492 * @see org.apache.oozie.BaseEngine#getJob(java.lang.String, int, int) 493 */ 494 @Override 495 public WorkflowJob getJob(String jobId, int start, int length) throws BaseEngineException { 496 throw new BaseEngineException(new XException(ErrorCode.E0301)); 497 } 498 499 private static final Set<String> FILTER_NAMES = new HashSet<String>(); 500 501 static { 502 FILTER_NAMES.add(OozieClient.FILTER_USER); 503 FILTER_NAMES.add(OozieClient.FILTER_NAME); 504 FILTER_NAMES.add(OozieClient.FILTER_GROUP); 505 FILTER_NAMES.add(OozieClient.FILTER_STATUS); 506 FILTER_NAMES.add(OozieClient.FILTER_ID); 507 FILTER_NAMES.add(OozieClient.FILTER_FREQUENCY); 508 } 509 510 /** 511 * @param filterStr 512 * @param start 513 * @param len 514 * @return CoordinatorJobInfo 515 * @throws CoordinatorEngineException 516 */ 517 public CoordinatorJobInfo getCoordJobs(String filterStr, int start, int len) throws CoordinatorEngineException { 518 Map<String, List<String>> filter = parseFilter(filterStr); 519 520 try { 521 if (useXCommand) { 522 return new CoordJobsXCommand(filter, start, len).call(); 523 } 524 else { 525 return new CoordJobsCommand(filter, start, len).call(); 526 } 527 } 528 catch (CommandException ex) { 529 throw new CoordinatorEngineException(ex); 530 } 531 } 532 533 /** 534 * @param filter 535 * @return Map<String, List<String>> 536 * @throws CoordinatorEngineException 537 */ 538 private Map<String, List<String>> parseFilter(String filter) throws CoordinatorEngineException { 539 Map<String, List<String>> map = new HashMap<String, List<String>>(); 540 if (filter != null) { 541 StringTokenizer st = new StringTokenizer(filter, ";"); 542 while (st.hasMoreTokens()) { 543 String token = st.nextToken(); 544 if (token.contains("=")) { 545 String[] pair = token.split("="); 546 if (pair.length != 2) { 547 throw new CoordinatorEngineException(ErrorCode.E0420, filter, 548 "elements must be name=value pairs"); 549 } 550 if (!FILTER_NAMES.contains(pair[0])) { 551 throw new CoordinatorEngineException(ErrorCode.E0420, filter, XLog.format("invalid name [{0}]", 552 pair[0])); 553 } 554 if (pair[0].equals("status")) { 555 try { 556 CoordinatorJob.Status.valueOf(pair[1]); 557 } 558 catch (IllegalArgumentException ex) { 559 throw new CoordinatorEngineException(ErrorCode.E0420, filter, XLog.format( 560 "invalid status [{0}]", pair[1])); 561 } 562 } 563 List<String> list = map.get(pair[0]); 564 if (list == null) { 565 list = new ArrayList<String>(); 566 map.put(pair[0], list); 567 } 568 list.add(pair[1]); 569 } 570 else { 571 throw new CoordinatorEngineException(ErrorCode.E0420, filter, "elements must be name=value pairs"); 572 } 573 } 574 } 575 return map; 576 } 577 }