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 }