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.coord; 020 021import com.google.common.collect.Lists; 022import org.apache.commons.lang.StringUtils; 023import org.apache.hadoop.conf.Configuration; 024import org.apache.oozie.ErrorCode; 025import org.apache.oozie.client.OozieClient; 026import org.apache.oozie.command.CommandException; 027import org.apache.oozie.dependency.URIHandler; 028import org.apache.oozie.dependency.URIHandler.Context; 029import org.apache.oozie.service.Services; 030import org.apache.oozie.service.URIHandlerService; 031import org.apache.oozie.util.DateUtils; 032import org.apache.oozie.util.ELEvaluator; 033import org.apache.oozie.util.ParamChecker; 034import org.apache.oozie.util.XLog; 035 036import java.net.URI; 037import java.util.ArrayList; 038import java.util.Calendar; 039import java.util.Date; 040import java.util.GregorianCalendar; 041import java.util.List; 042import java.util.TimeZone; 043 044/** 045 * This class implements the EL function related to coordinator 046 */ 047 048public class CoordELFunctions { 049 final public static String DATASET = "oozie.coord.el.dataset.bean"; 050 final public static String COORD_ACTION = "oozie.coord.el.app.bean"; 051 final public static String CONFIGURATION = "oozie.coord.el.conf"; 052 final public static String LATEST_EL_USE_CURRENT_TIME = "oozie.service.ELService.latest-el.use-current-time"; 053 // INSTANCE_SEPARATOR is used to separate multiple directories into one tag. 054 final public static String INSTANCE_SEPARATOR = "#"; 055 final public static String DIR_SEPARATOR = ","; 056 // TODO: in next release, support flexibility 057 private static String END_OF_OPERATION_INDICATOR_FILE = "_SUCCESS"; 058 059 public static final long MINUTE_MSEC = 60 * 1000L; 060 public static final long HOUR_MSEC = 60 * MINUTE_MSEC; 061 public static final long DAY_MSEC = 24 * HOUR_MSEC; 062 public static final long MONTH_MSEC = 30 * DAY_MSEC; 063 public static final long YEAR_MSEC = 365 * DAY_MSEC; 064 065 /** 066 * Used in defining the frequency in 'day' unit. <p/> domain: <code> val > 0</code> and should be integer. 067 * 068 * @param val frequency in number of days. 069 * @return number of days and also set the frequency timeunit to "day" 070 */ 071 public static int ph1_coord_days(int val) { 072 val = ParamChecker.checkGTZero(val, "n"); 073 ELEvaluator eval = ELEvaluator.getCurrent(); 074 eval.setVariable("timeunit", TimeUnit.DAY); 075 eval.setVariable("endOfDuration", TimeUnit.NONE); 076 return val; 077 } 078 079 /** 080 * Used in defining the frequency in 'month' unit. <p/> domain: <code> val > 0</code> and should be integer. 081 * 082 * @param val frequency in number of months. 083 * @return number of months and also set the frequency timeunit to "month" 084 */ 085 public static int ph1_coord_months(int val) { 086 val = ParamChecker.checkGTZero(val, "n"); 087 ELEvaluator eval = ELEvaluator.getCurrent(); 088 eval.setVariable("timeunit", TimeUnit.MONTH); 089 eval.setVariable("endOfDuration", TimeUnit.NONE); 090 return val; 091 } 092 093 /** 094 * Used in defining the frequency in 'hour' unit. <p/> parameter value domain: <code> val > 0</code> and should 095 * be integer. 096 * 097 * @param val frequency in number of hours. 098 * @return number of minutes and also set the frequency timeunit to "minute" 099 */ 100 public static int ph1_coord_hours(int val) { 101 val = ParamChecker.checkGTZero(val, "n"); 102 ELEvaluator eval = ELEvaluator.getCurrent(); 103 eval.setVariable("timeunit", TimeUnit.MINUTE); 104 eval.setVariable("endOfDuration", TimeUnit.NONE); 105 return val * 60; 106 } 107 108 /** 109 * Used in defining the frequency in 'minute' unit. <p/> domain: <code> val > 0</code> and should be integer. 110 * 111 * @param val frequency in number of minutes. 112 * @return number of minutes and also set the frequency timeunit to "minute" 113 */ 114 public static int ph1_coord_minutes(int val) { 115 val = ParamChecker.checkGTZero(val, "n"); 116 ELEvaluator eval = ELEvaluator.getCurrent(); 117 eval.setVariable("timeunit", TimeUnit.MINUTE); 118 eval.setVariable("endOfDuration", TimeUnit.NONE); 119 return val; 120 } 121 122 /** 123 * Used in defining the frequency in 'day' unit and specify the "end of day" property. <p/> Every instance will 124 * start at 00:00 hour of each day. <p/> domain: <code> val > 0</code> and should be integer. 125 * 126 * @param val frequency in number of days. 127 * @return number of days and also set the frequency timeunit to "day" and end_of_duration flag to "day" 128 */ 129 public static int ph1_coord_endOfDays(int val) { 130 val = ParamChecker.checkGTZero(val, "n"); 131 ELEvaluator eval = ELEvaluator.getCurrent(); 132 eval.setVariable("timeunit", TimeUnit.DAY); 133 eval.setVariable("endOfDuration", TimeUnit.END_OF_DAY); 134 return val; 135 } 136 137 /** 138 * Used in defining the frequency in 'month' unit and specify the "end of month" property. <p/> Every instance will 139 * start at first day of each month at 00:00 hour. <p/> domain: <code> val > 0</code> and should be integer. 140 * 141 * @param val: frequency in number of months. 142 * @return number of months and also set the frequency timeunit to "month" and end_of_duration flag to "month" 143 */ 144 public static int ph1_coord_endOfMonths(int val) { 145 val = ParamChecker.checkGTZero(val, "n"); 146 ELEvaluator eval = ELEvaluator.getCurrent(); 147 eval.setVariable("timeunit", TimeUnit.MONTH); 148 eval.setVariable("endOfDuration", TimeUnit.END_OF_MONTH); 149 return val; 150 } 151 152 /** 153 * Calculate the difference of timezone offset in minutes between dataset and coordinator job. <p/> Depends on: <p/> 154 * 1. Timezone of both dataset and job <p/> 2. Action creation Time 155 * 156 * @return difference in minutes (DataSet TZ Offset - Application TZ offset) 157 */ 158 public static int ph2_coord_tzOffset() { 159 long actionCreationTime = getActionCreationtime().getTime(); 160 TimeZone dsTZ = ParamChecker.notNull(getDatasetTZ(), "DatasetTZ"); 161 TimeZone jobTZ = ParamChecker.notNull(getJobTZ(), "JobTZ"); 162 return (dsTZ.getOffset(actionCreationTime) - jobTZ.getOffset(actionCreationTime)) / (1000 * 60); 163 } 164 165 public static int ph3_coord_tzOffset() { 166 return ph2_coord_tzOffset(); 167 } 168 169 /** 170 * Returns a date string that is offset from 'strBaseDate' by the amount specified. The unit can be one of 171 * DAY, MONTH, HOUR, MINUTE, MONTH. 172 * 173 * @param strBaseDate The base date 174 * @param offset any number 175 * @param unit one of DAY, MONTH, HOUR, MINUTE, MONTH 176 * @return the offset date string 177 * @throws Exception 178 */ 179 public static String ph2_coord_dateOffset(String strBaseDate, int offset, String unit) throws Exception { 180 Calendar baseCalDate = DateUtils.getCalendar(strBaseDate); 181 StringBuilder buffer = new StringBuilder(); 182 baseCalDate.add(TimeUnit.valueOf(unit).getCalendarUnit(), offset); 183 buffer.append(DateUtils.formatDateOozieTZ(baseCalDate)); 184 return buffer.toString(); 185 } 186 187 public static String ph3_coord_dateOffset(String strBaseDate, int offset, String unit) throws Exception { 188 return ph2_coord_dateOffset(strBaseDate, offset, unit); 189 } 190 191 /** 192 * Returns a date string that is offset from 'strBaseDate' by the difference from Oozie processing timezone to the given 193 * timezone. It will account for daylight saving time based on the given 'strBaseDate' and 'timezone'. 194 * 195 * @param strBaseDate The base date 196 * @param timezone 197 * @return the offset date string 198 * @throws Exception 199 */ 200 public static String ph2_coord_dateTzOffset(String strBaseDate, String timezone) throws Exception { 201 Calendar baseCalDate = DateUtils.getCalendar(strBaseDate); 202 StringBuilder buffer = new StringBuilder(); 203 baseCalDate.setTimeZone(DateUtils.getTimeZone(timezone)); 204 buffer.append(DateUtils.formatDate(baseCalDate)); 205 return buffer.toString(); 206 } 207 208 public static String ph3_coord_dateTzOffset(String strBaseDate, String timezone) throws Exception{ 209 return ph2_coord_dateTzOffset(strBaseDate, timezone); 210 } 211 212 /** 213 * Determine the date-time in Oozie processing timezone of n-th future available dataset instance 214 * from nominal Time but not beyond the instance specified as 'instance. 215 * <p/> 216 * It depends on: 217 * <p/> 218 * 1. Data set frequency 219 * <p/> 220 * 2. Data set Time unit (day, month, minute) 221 * <p/> 222 * 3. Data set Time zone/DST 223 * <p/> 224 * 4. End Day/Month flag 225 * <p/> 226 * 5. Data set initial instance 227 * <p/> 228 * 6. Action Creation Time 229 * <p/> 230 * 7. Existence of dataset's directory 231 * 232 * @param n :instance count 233 * <p/> 234 * domain: n >= 0, n is integer 235 * @param instance: How many future instance it should check? value should 236 * be >=0 237 * @return date-time in Oozie processing timezone of the n-th instance 238 * <p/> 239 * @throws Exception 240 */ 241 public static String ph3_coord_future(int n, int instance) throws Exception { 242 ParamChecker.checkGEZero(n, "future:n"); 243 ParamChecker.checkGTZero(instance, "future:instance"); 244 if (isSyncDataSet()) {// For Sync Dataset 245 return coord_future_sync(n, instance); 246 } 247 else { 248 throw new UnsupportedOperationException("Asynchronous Dataset is not supported yet"); 249 } 250 } 251 252 /** 253 * Determine the date-time in Oozie processing timezone of the future available dataset instances 254 * from start to end offsets from nominal Time but not beyond the instance specified as 'instance'. 255 * <p/> 256 * It depends on: 257 * <p/> 258 * 1. Data set frequency 259 * <p/> 260 * 2. Data set Time unit (day, month, minute) 261 * <p/> 262 * 3. Data set Time zone/DST 263 * <p/> 264 * 4. End Day/Month flag 265 * <p/> 266 * 5. Data set initial instance 267 * <p/> 268 * 6. Action Creation Time 269 * <p/> 270 * 7. Existence of dataset's directory 271 * 272 * @param start : start instance offset 273 * <p/> 274 * domain: start >= 0, start is integer 275 * @param end : end instance offset 276 * <p/> 277 * domain: end >= 0, end is integer 278 * @param instance: How many future instance it should check? value should 279 * be >=0 280 * @return date-time in Oozie processing timezone of the instances from start to end offsets 281 * delimited by comma. 282 * <p/> 283 * @throws Exception 284 */ 285 public static String ph3_coord_futureRange(int start, int end, int instance) throws Exception { 286 ParamChecker.checkGEZero(start, "future:n"); 287 ParamChecker.checkGEZero(end, "future:n"); 288 ParamChecker.checkGTZero(instance, "future:instance"); 289 if (isSyncDataSet()) {// For Sync Dataset 290 return coord_futureRange_sync(start, end, instance); 291 } 292 else { 293 throw new UnsupportedOperationException("Asynchronous Dataset is not supported yet"); 294 } 295 } 296 297 private static String coord_future_sync(int n, int instance) throws Exception { 298 return coord_futureRange_sync(n, n, instance); 299 } 300 301 private static String coord_futureRange_sync(int startOffset, int endOffset, int instance) throws Exception { 302 final XLog LOG = XLog.getLog(CoordELFunctions.class); 303 final Thread currentThread = Thread.currentThread(); 304 ELEvaluator eval = ELEvaluator.getCurrent(); 305 String retVal = ""; 306 int datasetFrequency = (int) getDSFrequency();// in minutes 307 TimeUnit dsTimeUnit = getDSTimeUnit(); 308 int[] instCount = new int[1]; 309 Calendar nominalInstanceCal = getCurrentInstance(getActionCreationtime(), instCount); 310 StringBuilder resolvedInstances = new StringBuilder(); 311 StringBuilder resolvedURIPaths = new StringBuilder(); 312 if (nominalInstanceCal != null) { 313 Calendar initInstance = getInitialInstanceCal(); 314 nominalInstanceCal = (Calendar) initInstance.clone(); 315 nominalInstanceCal.add(dsTimeUnit.getCalendarUnit(), instCount[0] * datasetFrequency); 316 317 SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET); 318 if (ds == null) { 319 throw new RuntimeException("Associated Dataset should be defined with key " + DATASET); 320 } 321 String uriTemplate = ds.getUriTemplate(); 322 Configuration conf = (Configuration) eval.getVariable(CONFIGURATION); 323 if (conf == null) { 324 throw new RuntimeException("Associated Configuration should be defined with key " + CONFIGURATION); 325 } 326 int available = 0, checkedInstance = 0; 327 boolean resolved = false; 328 String user = ParamChecker 329 .notEmpty((String) eval.getVariable(OozieClient.USER_NAME), OozieClient.USER_NAME); 330 String doneFlag = ds.getDoneFlag(); 331 URIHandlerService uriService = Services.get().get(URIHandlerService.class); 332 URIHandler uriHandler = null; 333 Context uriContext = null; 334 try { 335 while (instance >= checkedInstance && !currentThread.isInterrupted()) { 336 ELEvaluator uriEval = getUriEvaluator(nominalInstanceCal); 337 String uriPath = uriEval.evaluate(uriTemplate, String.class); 338 if (uriHandler == null) { 339 URI uri = new URI(uriPath); 340 uriHandler = uriService.getURIHandler(uri); 341 uriContext = uriHandler.getContext(uri, conf, user); 342 } 343 String uriWithDoneFlag = uriHandler.getURIWithDoneFlag(uriPath, doneFlag); 344 if (uriHandler.exists(new URI(uriWithDoneFlag), uriContext)) { 345 if (available == endOffset) { 346 LOG.debug("Matched future(" + available + "): " + uriWithDoneFlag); 347 resolved = true; 348 resolvedInstances.append(DateUtils.formatDateOozieTZ(nominalInstanceCal)); 349 resolvedURIPaths.append(uriPath); 350 retVal = resolvedInstances.toString(); 351 eval.setVariable("resolved_path", resolvedURIPaths.toString()); 352 break; 353 } 354 else if (available >= startOffset) { 355 LOG.debug("Matched future(" + available + "): " + uriWithDoneFlag); 356 resolvedInstances.append(DateUtils.formatDateOozieTZ(nominalInstanceCal)).append( 357 INSTANCE_SEPARATOR); 358 resolvedURIPaths.append(uriPath).append(INSTANCE_SEPARATOR); 359 } 360 available++; 361 } 362 // nominalInstanceCal.add(dsTimeUnit.getCalendarUnit(), datasetFrequency); 363 nominalInstanceCal = (Calendar) initInstance.clone(); 364 instCount[0]++; 365 nominalInstanceCal.add(dsTimeUnit.getCalendarUnit(), instCount[0] * datasetFrequency); 366 checkedInstance++; 367 // DateUtils.moveToEnd(nominalInstanceCal, getDSEndOfFlag()); 368 } 369 } 370 finally { 371 if (uriContext != null) { 372 uriContext.destroy(); 373 } 374 } 375 if (!resolved) { 376 // return unchanged future function with variable 'is_resolved' 377 // to 'false' 378 eval.setVariable("is_resolved", Boolean.FALSE); 379 if (startOffset == endOffset) { 380 retVal = "${coord:future(" + startOffset + ", " + instance + ")}"; 381 } 382 else { 383 retVal = "${coord:futureRange(" + startOffset + ", " + endOffset + ", " + instance + ")}"; 384 } 385 } 386 else { 387 eval.setVariable("is_resolved", Boolean.TRUE); 388 } 389 } 390 else {// No feasible nominal time 391 eval.setVariable("is_resolved", Boolean.TRUE); 392 retVal = ""; 393 } 394 return retVal; 395 } 396 397 /** 398 * Return nominal time or Action Creation Time. 399 * <p/> 400 * 401 * @return coordinator action creation or materialization date time 402 * @throws Exception if unable to format the Date object to String 403 */ 404 public static String ph2_coord_nominalTime() throws Exception { 405 ELEvaluator eval = ELEvaluator.getCurrent(); 406 SyncCoordAction action = ParamChecker.notNull((SyncCoordAction) eval.getVariable(COORD_ACTION), 407 "Coordinator Action"); 408 return DateUtils.formatDateOozieTZ(action.getNominalTime()); 409 } 410 411 public static String ph3_coord_nominalTime() throws Exception { 412 return ph2_coord_nominalTime(); 413 } 414 415 /** 416 * Convert from standard date-time formatting to a desired format. 417 * <p/> 418 * @param dateTimeStr - A timestamp in standard (ISO8601) format. 419 * @param format - A string representing the desired format. 420 * @return coordinator action creation or materialization date time 421 * @throws Exception if unable to format the Date object to String 422 */ 423 public static String ph2_coord_formatTime(String dateTimeStr, String format) 424 throws Exception { 425 Date dateTime = DateUtils.parseDateOozieTZ(dateTimeStr); 426 return DateUtils.formatDateCustom(dateTime, format); 427 } 428 429 public static String ph3_coord_formatTime(String dateTimeStr, String format) 430 throws Exception { 431 return ph2_coord_formatTime(dateTimeStr, format); 432 } 433 434 /** 435 * Return Action Id. <p/> 436 * 437 * @return coordinator action Id 438 */ 439 public static String ph2_coord_actionId() throws Exception { 440 ELEvaluator eval = ELEvaluator.getCurrent(); 441 SyncCoordAction action = ParamChecker.notNull((SyncCoordAction) eval.getVariable(COORD_ACTION), 442 "Coordinator Action"); 443 return action.getActionId(); 444 } 445 446 public static String ph3_coord_actionId() throws Exception { 447 return ph2_coord_actionId(); 448 } 449 450 /** 451 * Return Job Name. <p/> 452 * 453 * @return coordinator name 454 */ 455 public static String ph2_coord_name() throws Exception { 456 ELEvaluator eval = ELEvaluator.getCurrent(); 457 SyncCoordAction action = ParamChecker.notNull((SyncCoordAction) eval.getVariable(COORD_ACTION), 458 "Coordinator Action"); 459 return action.getName(); 460 } 461 462 public static String ph3_coord_name() throws Exception { 463 return ph2_coord_name(); 464 } 465 466 /** 467 * Return Action Start time. <p/> 468 * 469 * @return coordinator action start time 470 * @throws Exception if unable to format the Date object to String 471 */ 472 public static String ph2_coord_actualTime() throws Exception { 473 ELEvaluator eval = ELEvaluator.getCurrent(); 474 SyncCoordAction coordAction = (SyncCoordAction) eval.getVariable(COORD_ACTION); 475 if (coordAction == null) { 476 throw new RuntimeException("Associated Application instance should be defined with key " + COORD_ACTION); 477 } 478 return DateUtils.formatDateOozieTZ(coordAction.getActualTime()); 479 } 480 481 public static String ph3_coord_actualTime() throws Exception { 482 return ph2_coord_actualTime(); 483 } 484 485 /** 486 * Used to specify a list of URI's that are used as input dir to the workflow job. <p/> Look for two evaluator-level 487 * variables <p/> A) .datain.<DATAIN_NAME> B) .datain.<DATAIN_NAME>.unresolved <p/> A defines the current list of 488 * URI. <p/> B defines whether there are any unresolved EL-function (i.e latest) <p/> If there are something 489 * unresolved, this function will echo back the original function <p/> otherwise it sends the uris. 490 * 491 * @param dataInName : Datain name 492 * @return the list of URI's separated by INSTANCE_SEPARATOR <p/> if there are unresolved EL function (i.e. latest) 493 * , echo back <p/> the function without resolving the function. 494 */ 495 public static String ph3_coord_dataIn(String dataInName) { 496 String uris = ""; 497 ELEvaluator eval = ELEvaluator.getCurrent(); 498 uris = (String) eval.getVariable(".datain." + dataInName); 499 Boolean unresolved = (Boolean) eval.getVariable(".datain." + dataInName + ".unresolved"); 500 if (unresolved != null && unresolved.booleanValue() == true) { 501 return "${coord:dataIn('" + dataInName + "')}"; 502 } 503 return uris; 504 } 505 506 /** 507 * Used to specify a list of URI's that are output dir of the workflow job. <p/> Look for one evaluator-level 508 * variable <p/> dataout.<DATAOUT_NAME> <p/> It defines the current list of URI. <p/> otherwise it sends the uris. 509 * 510 * @param dataOutName : Dataout name 511 * @return the list of URI's separated by INSTANCE_SEPARATOR 512 */ 513 public static String ph3_coord_dataOut(String dataOutName) { 514 String uris = ""; 515 ELEvaluator eval = ELEvaluator.getCurrent(); 516 uris = (String) eval.getVariable(".dataout." + dataOutName); 517 return uris; 518 } 519 520 /** 521 * Determine the date-time in Oozie processing timezone of n-th dataset instance. <p/> It depends on: <p/> 1. 522 * Data set frequency <p/> 2. 523 * Data set Time unit (day, month, minute) <p/> 3. Data set Time zone/DST <p/> 4. End Day/Month flag <p/> 5. Data 524 * set initial instance <p/> 6. Action Creation Time 525 * 526 * @param n instance count domain: n is integer 527 * @return date-time in Oozie processing timezone of the n-th instance returns 'null' means n-th instance is 528 * earlier than Initial-Instance of DS 529 * @throws Exception 530 */ 531 public static String ph2_coord_current(int n) throws Exception { 532 if (isSyncDataSet()) { // For Sync Dataset 533 return coord_current_sync(n); 534 } 535 else { 536 throw new UnsupportedOperationException("Asynchronous Dataset is not supported yet"); 537 } 538 } 539 540 /** 541 * Determine the date-time in Oozie processing timezone of current dataset instances 542 * from start to end offsets from the nominal time. <p/> It depends 543 * on: <p/> 1. Data set frequency <p/> 2. Data set Time unit (day, month, minute) <p/> 3. Data set Time zone/DST 544 * <p/> 4. End Day/Month flag <p/> 5. Data set initial instance <p/> 6. Action Creation Time 545 * 546 * @param start :start instance offset <p/> domain: start <= 0, start is integer 547 * @param end :end instance offset <p/> domain: end <= 0, end is integer 548 * @return date-time in Oozie processing timezone of the instances from start to end offsets 549 * delimited by comma. <p/> If the current instance time of the dataset based on the Action Creation Time 550 * is earlier than the Initial-Instance of DS an empty string is returned. 551 * If an instance within the range is earlier than Initial-Instance of DS that instance is ignored 552 * @throws Exception 553 */ 554 public static String ph2_coord_currentRange(int start, int end) throws Exception { 555 if (isSyncDataSet()) { // For Sync Dataset 556 return coord_currentRange_sync(start, end); 557 } 558 else { 559 throw new UnsupportedOperationException("Asynchronous Dataset is not supported yet"); 560 } 561 } 562 /** 563 * Determine the date-time in Oozie processing timezone of the given offset from the dataset effective nominal time. <p/> It 564 * depends on: <p> 1. Data set frequency <p/> 2. Data set Time Unit <p/> 3. Data set Time zone/DST 565 * <p/> 4. Data set initial instance <p/> 5. Action Creation Time 566 * 567 * @param n offset amount (integer) 568 * @param timeUnit TimeUnit for offset n ("MINUTE", "HOUR", "DAY", "MONTH", "YEAR") 569 * @return date-time in Oozie processing timezone of the given offset from the dataset effective nominal time 570 * @throws Exception if there was a problem formatting 571 */ 572 public static String ph2_coord_offset(int n, String timeUnit) throws Exception { 573 if (isSyncDataSet()) { // For Sync Dataset 574 return coord_offset_sync(n, timeUnit); 575 } 576 else { 577 throw new UnsupportedOperationException("Asynchronous Dataset is not supported yet"); 578 } 579 } 580 581 /** 582 * Determine how many hours is on the date of n-th dataset instance. <p/> It depends on: <p/> 1. Data set frequency 583 * <p/> 2. Data set Time unit (day, month, minute) <p/> 3. Data set Time zone/DST <p/> 4. End Day/Month flag <p/> 5. 584 * Data set initial instance <p/> 6. Action Creation Time 585 * 586 * @param n instance count <p/> domain: n is integer 587 * @return number of hours on that day <p/> returns -1 means n-th instance is earlier than Initial-Instance of DS 588 * @throws Exception 589 */ 590 public static int ph2_coord_hoursInDay(int n) throws Exception { 591 int datasetFrequency = (int) getDSFrequency(); 592 // /Calendar nominalInstanceCal = 593 // getCurrentInstance(getActionCreationtime()); 594 Calendar nominalInstanceCal = getEffectiveNominalTime(); 595 if (nominalInstanceCal == null) { 596 return -1; 597 } 598 nominalInstanceCal.add(getDSTimeUnit().getCalendarUnit(), datasetFrequency * n); 599 /* 600 * if (nominalInstanceCal.getTime().compareTo(getInitialInstance()) < 0) 601 * { return -1; } 602 */ 603 nominalInstanceCal.setTimeZone(getDatasetTZ());// Use Dataset TZ 604 // DateUtils.moveToEnd(nominalInstanceCal, getDSEndOfFlag()); 605 return DateUtils.hoursInDay(nominalInstanceCal); 606 } 607 608 public static int ph3_coord_hoursInDay(int n) throws Exception { 609 return ph2_coord_hoursInDay(n); 610 } 611 612 /** 613 * Calculate number of days in one month for n-th dataset instance. <p/> It depends on: <p/> 1. Data set frequency . 614 * <p/> 2. Data set Time unit (day, month, minute) <p/> 3. Data set Time zone/DST <p/> 4. End Day/Month flag <p/> 5. 615 * Data set initial instance <p/> 6. Action Creation Time 616 * 617 * @param n instance count. domain: n is integer 618 * @return number of days in that month <p/> returns -1 means n-th instance is earlier than Initial-Instance of DS 619 * @throws Exception 620 */ 621 public static int ph2_coord_daysInMonth(int n) throws Exception { 622 int datasetFrequency = (int) getDSFrequency();// in minutes 623 // Calendar nominalInstanceCal = 624 // getCurrentInstance(getActionCreationtime()); 625 Calendar nominalInstanceCal = getEffectiveNominalTime(); 626 if (nominalInstanceCal == null) { 627 return -1; 628 } 629 nominalInstanceCal.add(getDSTimeUnit().getCalendarUnit(), datasetFrequency * n); 630 /* 631 * if (nominalInstanceCal.getTime().compareTo(getInitialInstance()) < 0) 632 * { return -1; } 633 */ 634 nominalInstanceCal.setTimeZone(getDatasetTZ());// Use Dataset TZ 635 // DateUtils.moveToEnd(nominalInstanceCal, getDSEndOfFlag()); 636 return nominalInstanceCal.getActualMaximum(Calendar.DAY_OF_MONTH); 637 } 638 639 public static int ph3_coord_daysInMonth(int n) throws Exception { 640 return ph2_coord_daysInMonth(n); 641 } 642 643 /** 644 * Determine the date-time in Oozie processing timezone of n-th latest available dataset instance. <p/> It depends 645 * on: <p/> 1. Data set frequency <p/> 2. Data set Time unit (day, month, minute) <p/> 3. Data set Time zone/DST 646 * <p/> 4. End Day/Month flag <p/> 5. Data set initial instance <p/> 6. Action Creation Time <p/> 7. Existence of 647 * dataset's directory 648 * 649 * @param n :instance count <p/> domain: n <= 0, n is integer 650 * @return date-time in Oozie processing timezone of the n-th instance <p/> returns 'null' means n-th instance is 651 * earlier than Initial-Instance of DS 652 * @throws Exception 653 */ 654 public static String ph3_coord_latest(int n) throws Exception { 655 ParamChecker.checkLEZero(n, "latest:n"); 656 if (isSyncDataSet()) {// For Sync Dataset 657 return coord_latest_sync(n); 658 } 659 else { 660 throw new UnsupportedOperationException("Asynchronous Dataset is not supported yet"); 661 } 662 } 663 664 /** 665 * Determine the date-time in Oozie processing timezone of latest available dataset instances 666 * from start to end offsets from the nominal time. <p/> It depends 667 * on: <p/> 1. Data set frequency <p/> 2. Data set Time unit (day, month, minute) <p/> 3. Data set Time zone/DST 668 * <p/> 4. End Day/Month flag <p/> 5. Data set initial instance <p/> 6. Action Creation Time <p/> 7. Existence of 669 * dataset's directory 670 * 671 * @param start :start instance offset <p/> domain: start <= 0, start is integer 672 * @param end :end instance offset <p/> domain: end <= 0, end is integer 673 * @return date-time in Oozie processing timezone of the instances from start to end offsets 674 * delimited by comma. <p/> returns 'null' means start offset instance is 675 * earlier than Initial-Instance of DS 676 * @throws Exception 677 */ 678 public static String ph3_coord_latestRange(int start, int end) throws Exception { 679 ParamChecker.checkLEZero(start, "latest:n"); 680 ParamChecker.checkLEZero(end, "latest:n"); 681 if (isSyncDataSet()) {// For Sync Dataset 682 return coord_latestRange_sync(start, end); 683 } 684 else { 685 throw new UnsupportedOperationException("Asynchronous Dataset is not supported yet"); 686 } 687 } 688 689 /** 690 * Configure an evaluator with data set and application specific information. <p/> Helper method of associating 691 * dataset and application object 692 * 693 * @param evaluator : to set variables 694 * @param ds : Data Set object 695 * @param coordAction : Application instance 696 */ 697 public static void configureEvaluator(ELEvaluator evaluator, SyncCoordDataset ds, SyncCoordAction coordAction) { 698 evaluator.setVariable(COORD_ACTION, coordAction); 699 evaluator.setVariable(DATASET, ds); 700 } 701 702 /** 703 * Helper method to wrap around with "${..}". <p/> 704 * 705 * 706 * @param eval :EL evaluator 707 * @param expr : expression to evaluate 708 * @return Resolved expression or echo back the same expression 709 * @throws Exception 710 */ 711 public static String evalAndWrap(ELEvaluator eval, String expr) throws Exception { 712 try { 713 eval.setVariable(".wrap", null); 714 String result = eval.evaluate(expr, String.class); 715 if (eval.getVariable(".wrap") != null) { 716 return "${" + result + "}"; 717 } 718 else { 719 return result; 720 } 721 } 722 catch (Exception e) { 723 throw new Exception("Unable to evaluate :" + expr + ":\n", e); 724 } 725 } 726 727 // Set of echo functions 728 729 public static String ph1_coord_current_echo(String n) { 730 return echoUnResolved("current", n); 731 } 732 733 public static String ph1_coord_absolute_echo(String date) { 734 return echoUnResolved("absolute", date); 735 } 736 737 public static String ph1_coord_currentRange_echo(String start, String end) { 738 return echoUnResolved("currentRange", start + ", " + end); 739 } 740 741 public static String ph1_coord_offset_echo(String n, String timeUnit) { 742 return echoUnResolved("offset", n + " , " + timeUnit); 743 } 744 745 public static String ph2_coord_current_echo(String n) { 746 return echoUnResolved("current", n); 747 } 748 749 public static String ph2_coord_currentRange_echo(String start, String end) { 750 return echoUnResolved("currentRange", start + ", " + end); 751 } 752 753 public static String ph2_coord_offset_echo(String n, String timeUnit) { 754 return echoUnResolved("offset", n + " , " + timeUnit); 755 } 756 757 public static String ph2_coord_absolute_echo(String date) { 758 return echoUnResolved("absolute", date); 759 } 760 761 public static String ph2_coord_absolute_range(String startInstance, int end) throws Exception { 762 int[] instanceCount = new int[1]; 763 764 // getCurrentInstance() returns null, which means startInstance is less 765 // than initial instance 766 if (getCurrentInstance(DateUtils.getCalendar(startInstance).getTime(), instanceCount) == null) { 767 throw new CommandException(ErrorCode.E1010, 768 "intial-instance should be equal or earlier than the start-instance. intial-instance is " 769 + getInitialInstance() + " and start-instance is " + startInstance); 770 } 771 int[] nominalCount = new int[1]; 772 if (getCurrentInstance(getActionCreationtime(), nominalCount) == null) { 773 throw new CommandException(ErrorCode.E1010, 774 "intial-instance should be equal or earlier than the nominal time. intial-instance is " 775 + getInitialInstance() + " and nominal time is " + getActionCreationtime()); 776 } 777 // getCurrentInstance return offset relative to initial instance. 778 // start instance offset - nominal offset = start offset relative to 779 // nominal time-stamp. 780 int start = instanceCount[0] - nominalCount[0]; 781 if (start > end) { 782 throw new CommandException(ErrorCode.E1010, 783 "start-instance should be equal or earlier than the end-instance. startInstance is " 784 + startInstance + " which is equivalent to current (" + instanceCount[0] 785 + ") but end is specified as current (" + end + ")"); 786 } 787 return ph2_coord_currentRange(start, end); 788 } 789 790 public static String ph1_coord_dateOffset_echo(String n, String offset, String unit) { 791 return echoUnResolved("dateOffset", n + " , " + offset + " , " + unit); 792 } 793 794 public static String ph1_coord_dateTzOffset_echo(String n, String timezone) { 795 return echoUnResolved("dateTzOffset", n + " , " + timezone); 796 } 797 798 public static String ph1_coord_formatTime_echo(String dateTime, String format) { 799 // Quote the dateTime value since it would contain a ':'. 800 return echoUnResolved("formatTime", "'"+dateTime+"'" + " , " + format); 801 } 802 803 public static String ph1_coord_latest_echo(String n) { 804 return echoUnResolved("latest", n); 805 } 806 807 public static String ph2_coord_latest_echo(String n) { 808 return ph1_coord_latest_echo(n); 809 } 810 811 public static String ph1_coord_future_echo(String n, String instance) { 812 return echoUnResolved("future", n + ", " + instance + ""); 813 } 814 815 public static String ph2_coord_future_echo(String n, String instance) { 816 return ph1_coord_future_echo(n, instance); 817 } 818 819 public static String ph1_coord_latestRange_echo(String start, String end) { 820 return echoUnResolved("latestRange", start + ", " + end); 821 } 822 823 public static String ph2_coord_latestRange_echo(String start, String end) { 824 return ph1_coord_latestRange_echo(start, end); 825 } 826 827 public static String ph1_coord_futureRange_echo(String start, String end, String instance) { 828 return echoUnResolved("futureRange", start + ", " + end + ", " + instance); 829 } 830 831 public static String ph2_coord_futureRange_echo(String start, String end, String instance) { 832 return ph1_coord_futureRange_echo(start, end, instance); 833 } 834 835 public static String ph1_coord_dataIn_echo(String n) { 836 ELEvaluator eval = ELEvaluator.getCurrent(); 837 String val = (String) eval.getVariable("oozie.dataname." + n); 838 if (val == null || val.equals("data-in") == false) { 839 XLog.getLog(CoordELFunctions.class).error("data_in_name " + n + " is not valid"); 840 throw new RuntimeException("data_in_name " + n + " is not valid"); 841 } 842 return echoUnResolved("dataIn", "'" + n + "'"); 843 } 844 845 public static String ph1_coord_dataOut_echo(String n) { 846 ELEvaluator eval = ELEvaluator.getCurrent(); 847 String val = (String) eval.getVariable("oozie.dataname." + n); 848 if (val == null || val.equals("data-out") == false) { 849 XLog.getLog(CoordELFunctions.class).error("data_out_name " + n + " is not valid"); 850 throw new RuntimeException("data_out_name " + n + " is not valid"); 851 } 852 return echoUnResolved("dataOut", "'" + n + "'"); 853 } 854 855 public static String ph1_coord_nominalTime_echo() { 856 return echoUnResolved("nominalTime", ""); 857 } 858 859 public static String ph1_coord_nominalTime_echo_wrap() { 860 // return "${coord:nominalTime()}"; // no resolution 861 return echoUnResolved("nominalTime", ""); 862 } 863 864 public static String ph1_coord_nominalTime_echo_fixed() { 865 return "2009-03-06T010:00"; // Dummy resolution 866 } 867 868 public static String ph1_coord_actualTime_echo_wrap() { 869 // return "${coord:actualTime()}"; // no resolution 870 return echoUnResolved("actualTime", ""); 871 } 872 873 public static String ph1_coord_actionId_echo() { 874 return echoUnResolved("actionId", ""); 875 } 876 877 public static String ph1_coord_name_echo() { 878 return echoUnResolved("name", ""); 879 } 880 881 // The following echo functions are not used in any phases yet 882 // They are here for future purpose. 883 public static String coord_minutes_echo(String n) { 884 return echoUnResolved("minutes", n); 885 } 886 887 public static String coord_hours_echo(String n) { 888 return echoUnResolved("hours", n); 889 } 890 891 public static String coord_days_echo(String n) { 892 return echoUnResolved("days", n); 893 } 894 895 public static String coord_endOfDay_echo(String n) { 896 return echoUnResolved("endOfDay", n); 897 } 898 899 public static String coord_months_echo(String n) { 900 return echoUnResolved("months", n); 901 } 902 903 public static String coord_endOfMonth_echo(String n) { 904 return echoUnResolved("endOfMonth", n); 905 } 906 907 public static String coord_actualTime_echo() { 908 return echoUnResolved("actualTime", ""); 909 } 910 911 // This echo function will always return "24" for validation only. 912 // This evaluation ****should not**** replace the original XML 913 // Create a temporary string and validate the function 914 // This is **required** for evaluating an expression like 915 // coord:HoursInDay(0) + 3 916 // actual evaluation will happen in phase 2 or phase 3. 917 public static String ph1_coord_hoursInDay_echo(String n) { 918 return "24"; 919 // return echoUnResolved("hoursInDay", n); 920 } 921 922 // This echo function will always return "30" for validation only. 923 // This evaluation ****should not**** replace the original XML 924 // Create a temporary string and validate the function 925 // This is **required** for evaluating an expression like 926 // coord:daysInMonth(0) + 3 927 // actual evaluation will happen in phase 2 or phase 3. 928 public static String ph1_coord_daysInMonth_echo(String n) { 929 // return echoUnResolved("daysInMonth", n); 930 return "30"; 931 } 932 933 // This echo function will always return "3" for validation only. 934 // This evaluation ****should not**** replace the original XML 935 // Create a temporary string and validate the function 936 // This is **required** for evaluating an expression like coord:tzOffset + 2 937 // actual evaluation will happen in phase 2 or phase 3. 938 public static String ph1_coord_tzOffset_echo() { 939 // return echoUnResolved("tzOffset", ""); 940 return "3"; 941 } 942 943 // Local methods 944 /** 945 * @param n 946 * @return n-th instance Date-Time from current instance for data-set <p/> return empty string ("") if the 947 * Action_Creation_time or the n-th instance <p/> is earlier than the Initial_Instance of dataset. 948 * @throws Exception 949 */ 950 private static String coord_current_sync(int n) throws Exception { 951 return coord_currentRange_sync(n, n); 952 } 953 954 private static String coord_currentRange_sync(int start, int end) throws Exception { 955 final XLog LOG = XLog.getLog(CoordELFunctions.class); 956 int datasetFrequency = getDSFrequency();// in minutes 957 TimeUnit dsTimeUnit = getDSTimeUnit(); 958 int[] instCount = new int[1];// used as pass by ref 959 Calendar nominalInstanceCal = getCurrentInstance(getActionCreationtime(), instCount); 960 if (nominalInstanceCal == null) { 961 LOG.warn("If the initial instance of the dataset is later than the nominal time, an empty string is" 962 + " returned. This means that no data is available at the current-instance specified by the user" 963 + " and the user could try modifying his initial-instance to an earlier time."); 964 return ""; 965 } else { 966 Calendar initInstance = getInitialInstanceCal(); 967 // Add in the reverse order - newest instance first. 968 nominalInstanceCal = (Calendar) initInstance.clone(); 969 nominalInstanceCal.add(dsTimeUnit.getCalendarUnit(), (instCount[0] + start) * datasetFrequency); 970 List<String> instances = new ArrayList<String>(); 971 for (int i = start; i <= end; i++) { 972 if (nominalInstanceCal.compareTo(initInstance) < 0) { 973 LOG.warn("If the initial instance of the dataset is later than the current-instance specified," 974 + " such as coord:current({0}) in this case, an empty string is returned. This means that" 975 + " no data is available at the current-instance specified by the user and the user could" 976 + " try modifying his initial-instance to an earlier time.", start); 977 } 978 else { 979 instances.add(DateUtils.formatDateOozieTZ(nominalInstanceCal)); 980 } 981 nominalInstanceCal.add(dsTimeUnit.getCalendarUnit(), datasetFrequency); 982 } 983 instances = Lists.reverse(instances); 984 return StringUtils.join(instances, CoordELFunctions.INSTANCE_SEPARATOR); 985 } 986 } 987 988 /** 989 * 990 * @param n offset amount (integer) 991 * @param timeUnit TimeUnit for offset n ("MINUTE", "HOUR", "DAY", "MONTH", "YEAR") 992 * @return the offset time from the effective nominal time <p/> return empty string ("") if the Action_Creation_time or the 993 * offset instance <p/> is earlier than the Initial_Instance of dataset. 994 * @throws Exception 995 */ 996 private static String coord_offset_sync(int n, String timeUnit) throws Exception { 997 Calendar rawCal = resolveOffsetRawTime(n, TimeUnit.valueOf(timeUnit), null); 998 if (rawCal == null) { 999 // warning already logged by resolveOffsetRawTime() 1000 return ""; 1001 } 1002 1003 int freq = getDSFrequency(); 1004 TimeUnit freqUnit = getDSTimeUnit(); 1005 int freqCount = 0; 1006 // We're going to manually turn back/forward cal by decrements/increments of freq and then check that it gives the same 1007 // time as rawCal; this is to check that the offset time resolves to a frequency offset of the effective nominal time 1008 // In other words, that there exists an integer x, such that coord:offset(n, timeUnit) == coord:current(x) is true 1009 // If not, then we'll "rewind" rawCal to the latest instance earlier than rawCal and use that. 1010 Calendar cal = getInitialInstanceCal(); 1011 if (rawCal.before(cal)) { 1012 while (cal.after(rawCal)) { 1013 cal.add(freqUnit.getCalendarUnit(), -freq); 1014 freqCount--; 1015 } 1016 } 1017 else if (rawCal.after(cal)) { 1018 while (cal.before(rawCal)) { 1019 cal.add(freqUnit.getCalendarUnit(), freq); 1020 freqCount++; 1021 } 1022 } 1023 if (cal.before(rawCal)) { 1024 rawCal = cal; 1025 } 1026 else if (cal.after(rawCal)) { 1027 cal.add(freqUnit.getCalendarUnit(), -freq); 1028 rawCal = cal; 1029 freqCount--; 1030 } 1031 String rawCalStr = DateUtils.formatDateOozieTZ(rawCal); 1032 1033 Calendar nominalInstanceCal = getInitialInstanceCal(); 1034 nominalInstanceCal.add(freqUnit.getCalendarUnit(), freq * freqCount); 1035 if (nominalInstanceCal.getTime().compareTo(getInitialInstance()) < 0) { 1036 XLog.getLog(CoordELFunctions.class).warn("If the initial instance of the dataset is later than the offset instance" 1037 + " specified, such as coord:offset({0}, {1}) in this case, an empty string is returned. This means that no" 1038 + " data is available at the offset instance specified by the user and the user could try modifying his" 1039 + " initial-instance to an earlier time.", n, timeUnit); 1040 return ""; 1041 } 1042 String nominalCalStr = DateUtils.formatDateOozieTZ(nominalInstanceCal); 1043 1044 if (!rawCalStr.equals(nominalCalStr)) { 1045 throw new RuntimeException("Shouldn't happen"); 1046 } 1047 return rawCalStr; 1048 } 1049 1050 /** 1051 * @param offset 1052 * @return n-th available latest instance Date-Time for SYNC data-set 1053 * @throws Exception 1054 */ 1055 private static String coord_latest_sync(int offset) throws Exception { 1056 return coord_latestRange_sync(offset, offset); 1057 } 1058 1059 private static String coord_latestRange_sync(int startOffset, int endOffset) throws Exception { 1060 final XLog LOG = XLog.getLog(CoordELFunctions.class); 1061 final Thread currentThread = Thread.currentThread(); 1062 ELEvaluator eval = ELEvaluator.getCurrent(); 1063 String retVal = ""; 1064 int datasetFrequency = (int) getDSFrequency();// in minutes 1065 TimeUnit dsTimeUnit = getDSTimeUnit(); 1066 int[] instCount = new int[1]; 1067 boolean useCurrentTime = Services.get().getConf().getBoolean(LATEST_EL_USE_CURRENT_TIME, false); 1068 Calendar nominalInstanceCal; 1069 if (useCurrentTime) { 1070 nominalInstanceCal = getCurrentInstance(new Date(), instCount); 1071 } 1072 else { 1073 nominalInstanceCal = getCurrentInstance(getActualTime(), instCount); 1074 } 1075 StringBuilder resolvedInstances = new StringBuilder(); 1076 StringBuilder resolvedURIPaths = new StringBuilder(); 1077 if (nominalInstanceCal != null) { 1078 Calendar initInstance = getInitialInstanceCal(); 1079 SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET); 1080 if (ds == null) { 1081 throw new RuntimeException("Associated Dataset should be defined with key " + DATASET); 1082 } 1083 String uriTemplate = ds.getUriTemplate(); 1084 Configuration conf = (Configuration) eval.getVariable(CONFIGURATION); 1085 if (conf == null) { 1086 throw new RuntimeException("Associated Configuration should be defined with key " + CONFIGURATION); 1087 } 1088 int available = 0; 1089 boolean resolved = false; 1090 String user = ParamChecker 1091 .notEmpty((String) eval.getVariable(OozieClient.USER_NAME), OozieClient.USER_NAME); 1092 String doneFlag = ds.getDoneFlag(); 1093 URIHandlerService uriService = Services.get().get(URIHandlerService.class); 1094 URIHandler uriHandler = null; 1095 Context uriContext = null; 1096 try { 1097 while (nominalInstanceCal.compareTo(initInstance) >= 0 && !currentThread.isInterrupted()) { 1098 ELEvaluator uriEval = getUriEvaluator(nominalInstanceCal); 1099 String uriPath = uriEval.evaluate(uriTemplate, String.class); 1100 if (uriHandler == null) { 1101 URI uri = new URI(uriPath); 1102 uriHandler = uriService.getURIHandler(uri); 1103 uriContext = uriHandler.getContext(uri, conf, user); 1104 } 1105 String uriWithDoneFlag = uriHandler.getURIWithDoneFlag(uriPath, doneFlag); 1106 if (uriHandler.exists(new URI(uriWithDoneFlag), uriContext)) { 1107 XLog.getLog(CoordELFunctions.class) 1108 .debug("Found latest(" + available + "): " + uriWithDoneFlag); 1109 if (available == startOffset) { 1110 LOG.debug("Matched latest(" + available + "): " + uriWithDoneFlag); 1111 resolved = true; 1112 resolvedInstances.append(DateUtils.formatDateOozieTZ(nominalInstanceCal)); 1113 resolvedURIPaths.append(uriPath); 1114 retVal = resolvedInstances.toString(); 1115 eval.setVariable("resolved_path", resolvedURIPaths.toString()); 1116 break; 1117 } 1118 else if (available <= endOffset) { 1119 LOG.debug("Matched latest(" + available + "): " + uriWithDoneFlag); 1120 resolvedInstances.append(DateUtils.formatDateOozieTZ(nominalInstanceCal)).append( 1121 INSTANCE_SEPARATOR); 1122 resolvedURIPaths.append(uriPath).append(INSTANCE_SEPARATOR); 1123 } 1124 1125 available--; 1126 } 1127 // nominalInstanceCal.add(dsTimeUnit.getCalendarUnit(), -datasetFrequency); 1128 nominalInstanceCal = (Calendar) initInstance.clone(); 1129 instCount[0]--; 1130 nominalInstanceCal.add(dsTimeUnit.getCalendarUnit(), instCount[0] * datasetFrequency); 1131 // DateUtils.moveToEnd(nominalInstanceCal, getDSEndOfFlag()); 1132 } 1133 } 1134 finally { 1135 if (uriContext != null) { 1136 uriContext.destroy(); 1137 } 1138 } 1139 if (!resolved) { 1140 // return unchanged latest function with variable 'is_resolved' 1141 // to 'false' 1142 eval.setVariable("is_resolved", Boolean.FALSE); 1143 if (startOffset == endOffset) { 1144 retVal = "${coord:latest(" + startOffset + ")}"; 1145 } 1146 else { 1147 retVal = "${coord:latestRange(" + startOffset + "," + endOffset + ")}"; 1148 } 1149 } 1150 else { 1151 eval.setVariable("is_resolved", Boolean.TRUE); 1152 } 1153 } 1154 else {// No feasible nominal time 1155 eval.setVariable("is_resolved", Boolean.FALSE); 1156 } 1157 return retVal; 1158 } 1159 1160 /** 1161 * @param tm 1162 * @return a new Evaluator to be used for URI-template evaluation 1163 */ 1164 private static ELEvaluator getUriEvaluator(Calendar tm) { 1165 tm.setTimeZone(DateUtils.getOozieProcessingTimeZone()); 1166 ELEvaluator retEval = new ELEvaluator(); 1167 retEval.setVariable("YEAR", tm.get(Calendar.YEAR)); 1168 retEval.setVariable("MONTH", (tm.get(Calendar.MONTH) + 1) < 10 ? "0" + (tm.get(Calendar.MONTH) + 1) : (tm 1169 .get(Calendar.MONTH) + 1)); 1170 retEval.setVariable("DAY", tm.get(Calendar.DAY_OF_MONTH) < 10 ? "0" + tm.get(Calendar.DAY_OF_MONTH) : tm 1171 .get(Calendar.DAY_OF_MONTH)); 1172 retEval.setVariable("HOUR", tm.get(Calendar.HOUR_OF_DAY) < 10 ? "0" + tm.get(Calendar.HOUR_OF_DAY) : tm 1173 .get(Calendar.HOUR_OF_DAY)); 1174 retEval.setVariable("MINUTE", tm.get(Calendar.MINUTE) < 10 ? "0" + tm.get(Calendar.MINUTE) : tm 1175 .get(Calendar.MINUTE)); 1176 return retEval; 1177 } 1178 1179 /** 1180 * @return whether a data set is SYNCH or ASYNC 1181 */ 1182 private static boolean isSyncDataSet() { 1183 ELEvaluator eval = ELEvaluator.getCurrent(); 1184 SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET); 1185 if (ds == null) { 1186 throw new RuntimeException("Associated Dataset should be defined with key " + DATASET); 1187 } 1188 return ds.getType().equalsIgnoreCase("SYNC"); 1189 } 1190 1191 /** 1192 * Check whether a function should be resolved. 1193 * 1194 * @param functionName 1195 * @param n 1196 * @return null if the functionName needs to be resolved otherwise return the calling function unresolved. 1197 */ 1198 private static String checkIfResolved(String functionName, String n) { 1199 ELEvaluator eval = ELEvaluator.getCurrent(); 1200 String replace = (String) eval.getVariable("resolve_" + functionName); 1201 if (replace == null || (replace != null && replace.equalsIgnoreCase("false"))) { // Don't 1202 // resolve 1203 // return "${coord:" + functionName + "(" + n +")}"; //Unresolved 1204 eval.setVariable(".wrap", "true"); 1205 return "coord:" + functionName + "(" + n + ")"; // Unresolved 1206 } 1207 return null; // Resolved it 1208 } 1209 1210 private static String echoUnResolved(String functionName, String n) { 1211 return echoUnResolvedPre(functionName, n, "coord:"); 1212 } 1213 1214 private static String echoUnResolvedPre(String functionName, String n, String prefix) { 1215 ELEvaluator eval = ELEvaluator.getCurrent(); 1216 eval.setVariable(".wrap", "true"); 1217 return prefix + functionName + "(" + n + ")"; // Unresolved 1218 } 1219 1220 /** 1221 * @return the initial instance of a DataSet in DATE 1222 */ 1223 private static Date getInitialInstance() { 1224 ELEvaluator eval = ELEvaluator.getCurrent(); 1225 return getInitialInstance(eval); 1226 } 1227 1228 /** 1229 * @return the initial instance of a DataSet in DATE 1230 */ 1231 private static Date getInitialInstance(ELEvaluator eval) { 1232 return getInitialInstanceCal(eval).getTime(); 1233 // return ds.getInitInstance(); 1234 } 1235 1236 /** 1237 * @return the initial instance of a DataSet in Calendar 1238 */ 1239 private static Calendar getInitialInstanceCal() { 1240 ELEvaluator eval = ELEvaluator.getCurrent(); 1241 return getInitialInstanceCal(eval); 1242 } 1243 1244 /** 1245 * @return the initial instance of a DataSet in Calendar 1246 */ 1247 private static Calendar getInitialInstanceCal(ELEvaluator eval) { 1248 SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET); 1249 if (ds == null) { 1250 throw new RuntimeException("Associated Dataset should be defined with key " + DATASET); 1251 } 1252 Calendar effInitTS = new GregorianCalendar(ds.getTimeZone()); 1253 effInitTS.setTime(ds.getInitInstance()); 1254 // To adjust EOD/EOM 1255 DateUtils.moveToEnd(effInitTS, getDSEndOfFlag(eval)); 1256 return effInitTS; 1257 // return ds.getInitInstance(); 1258 } 1259 1260 /** 1261 * @return Nominal or action creation Time when all the dependencies of an application instance are met. 1262 */ 1263 private static Date getActionCreationtime() { 1264 ELEvaluator eval = ELEvaluator.getCurrent(); 1265 return getActionCreationtime(eval); 1266 } 1267 1268 /** 1269 * @return Nominal or action creation Time when all the dependencies of an application instance are met. 1270 */ 1271 private static Date getActionCreationtime(ELEvaluator eval) { 1272 SyncCoordAction coordAction = (SyncCoordAction) eval.getVariable(COORD_ACTION); 1273 if (coordAction == null) { 1274 throw new RuntimeException("Associated Application instance should be defined with key " + COORD_ACTION); 1275 } 1276 return coordAction.getNominalTime(); 1277 } 1278 1279 /** 1280 * @return Actual Time when all the dependencies of an application instance are met. 1281 */ 1282 private static Date getActualTime() { 1283 ELEvaluator eval = ELEvaluator.getCurrent(); 1284 SyncCoordAction coordAction = (SyncCoordAction) eval.getVariable(COORD_ACTION); 1285 if (coordAction == null) { 1286 throw new RuntimeException("Associated Application instance should be defined with key " + COORD_ACTION); 1287 } 1288 return coordAction.getActualTime(); 1289 } 1290 1291 /** 1292 * @return TimeZone for the application or job. 1293 */ 1294 private static TimeZone getJobTZ() { 1295 ELEvaluator eval = ELEvaluator.getCurrent(); 1296 SyncCoordAction coordAction = (SyncCoordAction) eval.getVariable(COORD_ACTION); 1297 if (coordAction == null) { 1298 throw new RuntimeException("Associated Application instance should be defined with key " + COORD_ACTION); 1299 } 1300 return coordAction.getTimeZone(); 1301 } 1302 1303 /** 1304 * Find the current instance based on effectiveTime (i.e Action_Creation_Time or Action_Start_Time) 1305 * 1306 * @return current instance i.e. current(0) returns null if effectiveTime is earlier than Initial Instance time of 1307 * the dataset. 1308 */ 1309 public static Calendar getCurrentInstance(Date effectiveTime, int instanceCount[]) { 1310 ELEvaluator eval = ELEvaluator.getCurrent(); 1311 return getCurrentInstance(effectiveTime, instanceCount, eval); 1312 } 1313 1314 /** 1315 * Find the current instance based on effectiveTime (i.e Action_Creation_Time or Action_Start_Time) 1316 * 1317 * @return current instance i.e. current(0) returns null if effectiveTime is earlier than Initial Instance time of 1318 * the dataset. 1319 */ 1320 private static Calendar getCurrentInstance(Date effectiveTime, int instanceCount[], ELEvaluator eval) { 1321 Date datasetInitialInstance = getInitialInstance(eval); 1322 TimeUnit dsTimeUnit = getDSTimeUnit(eval); 1323 TimeZone dsTZ = getDatasetTZ(eval); 1324 int dsFreq = getDSFrequency(eval); 1325 // Convert Date to Calendar for corresponding TZ 1326 Calendar current = Calendar.getInstance(dsTZ); 1327 current.setTime(datasetInitialInstance); 1328 1329 Calendar calEffectiveTime = new GregorianCalendar(dsTZ); 1330 calEffectiveTime.setTime(effectiveTime); 1331 if (instanceCount == null) { // caller doesn't care about this value 1332 instanceCount = new int[1]; 1333 } 1334 instanceCount[0] = 0; 1335 if (current.compareTo(calEffectiveTime) > 0) { 1336 return null; 1337 } 1338 1339 switch(dsTimeUnit) { 1340 case MINUTE: 1341 instanceCount[0] = (int) ((effectiveTime.getTime() - datasetInitialInstance.getTime()) / MINUTE_MSEC); 1342 break; 1343 case HOUR: 1344 instanceCount[0] = (int) ((effectiveTime.getTime() - datasetInitialInstance.getTime()) / HOUR_MSEC); 1345 break; 1346 case DAY: 1347 case END_OF_DAY: 1348 instanceCount[0] = (int) ((effectiveTime.getTime() - datasetInitialInstance.getTime()) / DAY_MSEC); 1349 break; 1350 case MONTH: 1351 case END_OF_MONTH: 1352 instanceCount[0] = (int) ((effectiveTime.getTime() - datasetInitialInstance.getTime()) / MONTH_MSEC); 1353 break; 1354 case YEAR: 1355 instanceCount[0] = (int) ((effectiveTime.getTime() - datasetInitialInstance.getTime()) / YEAR_MSEC); 1356 break; 1357 default: 1358 throw new IllegalArgumentException("Unhandled dataset time unit " + dsTimeUnit); 1359 } 1360 1361 if (instanceCount[0] > 2) { 1362 instanceCount[0] = (instanceCount[0] / dsFreq); 1363 current.add(dsTimeUnit.getCalendarUnit(), instanceCount[0] * dsFreq); 1364 } else { 1365 instanceCount[0] = 0; 1366 } 1367 while (!current.getTime().after(effectiveTime)) { 1368 current.add(dsTimeUnit.getCalendarUnit(), dsFreq); 1369 instanceCount[0]++; 1370 } 1371 current.add(dsTimeUnit.getCalendarUnit(), -dsFreq); 1372 instanceCount[0]--; 1373 return current; 1374 } 1375 1376 /** 1377 * Find the current instance based on effectiveTime (i.e Action_Creation_Time or Action_Start_Time) 1378 * 1379 * @return current instance i.e. current(0) returns null if effectiveTime is earlier than Initial Instance time of 1380 * the dataset. 1381 */ 1382 private static Calendar getCurrentInstance_old(Date effectiveTime, int instanceCount[], ELEvaluator eval) { 1383 Date datasetInitialInstance = getInitialInstance(eval); 1384 TimeUnit dsTimeUnit = getDSTimeUnit(eval); 1385 TimeZone dsTZ = getDatasetTZ(eval); 1386 int dsFreq = getDSFrequency(eval); 1387 // Convert Date to Calendar for corresponding TZ 1388 Calendar current = Calendar.getInstance(); 1389 current.setTime(datasetInitialInstance); 1390 current.setTimeZone(dsTZ); 1391 1392 Calendar calEffectiveTime = Calendar.getInstance(); 1393 calEffectiveTime.setTime(effectiveTime); 1394 calEffectiveTime.setTimeZone(dsTZ); 1395 if (instanceCount == null) { // caller doesn't care about this value 1396 instanceCount = new int[1]; 1397 } 1398 instanceCount[0] = 0; 1399 if (current.compareTo(calEffectiveTime) > 0) { 1400 return null; 1401 } 1402 Calendar origCurrent = (Calendar) current.clone(); 1403 while (current.compareTo(calEffectiveTime) <= 0) { 1404 current = (Calendar) origCurrent.clone(); 1405 instanceCount[0]++; 1406 current.add(dsTimeUnit.getCalendarUnit(), instanceCount[0] * dsFreq); 1407 } 1408 instanceCount[0]--; 1409 1410 current = (Calendar) origCurrent.clone(); 1411 current.add(dsTimeUnit.getCalendarUnit(), instanceCount[0] * dsFreq); 1412 return current; 1413 } 1414 1415 public static Calendar getEffectiveNominalTime() { 1416 Date datasetInitialInstance = getInitialInstance(); 1417 TimeZone dsTZ = getDatasetTZ(); 1418 // Convert Date to Calendar for corresponding TZ 1419 Calendar current = Calendar.getInstance(); 1420 current.setTime(datasetInitialInstance); 1421 current.setTimeZone(dsTZ); 1422 1423 Calendar calEffectiveTime = Calendar.getInstance(); 1424 calEffectiveTime.setTime(getActionCreationtime()); 1425 calEffectiveTime.setTimeZone(dsTZ); 1426 if (current.compareTo(calEffectiveTime) > 0) { 1427 // Nominal Time < initial Instance 1428 // TODO: getClass() call doesn't work from static method. 1429 // XLog.getLog("CoordELFunction.class").warn("ACTION CREATED BEFORE INITIAL INSTACE "+ 1430 // current.getTime()); 1431 return null; 1432 } 1433 return calEffectiveTime; 1434 } 1435 1436 /** 1437 * @return dataset frequency in minutes 1438 */ 1439 private static int getDSFrequency() { 1440 ELEvaluator eval = ELEvaluator.getCurrent(); 1441 return getDSFrequency(eval); 1442 } 1443 1444 /** 1445 * @return dataset frequency in minutes 1446 */ 1447 private static int getDSFrequency(ELEvaluator eval) { 1448 SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET); 1449 if (ds == null) { 1450 throw new RuntimeException("Associated Dataset should be defined with key " + DATASET); 1451 } 1452 return ds.getFrequency(); 1453 } 1454 1455 /** 1456 * @return dataset TimeUnit 1457 */ 1458 private static TimeUnit getDSTimeUnit() { 1459 ELEvaluator eval = ELEvaluator.getCurrent(); 1460 return getDSTimeUnit(eval); 1461 } 1462 1463 /** 1464 * @return dataset TimeUnit 1465 */ 1466 public static TimeUnit getDSTimeUnit(ELEvaluator eval) { 1467 SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET); 1468 if (ds == null) { 1469 throw new RuntimeException("Associated Dataset should be defined with key " + DATASET); 1470 } 1471 return ds.getTimeUnit(); 1472 } 1473 1474 /** 1475 * @return dataset TimeZone 1476 */ 1477 public static TimeZone getDatasetTZ() { 1478 ELEvaluator eval = ELEvaluator.getCurrent(); 1479 return getDatasetTZ(eval); 1480 } 1481 1482 /** 1483 * @return dataset TimeZone 1484 */ 1485 private static TimeZone getDatasetTZ(ELEvaluator eval) { 1486 SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET); 1487 if (ds == null) { 1488 throw new RuntimeException("Associated Dataset should be defined with key " + DATASET); 1489 } 1490 return ds.getTimeZone(); 1491 } 1492 1493 /** 1494 * @return dataset TimeUnit 1495 */ 1496 private static TimeUnit getDSEndOfFlag() { 1497 ELEvaluator eval = ELEvaluator.getCurrent(); 1498 return getDSEndOfFlag(eval); 1499 } 1500 1501 /** 1502 * @return dataset TimeUnit 1503 */ 1504 private static TimeUnit getDSEndOfFlag(ELEvaluator eval) { 1505 SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET); 1506 if (ds == null) { 1507 throw new RuntimeException("Associated Dataset should be defined with key " + DATASET); 1508 } 1509 return ds.getEndOfDuration();// == null ? "": ds.getEndOfDuration(); 1510 } 1511 1512 /** 1513 * Return a job configuration property for the coordinator. 1514 * 1515 * @param property property name. 1516 * @return the value of the property, <code>null</code> if the property is undefined. 1517 */ 1518 public static String coord_conf(String property) { 1519 ELEvaluator eval = ELEvaluator.getCurrent(); 1520 return (String) eval.getVariable(property); 1521 } 1522 1523 /** 1524 * Return the user that submitted the coordinator job. 1525 * 1526 * @return the user that submitted the coordinator job. 1527 */ 1528 public static String coord_user() { 1529 ELEvaluator eval = ELEvaluator.getCurrent(); 1530 return (String) eval.getVariable(OozieClient.USER_NAME); 1531 } 1532 1533 /** 1534 * Takes two offset times and returns a list of multiples of the frequency offset from the effective nominal time that occur 1535 * between them. The caller should make sure that startCal is earlier than endCal. 1536 * <p> 1537 * As a simple example, assume its the same day: startCal is 1:00, endCal is 2:00, frequency is 20min, and effective nominal 1538 * time is 1:20 -- then this method would return a list containing: -20, 0, 20, 40, 60 1539 * 1540 * @param startCal The earlier offset time 1541 * @param endCal The later offset time 1542 * @param eval The ELEvaluator to use; cannot be null 1543 * @return A list of multiple of the frequency offset from the effective nominal time that occur between the startCal and endCal 1544 */ 1545 public static List<Integer> expandOffsetTimes(Calendar startCal, Calendar endCal, ELEvaluator eval) { 1546 List<Integer> expandedFreqs = new ArrayList<Integer>(); 1547 // Use eval because the "current" eval isn't set 1548 int freq = getDSFrequency(eval); 1549 TimeUnit freqUnit = getDSTimeUnit(eval); 1550 Calendar cal = getCurrentInstance(getActionCreationtime(eval), null, eval); 1551 int totalFreq = 0; 1552 if (startCal.before(cal)) { 1553 while (cal.after(startCal)) { 1554 cal.add(freqUnit.getCalendarUnit(), -freq); 1555 totalFreq += -freq; 1556 } 1557 if (cal.before(startCal)) { 1558 cal.add(freqUnit.getCalendarUnit(), freq); 1559 totalFreq += freq; 1560 } 1561 } 1562 else if (startCal.after(cal)) { 1563 while (cal.before(startCal)) { 1564 cal.add(freqUnit.getCalendarUnit(), freq); 1565 totalFreq += freq; 1566 } 1567 } 1568 // At this point, cal is the smallest multiple of the dataset frequency that is >= to the startCal and offset from the 1569 // effective nominal time. Now we can find all of the instances that occur between startCal and endCal, inclusive. 1570 while (cal.before(endCal) || cal.equals(endCal)) { 1571 expandedFreqs.add(totalFreq); 1572 cal.add(freqUnit.getCalendarUnit(), freq); 1573 totalFreq += freq; 1574 } 1575 return expandedFreqs; 1576 } 1577 1578 /** 1579 * Resolve the offset time from the effective nominal time 1580 * 1581 * @param n offset amount (integer) 1582 * @param timeUnit TimeUnit for offset n ("MINUTE", "HOUR", "DAY", "MONTH", "YEAR") 1583 * @param eval The ELEvaluator to use; or null to use the "current" eval 1584 * @return A Calendar of the offset time 1585 */ 1586 public static Calendar resolveOffsetRawTime(int n, TimeUnit timeUnit, ELEvaluator eval) { 1587 // Use eval if given (for when the "current" eval isn't set) 1588 Calendar cal; 1589 if (eval == null) { 1590 cal = getCurrentInstance(getActionCreationtime(), null); 1591 } 1592 else { 1593 cal = getCurrentInstance(getActionCreationtime(eval), null, eval); 1594 } 1595 if (cal == null) { 1596 XLog.getLog(CoordELFunctions.class).warn("If the initial instance of the dataset is later than the nominal time, an" 1597 + " empty string is returned. This means that no data is available at the offset instance specified by the user" 1598 + " and the user could try modifying his or her initial-instance to an earlier time."); 1599 return null; 1600 } 1601 cal.add(timeUnit.getCalendarUnit(), n); 1602 return cal; 1603 } 1604}