001 /** 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018 package org.apache.oozie.command.coord; 019 020 import java.io.IOException; 021 import java.io.StringReader; 022 import java.util.ArrayList; 023 import java.util.Calendar; 024 import java.util.Date; 025 import java.util.List; 026 import java.util.TimeZone; 027 028 import org.apache.hadoop.conf.Configuration; 029 import org.apache.oozie.AppType; 030 import org.apache.oozie.CoordinatorActionBean; 031 import org.apache.oozie.CoordinatorJobBean; 032 import org.apache.oozie.ErrorCode; 033 import org.apache.oozie.SLAEventBean; 034 import org.apache.oozie.client.CoordinatorJob; 035 import org.apache.oozie.client.SLAEvent.SlaAppType; 036 import org.apache.oozie.client.rest.JsonBean; 037 import org.apache.oozie.command.CommandException; 038 import org.apache.oozie.coord.TimeUnit; 039 import org.apache.oozie.executor.jpa.BulkUpdateInsertJPAExecutor; 040 import org.apache.oozie.executor.jpa.CoordJobGetJPAExecutor; 041 import org.apache.oozie.executor.jpa.JPAExecutorException; 042 import org.apache.oozie.service.JPAService; 043 import org.apache.oozie.service.Service; 044 import org.apache.oozie.service.Services; 045 import org.apache.oozie.store.CoordinatorStore; 046 import org.apache.oozie.store.StoreException; 047 import org.apache.oozie.util.DateUtils; 048 import org.apache.oozie.util.Instrumentation; 049 import org.apache.oozie.sla.SLAOperations; 050 import org.apache.oozie.util.XConfiguration; 051 import org.apache.oozie.util.XLog; 052 import org.apache.oozie.util.XmlUtils; 053 import org.apache.oozie.util.db.SLADbOperations; 054 import org.jdom.Element; 055 056 @SuppressWarnings("deprecation") 057 public class CoordActionMaterializeCommand extends CoordinatorCommand<Void> { 058 private String jobId; 059 private Date startTime; 060 private Date endTime; 061 private int lastActionNumber = 1; // over-ride by DB value 062 private final XLog log = XLog.getLog(getClass()); 063 private String user; 064 private String group; 065 private List<JsonBean> insertList = new ArrayList<JsonBean>(); 066 private List<JsonBean> updateList = new ArrayList<JsonBean>(); 067 068 /** 069 * Default timeout for catchup jobs, in minutes, after which coordinator input check will timeout 070 */ 071 public static final String CONF_DEFAULT_TIMEOUT_CATCHUP = Service.CONF_PREFIX + "coord.catchup.default.timeout"; 072 073 public CoordActionMaterializeCommand(String jobId, Date startTime, Date endTime) { 074 super("coord_action_mater", "coord_action_mater", 1, XLog.STD, false); 075 this.jobId = jobId; 076 this.startTime = startTime; 077 this.endTime = endTime; 078 } 079 080 @Override 081 protected Void call(CoordinatorStore store) throws CommandException { 082 CoordJobGetJPAExecutor getCoordJob = new CoordJobGetJPAExecutor(jobId); 083 CoordinatorJobBean job; 084 try { 085 job = Services.get().get(JPAService.class).execute(getCoordJob); 086 } 087 catch (JPAExecutorException jex) { 088 throw new CommandException(jex); 089 } 090 setLogInfo(job); 091 if (job.getLastActionTime() != null && job.getLastActionTime().compareTo(endTime) >= 0) { 092 log.info("ENDED Coordinator materialization for jobId = " + jobId 093 + " Action is *already* materialized for Materialization start time = " + startTime + " : Materialization end time = " + endTime + " Job status = " + job.getStatusStr()); 094 return null; 095 } 096 097 if (endTime.after(job.getEndTime())) { 098 log.info("ENDED Coordinator materialization for jobId = " + jobId + " Materialization end time = " + endTime 099 + " surpasses coordinator job's end time = " + job.getEndTime() + " Job status = " + job.getStatusStr()); 100 return null; 101 } 102 103 if (job.getPauseTime() != null && !startTime.before(job.getPauseTime())) { 104 log.info("ENDED Coordinator materialization for jobId = " + jobId + " Materialization start time = " + startTime 105 + " is after or equal to coordinator job's pause time = " + job.getPauseTime() + " Job status = " + job.getStatusStr()); 106 // pausetime blocks real materialization - we change job's status back to RUNNING; 107 if (job.getStatus() == CoordinatorJob.Status.PREMATER) { 108 job.setStatus(CoordinatorJob.Status.RUNNING); 109 } 110 updateList.add(job); 111 return null; 112 } 113 114 this.user = job.getUser(); 115 this.group = job.getGroup(); 116 117 if (job.getStatus().equals(CoordinatorJobBean.Status.PREMATER)) { 118 Configuration jobConf = null; 119 log.debug("start job :" + jobId + " Materialization "); 120 try { 121 jobConf = new XConfiguration(new StringReader(job.getConf())); 122 } 123 catch (IOException ioe) { 124 log.warn("Configuration parse error. read from DB :" + job.getConf(), ioe); 125 throw new CommandException(ErrorCode.E1005, ioe.getMessage(), ioe); 126 } 127 128 Instrumentation.Cron cron = new Instrumentation.Cron(); 129 cron.start(); 130 try { 131 materializeJobs(false, job, jobConf, store); 132 updateJobTable(job, store); 133 } 134 catch (CommandException ex) { 135 log.warn("Exception occurs:" + ex + " Making the job failed "); 136 job.setStatus(CoordinatorJobBean.Status.FAILED); 137 updateList.add(job); 138 } 139 catch (Exception e) { 140 log.error("Excepion thrown :", e); 141 throw new CommandException(ErrorCode.E1001, e.getMessage(), e); 142 } 143 cron.stop(); 144 } 145 else { 146 log.info("WARN: action is not in PREMATER state! It's in state=" + job.getStatus()); 147 } 148 return null; 149 } 150 151 /** 152 * Create action instances starting from "start-time" to end-time" and store them into Action table. 153 * 154 * @param dryrun 155 * @param jobBean 156 * @param conf 157 * @param store 158 * @throws Exception 159 */ 160 protected String materializeJobs(boolean dryrun, CoordinatorJobBean jobBean, Configuration conf, 161 CoordinatorStore store) throws Exception { 162 String jobXml = jobBean.getJobXml(); 163 Element eJob = XmlUtils.parseXml(jobXml); 164 // TODO: always UTC? 165 TimeZone appTz = DateUtils.getTimeZone(jobBean.getTimeZone()); 166 // TimeZone appTz = DateUtils.getTimeZone("UTC"); 167 int frequency = Integer.valueOf(jobBean.getFrequency()); 168 TimeUnit freqTU = TimeUnit.valueOf(eJob.getAttributeValue("freq_timeunit")); 169 TimeUnit endOfFlag = TimeUnit.valueOf(eJob.getAttributeValue("end_of_duration")); 170 Calendar start = Calendar.getInstance(appTz); 171 start.setTime(startTime); 172 DateUtils.moveToEnd(start, endOfFlag); 173 Calendar end = Calendar.getInstance(appTz); 174 end.setTime(endTime); 175 lastActionNumber = jobBean.getLastActionNumber(); 176 // DateUtils.moveToEnd(end, endOfFlag); 177 log.info(" *** materialize Actions for tz=" + appTz.getDisplayName() + ",\n start=" + start.getTime() 178 + ", end=" + end.getTime() + "\n TimeUNIT " + freqTU.getCalendarUnit() + " Frequency :" + frequency 179 + ":" + freqTU + " lastActionNumber " + lastActionNumber); 180 // Keep the actual start time 181 Calendar origStart = Calendar.getInstance(appTz); 182 origStart.setTime(jobBean.getStartTimestamp()); 183 // Move to the End of duration, if needed. 184 DateUtils.moveToEnd(origStart, endOfFlag); 185 // Cloning the start time to be used in loop iteration 186 Calendar effStart = (Calendar) origStart.clone(); 187 // Move the time when the previous action finished 188 effStart.add(freqTU.getCalendarUnit(), lastActionNumber * frequency); 189 190 String action = null; 191 StringBuilder actionStrings = new StringBuilder(); 192 Date jobPauseTime = jobBean.getPauseTime(); 193 Calendar pause = null; 194 if (jobPauseTime != null) { 195 pause = Calendar.getInstance(appTz); 196 pause.setTime(DateUtils.convertDateToTimestamp(jobPauseTime)); 197 } 198 199 while (effStart.compareTo(end) < 0) { 200 if (pause != null && effStart.compareTo(pause) >= 0) { 201 break; 202 } 203 CoordinatorActionBean actionBean = new CoordinatorActionBean(); 204 lastActionNumber++; 205 206 int timeout = jobBean.getTimeout(); 207 log.debug(origStart.getTime() + " Materializing action for time=" + effStart.getTime() 208 + ", lastactionnumber=" + lastActionNumber); 209 Date actualTime = new Date(); 210 action = CoordCommandUtils.materializeOneInstance(jobId, dryrun, (Element) eJob.clone(), 211 effStart.getTime(), actualTime, lastActionNumber, conf, actionBean); 212 int catchUpTOMultiplier = 1; // This value might be could be changed in future 213 if (actionBean.getNominalTimestamp().before(jobBean.getCreatedTimestamp())) { 214 // Catchup action 215 timeout = catchUpTOMultiplier * timeout; 216 // actionBean.setTimeOut(Services.get().getConf().getInt(CONF_DEFAULT_TIMEOUT_CATCHUP, 217 // -1)); 218 log.info("Catchup timeout is :" + actionBean.getTimeOut()); 219 } 220 actionBean.setTimeOut(timeout); 221 222 if (!dryrun) { 223 storeToDB(actionBean, action, store, jobBean.getAppName()); // Storing to table 224 } 225 else { 226 actionStrings.append("action for new instance"); 227 actionStrings.append(action); 228 } 229 // Restore the original start time 230 effStart = (Calendar) origStart.clone(); 231 effStart.add(freqTU.getCalendarUnit(), lastActionNumber * frequency); 232 } 233 234 endTime = new Date(effStart.getTimeInMillis()); 235 if (!dryrun) { 236 return action; 237 } 238 else { 239 return actionStrings.toString(); 240 } 241 } 242 243 /** 244 * Store an Action into database table. 245 * 246 * @param actionBean 247 * @param actionXml 248 * @param store 249 * @param appName 250 * @throws Exception 251 */ 252 private void storeToDB(CoordinatorActionBean actionBean, String actionXml, CoordinatorStore store, String appName) 253 throws Exception { 254 log.debug("In storeToDB() action Id " + actionBean.getId() + " Size of actionXml " + actionXml.length()); 255 actionBean.setActionXml(actionXml); 256 insertList.add(actionBean); 257 createActionRegistration(actionXml, actionBean, store, appName); 258 259 // TODO: time 100s should be configurable 260 queueCallable(new CoordActionNotificationXCommand(actionBean), 100); 261 queueCallable(new CoordActionInputCheckXCommand(actionBean.getId(), actionBean.getJobId()), 100); 262 } 263 264 /** 265 * @param actionXml 266 * @param actionBean 267 * @param store 268 * @param appName 269 * @throws Exception 270 */ 271 private void createActionRegistration(String actionXml, CoordinatorActionBean actionBean, CoordinatorStore store, 272 String appName) throws Exception { 273 Element eAction = XmlUtils.parseXml(actionXml); 274 Element eSla = eAction.getChild("action", eAction.getNamespace()).getChild("info", eAction.getNamespace("sla")); 275 SLAEventBean slaEvent = SLADbOperations.createSlaRegistrationEvent(eSla, store, actionBean.getId(), 276 SlaAppType.COORDINATOR_ACTION, user, group); 277 if(slaEvent != null) { 278 insertList.add(slaEvent); 279 } 280 // insert into new sla reg table too 281 SLAOperations.createSlaRegistrationEvent(eSla, actionBean.getId(), actionBean.getJobId(), 282 AppType.COORDINATOR_ACTION, user, appName, log, false); 283 } 284 285 /** 286 * @param job 287 * @param store 288 * @throws StoreException 289 */ 290 private void updateJobTable(CoordinatorJobBean job, CoordinatorStore store) { 291 // TODO: why do we need this? Isn't lastMatTime enough??? 292 job.setLastActionTime(endTime); 293 job.setLastActionNumber(lastActionNumber); 294 // if the job endtime == action endtime, then set status of job to 295 // succeeded 296 // we dont need to materialize this job anymore 297 Date jobEndTime = job.getEndTime(); 298 if (jobEndTime.compareTo(endTime) <= 0) { 299 job.setStatus(CoordinatorJob.Status.SUCCEEDED); 300 log.info("[" + job.getId() + "]: Update status from PREMATER to SUCCEEDED"); 301 } 302 else { 303 job.setStatus(CoordinatorJob.Status.RUNNING); 304 log.info("[" + job.getId() + "]: Update status from PREMATER to RUNNING"); 305 } 306 job.setNextMaterializedTime(endTime); 307 updateList.add(job); 308 } 309 310 @Override 311 protected Void execute(CoordinatorStore store) throws StoreException, CommandException { 312 log.info("STARTED CoordActionMaterializeCommand for jobId=" + jobId + ", startTime=" + startTime + ", endTime=" 313 + endTime); 314 try { 315 if (lock(jobId)) { 316 call(store); 317 JPAService jpaService = Services.get().get(JPAService.class); 318 if (jpaService != null) { 319 try { 320 jpaService.execute(new BulkUpdateInsertJPAExecutor(updateList, insertList)); 321 } 322 catch (JPAExecutorException je) { 323 throw new CommandException(je); 324 } 325 } 326 else { 327 throw new CommandException(ErrorCode.E0610); 328 } 329 } 330 else { 331 queueCallable(new CoordActionMaterializeCommand(jobId, startTime, endTime), 332 LOCK_FAILURE_REQUEUE_INTERVAL); 333 log.warn("CoordActionMaterializeCommand lock was not acquired - failed jobId=" + jobId 334 + ". Requeing the same."); 335 } 336 } 337 catch (InterruptedException e) { 338 queueCallable(new CoordActionMaterializeCommand(jobId, startTime, endTime), LOCK_FAILURE_REQUEUE_INTERVAL); 339 log.warn("CoordActionMaterializeCommand lock acquiring failed with exception " + e.getMessage() 340 + " for jobId=" + jobId + " Requeing the same."); 341 } 342 finally { 343 log.info(" ENDED CoordActionMaterializeCommand for jobId=" + jobId + ", startTime=" + startTime 344 + ", endTime=" + endTime); 345 } 346 return null; 347 } 348 349 350 351 /** 352 * For preliminery testing. Should be removed soon 353 * 354 * @param args 355 * @throws Exception 356 */ 357 public static void main(String[] args) throws Exception { 358 new Services().init(); 359 try { 360 Date startTime = DateUtils.parseDateUTC("2009-02-01T01:00Z"); 361 Date endTime = DateUtils.parseDateUTC("2009-02-02T01:00Z"); 362 String jobId = "0000000-091207151850551-oozie-dani-C"; 363 CoordActionMaterializeCommand matCmd = new CoordActionMaterializeCommand(jobId, startTime, endTime); 364 matCmd.call(); 365 } 366 finally { 367 try { 368 Thread.sleep(60000); 369 } 370 catch (Exception ex) { 371 } 372 new Services().destroy(); 373 } 374 } 375 376 }