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.service;
020
021import java.util.Date;
022
023import org.apache.hadoop.conf.Configuration;
024import org.apache.oozie.ErrorCode;
025import org.apache.oozie.client.event.JobEvent.EventStatus;
026import org.apache.oozie.executor.jpa.JPAExecutorException;
027import org.apache.oozie.service.ConfigurationService;
028import org.apache.oozie.service.EventHandlerService;
029import org.apache.oozie.service.SchedulerService;
030import org.apache.oozie.service.Service;
031import org.apache.oozie.service.ServiceException;
032import org.apache.oozie.service.Services;
033import org.apache.oozie.sla.SLACalculator;
034import org.apache.oozie.sla.SLACalculatorMemory;
035import org.apache.oozie.sla.SLARegistrationBean;
036import org.apache.oozie.util.XLog;
037
038import com.google.common.annotations.VisibleForTesting;
039
040public class SLAService implements Service {
041
042    public static final String CONF_PREFIX = "oozie.sla.service.SLAService.";
043    public static final String CONF_CALCULATOR_IMPL = CONF_PREFIX + "calculator.impl";
044    public static final String CONF_CAPACITY = CONF_PREFIX + "capacity";
045    public static final String CONF_ALERT_EVENTS = CONF_PREFIX + "alert.events";
046    public static final String CONF_EVENTS_MODIFIED_AFTER = CONF_PREFIX + "events.modified.after";
047    public static final String CONF_JOB_EVENT_LATENCY = CONF_PREFIX + "job.event.latency";
048    //Time interval, in seconds, at which SLA Worker will be scheduled to run
049    public static final String CONF_SLA_CHECK_INTERVAL = CONF_PREFIX + "check.interval";
050    public static final String CONF_SLA_CHECK_INITIAL_DELAY = CONF_PREFIX + "check.initial.delay";
051    public static final String CONF_SLA_CALC_LOCK_TIMEOUT = CONF_PREFIX + "oozie.sla.calc.default.lock.timeout";
052    public static final String CONF_SLA_HISTORY_PURGE_INTERVAL = CONF_PREFIX + "history.purge.interval";
053
054    private static SLACalculator calcImpl;
055    private static boolean slaEnabled = false;
056    private EventHandlerService eventHandler;
057    public static XLog LOG;
058    @Override
059    public void init(Services services) throws ServiceException {
060        try {
061            Configuration conf = services.getConf();
062            Class<? extends SLACalculator> calcClazz = (Class<? extends SLACalculator>) ConfigurationService.getClass(
063                    conf, CONF_CALCULATOR_IMPL);
064            calcImpl = calcClazz == null ? new SLACalculatorMemory() : (SLACalculator) calcClazz.newInstance();
065            calcImpl.init(conf);
066            eventHandler = Services.get().get(EventHandlerService.class);
067            if (eventHandler == null) {
068                throw new ServiceException(ErrorCode.E0103, "EventHandlerService", "Add it under config "
069                        + Services.CONF_SERVICE_EXT_CLASSES + " or declare it BEFORE SLAService");
070            }
071            LOG = XLog.getLog(getClass());
072            java.util.Set<String> appTypes = eventHandler.getAppTypes();
073            appTypes.add("workflow_action");
074            eventHandler.setAppTypes(appTypes);
075
076            Runnable slaThread = new SLAWorker(calcImpl);
077            // schedule runnable by default every 30 sec
078            int slaCheckInterval = ConfigurationService.getInt(conf, CONF_SLA_CHECK_INTERVAL);
079            int slaCheckInitialDelay = ConfigurationService.getInt(conf, CONF_SLA_CHECK_INITIAL_DELAY);
080            services.get(SchedulerService.class).schedule(slaThread, slaCheckInitialDelay, slaCheckInterval,
081                    SchedulerService.Unit.SEC);
082            slaEnabled = true;
083            LOG.info("SLAService initialized with impl [{0}] capacity [{1}]", calcImpl.getClass().getName(),
084                    conf.get(SLAService.CONF_CAPACITY));
085        }
086        catch (Exception ex) {
087            throw new ServiceException(ErrorCode.E0102, ex.getMessage(), ex);
088        }
089    }
090
091    @Override
092    public void destroy() {
093        slaEnabled = false;
094    }
095
096    @Override
097    public Class<? extends Service> getInterface() {
098        return SLAService.class;
099    }
100
101    public static boolean isEnabled() {
102        return slaEnabled;
103    }
104
105    @VisibleForTesting
106    public SLACalculator getSLACalculator() {
107        return calcImpl;
108    }
109
110    @VisibleForTesting
111    public void runSLAWorker() {
112        new SLAWorker(calcImpl).run();
113    }
114
115    private class SLAWorker implements Runnable {
116
117        SLACalculator calc;
118
119        public SLAWorker(SLACalculator calc) {
120            this.calc = calc;
121        }
122
123        @Override
124        public void run() {
125            if (Thread.currentThread().isInterrupted()) {
126                return;
127            }
128            try {
129                calc.updateAllSlaStatus();
130            }
131            catch (Throwable error) {
132                XLog.getLog(SLAService.class).debug("Throwable in SLAWorker thread run : ", error);
133            }
134        }
135    }
136
137    public boolean addRegistrationEvent(SLARegistrationBean reg) throws ServiceException {
138        try {
139            if (calcImpl.addRegistration(reg.getId(), reg)) {
140                return true;
141            }
142            else {
143                LOG.warn("SLA queue full. Unable to add new SLA entry for job [{0}]", reg.getId());
144            }
145        }
146        catch (JPAExecutorException ex) {
147            LOG.warn("Could not add new SLA entry for job [{0}]", reg.getId(), ex);
148        }
149        return false;
150    }
151
152    public boolean updateRegistrationEvent(SLARegistrationBean reg) throws ServiceException {
153        try {
154            if (calcImpl.updateRegistration(reg.getId(), reg)) {
155                return true;
156            }
157            else {
158                LOG.warn("SLA queue full. Unable to update the SLA entry for job [{0}]", reg.getId());
159            }
160        }
161        catch (JPAExecutorException ex) {
162            LOG.warn("Could not update SLA entry for job [{0}]", reg.getId(), ex);
163        }
164        return false;
165    }
166
167    public boolean addStatusEvent(String jobId, String status, EventStatus eventStatus, Date startTime, Date endTime)
168            throws ServiceException {
169        try {
170            if (calcImpl.addJobStatus(jobId, status, eventStatus, startTime, endTime)) {
171                return true;
172            }
173        }
174        catch (JPAExecutorException jpe) {
175            LOG.error("Exception while adding SLA Status event for Job [{0}]", jobId);
176        }
177        return false;
178    }
179
180    public void removeRegistration(String jobId) {
181        calcImpl.removeRegistration(jobId);
182    }
183
184}