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