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.client; 020 021import org.apache.oozie.BuildInfo; 022import org.apache.oozie.client.rest.JsonTags; 023import org.apache.oozie.client.rest.JsonToBean; 024import org.apache.oozie.client.rest.RestConstants; 025import org.apache.oozie.client.retry.ConnectionRetriableClient; 026import org.json.simple.JSONArray; 027import org.json.simple.JSONObject; 028import org.json.simple.JSONValue; 029import org.w3c.dom.Document; 030import org.w3c.dom.Element; 031 032import javax.xml.parsers.DocumentBuilderFactory; 033import javax.xml.transform.Transformer; 034import javax.xml.transform.TransformerFactory; 035import javax.xml.transform.dom.DOMSource; 036import javax.xml.transform.stream.StreamResult; 037import java.io.BufferedReader; 038import java.io.File; 039import java.io.FileInputStream; 040import java.io.IOException; 041import java.io.InputStream; 042import java.io.InputStreamReader; 043import java.io.OutputStream; 044import java.io.PrintStream; 045import java.io.Reader; 046import java.net.HttpURLConnection; 047import java.net.URL; 048import java.net.URLEncoder; 049import java.util.ArrayList; 050import java.util.Collections; 051import java.util.HashMap; 052import java.util.HashSet; 053import java.util.Iterator; 054import java.util.LinkedHashMap; 055import java.util.List; 056import java.util.Map; 057import java.util.Map.Entry; 058import java.util.Properties; 059import java.util.Set; 060import java.util.concurrent.Callable; 061 062/** 063 * Client API to submit and manage Oozie workflow jobs against an Oozie intance. 064 * <p/> 065 * This class is thread safe. 066 * <p/> 067 * Syntax for filter for the {@link #getJobsInfo(String)} {@link #getJobsInfo(String, int, int)} methods: 068 * <code>[NAME=VALUE][;NAME=VALUE]*</code>. 069 * <p/> 070 * Valid filter names are: 071 * <p/> 072 * <ul/> 073 * <li>name: the workflow application name from the workflow definition.</li> 074 * <li>user: the user that submitted the job.</li> 075 * <li>group: the group for the job.</li> 076 * <li>status: the status of the job.</li> 077 * </ul> 078 * <p/> 079 * The query will do an AND among all the filter names. The query will do an OR among all the filter values for the same 080 * name. Multiple values must be specified as different name value pairs. 081 */ 082public class OozieClient { 083 084 public static final long WS_PROTOCOL_VERSION_0 = 0; 085 086 public static final long WS_PROTOCOL_VERSION_1 = 1; 087 088 public static final long WS_PROTOCOL_VERSION = 2; // pointer to current version 089 090 public static final String USER_NAME = "user.name"; 091 092 @Deprecated 093 public static final String GROUP_NAME = "group.name"; 094 095 public static final String JOB_ACL = "oozie.job.acl"; 096 097 public static final String APP_PATH = "oozie.wf.application.path"; 098 099 public static final String COORDINATOR_APP_PATH = "oozie.coord.application.path"; 100 101 public static final String BUNDLE_APP_PATH = "oozie.bundle.application.path"; 102 103 public static final String BUNDLE_ID = "oozie.bundle.id"; 104 105 public static final String EXTERNAL_ID = "oozie.wf.external.id"; 106 107 public static final String WORKFLOW_NOTIFICATION_URL = "oozie.wf.workflow.notification.url"; 108 109 public static final String WORKFLOW_NOTIFICATION_PROXY = "oozie.wf.workflow.notification.proxy"; 110 111 public static final String ACTION_NOTIFICATION_URL = "oozie.wf.action.notification.url"; 112 113 public static final String COORD_ACTION_NOTIFICATION_URL = "oozie.coord.action.notification.url"; 114 115 public static final String COORD_ACTION_NOTIFICATION_PROXY = "oozie.coord.action.notification.proxy"; 116 117 public static final String RERUN_SKIP_NODES = "oozie.wf.rerun.skip.nodes"; 118 119 public static final String RERUN_FAIL_NODES = "oozie.wf.rerun.failnodes"; 120 121 public static final String LOG_TOKEN = "oozie.wf.log.token"; 122 123 public static final String ACTION_MAX_RETRIES = "oozie.wf.action.max.retries"; 124 125 public static final String ACTION_RETRY_INTERVAL = "oozie.wf.action.retry.interval"; 126 127 public static final String FILTER_USER = "user"; 128 129 public static final String FILTER_GROUP = "group"; 130 131 public static final String FILTER_NAME = "name"; 132 133 public static final String FILTER_STATUS = "status"; 134 135 public static final String FILTER_NOMINAL_TIME = "nominaltime"; 136 137 public static final String FILTER_FREQUENCY = "frequency"; 138 139 public static final String FILTER_ID = "id"; 140 141 public static final String FILTER_UNIT = "unit"; 142 143 public static final String FILTER_JOBID = "jobid"; 144 145 public static final String FILTER_APPNAME = "appname"; 146 147 public static final String FILTER_SLA_APPNAME = "app_name"; 148 149 public static final String FILTER_SLA_ID = "id"; 150 151 public static final String FILTER_SLA_PARENT_ID = "parent_id"; 152 153 public static final String FILTER_SLA_NOMINAL_START = "nominal_start"; 154 155 public static final String FILTER_SLA_NOMINAL_END = "nominal_end"; 156 157 public static final String FILTER_CREATED_TIME_START = "startcreatedtime"; 158 159 public static final String FILTER_CREATED_TIME_END = "endcreatedtime"; 160 161 public static final String CHANGE_VALUE_ENDTIME = "endtime"; 162 163 public static final String CHANGE_VALUE_PAUSETIME = "pausetime"; 164 165 public static final String CHANGE_VALUE_CONCURRENCY = "concurrency"; 166 167 public static final String CHANGE_VALUE_STATUS = "status"; 168 169 public static final String LIBPATH = "oozie.libpath"; 170 171 public static final String USE_SYSTEM_LIBPATH = "oozie.use.system.libpath"; 172 173 public static final String OOZIE_SUSPEND_ON_NODES = "oozie.suspend.on.nodes"; 174 175 public static enum SYSTEM_MODE { 176 NORMAL, NOWEBSERVICE, SAFEMODE 177 } 178 179 private static final Set<String> COMPLETED_WF_STATUSES = new HashSet<String>(); 180 private static final Set<String> COMPLETED_COORD_AND_BUNDLE_STATUSES = new HashSet<String>(); 181 private static final Set<String> COMPLETED_COORD_ACTION_STATUSES = new HashSet<String>(); 182 static { 183 COMPLETED_WF_STATUSES.add(WorkflowJob.Status.FAILED.toString()); 184 COMPLETED_WF_STATUSES.add(WorkflowJob.Status.KILLED.toString()); 185 COMPLETED_WF_STATUSES.add(WorkflowJob.Status.SUCCEEDED.toString()); 186 COMPLETED_COORD_AND_BUNDLE_STATUSES.add(Job.Status.FAILED.toString()); 187 COMPLETED_COORD_AND_BUNDLE_STATUSES.add(Job.Status.KILLED.toString()); 188 COMPLETED_COORD_AND_BUNDLE_STATUSES.add(Job.Status.SUCCEEDED.toString()); 189 COMPLETED_COORD_AND_BUNDLE_STATUSES.add(Job.Status.DONEWITHERROR.toString()); 190 COMPLETED_COORD_AND_BUNDLE_STATUSES.add(Job.Status.IGNORED.toString()); 191 COMPLETED_COORD_ACTION_STATUSES.add(CoordinatorAction.Status.FAILED.toString()); 192 COMPLETED_COORD_ACTION_STATUSES.add(CoordinatorAction.Status.IGNORED.toString()); 193 COMPLETED_COORD_ACTION_STATUSES.add(CoordinatorAction.Status.KILLED.toString()); 194 COMPLETED_COORD_ACTION_STATUSES.add(CoordinatorAction.Status.SKIPPED.toString()); 195 COMPLETED_COORD_ACTION_STATUSES.add(CoordinatorAction.Status.SUCCEEDED.toString()); 196 COMPLETED_COORD_ACTION_STATUSES.add(CoordinatorAction.Status.TIMEDOUT.toString()); 197 } 198 199 /** 200 * debugMode =0 means no debugging. > 0 means debugging on. 201 */ 202 public int debugMode = 0; 203 204 private int retryCount = 4; 205 206 207 private String baseUrl; 208 private String protocolUrl; 209 private boolean validatedVersion = false; 210 private JSONArray supportedVersions; 211 private final Map<String, String> headers = new HashMap<String, String>(); 212 213 private static final ThreadLocal<String> USER_NAME_TL = new ThreadLocal<String>(); 214 215 /** 216 * Allows to impersonate other users in the Oozie server. The current user 217 * must be configured as a proxyuser in Oozie. 218 * <p/> 219 * IMPORTANT: impersonation happens only with Oozie client requests done within 220 * doAs() calls. 221 * 222 * @param userName user to impersonate. 223 * @param callable callable with {@link OozieClient} calls impersonating the specified user. 224 * @return any response returned by the {@link Callable#call()} method. 225 * @throws Exception thrown by the {@link Callable#call()} method. 226 */ 227 public static <T> T doAs(String userName, Callable<T> callable) throws Exception { 228 notEmpty(userName, "userName"); 229 notNull(callable, "callable"); 230 try { 231 USER_NAME_TL.set(userName); 232 return callable.call(); 233 } 234 finally { 235 USER_NAME_TL.remove(); 236 } 237 } 238 239 protected OozieClient() { 240 } 241 242 /** 243 * Create a Workflow client instance. 244 * 245 * @param oozieUrl URL of the Oozie instance it will interact with. 246 */ 247 public OozieClient(String oozieUrl) { 248 this.baseUrl = notEmpty(oozieUrl, "oozieUrl"); 249 if (!this.baseUrl.endsWith("/")) { 250 this.baseUrl += "/"; 251 } 252 } 253 254 /** 255 * Return the Oozie URL of the workflow client instance. 256 * <p/> 257 * This URL is the base URL fo the Oozie system, with not protocol versioning. 258 * 259 * @return the Oozie URL of the workflow client instance. 260 */ 261 public String getOozieUrl() { 262 return baseUrl; 263 } 264 265 /** 266 * Return the Oozie URL used by the client and server for WS communications. 267 * <p/> 268 * This URL is the original URL plus the versioning element path. 269 * 270 * @return the Oozie URL used by the client and server for communication. 271 * @throws OozieClientException thrown in the client and the server are not protocol compatible. 272 */ 273 public String getProtocolUrl() throws OozieClientException { 274 validateWSVersion(); 275 return protocolUrl; 276 } 277 278 /** 279 * @return current debug Mode 280 */ 281 public int getDebugMode() { 282 return debugMode; 283 } 284 285 /** 286 * Set debug mode. 287 * 288 * @param debugMode : 0 means no debugging. > 0 means debugging 289 */ 290 public void setDebugMode(int debugMode) { 291 this.debugMode = debugMode; 292 } 293 294 public int getRetryCount() { 295 return retryCount; 296 } 297 298 299 public void setRetryCount(int retryCount) { 300 this.retryCount = retryCount; 301 } 302 303 private String getBaseURLForVersion(long protocolVersion) throws OozieClientException { 304 try { 305 if (supportedVersions == null) { 306 supportedVersions = getSupportedProtocolVersions(); 307 } 308 if (supportedVersions == null) { 309 throw new OozieClientException("HTTP error", "no response message"); 310 } 311 if (supportedVersions.contains(protocolVersion)) { 312 return baseUrl + "v" + protocolVersion + "/"; 313 } 314 else { 315 throw new OozieClientException(OozieClientException.UNSUPPORTED_VERSION, "Protocol version " 316 + protocolVersion + " is not supported"); 317 } 318 } 319 catch (IOException e) { 320 throw new OozieClientException(OozieClientException.IO_ERROR, e); 321 } 322 } 323 324 /** 325 * Validate that the Oozie client and server instances are protocol compatible. 326 * 327 * @throws OozieClientException thrown in the client and the server are not protocol compatible. 328 */ 329 public synchronized void validateWSVersion() throws OozieClientException { 330 if (!validatedVersion) { 331 try { 332 supportedVersions = getSupportedProtocolVersions(); 333 if (supportedVersions == null) { 334 throw new OozieClientException("HTTP error", "no response message"); 335 } 336 if (!supportedVersions.contains(WS_PROTOCOL_VERSION) 337 && !supportedVersions.contains(WS_PROTOCOL_VERSION_1) 338 && !supportedVersions.contains(WS_PROTOCOL_VERSION_0)) { 339 StringBuilder msg = new StringBuilder(); 340 msg.append("Supported version [").append(WS_PROTOCOL_VERSION) 341 .append("] or less, Unsupported versions["); 342 String separator = ""; 343 for (Object version : supportedVersions) { 344 msg.append(separator).append(version); 345 } 346 msg.append("]"); 347 throw new OozieClientException(OozieClientException.UNSUPPORTED_VERSION, msg.toString()); 348 } 349 if (supportedVersions.contains(WS_PROTOCOL_VERSION)) { 350 protocolUrl = baseUrl + "v" + WS_PROTOCOL_VERSION + "/"; 351 } 352 else if (supportedVersions.contains(WS_PROTOCOL_VERSION_1)) { 353 protocolUrl = baseUrl + "v" + WS_PROTOCOL_VERSION_1 + "/"; 354 } 355 else { 356 if (supportedVersions.contains(WS_PROTOCOL_VERSION_0)) { 357 protocolUrl = baseUrl + "v" + WS_PROTOCOL_VERSION_0 + "/"; 358 } 359 } 360 } 361 catch (IOException ex) { 362 throw new OozieClientException(OozieClientException.IO_ERROR, ex); 363 } 364 validatedVersion = true; 365 } 366 } 367 368 private JSONArray getSupportedProtocolVersions() throws IOException, OozieClientException { 369 JSONArray versions = null; 370 final URL url = new URL(baseUrl + RestConstants.VERSIONS); 371 372 HttpURLConnection conn = createRetryableConnection(url, "GET"); 373 374 if (conn.getResponseCode() == HttpURLConnection.HTTP_OK) { 375 versions = (JSONArray) JSONValue.parse(new InputStreamReader(conn.getInputStream())); 376 } 377 else { 378 handleError(conn); 379 } 380 return versions; 381 } 382 383 /** 384 * Create an empty configuration with just the {@link #USER_NAME} set to the JVM user name. 385 * 386 * @return an empty configuration. 387 */ 388 public Properties createConfiguration() { 389 Properties conf = new Properties(); 390 String userName = USER_NAME_TL.get(); 391 if (userName == null) { 392 userName = System.getProperty("user.name"); 393 } 394 conf.setProperty(USER_NAME, userName); 395 return conf; 396 } 397 398 /** 399 * Set a HTTP header to be used in the WS requests by the workflow instance. 400 * 401 * @param name header name. 402 * @param value header value. 403 */ 404 public void setHeader(String name, String value) { 405 headers.put(notEmpty(name, "name"), notNull(value, "value")); 406 } 407 408 /** 409 * Get the value of a set HTTP header from the workflow instance. 410 * 411 * @param name header name. 412 * @return header value, <code>null</code> if not set. 413 */ 414 public String getHeader(String name) { 415 return headers.get(notEmpty(name, "name")); 416 } 417 418 /** 419 * Get the set HTTP header 420 * 421 * @return map of header key and value 422 */ 423 public Map<String, String> getHeaders() { 424 return headers; 425 } 426 427 /** 428 * Remove a HTTP header from the workflow client instance. 429 * 430 * @param name header name. 431 */ 432 public void removeHeader(String name) { 433 headers.remove(notEmpty(name, "name")); 434 } 435 436 /** 437 * Return an iterator with all the header names set in the workflow instance. 438 * 439 * @return header names. 440 */ 441 public Iterator<String> getHeaderNames() { 442 return Collections.unmodifiableMap(headers).keySet().iterator(); 443 } 444 445 private URL createURL(Long protocolVersion, String collection, String resource, Map<String, String> parameters) 446 throws IOException, OozieClientException { 447 validateWSVersion(); 448 StringBuilder sb = new StringBuilder(); 449 if (protocolVersion == null) { 450 sb.append(protocolUrl); 451 } 452 else { 453 sb.append(getBaseURLForVersion(protocolVersion)); 454 } 455 sb.append(collection); 456 if (resource != null && resource.length() > 0) { 457 sb.append("/").append(resource); 458 } 459 if (parameters.size() > 0) { 460 String separator = "?"; 461 for (Map.Entry<String, String> param : parameters.entrySet()) { 462 if (param.getValue() != null) { 463 sb.append(separator).append(URLEncoder.encode(param.getKey(), "UTF-8")).append("=").append( 464 URLEncoder.encode(param.getValue(), "UTF-8")); 465 separator = "&"; 466 } 467 } 468 } 469 return new URL(sb.toString()); 470 } 471 472 private boolean validateCommand(String url) { 473 { 474 if (protocolUrl.contains(baseUrl + "v0")) { 475 if (url.contains("dryrun") || url.contains("jobtype=c") || url.contains("systemmode")) { 476 return false; 477 } 478 } 479 } 480 return true; 481 } 482 /** 483 * Create retryable http connection to oozie server. 484 * 485 * @param url 486 * @param method 487 * @return connection 488 * @throws IOException 489 * @throws OozieClientException 490 */ 491 protected HttpURLConnection createRetryableConnection(final URL url, final String method) throws IOException{ 492 return (HttpURLConnection) new ConnectionRetriableClient(getRetryCount()) { 493 @Override 494 public Object doExecute(URL url, String method) throws IOException, OozieClientException { 495 HttpURLConnection conn = createConnection(url, method); 496 return conn; 497 } 498 }.execute(url, method); 499 } 500 501 /** 502 * Create http connection to oozie server. 503 * 504 * @param url 505 * @param method 506 * @return connection 507 * @throws IOException 508 * @throws OozieClientException 509 */ 510 protected HttpURLConnection createConnection(URL url, String method) throws IOException, OozieClientException { 511 HttpURLConnection conn = (HttpURLConnection) url.openConnection(); 512 conn.setRequestMethod(method); 513 if (method.equals("POST") || method.equals("PUT")) { 514 conn.setDoOutput(true); 515 } 516 for (Map.Entry<String, String> header : headers.entrySet()) { 517 conn.setRequestProperty(header.getKey(), header.getValue()); 518 } 519 return conn; 520 } 521 522 protected abstract class ClientCallable<T> implements Callable<T> { 523 private final String method; 524 private final String collection; 525 private final String resource; 526 private final Map<String, String> params; 527 private final Long protocolVersion; 528 529 public ClientCallable(String method, String collection, String resource, Map<String, String> params) { 530 this(method, null, collection, resource, params); 531 } 532 533 public ClientCallable(String method, Long protocolVersion, String collection, String resource, Map<String, String> params) { 534 this.method = method; 535 this.protocolVersion = protocolVersion; 536 this.collection = collection; 537 this.resource = resource; 538 this.params = params; 539 } 540 541 public T call() throws OozieClientException { 542 try { 543 URL url = createURL(protocolVersion, collection, resource, params); 544 if (validateCommand(url.toString())) { 545 if (getDebugMode() > 0) { 546 System.out.println(method + " " + url); 547 } 548 return call(createRetryableConnection(url, method)); 549 } 550 else { 551 System.out.println("Option not supported in target server. Supported only on Oozie-2.0 or greater." 552 + " Use 'oozie help' for details"); 553 throw new OozieClientException(OozieClientException.UNSUPPORTED_VERSION, new Exception()); 554 } 555 } 556 catch (IOException ex) { 557 throw new OozieClientException(OozieClientException.IO_ERROR, ex); 558 } 559 } 560 561 protected abstract T call(HttpURLConnection conn) throws IOException, OozieClientException; 562 } 563 564 protected abstract class MapClientCallable extends ClientCallable<Map<String, String>> { 565 566 MapClientCallable(String method, String collection, String resource, Map<String, String> params) { 567 super(method, collection, resource, params); 568 } 569 570 @Override 571 protected Map<String, String> call(HttpURLConnection conn) throws IOException, OozieClientException { 572 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 573 Reader reader = new InputStreamReader(conn.getInputStream()); 574 JSONObject json = (JSONObject) JSONValue.parse(reader); 575 Map<String, String> map = new HashMap<String, String>(); 576 for (Object key : json.keySet()) { 577 map.put((String)key, (String)json.get(key)); 578 } 579 return map; 580 } 581 else { 582 handleError(conn); 583 } 584 return null; 585 } 586 } 587 588 static void handleError(HttpURLConnection conn) throws IOException, OozieClientException { 589 int status = conn.getResponseCode(); 590 String error = conn.getHeaderField(RestConstants.OOZIE_ERROR_CODE); 591 String message = conn.getHeaderField(RestConstants.OOZIE_ERROR_MESSAGE); 592 593 if (error == null) { 594 error = "HTTP error code: " + status; 595 } 596 597 if (message == null) { 598 message = conn.getResponseMessage(); 599 } 600 throw new OozieClientException(error, message); 601 } 602 603 static Map<String, String> prepareParams(String... params) { 604 Map<String, String> map = new LinkedHashMap<String, String>(); 605 for (int i = 0; i < params.length; i = i + 2) { 606 map.put(params[i], params[i + 1]); 607 } 608 String doAsUserName = USER_NAME_TL.get(); 609 if (doAsUserName != null) { 610 map.put(RestConstants.DO_AS_PARAM, doAsUserName); 611 } 612 return map; 613 } 614 615 public void writeToXml(Properties props, OutputStream out) throws IOException { 616 try { 617 Document doc = DocumentBuilderFactory.newInstance().newDocumentBuilder().newDocument(); 618 Element conf = doc.createElement("configuration"); 619 doc.appendChild(conf); 620 conf.appendChild(doc.createTextNode("\n")); 621 for (String name : props.stringPropertyNames()) { // Properties whose key or value is not of type String are omitted. 622 String value = props.getProperty(name); 623 Element propNode = doc.createElement("property"); 624 conf.appendChild(propNode); 625 626 Element nameNode = doc.createElement("name"); 627 nameNode.appendChild(doc.createTextNode(name.trim())); 628 propNode.appendChild(nameNode); 629 630 Element valueNode = doc.createElement("value"); 631 valueNode.appendChild(doc.createTextNode(value.trim())); 632 propNode.appendChild(valueNode); 633 634 conf.appendChild(doc.createTextNode("\n")); 635 } 636 637 DOMSource source = new DOMSource(doc); 638 StreamResult result = new StreamResult(out); 639 TransformerFactory transFactory = TransformerFactory.newInstance(); 640 Transformer transformer = transFactory.newTransformer(); 641 transformer.transform(source, result); 642 if (getDebugMode() > 0) { 643 result = new StreamResult(System.out); 644 transformer.transform(source, result); 645 System.out.println(); 646 } 647 } 648 catch (Exception e) { 649 throw new IOException(e); 650 } 651 } 652 653 private class JobSubmit extends ClientCallable<String> { 654 private final Properties conf; 655 656 JobSubmit(Properties conf, boolean start) { 657 super("POST", RestConstants.JOBS, "", (start) ? prepareParams(RestConstants.ACTION_PARAM, 658 RestConstants.JOB_ACTION_START) : prepareParams()); 659 this.conf = notNull(conf, "conf"); 660 } 661 662 JobSubmit(String jobId, Properties conf) { 663 super("PUT", RestConstants.JOB, notEmpty(jobId, "jobId"), prepareParams(RestConstants.ACTION_PARAM, 664 RestConstants.JOB_ACTION_RERUN)); 665 this.conf = notNull(conf, "conf"); 666 } 667 668 public JobSubmit(Properties conf, String jobActionDryrun) { 669 super("POST", RestConstants.JOBS, "", prepareParams(RestConstants.ACTION_PARAM, 670 RestConstants.JOB_ACTION_DRYRUN)); 671 this.conf = notNull(conf, "conf"); 672 } 673 674 @Override 675 protected String call(HttpURLConnection conn) throws IOException, OozieClientException { 676 conn.setRequestProperty("content-type", RestConstants.XML_CONTENT_TYPE); 677 writeToXml(conf, conn.getOutputStream()); 678 if (conn.getResponseCode() == HttpURLConnection.HTTP_CREATED) { 679 JSONObject json = (JSONObject) JSONValue.parse(new InputStreamReader(conn.getInputStream())); 680 return (String) json.get(JsonTags.JOB_ID); 681 } 682 if (conn.getResponseCode() != HttpURLConnection.HTTP_OK) { 683 handleError(conn); 684 } 685 return null; 686 } 687 } 688 689 /** 690 * Submit a workflow job. 691 * 692 * @param conf job configuration. 693 * @return the job Id. 694 * @throws OozieClientException thrown if the job could not be submitted. 695 */ 696 public String submit(Properties conf) throws OozieClientException { 697 return (new JobSubmit(conf, false)).call(); 698 } 699 700 private class JobAction extends ClientCallable<Void> { 701 702 JobAction(String jobId, String action) { 703 super("PUT", RestConstants.JOB, notEmpty(jobId, "jobId"), prepareParams(RestConstants.ACTION_PARAM, action)); 704 } 705 706 JobAction(String jobId, String action, String params) { 707 super("PUT", RestConstants.JOB, notEmpty(jobId, "jobId"), prepareParams(RestConstants.ACTION_PARAM, action, 708 RestConstants.JOB_CHANGE_VALUE, params)); 709 } 710 711 @Override 712 protected Void call(HttpURLConnection conn) throws IOException, OozieClientException { 713 if (!(conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 714 handleError(conn); 715 } 716 return null; 717 } 718 } 719 720 private class JobsAction extends ClientCallable<JSONObject> { 721 722 JobsAction(String action, String filter, String jobType, int start, int len) { 723 super("PUT", RestConstants.JOBS, "", 724 prepareParams(RestConstants.ACTION_PARAM, action, 725 RestConstants.JOB_FILTER_PARAM, filter, RestConstants.JOBTYPE_PARAM, jobType, 726 RestConstants.OFFSET_PARAM, Integer.toString(start), 727 RestConstants.LEN_PARAM, Integer.toString(len))); 728 } 729 730 @Override 731 protected JSONObject call(HttpURLConnection conn) throws IOException, OozieClientException { 732 conn.setRequestProperty("content-type", RestConstants.XML_CONTENT_TYPE); 733 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 734 Reader reader = new InputStreamReader(conn.getInputStream()); 735 JSONObject json = (JSONObject) JSONValue.parse(reader); 736 return json; 737 } 738 else { 739 handleError(conn); 740 } 741 return null; 742 } 743 } 744 /** 745 * Update coord definition. 746 * 747 * @param jobId the job id 748 * @param conf the conf 749 * @param dryrun the dryrun 750 * @param showDiff the show diff 751 * @return the string 752 * @throws OozieClientException the oozie client exception 753 */ 754 public String updateCoord(String jobId, Properties conf, String dryrun, String showDiff) 755 throws OozieClientException { 756 return (new UpdateCoord(jobId, conf, dryrun, showDiff)).call(); 757 } 758 759 /** 760 * Update coord definition without properties. 761 * 762 * @param jobId the job id 763 * @param dryrun the dryrun 764 * @param showDiff the show diff 765 * @return the string 766 * @throws OozieClientException the oozie client exception 767 */ 768 public String updateCoord(String jobId, String dryrun, String showDiff) throws OozieClientException { 769 return (new UpdateCoord(jobId, dryrun, showDiff)).call(); 770 } 771 772 /** 773 * The Class UpdateCoord. 774 */ 775 private class UpdateCoord extends ClientCallable<String> { 776 private final Properties conf; 777 778 public UpdateCoord(String jobId, Properties conf, String jobActionDryrun, String showDiff) { 779 super("PUT", RestConstants.JOB, notEmpty(jobId, "jobId"), prepareParams(RestConstants.ACTION_PARAM, 780 RestConstants.JOB_COORD_UPDATE, RestConstants.JOB_ACTION_DRYRUN, jobActionDryrun, 781 RestConstants.JOB_ACTION_SHOWDIFF, showDiff)); 782 this.conf = conf; 783 } 784 785 public UpdateCoord(String jobId, String jobActionDryrun, String showDiff) { 786 this(jobId, new Properties(), jobActionDryrun, showDiff); 787 } 788 789 @Override 790 protected String call(HttpURLConnection conn) throws IOException, OozieClientException { 791 conn.setRequestProperty("content-type", RestConstants.XML_CONTENT_TYPE); 792 writeToXml(conf, conn.getOutputStream()); 793 794 if (conn.getResponseCode() == HttpURLConnection.HTTP_OK) { 795 JSONObject json = (JSONObject) JSONValue.parse(new InputStreamReader(conn.getInputStream())); 796 JSONObject update = (JSONObject) json.get(JsonTags.COORD_UPDATE); 797 if (update != null) { 798 return (String) update.get(JsonTags.COORD_UPDATE_DIFF); 799 } 800 else { 801 return ""; 802 } 803 } 804 if (conn.getResponseCode() != HttpURLConnection.HTTP_OK) { 805 handleError(conn); 806 } 807 return null; 808 } 809 } 810 811 /** 812 * dryrun for a given job 813 * 814 * @param conf Job configuration. 815 */ 816 public String dryrun(Properties conf) throws OozieClientException { 817 return new JobSubmit(conf, RestConstants.JOB_ACTION_DRYRUN).call(); 818 } 819 820 /** 821 * Start a workflow job. 822 * 823 * @param jobId job Id. 824 * @throws OozieClientException thrown if the job could not be started. 825 */ 826 public void start(String jobId) throws OozieClientException { 827 new JobAction(jobId, RestConstants.JOB_ACTION_START).call(); 828 } 829 830 /** 831 * Submit and start a workflow job. 832 * 833 * @param conf job configuration. 834 * @return the job Id. 835 * @throws OozieClientException thrown if the job could not be submitted. 836 */ 837 public String run(Properties conf) throws OozieClientException { 838 return (new JobSubmit(conf, true)).call(); 839 } 840 841 /** 842 * Rerun a workflow job. 843 * 844 * @param jobId job Id to rerun. 845 * @param conf configuration information for the rerun. 846 * @throws OozieClientException thrown if the job could not be started. 847 */ 848 public void reRun(String jobId, Properties conf) throws OozieClientException { 849 new JobSubmit(jobId, conf).call(); 850 } 851 852 /** 853 * Suspend a workflow job. 854 * 855 * @param jobId job Id. 856 * @throws OozieClientException thrown if the job could not be suspended. 857 */ 858 public void suspend(String jobId) throws OozieClientException { 859 new JobAction(jobId, RestConstants.JOB_ACTION_SUSPEND).call(); 860 } 861 862 /** 863 * Resume a workflow job. 864 * 865 * @param jobId job Id. 866 * @throws OozieClientException thrown if the job could not be resume. 867 */ 868 public void resume(String jobId) throws OozieClientException { 869 new JobAction(jobId, RestConstants.JOB_ACTION_RESUME).call(); 870 } 871 872 /** 873 * Kill a workflow/coord/bundle job. 874 * 875 * @param jobId job Id. 876 * @throws OozieClientException thrown if the job could not be killed. 877 */ 878 public void kill(String jobId) throws OozieClientException { 879 new JobAction(jobId, RestConstants.JOB_ACTION_KILL).call(); 880 } 881 882 /** 883 * Kill coordinator actions 884 * @param jobId coordinator Job Id 885 * @param rangeType type 'date' if -date is used, 'action-num' if -action is used 886 * @param scope kill scope for date or action nums 887 * @return list of coordinator actions that underwent kill 888 * @throws OozieClientException thrown if some actions could not be killed. 889 */ 890 public List<CoordinatorAction> kill(String jobId, String rangeType, String scope) throws OozieClientException { 891 return new CoordActionsKill(jobId, rangeType, scope).call(); 892 } 893 894 public JSONObject bulkModifyJobs(String actionType, String filter, String jobType, int start, int len) 895 throws OozieClientException { 896 return new JobsAction(actionType, filter, jobType, start, len).call(); 897 } 898 899 public JSONObject killJobs(String filter, String jobType, int start, int len) 900 throws OozieClientException { 901 return bulkModifyJobs("kill", filter, jobType, start, len); 902 } 903 904 public JSONObject suspendJobs(String filter, String jobType, int start, int len) 905 throws OozieClientException { 906 return bulkModifyJobs("suspend", filter, jobType, start, len); 907 } 908 909 public JSONObject resumeJobs(String filter, String jobType, int start, int len) 910 throws OozieClientException { 911 return bulkModifyJobs("resume", filter, jobType, start, len); 912 } 913 /** 914 * Change a coordinator job. 915 * 916 * @param jobId job Id. 917 * @param changeValue change value. 918 * @throws OozieClientException thrown if the job could not be changed. 919 */ 920 public void change(String jobId, String changeValue) throws OozieClientException { 921 new JobAction(jobId, RestConstants.JOB_ACTION_CHANGE, changeValue).call(); 922 } 923 924 /** 925 * Ignore a coordinator job. 926 * 927 * @param jobId coord job Id. 928 * @param scope list of coord actions to be ignored 929 * @throws OozieClientException thrown if the job could not be changed. 930 */ 931 public List<CoordinatorAction> ignore(String jobId, String scope) throws OozieClientException { 932 return new CoordIgnore(jobId, RestConstants.JOB_COORD_SCOPE_ACTION, scope).call(); 933 } 934 935 private class JobInfo extends ClientCallable<WorkflowJob> { 936 937 JobInfo(String jobId, int start, int len) { 938 super("GET", RestConstants.JOB, notEmpty(jobId, "jobId"), prepareParams(RestConstants.JOB_SHOW_PARAM, 939 RestConstants.JOB_SHOW_INFO, RestConstants.OFFSET_PARAM, Integer.toString(start), 940 RestConstants.LEN_PARAM, Integer.toString(len))); 941 } 942 943 @Override 944 protected WorkflowJob call(HttpURLConnection conn) throws IOException, OozieClientException { 945 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 946 Reader reader = new InputStreamReader(conn.getInputStream()); 947 JSONObject json = (JSONObject) JSONValue.parse(reader); 948 return JsonToBean.createWorkflowJob(json); 949 } 950 else { 951 handleError(conn); 952 } 953 return null; 954 } 955 } 956 957 private class JMSInfo extends ClientCallable<JMSConnectionInfo> { 958 959 JMSInfo() { 960 super("GET", RestConstants.ADMIN, RestConstants.ADMIN_JMS_INFO, prepareParams()); 961 } 962 963 protected JMSConnectionInfo call(HttpURLConnection conn) throws IOException, OozieClientException { 964 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 965 Reader reader = new InputStreamReader(conn.getInputStream()); 966 JSONObject json = (JSONObject) JSONValue.parse(reader); 967 return JsonToBean.createJMSConnectionInfo(json); 968 } 969 else { 970 handleError(conn); 971 } 972 return null; 973 } 974 } 975 976 private class WorkflowActionInfo extends ClientCallable<WorkflowAction> { 977 WorkflowActionInfo(String actionId) { 978 super("GET", RestConstants.JOB, notEmpty(actionId, "id"), prepareParams(RestConstants.JOB_SHOW_PARAM, 979 RestConstants.JOB_SHOW_INFO)); 980 } 981 982 @Override 983 protected WorkflowAction call(HttpURLConnection conn) throws IOException, OozieClientException { 984 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 985 Reader reader = new InputStreamReader(conn.getInputStream()); 986 JSONObject json = (JSONObject) JSONValue.parse(reader); 987 return JsonToBean.createWorkflowAction(json); 988 } 989 else { 990 handleError(conn); 991 } 992 return null; 993 } 994 } 995 996 /** 997 * Get the info of a workflow job. 998 * 999 * @param jobId job Id. 1000 * @return the job info. 1001 * @throws OozieClientException thrown if the job info could not be retrieved. 1002 */ 1003 public WorkflowJob getJobInfo(String jobId) throws OozieClientException { 1004 return getJobInfo(jobId, 0, 0); 1005 } 1006 1007 /** 1008 * Get the JMS Connection info 1009 * @return JMSConnectionInfo object 1010 * @throws OozieClientException 1011 */ 1012 public JMSConnectionInfo getJMSConnectionInfo() throws OozieClientException { 1013 return new JMSInfo().call(); 1014 } 1015 1016 /** 1017 * Get the info of a workflow job and subset actions. 1018 * 1019 * @param jobId job Id. 1020 * @param start starting index in the list of actions belonging to the job 1021 * @param len number of actions to be returned 1022 * @return the job info. 1023 * @throws OozieClientException thrown if the job info could not be retrieved. 1024 */ 1025 public WorkflowJob getJobInfo(String jobId, int start, int len) throws OozieClientException { 1026 return new JobInfo(jobId, start, len).call(); 1027 } 1028 1029 /** 1030 * Get the info of a workflow action. 1031 * 1032 * @param actionId Id. 1033 * @return the workflow action info. 1034 * @throws OozieClientException thrown if the job info could not be retrieved. 1035 */ 1036 public WorkflowAction getWorkflowActionInfo(String actionId) throws OozieClientException { 1037 return new WorkflowActionInfo(actionId).call(); 1038 } 1039 1040 /** 1041 * Get the log of a workflow job. 1042 * 1043 * @param jobId job Id. 1044 * @return the job log. 1045 * @throws OozieClientException thrown if the job info could not be retrieved. 1046 */ 1047 public String getJobLog(String jobId) throws OozieClientException { 1048 return new JobLog(jobId).call(); 1049 } 1050 1051 /** 1052 * Get the log of a job. 1053 * 1054 * @param jobId job Id. 1055 * @param logRetrievalType Based on which filter criteria the log is retrieved 1056 * @param logRetrievalScope Value for the retrieval type 1057 * @param logFilter log filter 1058 * @param ps Printstream of command line interface 1059 * @throws OozieClientException thrown if the job info could not be retrieved. 1060 */ 1061 public void getJobLog(String jobId, String logRetrievalType, String logRetrievalScope, String logFilter, 1062 PrintStream ps) throws OozieClientException { 1063 new JobLog(jobId, logRetrievalType, logRetrievalScope, logFilter, ps).call(); 1064 } 1065 1066 /** 1067 * Get the log of a job. 1068 * 1069 * @param jobId job Id. 1070 * @param logRetrievalType Based on which filter criteria the log is retrieved 1071 * @param logRetrievalScope Value for the retrieval type 1072 * @param ps Printstream of command line interface 1073 * @throws OozieClientException thrown if the job info could not be retrieved. 1074 */ 1075 public void getJobLog(String jobId, String logRetrievalType, String logRetrievalScope, PrintStream ps) 1076 throws OozieClientException { 1077 getJobLog(jobId, logRetrievalType, logRetrievalScope, null, ps); 1078 } 1079 1080 private class JobLog extends JobMetadata { 1081 JobLog(String jobId) { 1082 super(jobId, RestConstants.JOB_SHOW_LOG); 1083 } 1084 JobLog(String jobId, String logRetrievalType, String logRetrievalScope, String logFilter, PrintStream ps) { 1085 super(jobId, logRetrievalType, logRetrievalScope, RestConstants.JOB_SHOW_LOG, logFilter, ps); 1086 } 1087 } 1088 1089 /** 1090 * Gets the JMS topic name for a particular job 1091 * @param jobId given jobId 1092 * @return the JMS topic name 1093 * @throws OozieClientException 1094 */ 1095 public String getJMSTopicName(String jobId) throws OozieClientException { 1096 return new JMSTopic(jobId).call(); 1097 } 1098 1099 private class JMSTopic extends ClientCallable<String> { 1100 1101 JMSTopic(String jobId) { 1102 super("GET", RestConstants.JOB, notEmpty(jobId, "jobId"), prepareParams(RestConstants.JOB_SHOW_PARAM, 1103 RestConstants.JOB_SHOW_JMS_TOPIC)); 1104 } 1105 1106 protected String call(HttpURLConnection conn) throws IOException, OozieClientException { 1107 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1108 Reader reader = new InputStreamReader(conn.getInputStream()); 1109 JSONObject json = (JSONObject) JSONValue.parse(reader); 1110 return (String) json.get(JsonTags.JMS_TOPIC_NAME); 1111 } 1112 else { 1113 handleError(conn); 1114 } 1115 return null; 1116 } 1117 } 1118 1119 /** 1120 * Get the definition of a workflow job. 1121 * 1122 * @param jobId job Id. 1123 * @return the job log. 1124 * @throws OozieClientException thrown if the job info could not be retrieved. 1125 */ 1126 public String getJobDefinition(String jobId) throws OozieClientException { 1127 return new JobDefinition(jobId).call(); 1128 } 1129 1130 private class JobDefinition extends JobMetadata { 1131 1132 JobDefinition(String jobId) { 1133 super(jobId, RestConstants.JOB_SHOW_DEFINITION); 1134 } 1135 } 1136 1137 private class JobMetadata extends ClientCallable<String> { 1138 PrintStream printStream; 1139 1140 JobMetadata(String jobId, String metaType) { 1141 super("GET", RestConstants.JOB, notEmpty(jobId, "jobId"), prepareParams(RestConstants.JOB_SHOW_PARAM, 1142 metaType)); 1143 } 1144 1145 JobMetadata(String jobId, String logRetrievalType, String logRetrievalScope, String metaType, String logFilter, 1146 PrintStream ps) { 1147 super("GET", RestConstants.JOB, notEmpty(jobId, "jobId"), prepareParams(RestConstants.JOB_SHOW_PARAM, 1148 metaType, RestConstants.JOB_LOG_TYPE_PARAM, logRetrievalType, RestConstants.JOB_LOG_SCOPE_PARAM, 1149 logRetrievalScope, RestConstants.LOG_FILTER_OPTION, logFilter)); 1150 printStream = ps; 1151 } 1152 1153 @Override 1154 protected String call(HttpURLConnection conn) throws IOException, OozieClientException { 1155 String returnVal = null; 1156 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1157 InputStream is = conn.getInputStream(); 1158 InputStreamReader isr = new InputStreamReader(is); 1159 try { 1160 if (printStream != null) { 1161 sendToOutputStream(isr, -1); 1162 } 1163 else { 1164 returnVal = getReaderAsString(isr, -1); 1165 } 1166 } 1167 finally { 1168 isr.close(); 1169 } 1170 } 1171 else { 1172 handleError(conn); 1173 } 1174 return returnVal; 1175 } 1176 1177 /** 1178 * Output the log to command line interface 1179 * 1180 * @param reader reader to read into a string. 1181 * @param maxLen max content length allowed, if -1 there is no limit. 1182 * @throws IOException 1183 */ 1184 private void sendToOutputStream(Reader reader, int maxLen) throws IOException { 1185 if (reader == null) { 1186 throw new IllegalArgumentException("reader cannot be null"); 1187 } 1188 StringBuilder sb = new StringBuilder(); 1189 char[] buffer = new char[2048]; 1190 int read; 1191 int count = 0; 1192 int noOfCharstoFlush = 1024; 1193 while ((read = reader.read(buffer)) > -1) { 1194 count += read; 1195 if ((maxLen > -1) && (count > maxLen)) { 1196 break; 1197 } 1198 sb.append(buffer, 0, read); 1199 if (sb.length() > noOfCharstoFlush) { 1200 printStream.print(sb.toString()); 1201 sb = new StringBuilder(""); 1202 } 1203 } 1204 printStream.print(sb.toString()); 1205 } 1206 1207 /** 1208 * Return a reader as string. 1209 * <p/> 1210 * 1211 * @param reader reader to read into a string. 1212 * @param maxLen max content length allowed, if -1 there is no limit. 1213 * @return the reader content. 1214 * @throws IOException thrown if the resource could not be read. 1215 */ 1216 private String getReaderAsString(Reader reader, int maxLen) throws IOException { 1217 if (reader == null) { 1218 throw new IllegalArgumentException("reader cannot be null"); 1219 } 1220 StringBuffer sb = new StringBuffer(); 1221 char[] buffer = new char[2048]; 1222 int read; 1223 int count = 0; 1224 while ((read = reader.read(buffer)) > -1) { 1225 count += read; 1226 1227 // read up to maxLen chars; 1228 if ((maxLen > -1) && (count > maxLen)) { 1229 break; 1230 } 1231 sb.append(buffer, 0, read); 1232 } 1233 reader.close(); 1234 return sb.toString(); 1235 } 1236 } 1237 1238 private class CoordJobInfo extends ClientCallable<CoordinatorJob> { 1239 1240 CoordJobInfo(String jobId, String filter, int start, int len, String order) { 1241 super("GET", RestConstants.JOB, notEmpty(jobId, "jobId"), prepareParams(RestConstants.JOB_SHOW_PARAM, 1242 RestConstants.JOB_SHOW_INFO, RestConstants.JOB_FILTER_PARAM, filter, RestConstants.OFFSET_PARAM, 1243 Integer.toString(start), RestConstants.LEN_PARAM, Integer.toString(len), RestConstants.ORDER_PARAM, 1244 order)); 1245 } 1246 1247 @Override 1248 protected CoordinatorJob call(HttpURLConnection conn) throws IOException, OozieClientException { 1249 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1250 Reader reader = new InputStreamReader(conn.getInputStream()); 1251 JSONObject json = (JSONObject) JSONValue.parse(reader); 1252 return JsonToBean.createCoordinatorJob(json); 1253 } 1254 else { 1255 handleError(conn); 1256 } 1257 return null; 1258 } 1259 } 1260 1261 private class WfsForCoordAction extends ClientCallable<List<WorkflowJob>> { 1262 1263 WfsForCoordAction(String coordActionId) { 1264 super("GET", RestConstants.JOB, notEmpty(coordActionId, "coordActionId"), prepareParams( 1265 RestConstants.JOB_SHOW_PARAM, RestConstants.ALL_WORKFLOWS_FOR_COORD_ACTION)); 1266 } 1267 1268 @Override 1269 protected List<WorkflowJob> call(HttpURLConnection conn) throws IOException, OozieClientException { 1270 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1271 Reader reader = new InputStreamReader(conn.getInputStream()); 1272 JSONObject json = (JSONObject) JSONValue.parse(reader); 1273 JSONArray workflows = (JSONArray) json.get(JsonTags.WORKFLOWS_JOBS); 1274 if (workflows == null) { 1275 workflows = new JSONArray(); 1276 } 1277 return JsonToBean.createWorkflowJobList(workflows); 1278 } 1279 else { 1280 handleError(conn); 1281 } 1282 return null; 1283 } 1284 } 1285 1286 1287 private class BundleJobInfo extends ClientCallable<BundleJob> { 1288 1289 BundleJobInfo(String jobId) { 1290 super("GET", RestConstants.JOB, notEmpty(jobId, "jobId"), prepareParams(RestConstants.JOB_SHOW_PARAM, 1291 RestConstants.JOB_SHOW_INFO)); 1292 } 1293 1294 @Override 1295 protected BundleJob call(HttpURLConnection conn) throws IOException, OozieClientException { 1296 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1297 Reader reader = new InputStreamReader(conn.getInputStream()); 1298 JSONObject json = (JSONObject) JSONValue.parse(reader); 1299 return JsonToBean.createBundleJob(json); 1300 } 1301 else { 1302 handleError(conn); 1303 } 1304 return null; 1305 } 1306 } 1307 1308 private class CoordActionInfo extends ClientCallable<CoordinatorAction> { 1309 CoordActionInfo(String actionId) { 1310 super("GET", RestConstants.JOB, notEmpty(actionId, "id"), prepareParams(RestConstants.JOB_SHOW_PARAM, 1311 RestConstants.JOB_SHOW_INFO)); 1312 } 1313 1314 @Override 1315 protected CoordinatorAction call(HttpURLConnection conn) throws IOException, OozieClientException { 1316 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1317 Reader reader = new InputStreamReader(conn.getInputStream()); 1318 JSONObject json = (JSONObject) JSONValue.parse(reader); 1319 return JsonToBean.createCoordinatorAction(json); 1320 } 1321 else { 1322 handleError(conn); 1323 } 1324 return null; 1325 } 1326 } 1327 1328 /** 1329 * Get the info of a bundle job. 1330 * 1331 * @param jobId job Id. 1332 * @return the job info. 1333 * @throws OozieClientException thrown if the job info could not be retrieved. 1334 */ 1335 public BundleJob getBundleJobInfo(String jobId) throws OozieClientException { 1336 return new BundleJobInfo(jobId).call(); 1337 } 1338 1339 /** 1340 * Get the info of a coordinator job. 1341 * 1342 * @param jobId job Id. 1343 * @return the job info. 1344 * @throws OozieClientException thrown if the job info could not be retrieved. 1345 */ 1346 public CoordinatorJob getCoordJobInfo(String jobId) throws OozieClientException { 1347 return new CoordJobInfo(jobId, null, -1, -1, "asc").call(); 1348 } 1349 1350 /** 1351 * Get the info of a coordinator job and subset actions. 1352 * 1353 * @param jobId job Id. 1354 * @param filter filter the status filter 1355 * @param start starting index in the list of actions belonging to the job 1356 * @param len number of actions to be returned 1357 * @return the job info. 1358 * @throws OozieClientException thrown if the job info could not be retrieved. 1359 */ 1360 public CoordinatorJob getCoordJobInfo(String jobId, String filter, int start, int len) 1361 throws OozieClientException { 1362 return new CoordJobInfo(jobId, filter, start, len, "asc").call(); 1363 } 1364 1365 /** 1366 * Get the info of a coordinator job and subset actions. 1367 * 1368 * @param jobId job Id. 1369 * @param filter filter the status filter 1370 * @param start starting index in the list of actions belonging to the job 1371 * @param len number of actions to be returned 1372 * @param order order to list coord actions (e.g, desc) 1373 * @return the job info. 1374 * @throws OozieClientException thrown if the job info could not be retrieved. 1375 */ 1376 public CoordinatorJob getCoordJobInfo(String jobId, String filter, int start, int len, String order) 1377 throws OozieClientException { 1378 return new CoordJobInfo(jobId, filter, start, len, order).call(); 1379 } 1380 1381 public List<WorkflowJob> getWfsForCoordAction(String coordActionId) throws OozieClientException { 1382 return new WfsForCoordAction(coordActionId).call(); 1383 } 1384 1385 /** 1386 * Get the info of a coordinator action. 1387 * 1388 * @param actionId Id. 1389 * @return the coordinator action info. 1390 * @throws OozieClientException thrown if the job info could not be retrieved. 1391 */ 1392 public CoordinatorAction getCoordActionInfo(String actionId) throws OozieClientException { 1393 return new CoordActionInfo(actionId).call(); 1394 } 1395 1396 private class JobsStatus extends ClientCallable<List<WorkflowJob>> { 1397 1398 JobsStatus(String filter, int start, int len) { 1399 super("GET", RestConstants.JOBS, "", prepareParams(RestConstants.JOBS_FILTER_PARAM, filter, 1400 RestConstants.JOBTYPE_PARAM, "wf", RestConstants.OFFSET_PARAM, Integer.toString(start), 1401 RestConstants.LEN_PARAM, Integer.toString(len))); 1402 } 1403 1404 @Override 1405 protected List<WorkflowJob> call(HttpURLConnection conn) throws IOException, OozieClientException { 1406 conn.setRequestProperty("content-type", RestConstants.XML_CONTENT_TYPE); 1407 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1408 Reader reader = new InputStreamReader(conn.getInputStream()); 1409 JSONObject json = (JSONObject) JSONValue.parse(reader); 1410 JSONArray workflows = (JSONArray) json.get(JsonTags.WORKFLOWS_JOBS); 1411 if (workflows == null) { 1412 workflows = new JSONArray(); 1413 } 1414 return JsonToBean.createWorkflowJobList(workflows); 1415 } 1416 else { 1417 handleError(conn); 1418 } 1419 return null; 1420 } 1421 } 1422 1423 private class CoordJobsStatus extends ClientCallable<List<CoordinatorJob>> { 1424 1425 CoordJobsStatus(String filter, int start, int len) { 1426 super("GET", RestConstants.JOBS, "", prepareParams(RestConstants.JOBS_FILTER_PARAM, filter, 1427 RestConstants.JOBTYPE_PARAM, "coord", RestConstants.OFFSET_PARAM, Integer.toString(start), 1428 RestConstants.LEN_PARAM, Integer.toString(len))); 1429 } 1430 1431 @Override 1432 protected List<CoordinatorJob> call(HttpURLConnection conn) throws IOException, OozieClientException { 1433 conn.setRequestProperty("content-type", RestConstants.XML_CONTENT_TYPE); 1434 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1435 Reader reader = new InputStreamReader(conn.getInputStream()); 1436 JSONObject json = (JSONObject) JSONValue.parse(reader); 1437 JSONArray jobs = (JSONArray) json.get(JsonTags.COORDINATOR_JOBS); 1438 if (jobs == null) { 1439 jobs = new JSONArray(); 1440 } 1441 return JsonToBean.createCoordinatorJobList(jobs); 1442 } 1443 else { 1444 handleError(conn); 1445 } 1446 return null; 1447 } 1448 } 1449 1450 private class BundleJobsStatus extends ClientCallable<List<BundleJob>> { 1451 1452 BundleJobsStatus(String filter, int start, int len) { 1453 super("GET", RestConstants.JOBS, "", prepareParams(RestConstants.JOBS_FILTER_PARAM, filter, 1454 RestConstants.JOBTYPE_PARAM, "bundle", RestConstants.OFFSET_PARAM, Integer.toString(start), 1455 RestConstants.LEN_PARAM, Integer.toString(len))); 1456 } 1457 1458 @Override 1459 protected List<BundleJob> call(HttpURLConnection conn) throws IOException, OozieClientException { 1460 conn.setRequestProperty("content-type", RestConstants.XML_CONTENT_TYPE); 1461 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1462 Reader reader = new InputStreamReader(conn.getInputStream()); 1463 JSONObject json = (JSONObject) JSONValue.parse(reader); 1464 JSONArray jobs = (JSONArray) json.get(JsonTags.BUNDLE_JOBS); 1465 if (jobs == null) { 1466 jobs = new JSONArray(); 1467 } 1468 return JsonToBean.createBundleJobList(jobs); 1469 } 1470 else { 1471 handleError(conn); 1472 } 1473 return null; 1474 } 1475 } 1476 1477 private class BulkResponseStatus extends ClientCallable<List<BulkResponse>> { 1478 1479 BulkResponseStatus(String filter, int start, int len) { 1480 super("GET", RestConstants.JOBS, "", prepareParams(RestConstants.JOBS_BULK_PARAM, filter, 1481 RestConstants.OFFSET_PARAM, Integer.toString(start), RestConstants.LEN_PARAM, Integer.toString(len))); 1482 } 1483 1484 @Override 1485 protected List<BulkResponse> call(HttpURLConnection conn) throws IOException, OozieClientException { 1486 conn.setRequestProperty("content-type", RestConstants.XML_CONTENT_TYPE); 1487 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1488 Reader reader = new InputStreamReader(conn.getInputStream()); 1489 JSONObject json = (JSONObject) JSONValue.parse(reader); 1490 JSONArray results = (JSONArray) json.get(JsonTags.BULK_RESPONSES); 1491 if (results == null) { 1492 results = new JSONArray(); 1493 } 1494 return JsonToBean.createBulkResponseList(results); 1495 } 1496 else { 1497 handleError(conn); 1498 } 1499 return null; 1500 } 1501 } 1502 1503 private class CoordActionsKill extends ClientCallable<List<CoordinatorAction>> { 1504 1505 CoordActionsKill(String jobId, String rangeType, String scope) { 1506 super("PUT", RestConstants.JOB, notEmpty(jobId, "jobId"), prepareParams(RestConstants.ACTION_PARAM, 1507 RestConstants.JOB_ACTION_KILL, RestConstants.JOB_COORD_RANGE_TYPE_PARAM, rangeType, 1508 RestConstants.JOB_COORD_SCOPE_PARAM, scope)); 1509 } 1510 1511 @Override 1512 protected List<CoordinatorAction> call(HttpURLConnection conn) throws IOException, OozieClientException { 1513 conn.setRequestProperty("content-type", RestConstants.XML_CONTENT_TYPE); 1514 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1515 Reader reader = new InputStreamReader(conn.getInputStream()); 1516 JSONObject json = (JSONObject) JSONValue.parse(reader); 1517 JSONArray coordActions = (JSONArray) json.get(JsonTags.COORDINATOR_ACTIONS); 1518 return JsonToBean.createCoordinatorActionList(coordActions); 1519 } 1520 else { 1521 handleError(conn); 1522 } 1523 return null; 1524 } 1525 } 1526 1527 private class CoordIgnore extends ClientCallable<List<CoordinatorAction>> { 1528 CoordIgnore(String jobId, String rerunType, String scope) { 1529 super("PUT", RestConstants.JOB, notEmpty(jobId, "jobId"), prepareParams(RestConstants.ACTION_PARAM, 1530 RestConstants.JOB_ACTION_IGNORE, RestConstants.JOB_COORD_RANGE_TYPE_PARAM, 1531 rerunType, RestConstants.JOB_COORD_SCOPE_PARAM, scope)); 1532 } 1533 1534 @Override 1535 protected List<CoordinatorAction> call(HttpURLConnection conn) throws IOException, OozieClientException { 1536 conn.setRequestProperty("content-type", RestConstants.XML_CONTENT_TYPE); 1537 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1538 Reader reader = new InputStreamReader(conn.getInputStream()); 1539 JSONObject json = (JSONObject) JSONValue.parse(reader); 1540 if(json != null) { 1541 JSONArray coordActions = (JSONArray) json.get(JsonTags.COORDINATOR_ACTIONS); 1542 return JsonToBean.createCoordinatorActionList(coordActions); 1543 } 1544 } 1545 else { 1546 handleError(conn); 1547 } 1548 return null; 1549 } 1550 } 1551 private class CoordRerun extends ClientCallable<List<CoordinatorAction>> { 1552 1553 CoordRerun(String jobId, String rerunType, String scope, boolean refresh, boolean noCleanup, boolean failed) { 1554 super("PUT", RestConstants.JOB, notEmpty(jobId, "jobId"), prepareParams(RestConstants.ACTION_PARAM, 1555 RestConstants.JOB_COORD_ACTION_RERUN, RestConstants.JOB_COORD_RANGE_TYPE_PARAM, rerunType, 1556 RestConstants.JOB_COORD_SCOPE_PARAM, scope, RestConstants.JOB_COORD_RERUN_REFRESH_PARAM, 1557 Boolean.toString(refresh), RestConstants.JOB_COORD_RERUN_NOCLEANUP_PARAM, Boolean 1558 .toString(noCleanup), RestConstants.JOB_COORD_RERUN_FAILED_PARAM, Boolean.toString(failed))); 1559 } 1560 1561 @Override 1562 protected List<CoordinatorAction> call(HttpURLConnection conn) throws IOException, OozieClientException { 1563 conn.setRequestProperty("content-type", RestConstants.XML_CONTENT_TYPE); 1564 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1565 Reader reader = new InputStreamReader(conn.getInputStream()); 1566 JSONObject json = (JSONObject) JSONValue.parse(reader); 1567 JSONArray coordActions = (JSONArray) json.get(JsonTags.COORDINATOR_ACTIONS); 1568 return JsonToBean.createCoordinatorActionList(coordActions); 1569 } 1570 else { 1571 handleError(conn); 1572 } 1573 return null; 1574 } 1575 } 1576 1577 private class BundleRerun extends ClientCallable<Void> { 1578 1579 BundleRerun(String jobId, String coordScope, String dateScope, boolean refresh, boolean noCleanup) { 1580 super("PUT", RestConstants.JOB, notEmpty(jobId, "jobId"), prepareParams(RestConstants.ACTION_PARAM, 1581 RestConstants.JOB_BUNDLE_ACTION_RERUN, RestConstants.JOB_BUNDLE_RERUN_COORD_SCOPE_PARAM, 1582 coordScope, RestConstants.JOB_BUNDLE_RERUN_DATE_SCOPE_PARAM, dateScope, 1583 RestConstants.JOB_COORD_RERUN_REFRESH_PARAM, Boolean.toString(refresh), 1584 RestConstants.JOB_COORD_RERUN_NOCLEANUP_PARAM, Boolean.toString(noCleanup))); 1585 } 1586 1587 @Override 1588 protected Void call(HttpURLConnection conn) throws IOException, OozieClientException { 1589 conn.setRequestProperty("content-type", RestConstants.XML_CONTENT_TYPE); 1590 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1591 return null; 1592 } 1593 else { 1594 handleError(conn); 1595 } 1596 return null; 1597 } 1598 } 1599 1600 /** 1601 * Rerun coordinator actions. 1602 * 1603 * @param jobId coordinator jobId 1604 * @param rerunType rerun type 'date' if -date is used, 'action-id' if -action is used 1605 * @param scope rerun scope for date or actionIds 1606 * @param refresh true if -refresh is given in command option 1607 * @param noCleanup true if -nocleanup is given in command option 1608 * @throws OozieClientException 1609 */ 1610 public List<CoordinatorAction> reRunCoord(String jobId, String rerunType, String scope, boolean refresh, 1611 boolean noCleanup) throws OozieClientException { 1612 return new CoordRerun(jobId, rerunType, scope, refresh, noCleanup, false).call(); 1613 } 1614 1615 /** 1616 * Rerun coordinator actions with failed option. 1617 * 1618 * @param jobId coordinator jobId 1619 * @param rerunType rerun type 'date' if -date is used, 'action-id' if -action is used 1620 * @param scope rerun scope for date or actionIds 1621 * @param refresh true if -refresh is given in command option 1622 * @param noCleanup true if -nocleanup is given in command option 1623 * @param failed true if -failed is given in command option 1624 * @throws OozieClientException 1625 */ 1626 public List<CoordinatorAction> reRunCoord(String jobId, String rerunType, String scope, boolean refresh, 1627 boolean noCleanup, boolean failed) throws OozieClientException { 1628 return new CoordRerun(jobId, rerunType, scope, refresh, noCleanup, failed).call(); 1629 } 1630 1631 /** 1632 * Rerun bundle coordinators. 1633 * 1634 * @param jobId bundle jobId 1635 * @param coordScope rerun scope for coordinator jobs 1636 * @param dateScope rerun scope for date 1637 * @param refresh true if -refresh is given in command option 1638 * @param noCleanup true if -nocleanup is given in command option 1639 * @throws OozieClientException 1640 */ 1641 public Void reRunBundle(String jobId, String coordScope, String dateScope, boolean refresh, boolean noCleanup) 1642 throws OozieClientException { 1643 return new BundleRerun(jobId, coordScope, dateScope, refresh, noCleanup).call(); 1644 } 1645 1646 /** 1647 * Return the info of the workflow jobs that match the filter. 1648 * 1649 * @param filter job filter. Refer to the {@link OozieClient} for the filter syntax. 1650 * @param start jobs offset, base 1. 1651 * @param len number of jobs to return. 1652 * @return a list with the workflow jobs info, without node details. 1653 * @throws OozieClientException thrown if the jobs info could not be retrieved. 1654 */ 1655 public List<WorkflowJob> getJobsInfo(String filter, int start, int len) throws OozieClientException { 1656 return new JobsStatus(filter, start, len).call(); 1657 } 1658 1659 /** 1660 * Return the info of the workflow jobs that match the filter. 1661 * <p/> 1662 * It returns the first 100 jobs that match the filter. 1663 * 1664 * @param filter job filter. Refer to the {@link OozieClient} for the filter syntax. 1665 * @return a list with the workflow jobs info, without node details. 1666 * @throws OozieClientException thrown if the jobs info could not be retrieved. 1667 */ 1668 public List<WorkflowJob> getJobsInfo(String filter) throws OozieClientException { 1669 return getJobsInfo(filter, 1, 50); 1670 } 1671 1672 /** 1673 * Print sla info about coordinator and workflow jobs and actions. 1674 * 1675 * @param start starting offset 1676 * @param len number of results 1677 * @throws OozieClientException 1678 */ 1679 public void getSlaInfo(int start, int len, String filter) throws OozieClientException { 1680 new SlaInfo(start, len, filter).call(); 1681 } 1682 1683 private class SlaInfo extends ClientCallable<Void> { 1684 1685 SlaInfo(int start, int len, String filter) { 1686 super("GET", WS_PROTOCOL_VERSION_1, RestConstants.SLA, "", prepareParams(RestConstants.SLA_GT_SEQUENCE_ID, 1687 Integer.toString(start), RestConstants.MAX_EVENTS, Integer.toString(len), 1688 RestConstants.JOBS_FILTER_PARAM, filter)); 1689 } 1690 1691 @Override 1692 protected Void call(HttpURLConnection conn) throws IOException, OozieClientException { 1693 conn.setRequestProperty("content-type", RestConstants.XML_CONTENT_TYPE); 1694 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1695 BufferedReader br = new BufferedReader(new InputStreamReader(conn.getInputStream())); 1696 String line = null; 1697 while ((line = br.readLine()) != null) { 1698 System.out.println(line); 1699 } 1700 } 1701 else { 1702 handleError(conn); 1703 } 1704 return null; 1705 } 1706 } 1707 1708 private class JobIdAction extends ClientCallable<String> { 1709 1710 JobIdAction(String externalId) { 1711 super("GET", RestConstants.JOBS, "", prepareParams(RestConstants.JOBTYPE_PARAM, "wf", 1712 RestConstants.JOBS_EXTERNAL_ID_PARAM, externalId)); 1713 } 1714 1715 @Override 1716 protected String call(HttpURLConnection conn) throws IOException, OozieClientException { 1717 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1718 Reader reader = new InputStreamReader(conn.getInputStream()); 1719 JSONObject json = (JSONObject) JSONValue.parse(reader); 1720 return (String) json.get(JsonTags.JOB_ID); 1721 } 1722 else { 1723 handleError(conn); 1724 } 1725 return null; 1726 } 1727 } 1728 1729 /** 1730 * Return the workflow job Id for an external Id. 1731 * <p/> 1732 * The external Id must have provided at job creation time. 1733 * 1734 * @param externalId external Id given at job creation time. 1735 * @return the workflow job Id for an external Id, <code>null</code> if none. 1736 * @throws OozieClientException thrown if the operation could not be done. 1737 */ 1738 public String getJobId(String externalId) throws OozieClientException { 1739 return new JobIdAction(externalId).call(); 1740 } 1741 1742 private class SetSystemMode extends ClientCallable<Void> { 1743 1744 public SetSystemMode(SYSTEM_MODE status) { 1745 super("PUT", RestConstants.ADMIN, RestConstants.ADMIN_STATUS_RESOURCE, prepareParams( 1746 RestConstants.ADMIN_SYSTEM_MODE_PARAM, status + "")); 1747 } 1748 1749 @Override 1750 public Void call(HttpURLConnection conn) throws IOException, OozieClientException { 1751 if (conn.getResponseCode() != HttpURLConnection.HTTP_OK) { 1752 handleError(conn); 1753 } 1754 return null; 1755 } 1756 } 1757 1758 /** 1759 * Enable or disable safe mode. Used by OozieCLI. In safe mode, Oozie would not accept any commands except status 1760 * command to change and view the safe mode status. 1761 * 1762 * @param status true to enable safe mode, false to disable safe mode. 1763 * @throws OozieClientException if it fails to set the safe mode status. 1764 */ 1765 public void setSystemMode(SYSTEM_MODE status) throws OozieClientException { 1766 new SetSystemMode(status).call(); 1767 } 1768 1769 private class GetSystemMode extends ClientCallable<SYSTEM_MODE> { 1770 1771 GetSystemMode() { 1772 super("GET", RestConstants.ADMIN, RestConstants.ADMIN_STATUS_RESOURCE, prepareParams()); 1773 } 1774 1775 @Override 1776 protected SYSTEM_MODE call(HttpURLConnection conn) throws IOException, OozieClientException { 1777 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1778 Reader reader = new InputStreamReader(conn.getInputStream()); 1779 JSONObject json = (JSONObject) JSONValue.parse(reader); 1780 return SYSTEM_MODE.valueOf((String) json.get(JsonTags.OOZIE_SYSTEM_MODE)); 1781 } 1782 else { 1783 handleError(conn); 1784 } 1785 return SYSTEM_MODE.NORMAL; 1786 } 1787 } 1788 1789 /** 1790 * Returns if Oozie is in safe mode or not. 1791 * 1792 * @return true if safe mode is ON<br> 1793 * false if safe mode is OFF 1794 * @throws OozieClientException throw if it could not obtain the safe mode status. 1795 */ 1796 /* 1797 * public boolean isInSafeMode() throws OozieClientException { return new GetSafeMode().call(); } 1798 */ 1799 public SYSTEM_MODE getSystemMode() throws OozieClientException { 1800 return new GetSystemMode().call(); 1801 } 1802 1803 public String updateShareLib() throws OozieClientException { 1804 return new UpdateSharelib().call(); 1805 } 1806 1807 public String listShareLib(String sharelibKey) throws OozieClientException { 1808 return new ListShareLib(sharelibKey).call(); 1809 } 1810 1811 private class GetBuildVersion extends ClientCallable<String> { 1812 1813 GetBuildVersion() { 1814 super("GET", RestConstants.ADMIN, RestConstants.ADMIN_BUILD_VERSION_RESOURCE, prepareParams()); 1815 } 1816 1817 @Override 1818 protected String call(HttpURLConnection conn) throws IOException, OozieClientException { 1819 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1820 Reader reader = new InputStreamReader(conn.getInputStream()); 1821 JSONObject json = (JSONObject) JSONValue.parse(reader); 1822 return (String) json.get(JsonTags.BUILD_VERSION); 1823 } 1824 else { 1825 handleError(conn); 1826 } 1827 return null; 1828 } 1829 } 1830 1831 private class ValidateXML extends ClientCallable<String> { 1832 1833 String file = null; 1834 1835 ValidateXML(String file, String user) { 1836 super("POST", RestConstants.VALIDATE, "", 1837 prepareParams(RestConstants.FILE_PARAM, file, RestConstants.USER_PARAM, user)); 1838 this.file = file; 1839 } 1840 1841 @Override 1842 protected String call(HttpURLConnection conn) throws IOException, OozieClientException { 1843 conn.setRequestProperty("content-type", RestConstants.XML_CONTENT_TYPE); 1844 if (file.startsWith("/")) { 1845 FileInputStream fi = new FileInputStream(new File(file)); 1846 byte[] buffer = new byte[1024]; 1847 int n = 0; 1848 while (-1 != (n = fi.read(buffer))) { 1849 conn.getOutputStream().write(buffer, 0, n); 1850 } 1851 } 1852 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1853 Reader reader = new InputStreamReader(conn.getInputStream()); 1854 JSONObject json = (JSONObject) JSONValue.parse(reader); 1855 return (String) json.get(JsonTags.VALIDATE); 1856 } 1857 else if ((conn.getResponseCode() == HttpURLConnection.HTTP_NOT_FOUND)) { 1858 return null; 1859 } 1860 else { 1861 handleError(conn); 1862 } 1863 return null; 1864 } 1865 } 1866 1867 1868 private class UpdateSharelib extends ClientCallable<String> { 1869 1870 UpdateSharelib() { 1871 super("GET", RestConstants.ADMIN, RestConstants.ADMIN_UPDATE_SHARELIB, prepareParams( 1872 RestConstants.ALL_SERVER_REQUEST, "true")); 1873 } 1874 1875 @Override 1876 protected String call(HttpURLConnection conn) throws IOException, OozieClientException { 1877 StringBuffer bf = new StringBuffer(); 1878 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1879 Reader reader = new InputStreamReader(conn.getInputStream()); 1880 Object sharelib = (Object) JSONValue.parse(reader); 1881 bf.append("[ShareLib update status]").append(System.getProperty("line.separator")); 1882 if (sharelib instanceof JSONArray) { 1883 for (Object o : ((JSONArray) sharelib)) { 1884 JSONObject obj = (JSONObject) ((JSONObject) o).get(JsonTags.SHARELIB_LIB_UPDATE); 1885 for (Object key : obj.keySet()) { 1886 bf.append("\t").append(key).append(" = ").append(obj.get(key)) 1887 .append(System.getProperty("line.separator")); 1888 } 1889 bf.append(System.getProperty("line.separator")); 1890 } 1891 } 1892 else{ 1893 JSONObject obj = (JSONObject) ((JSONObject) sharelib).get(JsonTags.SHARELIB_LIB_UPDATE); 1894 for (Object key : obj.keySet()) { 1895 bf.append("\t").append(key).append(" = ").append(obj.get(key)) 1896 .append(System.getProperty("line.separator")); 1897 } 1898 bf.append(System.getProperty("line.separator")); 1899 } 1900 return bf.toString(); 1901 } 1902 else { 1903 handleError(conn); 1904 } 1905 return null; 1906 } 1907 } 1908 1909 private class ListShareLib extends ClientCallable<String> { 1910 1911 ListShareLib(String sharelibKey) { 1912 super("GET", RestConstants.ADMIN, RestConstants.ADMIN_LIST_SHARELIB, prepareParams( 1913 RestConstants.SHARE_LIB_REQUEST_KEY, sharelibKey)); 1914 } 1915 1916 @Override 1917 protected String call(HttpURLConnection conn) throws IOException, OozieClientException { 1918 1919 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 1920 StringBuffer bf = new StringBuffer(); 1921 Reader reader = new InputStreamReader(conn.getInputStream()); 1922 JSONObject json = (JSONObject) JSONValue.parse(reader); 1923 Object sharelib = json.get(JsonTags.SHARELIB_LIB); 1924 bf.append("[Available ShareLib]").append(System.getProperty("line.separator")); 1925 if (sharelib instanceof JSONArray) { 1926 for (Object o : ((JSONArray) sharelib)) { 1927 JSONObject obj = (JSONObject) o; 1928 bf.append(obj.get(JsonTags.SHARELIB_LIB_NAME)) 1929 .append(System.getProperty("line.separator")); 1930 if (obj.get(JsonTags.SHARELIB_LIB_FILES) != null) { 1931 for (Object file : ((JSONArray) obj.get(JsonTags.SHARELIB_LIB_FILES))) { 1932 bf.append("\t").append(file).append(System.getProperty("line.separator")); 1933 } 1934 } 1935 } 1936 return bf.toString(); 1937 } 1938 } 1939 else { 1940 handleError(conn); 1941 } 1942 return null; 1943 } 1944 1945 } 1946 1947 /** 1948 * Return the Oozie server build version. 1949 * 1950 * @return the Oozie server build version. 1951 * @throws OozieClientException throw if it the server build version could not be retrieved. 1952 */ 1953 public String getServerBuildVersion() throws OozieClientException { 1954 return new GetBuildVersion().call(); 1955 } 1956 1957 /** 1958 * Return the Oozie client build version. 1959 * 1960 * @return the Oozie client build version. 1961 */ 1962 public String getClientBuildVersion() { 1963 return BuildInfo.getBuildInfo().getProperty(BuildInfo.BUILD_VERSION); 1964 } 1965 1966 /** 1967 * Return the workflow application is valid. 1968 * 1969 * @param file local file or hdfs file. 1970 * @return the workflow application is valid. 1971 * @throws OozieClientException throw if it the workflow application's validation could not be retrieved. 1972 */ 1973 public String validateXML(String file) throws OozieClientException { 1974 String fileName = file; 1975 if (file.startsWith("file://")) { 1976 fileName = file.substring(7, file.length()); 1977 } 1978 if (!fileName.contains("://")) { 1979 File f = new File(fileName); 1980 if (!f.isFile()) { 1981 throw new OozieClientException("File error", "File does not exist : " + f.getAbsolutePath()); 1982 } 1983 fileName = f.getAbsolutePath(); 1984 } 1985 String user = USER_NAME_TL.get(); 1986 if (user == null) { 1987 user = System.getProperty("user.name"); 1988 } 1989 return new ValidateXML(fileName, user).call(); 1990 } 1991 1992 /** 1993 * Return the info of the coordinator jobs that match the filter. 1994 * 1995 * @param filter job filter. Refer to the {@link OozieClient} for the filter syntax. 1996 * @param start jobs offset, base 1. 1997 * @param len number of jobs to return. 1998 * @return a list with the coordinator jobs info 1999 * @throws OozieClientException thrown if the jobs info could not be retrieved. 2000 */ 2001 public List<CoordinatorJob> getCoordJobsInfo(String filter, int start, int len) throws OozieClientException { 2002 return new CoordJobsStatus(filter, start, len).call(); 2003 } 2004 2005 /** 2006 * Return the info of the bundle jobs that match the filter. 2007 * 2008 * @param filter job filter. Refer to the {@link OozieClient} for the filter syntax. 2009 * @param start jobs offset, base 1. 2010 * @param len number of jobs to return. 2011 * @return a list with the bundle jobs info 2012 * @throws OozieClientException thrown if the jobs info could not be retrieved. 2013 */ 2014 public List<BundleJob> getBundleJobsInfo(String filter, int start, int len) throws OozieClientException { 2015 return new BundleJobsStatus(filter, start, len).call(); 2016 } 2017 2018 public List<BulkResponse> getBulkInfo(String filter, int start, int len) throws OozieClientException { 2019 return new BulkResponseStatus(filter, start, len).call(); 2020 } 2021 2022 /** 2023 * Poll a job (Workflow Job ID, Coordinator Job ID, Coordinator Action ID, or Bundle Job ID) and return when it has reached a 2024 * terminal state. 2025 * (i.e. FAILED, KILLED, SUCCEEDED) 2026 * 2027 * @param id The Job ID 2028 * @param timeout timeout in minutes (negative values indicate no timeout) 2029 * @param interval polling interval in minutes (must be positive) 2030 * @param verbose if true, the current status will be printed out at each poll; if false, no output 2031 * @throws OozieClientException thrown if the job's status could not be retrieved 2032 */ 2033 public void pollJob(String id, int timeout, int interval, boolean verbose) throws OozieClientException { 2034 notEmpty("id", id); 2035 if (interval < 1) { 2036 throw new IllegalArgumentException("interval must be a positive integer"); 2037 } 2038 boolean noTimeout = (timeout < 1); 2039 long endTime = System.currentTimeMillis() + timeout * 60 * 1000; 2040 interval *= 60 * 1000; 2041 2042 final Set<String> completedStatuses; 2043 if (id.endsWith("-W")) { 2044 completedStatuses = COMPLETED_WF_STATUSES; 2045 } else if (id.endsWith("-C")) { 2046 completedStatuses = COMPLETED_COORD_AND_BUNDLE_STATUSES; 2047 } else if (id.endsWith("-B")) { 2048 completedStatuses = COMPLETED_COORD_AND_BUNDLE_STATUSES; 2049 } else if (id.contains("-C@")) { 2050 completedStatuses = COMPLETED_COORD_ACTION_STATUSES; 2051 } else { 2052 throw new IllegalArgumentException("invalid job type"); 2053 } 2054 2055 String status = getStatus(id); 2056 if (verbose) { 2057 System.out.println(status); 2058 } 2059 while(!completedStatuses.contains(status) && (noTimeout || System.currentTimeMillis() <= endTime)) { 2060 try { 2061 Thread.sleep(interval); 2062 } catch (InterruptedException ie) { 2063 // ignore 2064 } 2065 status = getStatus(id); 2066 if (verbose) { 2067 System.out.println(status); 2068 } 2069 } 2070 } 2071 2072 /** 2073 * Gets the status for a particular job (Workflow Job ID, Coordinator Job ID, Coordinator Action ID, or Bundle Job ID). 2074 * 2075 * @param jobId given jobId 2076 * @return the status 2077 * @throws OozieClientException 2078 */ 2079 public String getStatus(String jobId) throws OozieClientException { 2080 return new Status(jobId).call(); 2081 } 2082 2083 private class Status extends ClientCallable<String> { 2084 2085 Status(String jobId) { 2086 super("GET", RestConstants.JOB, notEmpty(jobId, "jobId"), prepareParams(RestConstants.JOB_SHOW_PARAM, 2087 RestConstants.JOB_SHOW_STATUS)); 2088 } 2089 2090 @Override 2091 protected String call(HttpURLConnection conn) throws IOException, OozieClientException { 2092 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 2093 Reader reader = new InputStreamReader(conn.getInputStream()); 2094 JSONObject json = (JSONObject) JSONValue.parse(reader); 2095 return (String) json.get(JsonTags.STATUS); 2096 } 2097 else { 2098 handleError(conn); 2099 } 2100 return null; 2101 } 2102 } 2103 2104 private class GetQueueDump extends ClientCallable<List<String>> { 2105 GetQueueDump() { 2106 super("GET", RestConstants.ADMIN, RestConstants.ADMIN_QUEUE_DUMP_RESOURCE, prepareParams()); 2107 } 2108 2109 @Override 2110 protected List<String> call(HttpURLConnection conn) throws IOException, OozieClientException { 2111 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 2112 Reader reader = new InputStreamReader(conn.getInputStream()); 2113 JSONObject json = (JSONObject) JSONValue.parse(reader); 2114 JSONArray queueDumpArray = (JSONArray) json.get(JsonTags.QUEUE_DUMP); 2115 2116 List<String> list = new ArrayList<String>(); 2117 list.add("[Server Queue Dump]:"); 2118 for (Object o : queueDumpArray) { 2119 JSONObject entry = (JSONObject) o; 2120 if (entry.get(JsonTags.CALLABLE_DUMP) != null) { 2121 String value = (String) entry.get(JsonTags.CALLABLE_DUMP); 2122 list.add(value); 2123 } 2124 } 2125 if (queueDumpArray.size() == 0) { 2126 list.add("Queue dump is null!"); 2127 } 2128 2129 list.add("******************************************"); 2130 list.add("[Server Uniqueness Map Dump]:"); 2131 2132 JSONArray uniqueDumpArray = (JSONArray) json.get(JsonTags.UNIQUE_MAP_DUMP); 2133 for (Object o : uniqueDumpArray) { 2134 JSONObject entry = (JSONObject) o; 2135 if (entry.get(JsonTags.UNIQUE_ENTRY_DUMP) != null) { 2136 String value = (String) entry.get(JsonTags.UNIQUE_ENTRY_DUMP); 2137 list.add(value); 2138 } 2139 } 2140 if (uniqueDumpArray.size() == 0) { 2141 list.add("Uniqueness dump is null!"); 2142 } 2143 return list; 2144 } 2145 else { 2146 handleError(conn); 2147 } 2148 return null; 2149 } 2150 } 2151 2152 /** 2153 * Return the Oozie queue's commands' dump 2154 * 2155 * @return the list of strings of callable identification in queue 2156 * @throws OozieClientException throw if it the queue dump could not be retrieved. 2157 */ 2158 public List<String> getQueueDump() throws OozieClientException { 2159 return new GetQueueDump().call(); 2160 } 2161 2162 private class GetAvailableOozieServers extends MapClientCallable { 2163 2164 GetAvailableOozieServers() { 2165 super("GET", RestConstants.ADMIN, RestConstants.ADMIN_AVAILABLE_OOZIE_SERVERS_RESOURCE, prepareParams()); 2166 } 2167 } 2168 2169 /** 2170 * Return the list of available Oozie servers. 2171 * 2172 * @return the list of available Oozie servers. 2173 * @throws OozieClientException throw if it the list of available Oozie servers could not be retrieved. 2174 */ 2175 public Map<String, String> getAvailableOozieServers() throws OozieClientException { 2176 return new GetAvailableOozieServers().call(); 2177 } 2178 2179 private class GetServerConfiguration extends MapClientCallable { 2180 2181 GetServerConfiguration() { 2182 super("GET", RestConstants.ADMIN, RestConstants.ADMIN_CONFIG_RESOURCE, prepareParams()); 2183 } 2184 } 2185 2186 /** 2187 * Return the Oozie system configuration. 2188 * 2189 * @return the Oozie system configuration. 2190 * @throws OozieClientException throw if the system configuration could not be retrieved. 2191 */ 2192 public Map<String, String> getServerConfiguration() throws OozieClientException { 2193 return new GetServerConfiguration().call(); 2194 } 2195 2196 private class GetJavaSystemProperties extends MapClientCallable { 2197 2198 GetJavaSystemProperties() { 2199 super("GET", RestConstants.ADMIN, RestConstants.ADMIN_JAVA_SYS_PROPS_RESOURCE, prepareParams()); 2200 } 2201 } 2202 2203 /** 2204 * Return the Oozie Java system properties. 2205 * 2206 * @return the Oozie Java system properties. 2207 * @throws OozieClientException throw if the system properties could not be retrieved. 2208 */ 2209 public Map<String, String> getJavaSystemProperties() throws OozieClientException { 2210 return new GetJavaSystemProperties().call(); 2211 } 2212 2213 private class GetOSEnv extends MapClientCallable { 2214 2215 GetOSEnv() { 2216 super("GET", RestConstants.ADMIN, RestConstants.ADMIN_OS_ENV_RESOURCE, prepareParams()); 2217 } 2218 } 2219 2220 /** 2221 * Return the Oozie system OS environment. 2222 * 2223 * @return the Oozie system OS environment. 2224 * @throws OozieClientException throw if the system OS environment could not be retrieved. 2225 */ 2226 public Map<String, String> getOSEnv() throws OozieClientException { 2227 return new GetOSEnv().call(); 2228 } 2229 2230 private class GetMetrics extends ClientCallable<Metrics> { 2231 2232 GetMetrics() { 2233 super("GET", RestConstants.ADMIN, RestConstants.ADMIN_METRICS_RESOURCE, prepareParams()); 2234 } 2235 2236 @Override 2237 protected Metrics call(HttpURLConnection conn) throws IOException, OozieClientException { 2238 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 2239 Reader reader = new InputStreamReader(conn.getInputStream()); 2240 JSONObject json = (JSONObject) JSONValue.parse(reader); 2241 Metrics metrics = new Metrics(json); 2242 return metrics; 2243 } 2244 else if ((conn.getResponseCode() == HttpURLConnection.HTTP_UNAVAILABLE)) { 2245 // Use Instrumentation endpoint 2246 return null; 2247 } 2248 else { 2249 handleError(conn); 2250 } 2251 return null; 2252 } 2253 } 2254 2255 public class Metrics { 2256 private Map<String, Long> counters; 2257 private Map<String, Object> gauges; 2258 private Map<String, Timer> timers; 2259 private Map<String, Histogram> histograms; 2260 2261 @SuppressWarnings("unchecked") 2262 public Metrics(JSONObject json) { 2263 JSONObject jCounters = (JSONObject) json.get("counters"); 2264 counters = new HashMap<String, Long>(jCounters.size()); 2265 for (Object entO : jCounters.entrySet()) { 2266 Entry<String, JSONObject> ent = (Entry<String, JSONObject>) entO; 2267 counters.put(ent.getKey(), (Long)ent.getValue().get("count")); 2268 } 2269 2270 JSONObject jGuages = (JSONObject) json.get("gauges"); 2271 gauges = new HashMap<String, Object>(jGuages.size()); 2272 for (Object entO : jGuages.entrySet()) { 2273 Entry<String, JSONObject> ent = (Entry<String, JSONObject>) entO; 2274 gauges.put(ent.getKey(), ent.getValue().get("value")); 2275 } 2276 2277 JSONObject jTimers = (JSONObject) json.get("timers"); 2278 timers = new HashMap<String, Timer>(jTimers.size()); 2279 for (Object entO : jTimers.entrySet()) { 2280 Entry<String, JSONObject> ent = (Entry<String, JSONObject>) entO; 2281 timers.put(ent.getKey(), new Timer(ent.getValue())); 2282 } 2283 2284 JSONObject jHistograms = (JSONObject) json.get("histograms"); 2285 histograms = new HashMap<String, Histogram>(jHistograms.size()); 2286 for (Object entO : jHistograms.entrySet()) { 2287 Entry<String, JSONObject> ent = (Entry<String, JSONObject>) entO; 2288 histograms.put(ent.getKey(), new Histogram(ent.getValue())); 2289 } 2290 } 2291 2292 public Map<String, Long> getCounters() { 2293 return counters; 2294 } 2295 2296 public Map<String, Object> getGauges() { 2297 return gauges; 2298 } 2299 2300 public Map<String, Timer> getTimers() { 2301 return timers; 2302 } 2303 2304 public Map<String, Histogram> getHistograms() { 2305 return histograms; 2306 } 2307 2308 public class Timer extends Histogram { 2309 private double m15Rate; 2310 private double m5Rate; 2311 private double m1Rate; 2312 private double meanRate; 2313 private String durationUnits; 2314 private String rateUnits; 2315 2316 public Timer(JSONObject json) { 2317 super(json); 2318 m15Rate = Double.valueOf(json.get("m15_rate").toString()); 2319 m5Rate = Double.valueOf(json.get("m5_rate").toString()); 2320 m1Rate = Double.valueOf(json.get("m1_rate").toString()); 2321 meanRate = Double.valueOf(json.get("mean_rate").toString()); 2322 durationUnits = json.get("duration_units").toString(); 2323 rateUnits = json.get("rate_units").toString(); 2324 } 2325 2326 public double get15MinuteRate() { 2327 return m15Rate; 2328 } 2329 2330 public double get5MinuteRate() { 2331 return m5Rate; 2332 } 2333 2334 public double get1MinuteRate() { 2335 return m1Rate; 2336 } 2337 2338 public double getMeanRate() { 2339 return meanRate; 2340 } 2341 2342 public String getDurationUnits() { 2343 return durationUnits; 2344 } 2345 2346 public String getRateUnits() { 2347 return rateUnits; 2348 } 2349 2350 @Override 2351 public String toString() { 2352 StringBuilder sb = new StringBuilder(super.toString()); 2353 sb.append("\n\t15 minute rate : ").append(m15Rate); 2354 sb.append("\n\t5 minute rate : ").append(m5Rate); 2355 sb.append("\n\t1 minute rate : ").append(m15Rate); 2356 sb.append("\n\tmean rate : ").append(meanRate); 2357 sb.append("\n\tduration units : ").append(durationUnits); 2358 sb.append("\n\trate units : ").append(rateUnits); 2359 return sb.toString(); 2360 } 2361 } 2362 2363 public class Histogram { 2364 private double p999; 2365 private double p99; 2366 private double p98; 2367 private double p95; 2368 private double p75; 2369 private double p50; 2370 private double mean; 2371 private double min; 2372 private double max; 2373 private double stdDev; 2374 private long count; 2375 2376 public Histogram(JSONObject json) { 2377 p999 = Double.valueOf(json.get("p999").toString()); 2378 p99 = Double.valueOf(json.get("p99").toString()); 2379 p98 = Double.valueOf(json.get("p98").toString()); 2380 p95 = Double.valueOf(json.get("p95").toString()); 2381 p75 = Double.valueOf(json.get("p75").toString()); 2382 p50 = Double.valueOf(json.get("p50").toString()); 2383 mean = Double.valueOf(json.get("mean").toString()); 2384 min = Double.valueOf(json.get("min").toString()); 2385 max = Double.valueOf(json.get("max").toString()); 2386 stdDev = Double.valueOf(json.get("stddev").toString()); 2387 count = Long.valueOf(json.get("count").toString()); 2388 } 2389 2390 public double get999thPercentile() { 2391 return p999; 2392 } 2393 2394 public double get99thPercentile() { 2395 return p99; 2396 } 2397 2398 public double get98thPercentile() { 2399 return p98; 2400 } 2401 2402 public double get95thPercentile() { 2403 return p95; 2404 } 2405 2406 public double get75thPercentile() { 2407 return p75; 2408 } 2409 2410 public double get50thPercentile() { 2411 return p50; 2412 } 2413 2414 public double getMean() { 2415 return mean; 2416 } 2417 2418 public double getMin() { 2419 return min; 2420 } 2421 2422 public double getMax() { 2423 return max; 2424 } 2425 2426 public double getStandardDeviation() { 2427 return stdDev; 2428 } 2429 2430 public long getCount() { 2431 return count; 2432 } 2433 2434 @Override 2435 public String toString() { 2436 StringBuilder sb = new StringBuilder(); 2437 sb.append("\t999th percentile : ").append(p999); 2438 sb.append("\n\t99th percentile : ").append(p99); 2439 sb.append("\n\t98th percentile : ").append(p98); 2440 sb.append("\n\t95th percentile : ").append(p95); 2441 sb.append("\n\t75th percentile : ").append(p75); 2442 sb.append("\n\t50th percentile : ").append(p50); 2443 sb.append("\n\tmean : ").append(mean); 2444 sb.append("\n\tmax : ").append(max); 2445 sb.append("\n\tmin : ").append(min); 2446 sb.append("\n\tcount : ").append(count); 2447 sb.append("\n\tstandard deviation : ").append(stdDev); 2448 return sb.toString(); 2449 } 2450 } 2451 } 2452 2453 /** 2454 * Return the Oozie metrics. If null is returned, then try {@link #getInstrumentation()}. 2455 * 2456 * @return the Oozie metrics or null. 2457 * @throws OozieClientException throw if the metrics could not be retrieved. 2458 */ 2459 public Metrics getMetrics() throws OozieClientException { 2460 return new GetMetrics().call(); 2461 } 2462 2463 private class GetInstrumentation extends ClientCallable<Instrumentation> { 2464 2465 GetInstrumentation() { 2466 super("GET", RestConstants.ADMIN, RestConstants.ADMIN_INSTRUMENTATION_RESOURCE, prepareParams()); 2467 } 2468 2469 @Override 2470 protected Instrumentation call(HttpURLConnection conn) throws IOException, OozieClientException { 2471 if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) { 2472 Reader reader = new InputStreamReader(conn.getInputStream()); 2473 JSONObject json = (JSONObject) JSONValue.parse(reader); 2474 Instrumentation instrumentation = new Instrumentation(json); 2475 return instrumentation; 2476 } 2477 else if ((conn.getResponseCode() == HttpURLConnection.HTTP_UNAVAILABLE)) { 2478 // Use Metrics endpoint 2479 return null; 2480 } 2481 else { 2482 handleError(conn); 2483 } 2484 return null; 2485 } 2486 } 2487 2488 public class Instrumentation { 2489 private Map<String, Long> counters; 2490 private Map<String, Object> variables; 2491 private Map<String, Double> samplers; 2492 private Map<String, Timer> timers; 2493 2494 public Instrumentation(JSONObject json) { 2495 JSONArray jCounters = (JSONArray) json.get("counters"); 2496 counters = new HashMap<String, Long>(jCounters.size()); 2497 for (Object groupO : jCounters) { 2498 JSONObject group = (JSONObject) groupO; 2499 String groupName = group.get("group").toString() + "."; 2500 JSONArray data = (JSONArray) group.get("data"); 2501 for (Object datO : data) { 2502 JSONObject dat = (JSONObject) datO; 2503 counters.put(groupName + dat.get("name").toString(), Long.valueOf(dat.get("value").toString())); 2504 } 2505 } 2506 2507 JSONArray jVariables = (JSONArray) json.get("variables"); 2508 variables = new HashMap<String, Object>(jVariables.size()); 2509 for (Object groupO : jVariables) { 2510 JSONObject group = (JSONObject) groupO; 2511 String groupName = group.get("group").toString() + "."; 2512 JSONArray data = (JSONArray) group.get("data"); 2513 for (Object datO : data) { 2514 JSONObject dat = (JSONObject) datO; 2515 variables.put(groupName + dat.get("name").toString(), dat.get("value")); 2516 } 2517 } 2518 2519 JSONArray jSamplers = (JSONArray) json.get("samplers"); 2520 samplers = new HashMap<String, Double>(jSamplers.size()); 2521 for (Object groupO : jSamplers) { 2522 JSONObject group = (JSONObject) groupO; 2523 String groupName = group.get("group").toString() + "."; 2524 JSONArray data = (JSONArray) group.get("data"); 2525 for (Object datO : data) { 2526 JSONObject dat = (JSONObject) datO; 2527 samplers.put(groupName + dat.get("name").toString(), Double.valueOf(dat.get("value").toString())); 2528 } 2529 } 2530 2531 JSONArray jTimers = (JSONArray) json.get("timers"); 2532 timers = new HashMap<String, Timer>(jTimers.size()); 2533 for (Object groupO : jTimers) { 2534 JSONObject group = (JSONObject) groupO; 2535 String groupName = group.get("group").toString() + "."; 2536 JSONArray data = (JSONArray) group.get("data"); 2537 for (Object datO : data) { 2538 JSONObject dat = (JSONObject) datO; 2539 timers.put(groupName + dat.get("name").toString(), new Timer(dat)); 2540 } 2541 } 2542 } 2543 2544 public class Timer { 2545 private double ownTimeStdDev; 2546 private long ownTimeAvg; 2547 private long ownMaxTime; 2548 private long ownMinTime; 2549 private double totalTimeStdDev; 2550 private long totalTimeAvg; 2551 private long totalMaxTime; 2552 private long totalMinTime; 2553 private long ticks; 2554 2555 public Timer(JSONObject json) { 2556 ownTimeStdDev = Double.valueOf(json.get("ownTimeStdDev").toString()); 2557 ownTimeAvg = Long.valueOf(json.get("ownTimeAvg").toString()); 2558 ownMaxTime = Long.valueOf(json.get("ownMaxTime").toString()); 2559 ownMinTime = Long.valueOf(json.get("ownMinTime").toString()); 2560 totalTimeStdDev = Double.valueOf(json.get("totalTimeStdDev").toString()); 2561 totalTimeAvg = Long.valueOf(json.get("totalTimeAvg").toString()); 2562 totalMaxTime = Long.valueOf(json.get("totalMaxTime").toString()); 2563 totalMinTime = Long.valueOf(json.get("totalMinTime").toString()); 2564 ticks = Long.valueOf(json.get("ticks").toString()); 2565 } 2566 2567 public double getOwnTimeStandardDeviation() { 2568 return ownTimeStdDev; 2569 } 2570 2571 public long getOwnTimeAverage() { 2572 return ownTimeAvg; 2573 } 2574 2575 public long getOwnMaxTime() { 2576 return ownMaxTime; 2577 } 2578 2579 public long getOwnMinTime() { 2580 return ownMinTime; 2581 } 2582 2583 public double getTotalTimeStandardDeviation() { 2584 return totalTimeStdDev; 2585 } 2586 2587 public long getTotalTimeAverage() { 2588 return totalTimeAvg; 2589 } 2590 2591 public long getTotalMaxTime() { 2592 return totalMaxTime; 2593 } 2594 2595 public long getTotalMinTime() { 2596 return totalMinTime; 2597 } 2598 2599 public long getTicks() { 2600 return ticks; 2601 } 2602 2603 @Override 2604 public String toString() { 2605 StringBuilder sb = new StringBuilder(); 2606 sb.append("\town time standard deviation : ").append(ownTimeStdDev); 2607 sb.append("\n\town average time : ").append(ownTimeAvg); 2608 sb.append("\n\town max time : ").append(ownMaxTime); 2609 sb.append("\n\town min time : ").append(ownMinTime); 2610 sb.append("\n\ttotal time standard deviation : ").append(totalTimeStdDev); 2611 sb.append("\n\ttotal average time : ").append(totalTimeAvg); 2612 sb.append("\n\ttotal max time : ").append(totalMaxTime); 2613 sb.append("\n\ttotal min time : ").append(totalMinTime); 2614 sb.append("\n\tticks : ").append(ticks); 2615 return sb.toString(); 2616 } 2617 } 2618 2619 public Map<String, Long> getCounters() { 2620 return counters; 2621 } 2622 2623 public Map<String, Object> getVariables() { 2624 return variables; 2625 } 2626 2627 public Map<String, Double> getSamplers() { 2628 return samplers; 2629 } 2630 2631 public Map<String, Timer> getTimers() { 2632 return timers; 2633 } 2634 } 2635 2636 /** 2637 * Return the Oozie instrumentation. If null is returned, then try {@link #getMetrics()}. 2638 * 2639 * @return the Oozie intstrumentation or null. 2640 * @throws OozieClientException throw if the intstrumentation could not be retrieved. 2641 */ 2642 public Instrumentation getInstrumentation() throws OozieClientException { 2643 return new GetInstrumentation().call(); 2644 } 2645 2646 /** 2647 * Check if the string is not null or not empty. 2648 * 2649 * @param str 2650 * @param name 2651 * @return string 2652 */ 2653 public static String notEmpty(String str, String name) { 2654 if (str == null) { 2655 throw new IllegalArgumentException(name + " cannot be null"); 2656 } 2657 if (str.length() == 0) { 2658 throw new IllegalArgumentException(name + " cannot be empty"); 2659 } 2660 return str; 2661 } 2662 2663 /** 2664 * Check if the object is not null. 2665 * 2666 * @param <T> 2667 * @param obj 2668 * @param name 2669 * @return string 2670 */ 2671 public static <T> T notNull(T obj, String name) { 2672 if (obj == null) { 2673 throw new IllegalArgumentException(name + " cannot be null"); 2674 } 2675 return obj; 2676 } 2677 2678}