Source code

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.sla;
020
021import java.sql.Timestamp;
022import java.util.ArrayList;
023import java.util.Collections;
024import java.util.Date;
025import java.util.HashSet;
026import java.util.Iterator;
027import java.util.List;
028import java.util.Map;
029import java.util.Set;
030import java.util.concurrent.ConcurrentHashMap;
031
032import org.apache.hadoop.conf.Configuration;
033import org.apache.oozie.AppType;
034import org.apache.oozie.CoordinatorActionBean;
035import org.apache.oozie.CoordinatorJobBean;
036import org.apache.oozie.ErrorCode;
037import org.apache.oozie.WorkflowActionBean;
038import org.apache.oozie.WorkflowJobBean;
039import org.apache.oozie.XException;
040import org.apache.oozie.client.CoordinatorAction;
041import org.apache.oozie.client.WorkflowAction;
042import org.apache.oozie.client.WorkflowJob;
043import org.apache.oozie.client.event.JobEvent;
044import org.apache.oozie.client.event.SLAEvent.EventStatus;
045import org.apache.oozie.client.event.SLAEvent.SLAStatus;
046import org.apache.oozie.client.rest.JsonBean;
047import org.apache.oozie.executor.jpa.BatchQueryExecutor;
048import org.apache.oozie.executor.jpa.CoordActionGetForSLAJPAExecutor;
049import org.apache.oozie.executor.jpa.CoordActionQueryExecutor;
050import org.apache.oozie.executor.jpa.CoordActionQueryExecutor.CoordActionQuery;
051import org.apache.oozie.executor.jpa.CoordJobQueryExecutor;
052import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery;
053import org.apache.oozie.executor.jpa.JPAExecutorException;
054import org.apache.oozie.executor.jpa.SLARegistrationQueryExecutor;
055import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor;
056import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor;
057import org.apache.oozie.executor.jpa.SLARegistrationQueryExecutor.SLARegQuery;
058import org.apache.oozie.executor.jpa.SLASummaryQueryExecutor;
059import org.apache.oozie.executor.jpa.WorkflowActionGetForSLAJPAExecutor;
060import org.apache.oozie.executor.jpa.WorkflowJobGetForSLAJPAExecutor;
061import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor.WorkflowActionQuery;
062import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor.WorkflowJobQuery;
063import org.apache.oozie.executor.jpa.sla.SLASummaryGetRecordsOnRestartJPAExecutor;
064import org.apache.oozie.executor.jpa.SLASummaryQueryExecutor.SLASummaryQuery;
065import org.apache.oozie.executor.jpa.BatchQueryExecutor.UpdateEntry;
066import org.apache.oozie.lock.LockToken;
067import org.apache.oozie.service.ConfigurationService;
068import org.apache.oozie.service.EventHandlerService;
069import org.apache.oozie.service.JPAService;
070import org.apache.oozie.service.JobsConcurrencyService;
071import org.apache.oozie.service.MemoryLocksService;
072import org.apache.oozie.service.SchedulerService;
073import org.apache.oozie.service.ServiceException;
074import org.apache.oozie.service.Services;
075import org.apache.oozie.sla.service.SLAService;
076import org.apache.oozie.util.DateUtils;
077import org.apache.oozie.util.LogUtils;
078import org.apache.oozie.util.XLog;
079
080import com.google.common.annotations.VisibleForTesting;
081
082
083/**
084 * Implementation class for SLACalculator that calculates SLA related to
085 * start/end/duration of jobs using a memory-based map
086 */
087public class SLACalculatorMemory implements SLACalculator {
088
089    private static XLog LOG = XLog.getLog(SLACalculatorMemory.class);
090    // TODO optimization priority based insertion/processing/bumping up-down
091    protected Map<String, SLACalcStatus> slaMap;
092    protected Set<String> historySet;
093    private static int capacity;
094    private static JPAService jpaService;
095    protected EventHandlerService eventHandler;
096    private static int modifiedAfter;
097    private static long jobEventLatency;
098
099    @Override
100    public void init(Configuration conf) throws ServiceException {
101        capacity = ConfigurationService.getInt(conf, SLAService.CONF_CAPACITY);
102        jobEventLatency = ConfigurationService.getInt(conf, SLAService.CONF_JOB_EVENT_LATENCY);
103        slaMap = new ConcurrentHashMap<String, SLACalcStatus>();
104        historySet = Collections.synchronizedSet(new HashSet<String>());
105        jpaService = Services.get().get(JPAService.class);
106        eventHandler = Services.get().get(EventHandlerService.class);
107        // load events modified after
108        modifiedAfter = conf.getInt(SLAService.CONF_EVENTS_MODIFIED_AFTER, 7);
109        loadOnRestart();
110        Runnable purgeThread = new HistoryPurgeWorker();
111        // schedule runnable by default 1 day
112        Services.get()
113                .get(SchedulerService.class)
114                .schedule(purgeThread, 86400, Services.get().getConf().getInt(SLAService.CONF_SLA_HISTORY_PURGE_INTERVAL, 86400),
115                        SchedulerService.Unit.SEC);
116    }
117
118    public class HistoryPurgeWorker implements Runnable {
119
120        public HistoryPurgeWorker() {
121        }
122
123        @Override
124        public void run() {
125            if (Thread.currentThread().isInterrupted()) {
126                return;
127            }
128            Iterator<String> jobItr = historySet.iterator();
129            while (jobItr.hasNext()) {
130                String jobId = jobItr.next();
131
132                if (jobId.endsWith("-W")) {
133                    WorkflowJobBean wfJob = null;
134                    try {
135                        wfJob = WorkflowJobQueryExecutor.getInstance().get(WorkflowJobQuery.GET_WORKFLOW_STATUS, jobId);
136                    }
137                    catch (JPAExecutorException e) {
138                        if (e.getErrorCode().equals(ErrorCode.E0604)) {
139                            jobItr.remove();
140                        }
141                        else {
142                            LOG.info("Failed to fetch the workflow job: " + jobId, e);
143                        }
144                    }
145                    if (wfJob != null && wfJob.inTerminalState()) {
146                        try {
147                            updateSLASummary(wfJob.getId(), wfJob.getStartTime(), wfJob.getEndTime());
148                            jobItr.remove();
149                        }
150                        catch (JPAExecutorException e) {
151                            LOG.info("Failed to update SLASummaryBean when purging history set entry for " + jobId, e);
152                        }
153
154                    }
155                }
156                else if (jobId.contains("-W@")) {
157                    WorkflowActionBean wfAction = null;
158                    try {
159                        wfAction = WorkflowActionQueryExecutor.getInstance().get(
160                                WorkflowActionQuery.GET_ACTION_COMPLETED, jobId);
161                    }
162                    catch (JPAExecutorException e) {
163                        if (e.getErrorCode().equals(ErrorCode.E0605)) {
164                            jobItr.remove();
165                        }
166                        else {
167                            LOG.info("Failed to fetch the workflow action: " + jobId, e);
168                        }
169                    }
170                    if (wfAction != null && (wfAction.isComplete() || wfAction.isTerminalWithFailure())) {
171                        try {
172                            updateSLASummary(wfAction.getId(), wfAction.getStartTime(), wfAction.getEndTime());
173                            jobItr.remove();
174                        }
175                        catch (JPAExecutorException e) {
176                            LOG.info("Failed to update SLASummaryBean when purging history set entry for " + jobId, e);
177                        }
178                    }
179                }
180                else if (jobId.contains("-C@")) {
181                    CoordinatorActionBean cAction = null;
182                    try {
183                        cAction = CoordActionQueryExecutor.getInstance().get(CoordActionQuery.GET_COORD_ACTION, jobId);
184                    }
185                    catch (JPAExecutorException e) {
186                        if (e.getErrorCode().equals(ErrorCode.E0605)) {
187                            jobItr.remove();
188                        }
189                        else {
190                            LOG.info("Failed to fetch the coord action: " + jobId, e);
191                        }
192                    }
193                    if (cAction != null && cAction.isTerminalStatus()) {
194                        try {
195                            updateSLASummaryForCoordAction(cAction);
196                            jobItr.remove();
197                        }
198                        catch (JPAExecutorException e) {
199                            XLog.getLog(SLACalculatorMemory.class).info(
200                                    "Failed to update SLASummaryBean when purging history set entry for " + jobId, e);
201                        }
202
203                    }
204                }
205                else if (jobId.endsWith("-C")) {
206                    CoordinatorJobBean cJob = null;
207                    try {
208                        cJob = CoordJobQueryExecutor.getInstance().get(CoordJobQuery.GET_COORD_JOB_STATUS_PARENTID,
209                                jobId);
210                    }
211                    catch (JPAExecutorException e) {
212                        if (e.getErrorCode().equals(ErrorCode.E0604)) {
213                            jobItr.remove();
214                        }
215                        else {
216                            LOG.info("Failed to fetch the coord job: " + jobId, e);
217                        }
218                    }
219                    if (cJob != null && cJob.isTerminalStatus()) {
220                        try {
221                            updateSLASummary(cJob.getId(), cJob.getStartTime(), cJob.getEndTime());
222                            jobItr.remove();
223                        }
224                        catch (JPAExecutorException e) {
225                            LOG.info("Failed to update SLASummaryBean when purging history set entry for " + jobId, e);
226                        }
227
228                    }
229                }
230            }
231        }
232
233        private void updateSLASummary(String id, Date startTime, Date endTime) throws JPAExecutorException {
234            SLASummaryBean sla = SLASummaryQueryExecutor.getInstance().get(SLASummaryQuery.GET_SLA_SUMMARY, id);
235            if (sla != null) {
236                sla.setActualStart(startTime);
237                sla.setActualEnd(endTime);
238                if (startTime != null && endTime != null) {
239                    sla.setActualDuration(endTime.getTime() - startTime.getTime());
240                }
241                sla.setLastModifiedTime(new Date());
242                sla.setEventProcessed(8);
243                SLASummaryQueryExecutor.getInstance().executeUpdate(
244                        SLASummaryQuery.UPDATE_SLA_SUMMARY_FOR_ACTUAL_TIMES, sla);
245            }
246        }
247
248        private void updateSLASummaryForCoordAction(CoordinatorActionBean bean) throws JPAExecutorException {
249            String wrkflowId = bean.getExternalId();
250            if (wrkflowId != null) {
251                WorkflowJobBean wrkflow = WorkflowJobQueryExecutor.getInstance().get(
252                        WorkflowJobQuery.GET_WORKFLOW_START_END_TIME, wrkflowId);
253                if (wrkflow != null) {
254                    updateSLASummary(bean.getId(), wrkflow.getStartTime(), wrkflow.getEndTime());
255                }
256            }
257        }
258    }
259
260    private void loadOnRestart() {
261        boolean isJobModified = false;
262        try {
263            long slaPendingCount = 0;
264            long statusPendingCount = 0;
265            List<SLASummaryBean> summaryBeans = jpaService.execute(new SLASummaryGetRecordsOnRestartJPAExecutor(
266                    modifiedAfter));
267            for (SLASummaryBean summaryBean : summaryBeans) {
268                String jobId = summaryBean.getId();
269                LockToken lock = null;
270                switch (summaryBean.getAppType()) {
271                    case COORDINATOR_ACTION:
272                        isJobModified = processSummaryBeanForCoordAction(summaryBean, jobId);
273                        break;
274                    case WORKFLOW_ACTION:
275                        isJobModified = processSummaryBeanForWorkflowAction(summaryBean, jobId);
276                        break;
277                    case WORKFLOW_JOB:
278                        isJobModified = processSummaryBeanForWorkflowJob(summaryBean, jobId);
279                        break;
280                    default:
281                        break;
282                }
283                if (isJobModified) {
284                    try {
285                        boolean update = true;
286                        if (Services.get().get(JobsConcurrencyService.class).isHighlyAvailableMode()) {
287                            lock = Services
288                                    .get()
289                                    .get(MemoryLocksService.class)
290                                    .getWriteLock(
291                                            SLACalcStatus.SLA_ENTITYKEY_PREFIX + jobId,
292                                            Services.get().getConf()
293                                                    .getLong(SLAService.CONF_SLA_CALC_LOCK_TIMEOUT, 5 * 1000));
294                            if (lock == null) {
295                                update = false;
296                            }
297                        }
298                        if (update) {
299                            summaryBean.setLastModifiedTime(new Date());
300                            SLASummaryQueryExecutor.getInstance().executeUpdate(
301                                    SLASummaryQuery.UPDATE_SLA_SUMMARY_FOR_STATUS_ACTUAL_TIMES, summaryBean);
302                        }
303                    }
304                    catch (Exception e) {
305                        LOG.warn("Failed to load records for " + jobId, e);
306                    }
307                    finally {
308                        if (lock != null) {
309                            lock.release();
310                            lock = null;
311                        }
312                    }
313                }
314                try {
315                    if (summaryBean.getEventProcessed() == 7) {
316                        historySet.add(jobId);
317                        statusPendingCount++;
318                    }
319                    else if (summaryBean.getEventProcessed() <= 7) {
320                        SLARegistrationBean slaRegBean = SLARegistrationQueryExecutor.getInstance().get(
321                                SLARegQuery.GET_SLA_REG_ON_RESTART, jobId);
322                        SLACalcStatus slaCalcStatus = new SLACalcStatus(summaryBean, slaRegBean);
323                        slaMap.put(jobId, slaCalcStatus);
324                        slaPendingCount++;
325                    }
326                }
327                catch (Exception e) {
328                    LOG.warn("Failed to fetch/update records for " + jobId, e);
329                }
330
331            }
332            LOG.info("Loaded SLASummary pendingSLA=" + slaPendingCount + ", pendingStatusUpdate=" + statusPendingCount);
333
334        }
335        catch (Exception e) {
336            LOG.warn("Failed to retrieve SLASummary records on restart", e);
337        }
338    }
339
340    private boolean processSummaryBeanForCoordAction(SLASummaryBean summaryBean, String jobId)
341            throws JPAExecutorException {
342        boolean isJobModified = false;
343        CoordinatorActionBean coordAction = null;
344        coordAction = jpaService.execute(new CoordActionGetForSLAJPAExecutor(jobId));
345        if (!coordAction.getStatusStr().equals(summaryBean.getJobStatus())) {
346            LOG.trace("Coordinator action status is " + coordAction.getStatusStr() + " and summary bean status is "
347                    + summaryBean.getJobStatus());
348            isJobModified = true;
349            summaryBean.setJobStatus(coordAction.getStatusStr());
350            if (coordAction.isTerminalStatus()) {
351                WorkflowJobBean wfJob = jpaService.execute(new WorkflowJobGetForSLAJPAExecutor(coordAction
352                        .getExternalId()));
353                setEndForSLASummaryBean(summaryBean, wfJob.getStartTime(), coordAction.getLastModifiedTime(),
354                        coordAction.getStatusStr());
355            }
356            else if (coordAction.getStatus() != CoordinatorAction.Status.WAITING) {
357                WorkflowJobBean wfJob = jpaService.execute(new WorkflowJobGetForSLAJPAExecutor(coordAction
358                        .getExternalId()));
359                setStartForSLASummaryBean(summaryBean, summaryBean.getEventProcessed(), wfJob.getStartTime());
360            }
361        }
362        return isJobModified;
363    }
364
365    private boolean processSummaryBeanForWorkflowAction(SLASummaryBean summaryBean, String jobId)
366            throws JPAExecutorException {
367        boolean isJobModified = false;
368        WorkflowActionBean wfAction = null;
369        wfAction = jpaService.execute(new WorkflowActionGetForSLAJPAExecutor(jobId));
370        if (!wfAction.getStatusStr().equals(summaryBean.getJobStatus())) {
371            LOG.trace("Workflow action status is " + wfAction.getStatusStr() + "and summary bean status is "
372                    + summaryBean.getJobStatus());
373            isJobModified = true;
374            summaryBean.setJobStatus(wfAction.getStatusStr());
375            if (wfAction.inTerminalState()) {
376                setEndForSLASummaryBean(summaryBean, wfAction.getStartTime(), wfAction.getEndTime(), wfAction.getStatusStr());
377            }
378            else if (wfAction.getStatus() != WorkflowAction.Status.PREP) {
379                setStartForSLASummaryBean(summaryBean, summaryBean.getEventProcessed(), wfAction.getStartTime());
380            }
381        }
382        return isJobModified;
383    }
384
385    private boolean processSummaryBeanForWorkflowJob(SLASummaryBean summaryBean, String jobId)
386            throws JPAExecutorException {
387        boolean isJobModified = false;
388        WorkflowJobBean wfJob = null;
389        wfJob = jpaService.execute(new WorkflowJobGetForSLAJPAExecutor(jobId));
390        if (!wfJob.getStatusStr().equals(summaryBean.getJobStatus())) {
391            LOG.trace("Workflow job status is " + wfJob.getStatusStr() + "and summary bean status is "
392                    + summaryBean.getJobStatus());
393            isJobModified = true;
394            summaryBean.setJobStatus(wfJob.getStatusStr());
395            if (wfJob.inTerminalState()) {
396                setEndForSLASummaryBean(summaryBean, wfJob.getStartTime(), wfJob.getEndTime(), wfJob.getStatusStr());
397            }
398            else if (wfJob.getStatus() != WorkflowJob.Status.PREP) {
399                setStartForSLASummaryBean(summaryBean, summaryBean.getEventProcessed(), wfJob.getStartTime());
400            }
401        }
402        return isJobModified;
403    }
404
405    private void setEndForSLASummaryBean(SLASummaryBean summaryBean, Date startTime, Date endTime, String status) {
406        byte eventProc = summaryBean.getEventProcessed();
407        summaryBean.setEventProcessed(8);
408        summaryBean.setActualStart(startTime);
409        summaryBean.setActualEnd(endTime);
410        long actualDuration = endTime.getTime() - startTime.getTime();
411        summaryBean.setActualDuration(actualDuration);
412        if (eventProc < 4) {
413            if (status.equals(WorkflowJob.Status.SUCCEEDED.name()) || status.equals(WorkflowAction.Status.OK.name())
414                    || status.equals(CoordinatorAction.Status.SUCCEEDED.name())) {
415                if (endTime.getTime() <= summaryBean.getExpectedEnd().getTime()) {
416                    summaryBean.setSLAStatus(SLAStatus.MET);
417                }
418                else {
419                    summaryBean.setSLAStatus(SLAStatus.MISS);
420                }
421            }
422            else {
423                summaryBean.setSLAStatus(SLAStatus.MISS);
424            }
425        }
426
427    }
428
429    private void setStartForSLASummaryBean(SLASummaryBean summaryBean, byte eventProc, Date startTime) {
430        if (((eventProc & 1) == 0)) {
431            eventProc += 1;
432            summaryBean.setEventProcessed(eventProc);
433        }
434        if (summaryBean.getSLAStatus().equals(SLAStatus.NOT_STARTED)) {
435            summaryBean.setSLAStatus(SLAStatus.IN_PROCESS);
436        }
437        summaryBean.setActualStart(startTime);
438    }
439
440    @Override
441    public int size() {
442        return slaMap.size();
443    }
444
445    @Override
446    public SLACalcStatus get(String jobId) throws JPAExecutorException {
447        SLACalcStatus memObj;
448        memObj = slaMap.get(jobId);
449        if (memObj == null && historySet.contains(jobId)) {
450            memObj = new SLACalcStatus(SLASummaryQueryExecutor.getInstance().get(SLASummaryQuery.GET_SLA_SUMMARY, jobId),
451                    SLARegistrationQueryExecutor.getInstance().get(SLARegQuery.GET_SLA_REG_ON_RESTART, jobId));
452        }
453        return memObj;
454    }
455
456    @Override
457    public Iterator<String> iterator() {
458        return slaMap.keySet().iterator();
459    }
460
461    @Override
462    public boolean isEmpty() {
463        return slaMap.isEmpty();
464    }
465
466    @Override
467    public void clear() {
468        slaMap.clear();
469        historySet.clear();
470    }
471
472    /**
473     * Invoked via periodic run, update the SLA for registered jobs
474     */
475    protected void updateJobSla(String jobId) throws Exception {
476        SLACalcStatus slaCalc = slaMap.get(jobId);
477        synchronized (slaCalc) {
478            boolean change = false;
479            // get eventProcessed on DB for validation in HA
480            Object eventProcObj = ((SLASummaryQueryExecutor) SLASummaryQueryExecutor.getInstance()).getSingleValue(
481                    SLASummaryQuery.GET_SLA_SUMMARY_EVENTPROCESSED, jobId);
482            byte eventProc = ((Byte) eventProcObj).byteValue();
483            if (eventProc >= 7) {
484                if (eventProc == 7) {
485                    historySet.add(jobId);
486                }
487                slaMap.remove(jobId);
488                LOG.trace("Removed Job [{0}] from map as SLA processed", jobId);
489            }
490            else {
491                slaCalc.setEventProcessed(eventProc);
492                SLARegistrationBean reg = slaCalc.getSLARegistrationBean();
493                // calculation w.r.t current time and status
494                if ((eventProc & 1) == 0) { // first bit (start-processed) unset
495                    if (reg.getExpectedStart() != null) {
496                        if (reg.getExpectedStart().getTime() + jobEventLatency < System.currentTimeMillis()) {
497                            confirmWithDB(slaCalc);
498                            eventProc = slaCalc.getEventProcessed();
499                            if (eventProc != 8 && (eventProc & 1) == 0) {
500                                // Some DB exception
501                                slaCalc.setEventStatus(EventStatus.START_MISS);
502                                eventHandler.queueEvent(new SLACalcStatus(slaCalc));
503                                eventProc++;
504                            }
505                            change = true;
506                        }
507                    }
508                    else {
509                        eventProc++; // disable further processing for optional start sla condition
510                        change = true;
511                    }
512                }
513                // check if second bit (duration-processed) is unset
514                if (eventProc != 8 && ((eventProc >> 1) & 1) == 0) {
515                    if (reg.getExpectedDuration() == -1) {
516                        eventProc += 2;
517                        change = true;
518                    }
519                    else if (slaCalc.getActualStart() != null) {
520                        if ((reg.getExpectedDuration() + jobEventLatency) < (System.currentTimeMillis() - slaCalc
521                                .getActualStart().getTime())) {
522                            slaCalc.setEventProcessed(eventProc);
523                            confirmWithDB(slaCalc);
524                            eventProc = slaCalc.getEventProcessed();
525                            if (eventProc != 8 && ((eventProc >> 1) & 1) == 0) {
526                                // Some DB exception
527                                slaCalc.setEventStatus(EventStatus.DURATION_MISS);
528                                eventHandler.queueEvent(new SLACalcStatus(slaCalc));
529                                eventProc += 2;
530                            }
531                            change = true;
532                        }
533                    }
534                }
535                if (eventProc < 4) {
536                    if (reg.getExpectedEnd().getTime() + jobEventLatency < System.currentTimeMillis()) {
537                        slaCalc.setEventProcessed(eventProc);
538                        confirmWithDB(slaCalc);
539                        eventProc = slaCalc.getEventProcessed();
540                        change = true;
541                    }
542                }
543                if (change) {
544                    try {
545                        boolean locked = true;
546                        slaCalc.acquireLock();
547                        locked = slaCalc.isLocked();
548                        if (locked) {
549                            // no more processing, no transfer to history set
550                            if (slaCalc.getEventProcessed() >= 8) {
551                                eventProc = 8;
552                                // Should not be > 8. But to handle any corner cases
553                                slaCalc.setEventProcessed(8);
554                                slaMap.remove(jobId);
555                            }
556                            else {
557                                slaCalc.setEventProcessed(eventProc);
558                            }
559                            SLASummaryBean slaSummaryBean = new SLASummaryBean();
560                            slaSummaryBean.setId(slaCalc.getId());
561                            slaSummaryBean.setEventProcessed(eventProc);
562                            slaSummaryBean.setSLAStatus(slaCalc.getSLAStatus());
563                            slaSummaryBean.setEventStatus(slaCalc.getEventStatus());
564                            slaSummaryBean.setActualEnd(slaCalc.getActualEnd());
565                            slaSummaryBean.setActualStart(slaCalc.getActualStart());
566                            slaSummaryBean.setActualDuration(slaCalc.getActualDuration());
567                            slaSummaryBean.setJobStatus(slaCalc.getJobStatus());
568                            slaSummaryBean.setLastModifiedTime(new Date());
569                            SLASummaryQueryExecutor.getInstance().executeUpdate(
570                                    SLASummaryQuery.UPDATE_SLA_SUMMARY_FOR_STATUS_ACTUAL_TIMES, slaSummaryBean);
571                            if (eventProc == 7) {
572                                historySet.add(jobId);
573                                slaMap.remove(jobId);
574                                LOG.trace("Removed Job [{0}] from map after End-processed", jobId);
575                            }
576                        }
577                    }
578                    catch (InterruptedException e) {
579                        throw new XException(ErrorCode.E0606, slaCalc.getId(), slaCalc.getLockTimeOut());
580                    }
581                    finally {
582                        slaCalc.releaseLock();
583                    }
584                }
585            }
586        }
587    }
588
589    /**
590     * Periodically run by the SLAService worker threads to update SLA status by
591     * iterating through all the jobs in the map
592     */
593    @Override
594    public void updateAllSlaStatus() {
595        LOG.info("Running periodic SLA check");
596        Iterator<String> iterator = slaMap.keySet().iterator();
597        while (iterator.hasNext()) {
598            String jobId = iterator.next();
599            try {
600                LOG.trace("Processing SLA for jobid={0}", jobId);
601                updateJobSla(jobId);
602            }
603            catch (Exception e) {
604                setLogPrefix(jobId);
605                LOG.error("Exception in SLA processing for job [{0}]", jobId, e);
606                LogUtils.clearLogPrefix();
607            }
608        }
609    }
610
611    /**
612     * Register a new job into the map for SLA tracking
613     */
614    @Override
615    public boolean addRegistration(String jobId, SLARegistrationBean reg) throws JPAExecutorException {
616        try {
617            if (slaMap.size() < capacity) {
618                SLACalcStatus slaCalc = new SLACalcStatus(reg);
619                slaCalc.setSLAStatus(SLAStatus.NOT_STARTED);
620                slaCalc.setJobStatus(getJobStatus(reg.getAppType()));
621                slaMap.put(jobId, slaCalc);
622                List<JsonBean> insertList = new ArrayList<JsonBean>();
623                final SLASummaryBean summaryBean = new SLASummaryBean(slaCalc);
624                final Timestamp currentTime = DateUtils.convertDateToTimestamp(new Date());
625                reg.setCreatedTimestamp(currentTime);
626                summaryBean.setCreatedTimestamp(currentTime);
627                insertList.add(reg);
628                insertList.add(summaryBean);
629                BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(insertList, null, null);
630                LOG.trace("SLA Registration Event - Job:" + jobId);
631                return true;
632            }
633            else {
634                setLogPrefix(reg.getId());
635                LOG.error(
636                        "SLACalculator memory capacity reached. Cannot add or update new SLA Registration entry for job [{0}]",
637                        reg.getId());
638                LogUtils.clearLogPrefix();
639            }
640        }
641        catch (JPAExecutorException jpa) {
642            throw jpa;
643        }
644        return false;
645    }
646
647    private String getJobStatus(AppType appType) {
648        String status = null;
649        switch (appType) {
650            case COORDINATOR_ACTION:
651                status = CoordinatorAction.Status.WAITING.name();
652                break;
653            case WORKFLOW_ACTION:
654                status = WorkflowAction.Status.PREP.name();
655                break;
656            case WORKFLOW_JOB:
657                status = WorkflowJob.Status.PREP.name();
658                break;
659            default:
660                break;
661        }
662        return status;
663    }
664
665    /**
666     * Update job into the map for SLA tracking
667     */
668    @Override
669    public boolean updateRegistration(String jobId, SLARegistrationBean reg) throws JPAExecutorException {
670        try {
671            if (slaMap.size() < capacity) {
672                SLACalcStatus slaCalc = new SLACalcStatus(reg);
673                slaCalc.setSLAStatus(SLAStatus.NOT_STARTED);
674                slaCalc.setJobStatus(getJobStatus(reg.getAppType()));
675                slaMap.put(jobId, slaCalc);
676                List<UpdateEntry> updateList = new ArrayList<UpdateEntry>();
677                updateList.add(new UpdateEntry<SLARegQuery>(SLARegQuery.UPDATE_SLA_REG_ALL, reg));
678                updateList.add(new UpdateEntry<SLASummaryQuery>(SLASummaryQuery.UPDATE_SLA_SUMMARY_ALL,
679                        new SLASummaryBean(slaCalc)));
680                BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(null, updateList, null);
681                LOG.trace("SLA Registration Event - Job:" + jobId);
682                return true;
683            }
684            else {
685                setLogPrefix(reg.getId());
686                LOG.error(
687                        "SLACalculator memory capacity reached. Cannot add or update new SLA Registration entry for job [{0}]",
688                        reg.getId());
689                LogUtils.clearLogPrefix();
690            }
691        }
692        catch (JPAExecutorException jpa) {
693            throw jpa;
694        }
695        return false;
696    }
697
698    /**
699     * Remove job from being tracked in map
700     */
701    @Override
702    public void removeRegistration(String jobId) {
703        if (slaMap.remove(jobId) == null) {
704            historySet.remove(jobId);
705        }
706    }
707
708    /**
709     * Triggered after receiving Job status change event, update SLA status
710     * accordingly
711     */
712    @Override
713    public boolean addJobStatus(String jobId, String jobStatus, JobEvent.EventStatus jobEventStatus, Date startTime,
714            Date endTime) throws JPAExecutorException, ServiceException {
715        SLACalcStatus slaCalc = slaMap.get(jobId);
716        SLASummaryBean slaInfo = null;
717        boolean hasSla = false;
718        if (slaCalc == null) {
719            if (historySet.contains(jobId)) {
720                slaInfo = SLASummaryQueryExecutor.getInstance().get(SLASummaryQuery.GET_SLA_SUMMARY, jobId);
721                if (slaInfo == null) {
722                    throw new JPAExecutorException(ErrorCode.E0604, jobId);
723                }
724                slaInfo.setJobStatus(jobStatus);
725                slaInfo.setActualStart(startTime);
726                slaInfo.setActualEnd(endTime);
727                if (endTime != null) {
728                    slaInfo.setActualDuration(endTime.getTime() - startTime.getTime());
729                }
730                slaInfo.setEventProcessed(8);
731                historySet.remove(jobId);
732                slaInfo.setLastModifiedTime(new Date());
733                SLASummaryQueryExecutor.getInstance().executeUpdate(
734                        SLASummaryQuery.UPDATE_SLA_SUMMARY_FOR_STATUS_ACTUAL_TIMES, slaInfo);
735                hasSla = true;
736            }
737            else if (Services.get().get(JobsConcurrencyService.class).isHighlyAvailableMode()) {
738                // jobid might not exist in slaMap in HA Setting
739                SLARegistrationBean slaRegBean = SLARegistrationQueryExecutor.getInstance().get(
740                        SLARegQuery.GET_SLA_REG_ALL, jobId);
741                if (slaRegBean != null) { // filter out jobs picked by SLA job event listener
742                                          // but not actually configured for SLA
743                    SLASummaryBean slaSummaryBean = SLASummaryQueryExecutor.getInstance().get(
744                            SLASummaryQuery.GET_SLA_SUMMARY, jobId);
745                    if (slaSummaryBean.getEventProcessed() < 7) {
746                        slaCalc = new SLACalcStatus(slaSummaryBean, slaRegBean);
747                        slaMap.put(jobId, slaCalc);
748                    }
749                }
750            }
751        }
752        if (slaCalc != null) {
753            synchronized (slaCalc) {
754                try {
755                    // only get ZK lock when multiple servers running
756                    boolean locked = true;
757                    slaCalc.acquireLock();
758                    locked = slaCalc.isLocked();
759                    if (locked) {
760                        // get eventProcessed on DB for validation in HA
761                        Object eventProcObj = ((SLASummaryQueryExecutor) SLASummaryQueryExecutor.getInstance())
762                                .getSingleValue(SLASummaryQuery.GET_SLA_SUMMARY_EVENTPROCESSED, jobId);
763                        byte eventProc = ((Byte) eventProcObj).byteValue();
764                        slaCalc.setEventProcessed(eventProc);
765                        slaCalc.setJobStatus(jobStatus);
766                        switch (jobEventStatus) {
767                            case STARTED:
768                                slaInfo = processJobStartSLA(slaCalc, startTime);
769                                break;
770                            case SUCCESS:
771                                slaInfo = processJobEndSuccessSLA(slaCalc, startTime, endTime);
772                                break;
773                            case FAILURE:
774                                slaInfo = processJobEndFailureSLA(slaCalc, startTime, endTime);
775                                break;
776                            default:
777                                LOG.debug("Unknown Job Status for SLA purpose[{0}]", jobEventStatus);
778                                slaInfo = getSLASummaryBean(slaCalc);
779                        }
780                        if (slaCalc.getEventProcessed() == 7) {
781                            slaInfo.setEventProcessed(8);
782                            slaMap.remove(jobId);
783                        }
784                        slaInfo.setLastModifiedTime(new Date());
785                        SLASummaryQueryExecutor.getInstance().executeUpdate(
786                                SLASummaryQuery.UPDATE_SLA_SUMMARY_FOR_STATUS_ACTUAL_TIMES, slaInfo);
787                        hasSla = true;
788                    }
789                }
790                catch (InterruptedException e) {
791                    throw new ServiceException(ErrorCode.E0606, slaCalc.getEntityKey(), slaCalc.getLockTimeOut());
792                }
793                finally {
794                    slaCalc.releaseLock();
795                }
796            }
797            LOG.trace("SLA Status Event - Job:" + jobId + " Status:" + slaCalc.getSLAStatus());
798        }
799
800        return hasSla;
801    }
802
803    /**
804     * Process SLA for jobs that started running. Also update actual-start time
805     *
806     * @param slaCalc
807     * @param actualStart
808     * @return SLASummaryBean
809     */
810    private SLASummaryBean processJobStartSLA(SLACalcStatus slaCalc, Date actualStart) {
811        slaCalc.setActualStart(actualStart);
812        if (slaCalc.getSLAStatus().equals(SLAStatus.NOT_STARTED)) {
813            slaCalc.setSLAStatus(SLAStatus.IN_PROCESS);
814        }
815        SLARegistrationBean reg = slaCalc.getSLARegistrationBean();
816        Date expecStart = reg.getExpectedStart();
817        byte eventProc = slaCalc.getEventProcessed();
818        // set event proc here
819        if (((eventProc & 1) == 0)) {
820            if (expecStart != null) {
821                if (actualStart.getTime() > expecStart.getTime()) {
822                    slaCalc.setEventStatus(EventStatus.START_MISS);
823                }
824                else {
825                    slaCalc.setEventStatus(EventStatus.START_MET);
826                }
827                eventHandler.queueEvent(new SLACalcStatus(slaCalc));
828            }
829            eventProc += 1;
830            slaCalc.setEventProcessed(eventProc);
831        }
832        return getSLASummaryBean(slaCalc);
833    }
834
835    /**
836     * Process SLA for jobs that ended successfully. Also update actual-start
837     * and end time
838     *
839     * @param slaCalc
840     * @param actualStart
841     * @param actualEnd
842     * @return SLASummaryBean
843     * @throws JPAExecutorException
844     */
845    private SLASummaryBean processJobEndSuccessSLA(SLACalcStatus slaCalc, Date actualStart, Date actualEnd) throws JPAExecutorException {
846        SLARegistrationBean reg = slaCalc.getSLARegistrationBean();
847        slaCalc.setActualStart(actualStart);
848        slaCalc.setActualEnd(actualEnd);
849        long expectedDuration = reg.getExpectedDuration();
850        long actualDuration = actualEnd.getTime() - actualStart.getTime();
851        slaCalc.setActualDuration(actualDuration);
852        //check event proc
853        byte eventProc = slaCalc.getEventProcessed();
854        if (((eventProc >> 1) & 1) == 0) {
855            processDurationSLA(expectedDuration, actualDuration, slaCalc);
856            eventProc += 2;
857            slaCalc.setEventProcessed(eventProc);
858        }
859
860        if (eventProc < 4) {
861            Date expectedEnd = reg.getExpectedEnd();
862            if (actualEnd.getTime() > expectedEnd.getTime()) {
863                slaCalc.setEventStatus(EventStatus.END_MISS);
864                slaCalc.setSLAStatus(SLAStatus.MISS);
865            }
866            else {
867                slaCalc.setEventStatus(EventStatus.END_MET);
868                slaCalc.setSLAStatus(SLAStatus.MET);
869            }
870            eventProc += 4;
871            slaCalc.setEventProcessed(eventProc);
872            eventHandler.queueEvent(new SLACalcStatus(slaCalc));
873        }
874        return getSLASummaryBean(slaCalc);
875    }
876
877    /**
878     * Process SLA for jobs that ended in failure. Also update actual-start and
879     * end time
880     *
881     * @param slaCalc
882     * @param actualStart
883     * @param actualEnd
884     * @return SLASummaryBean
885     * @throws JPAExecutorException
886     */
887    private SLASummaryBean processJobEndFailureSLA(SLACalcStatus slaCalc, Date actualStart, Date actualEnd) throws JPAExecutorException {
888        slaCalc.setActualStart(actualStart);
889        slaCalc.setActualEnd(actualEnd);
890        if (actualStart == null) { // job failed before starting
891            if (slaCalc.getEventProcessed() < 4) {
892                slaCalc.setEventStatus(EventStatus.END_MISS);
893                slaCalc.setSLAStatus(SLAStatus.MISS);
894                eventHandler.queueEvent(new SLACalcStatus(slaCalc));
895                slaCalc.setEventProcessed(7);
896                return getSLASummaryBean(slaCalc);
897            }
898        }
899        SLARegistrationBean reg = slaCalc.getSLARegistrationBean();
900        long expectedDuration = reg.getExpectedDuration();
901        long actualDuration = actualEnd.getTime() - actualStart.getTime();
902        slaCalc.setActualDuration(actualDuration);
903
904        byte eventProc = slaCalc.getEventProcessed();
905        if (((eventProc >> 1) & 1) == 0) {
906            if (expectedDuration != -1) {
907                slaCalc.setEventStatus(EventStatus.DURATION_MISS);
908                eventHandler.queueEvent(new SLACalcStatus(slaCalc));
909            }
910            eventProc += 2;
911            slaCalc.setEventProcessed(eventProc);
912        }
913        if (eventProc < 4) {
914            slaCalc.setEventStatus(EventStatus.END_MISS);
915            slaCalc.setSLAStatus(SLAStatus.MISS);
916            eventProc += 4;
917            slaCalc.setEventProcessed(eventProc);
918            eventHandler.queueEvent(new SLACalcStatus(slaCalc));
919        }
920        return getSLASummaryBean(slaCalc);
921    }
922
923    private SLASummaryBean getSLASummaryBean (SLACalcStatus slaCalc) {
924        SLASummaryBean slaSummaryBean = new SLASummaryBean();
925        slaSummaryBean.setActualStart(slaCalc.getActualStart());
926        slaSummaryBean.setActualEnd(slaCalc.getActualEnd());
927        slaSummaryBean.setActualDuration(slaCalc.getActualDuration());
928        slaSummaryBean.setSLAStatus(slaCalc.getSLAStatus());
929        slaSummaryBean.setEventStatus(slaCalc.getEventStatus());
930        slaSummaryBean.setEventProcessed(slaCalc.getEventProcessed());
931        slaSummaryBean.setId(slaCalc.getId());
932        slaSummaryBean.setJobStatus(slaCalc.getJobStatus());
933        return slaSummaryBean;
934    }
935
936    private void processDurationSLA(long expected, long actual, SLACalcStatus slaCalc) {
937        if (expected != -1 && actual > expected) {
938            slaCalc.setEventStatus(EventStatus.DURATION_MISS);
939            eventHandler.queueEvent(new SLACalcStatus(slaCalc));
940        }
941        else if (expected != -1 && actual <= expected) {
942            slaCalc.setEventStatus(EventStatus.DURATION_MET);
943            eventHandler.queueEvent(new SLACalcStatus(slaCalc));
944        }
945    }
946
947    /*
948     * Confirm alerts against source of truth - DB. Also required in case of High Availability
949     */
950    private void confirmWithDB(SLACalcStatus slaCalc) {
951        boolean ended = false, isEndMiss = false;
952        try {
953            switch (slaCalc.getAppType()) {
954                case WORKFLOW_JOB:
955                    WorkflowJobBean wf = jpaService.execute(new WorkflowJobGetForSLAJPAExecutor(slaCalc.getId()));
956                    if (wf.getEndTime() != null) {
957                        ended = true;
958                        if (wf.getStatus() == WorkflowJob.Status.KILLED || wf.getStatus() == WorkflowJob.Status.FAILED
959                                || wf.getEndTime().getTime() > slaCalc.getExpectedEnd().getTime()) {
960                            isEndMiss = true;
961                        }
962                    }
963                    slaCalc.setActualStart(wf.getStartTime());
964                    slaCalc.setActualEnd(wf.getEndTime());
965                    slaCalc.setJobStatus(wf.getStatusStr());
966                    break;
967                case WORKFLOW_ACTION:
968                    WorkflowActionBean wa = jpaService.execute(new WorkflowActionGetForSLAJPAExecutor(slaCalc.getId()));
969                    if (wa.getEndTime() != null) {
970                        ended = true;
971                        if (wa.isTerminalWithFailure()
972                                || wa.getEndTime().getTime() > slaCalc.getExpectedEnd().getTime()) {
973                            isEndMiss = true;
974                        }
975                    }
976                    slaCalc.setActualStart(wa.getStartTime());
977                    slaCalc.setActualEnd(wa.getEndTime());
978                    slaCalc.setJobStatus(wa.getStatusStr());
979                    break;
980                case COORDINATOR_ACTION:
981                    CoordinatorActionBean ca = jpaService.execute(new CoordActionGetForSLAJPAExecutor(slaCalc.getId()));
982                    if (ca.isTerminalWithFailure()) {
983                        isEndMiss = ended = true;
984                        slaCalc.setActualEnd(ca.getLastModifiedTime());
985                    }
986                    if (ca.getExternalId() != null) {
987                        wf = jpaService.execute(new WorkflowJobGetForSLAJPAExecutor(ca.getExternalId()));
988                        if (wf.getEndTime() != null) {
989                            ended = true;
990                            if (wf.getEndTime().getTime() > slaCalc.getExpectedEnd().getTime()) {
991                                isEndMiss = true;
992                            }
993                        }
994                        slaCalc.setActualEnd(wf.getEndTime());
995                        slaCalc.setActualStart(wf.getStartTime());
996                    }
997                    slaCalc.setJobStatus(ca.getStatusStr());
998                    break;
999                default:
1000                    LOG.debug("Unsupported App-type for SLA - " + slaCalc.getAppType());
1001            }
1002
1003            byte eventProc = slaCalc.getEventProcessed();
1004            if (ended) {
1005                if (isEndMiss) {
1006                    slaCalc.setSLAStatus(SLAStatus.MISS);
1007                }
1008                else {
1009                    slaCalc.setSLAStatus(SLAStatus.MET);
1010                }
1011                if (slaCalc.getActualStart() != null) {
1012                    if ((eventProc & 1) == 0) {
1013                        if (slaCalc.getExpectedStart().getTime() < slaCalc.getActualStart().getTime()) {
1014                            slaCalc.setEventStatus(EventStatus.START_MISS);
1015                        }
1016                        else {
1017                            slaCalc.setEventStatus(EventStatus.START_MET);
1018                        }
1019                        eventHandler.queueEvent(new SLACalcStatus(slaCalc));
1020                    }
1021                    slaCalc.setActualDuration(slaCalc.getActualEnd().getTime() - slaCalc.getActualStart().getTime());
1022                    if (((eventProc >> 1) & 1) == 0) {
1023                        processDurationSLA(slaCalc.getExpectedDuration(), slaCalc.getActualDuration(), slaCalc);
1024                    }
1025                }
1026                if (eventProc < 4) {
1027                    if (isEndMiss) {
1028                        slaCalc.setEventStatus(EventStatus.END_MISS);
1029                    }
1030                    else {
1031                        slaCalc.setEventStatus(EventStatus.END_MET);
1032                    }
1033                    eventHandler.queueEvent(new SLACalcStatus(slaCalc));
1034                }
1035                slaCalc.setEventProcessed(8);
1036            }
1037            else {
1038                if (slaCalc.getActualStart() != null) {
1039                    slaCalc.setSLAStatus(SLAStatus.IN_PROCESS);
1040                }
1041                if ((eventProc & 1) == 0) {
1042                    if (slaCalc.getActualStart() != null) {
1043                        if (slaCalc.getExpectedStart().getTime() < slaCalc.getActualStart().getTime()) {
1044                            slaCalc.setEventStatus(EventStatus.START_MISS);
1045                        }
1046                        else {
1047                            slaCalc.setEventStatus(EventStatus.START_MET);
1048                        }
1049                        eventHandler.queueEvent(new SLACalcStatus(slaCalc));
1050                        eventProc++;
1051                    }
1052                    else if (slaCalc.getExpectedStart().getTime() < System.currentTimeMillis()) {
1053                        slaCalc.setEventStatus(EventStatus.START_MISS);
1054                        eventHandler.queueEvent(new SLACalcStatus(slaCalc));
1055                        eventProc++;
1056                    }
1057                }
1058                if (((eventProc >> 1) & 1) == 0 && slaCalc.getActualStart() != null
1059                        && slaCalc.getExpectedDuration() != -1) {
1060                    if (System.currentTimeMillis() - slaCalc.getActualStart().getTime() > slaCalc.getExpectedDuration()) {
1061                        slaCalc.setEventStatus(EventStatus.DURATION_MISS);
1062                        eventHandler.queueEvent(new SLACalcStatus(slaCalc));
1063                        eventProc += 2;
1064                    }
1065                }
1066                if (eventProc < 4 && slaCalc.getExpectedEnd().getTime() < System.currentTimeMillis()) {
1067                    slaCalc.setEventStatus(EventStatus.END_MISS);
1068                    slaCalc.setSLAStatus(SLAStatus.MISS);
1069                    eventHandler.queueEvent(new SLACalcStatus(slaCalc));
1070                    eventProc += 4;
1071                }
1072                slaCalc.setEventProcessed(eventProc);
1073            }
1074        }
1075        catch (Exception e) {
1076            LOG.warn("Error while confirming SLA against DB for jobid= " + slaCalc.getId() + ". Exception is "
1077                    + e.getClass().getName() + ": " + e.getMessage());
1078            if (slaCalc.getEventProcessed() < 4 && slaCalc.getExpectedEnd().getTime() < System.currentTimeMillis()) {
1079                slaCalc.setEventStatus(EventStatus.END_MISS);
1080                slaCalc.setSLAStatus(SLAStatus.MISS);
1081                eventHandler.queueEvent(new SLACalcStatus(slaCalc));
1082                slaCalc.setEventProcessed(slaCalc.getEventProcessed() + 4);
1083            }
1084        }
1085    }
1086
1087    @VisibleForTesting
1088    public boolean isJobIdInSLAMap(String jobId) {
1089        return this.slaMap.containsKey(jobId);
1090    }
1091
1092    @VisibleForTesting
1093    public boolean isJobIdInHistorySet(String jobId) {
1094        return this.historySet.contains(jobId);
1095    }
1096
1097    private void setLogPrefix(String jobId) {
1098        LOG = LogUtils.setLogInfo(LOG, jobId, null, null);
1099    }
1100}