/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.falcon.service;

import com.google.common.annotations.VisibleForTesting;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.falcon.FalconException;
import org.apache.falcon.LifeCycle;
import org.apache.falcon.Pair;
import org.apache.falcon.entity.ClusterHelper;
import org.apache.falcon.entity.EntityUtil;
import org.apache.falcon.entity.FeedHelper;
import org.apache.falcon.entity.FeedInstanceStatus;
import org.apache.falcon.entity.ProcessHelper;
import org.apache.falcon.entity.v0.Entity;
import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.Frequency;
import org.apache.falcon.entity.v0.feed.Cluster;
import org.apache.falcon.entity.v0.feed.Feed;
import org.apache.falcon.entity.v0.feed.Sla;
import org.apache.falcon.entity.v0.process.Clusters;
import org.apache.falcon.entity.v0.process.Process;
import org.apache.falcon.expression.ExpressionHelper;
import org.apache.falcon.hadoop.HadoopClientFactory;
import org.apache.falcon.jdbc.MonitoringJdbcStateStore;
import org.apache.falcon.persistence.MonitoredEntityBean;
import org.apache.falcon.persistence.PendingInstanceBean;
import org.apache.falcon.resource.InstancesResult;
import org.apache.falcon.resource.SchedulableEntityInstance;
import org.apache.falcon.security.CurrentUser;
import org.apache.falcon.util.DateUtil;
import org.apache.falcon.util.DeploymentUtil;
import org.apache.falcon.util.RuntimeProperties;
import org.apache.falcon.util.StartupProperties;
import org.apache.falcon.workflow.WorkflowEngineFactory;
import org.apache.falcon.workflow.engine.AbstractWorkflowEngine;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.falcon.entity.EntityUtil.getStartTime;
import static org.apache.falcon.util.DateUtil.now;

/**
 * Service to monitor Feed SLAs.
 */
public final class EntitySLAMonitoringService implements ConfigurationChangeListener, FalconService {
    private static final Logger LOG = LoggerFactory.getLogger(EntitySLAMonitoringService.class);

    private static final MonitoringJdbcStateStore MONITORING_JDBC_STATE_STORE = new MonitoringJdbcStateStore();

    private static final int ONE_MS = 1;

    private static final EntitySLAMonitoringService SERVICE = new EntitySLAMonitoringService();

    static final String TAG_CRITICAL = "Missed-SLA-High";
    static final String TAG_WARN = "Missed-SLA-Low";
    private static final long MINUTE_DELAY = 60000L;

    private static final List<LifeCycle> PROCESS_LIFE_CYCLE =
            Collections.singletonList(LifeCycle.valueOf(LifeCycle.EXECUTION.name()));

    private EntitySLAMonitoringService() {

    }

    public static EntitySLAMonitoringService get() {
        return SERVICE;
    }

    /**
     * Permissions for storePath.
     */
    private static final FsPermission STORE_PERMISSION = new FsPermission(FsAction.ALL, FsAction.NONE, FsAction.NONE);


    /**
     * Frequency in seconds of "status check" for pending entity instances.
     */
    private int statusCheckFrequencySeconds; // 10 minutes


    /**
     * Time Duration (in milliseconds) in future for generating pending entity instances.
     *
     * In every cycle pending entity instances are added for monitoring, till this time in future.
     */
    private int lookAheadWindowMillis; // 15 MINUTES


    /**
     * Filesystem used for serializing and deserializing.
     */
    private FileSystem fileSystem;

    /**
     * Working directory for the entity sla monitoring service.
     */
    private Path storePath;

    /**
     * Path to store the state of the monitoring service.
     */
    private Path filePath;

    @Override
    public void onAdd(Entity entity) throws FalconException {
        startEntityMonitoring(entity, false);
    }

    private void startEntityMonitoring(Entity entity, boolean isEntityUpdated) throws FalconException{
        Set<String> currentClusters = DeploymentUtil.getCurrentClusters();
        Set<String> clustersDefined = EntityUtil.getClustersDefined(entity);
        if (entity.getEntityType() == EntityType.FEED) {
            Feed feed = (Feed) entity;
            // currently sla service for feed is enabled only for fileSystemStorage
            if (feed.getLocations() != null || feed.getSla() != null || checkFeedClusterSLA(feed)) {
                for (String cluster : clustersDefined) {
                    if (currentClusters.contains(cluster)) {
                        if (FeedHelper.getSLA(cluster, feed) != null) {
                            LOG.debug("Adding feed:{} for monitoring", feed.getName());
                            if (isEntityUpdated) {
                                MONITORING_JDBC_STATE_STORE.putMonitoredEntity(feed.getName(),
                                        EntityType.FEED.toString(), now());
                            } else {
                                MONITORING_JDBC_STATE_STORE.putMonitoredEntity(feed.getName(),
                                        EntityType.FEED.toString(), getStartTime(entity, cluster));
                            }
                        }
                    }
                }
            }
        } else if (entity.getEntityType() == EntityType.PROCESS) {
            Process process = (Process) entity;
            if (process.getSla() != null || checkProcessClusterSLA(process)) {
                for (String cluster : clustersDefined) {
                    if (currentClusters.contains(cluster)) {
                        LOG.debug("Adding process:{} for monitoring", process.getName());
                        if (isEntityUpdated) {
                            MONITORING_JDBC_STATE_STORE.putMonitoredEntity(process.getName(),
                                    EntityType.PROCESS.toString(), now());
                        } else {
                            MONITORING_JDBC_STATE_STORE.putMonitoredEntity(process.getName(),
                                    EntityType.PROCESS.toString(), getStartTime(entity, cluster));
                        }
                    }
                }
            }
        }
    }

    private Boolean checkFeedClusterSLA(Feed feed){
        for(Cluster  cluster : feed.getClusters().getClusters()){
            Sla sla =  FeedHelper.getSLA(cluster, feed);
            if (sla != null){
                return true;
            }
        }
        return false;
    }


    private Boolean checkProcessClusterSLA(Process process){
        Clusters clusters = process.getClusters();
        for(org.apache.falcon.entity.v0.process.Cluster  cluster : clusters.getClusters()){
            org.apache.falcon.entity.v0.process.Sla sla =  ProcessHelper.getSLA(cluster, process);
            if (sla != null){
                return true;
            }
        }
        return false;
    }

    @Override
    public void onRemove(Entity entity) throws FalconException {
        Set<String> currentClusters = DeploymentUtil.getCurrentClusters();
        if (entity.getEntityType() == EntityType.FEED) {
            Feed feed = (Feed) entity;
            // currently sla service is enabled only for fileSystemStorage
            if (feed.getSla() != null && feed.getLocations() != null) {
                for (Cluster cluster : feed.getClusters().getClusters()) {
                    if (currentClusters.contains(cluster.getName()) && FeedHelper.getSLA(cluster, feed) != null) {
                        LOG.debug("Removing feed:{} for monitoring", feed.getName());
                        MONITORING_JDBC_STATE_STORE.deleteMonitoringEntity(feed.getName(), EntityType.FEED.toString());
                        MONITORING_JDBC_STATE_STORE.deletePendingInstances(feed.getName(), cluster.getName(),
                                EntityType.FEED.toString());
                        LOG.debug("Removing feed:{} for monitoring", feed.getName());
                    }
                }
            }
        }
        if (entity.getEntityType() == EntityType.PROCESS){
            Process process = (Process) entity;
            if (process.getSla() != null){
                for (org.apache.falcon.entity.v0.process.Cluster  cluster : process.getClusters().getClusters()) {
                    if (currentClusters.contains(cluster.getName())) {
                        LOG.debug("Removing feed:{} for monitoring", process.getName());
                        MONITORING_JDBC_STATE_STORE.deleteMonitoringEntity(process.getName(),
                                EntityType.PROCESS.toString());
                        MONITORING_JDBC_STATE_STORE.deletePendingInstances(process.getName(), cluster.getName(),
                                EntityType.PROCESS.toString());
                    }
                }
            }
        }
    }

    private boolean isSLAMonitoringEnabledInCurrentColo(Feed feed) {
        if (feed.getLocations() != null) {
            Set<String> currentClusters = DeploymentUtil.getCurrentClusters();
            for (Cluster cluster : feed.getClusters().getClusters()) {
                if (currentClusters.contains(cluster.getName()) && FeedHelper.getSLA(cluster, feed) != null) {
                    return true;
                }
            }
        }
        return false;
    }

    private boolean isSLAMonitoringEnabledInCurrentColo(Process process) {

        Set<String> currentClusters = DeploymentUtil.getCurrentClusters();
        for (org.apache.falcon.entity.v0.process.Cluster  cluster : process.getClusters().getClusters()) {
            if (currentClusters.contains(cluster.getName()) && ProcessHelper.getSLA(cluster, process) != null) {
                return true;
            }
        }
        return false;
    }

    @Override
    public void onChange(Entity oldEntity, Entity newEntity) throws FalconException {
        if (newEntity.getEntityType() == EntityType.FEED) {
            Feed oldFeed = (Feed) oldEntity;
            Feed newFeed = (Feed) newEntity;
            if (!isSLAMonitoringEnabledInCurrentColo(newFeed)) {
                onRemove(oldFeed);
            } else if (!isSLAMonitoringEnabledInCurrentColo(oldFeed)) {
                onAdd(newFeed);
            } else {
                List<String> slaRemovedClusters = new ArrayList<>();
                for (String oldCluster : EntityUtil.getClustersDefinedInColos(oldFeed)) {
                    if (FeedHelper.getSLA(oldCluster, oldFeed) != null
                        && FeedHelper.getSLA(oldCluster, newFeed) == null) {
                        slaRemovedClusters.add(oldCluster);
                    }
                }
                updatePendingInstances(newFeed.getName(), slaRemovedClusters, EntityType.FEED.toString());
            }
        }
        if (newEntity.getEntityType() == EntityType.PROCESS) {
            Process oldProcess = (Process) oldEntity;
            Process newProcess = (Process) newEntity;
            if (!isSLAMonitoringEnabledInCurrentColo(newProcess)){
                onRemove(oldProcess);
            } else if (!isSLAMonitoringEnabledInCurrentColo(oldProcess)){
                onAdd(newProcess);
            } else {
                List<String> slaRemovedClusters = new ArrayList<>();
                for (String oldCluster : EntityUtil.getClustersDefined(oldProcess)){
                    if (ProcessHelper.getSLA(oldCluster, oldProcess) != null
                            && ProcessHelper.getSLA(oldCluster, newProcess) == null){
                        slaRemovedClusters.add(oldCluster);
                    }
                }
                updatePendingInstances(newProcess.getName(), slaRemovedClusters, EntityType.PROCESS.toString());
            }
        }
    }

    private void updatePendingInstances(String entityName, List<String> slaRemovedClusters, String entityType){
        for(String clusterName :slaRemovedClusters){
            MONITORING_JDBC_STATE_STORE.deletePendingInstances(entityName, clusterName,
                    entityType);
        }
    }

    @Override
    public void onReload(Entity entity) throws FalconException {
    }

    @Override
    public String getName() {
        return EntitySLAMonitoringService.class.getSimpleName();
    }

    @Override
    public void init() throws FalconException {

        String freq = StartupProperties.get().getProperty("entity.sla.statusCheck.frequency.seconds", "600");
        statusCheckFrequencySeconds = Integer.parseInt(freq);

        freq = StartupProperties.get().getProperty("entity.sla.lookAheadWindow.millis", "900000");
        lookAheadWindowMillis = Integer.parseInt(freq);
        ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
        addPendingEntityInstances(now());
        executor.scheduleWithFixedDelay(new Monitor(), 0, statusCheckFrequencySeconds, TimeUnit.SECONDS);
    }

    public void makeFeedInstanceAvailable(String feedName, String clusterName, Date nominalTime)
        throws FalconException {
        LOG.debug("Removing {} feed's instance {} in cluster {} from pendingSLA", feedName,
                clusterName, nominalTime);
        List<Date> instances = (MONITORING_JDBC_STATE_STORE.getNominalInstances(feedName, clusterName,
                EntityType.FEED.toString()));
        // Slas for feeds not having sla tag are not stored.
        if (CollectionUtils.isEmpty(instances)){
            MONITORING_JDBC_STATE_STORE.deletePendingInstance(feedName, clusterName, nominalTime,
                    EntityType.FEED.toString());
        }
    }

    private FileSystem initializeFileSystem() {
        try {
            fileSystem = HadoopClientFactory.get().createFalconFileSystem(storePath.toUri());
            if (!fileSystem.exists(storePath)) {
                LOG.info("Creating directory for pending entity instances: {}", storePath);
                // set permissions so config store dir is owned by falcon alone
                HadoopClientFactory.mkdirs(fileSystem, storePath, STORE_PERMISSION);
            }
            return fileSystem;
        } catch (Exception e) {
            throw new RuntimeException("Unable to bring up entity sla store for path: " + storePath, e);
        }
    }

    @Override
    public void destroy() throws FalconException {
    }

    //Periodically update status of pending instances, add new instances and take backup.
    private class Monitor implements Runnable {

        @Override
        public void run() {
            try {
                if (MONITORING_JDBC_STATE_STORE.getAllMonitoredEntities().size() > 0) {
                    checkPendingInstanceAvailability();

                    // add Instances from last checked time to 10 minutes from now(some buffer for status check)
                    Date newCheckPointTime = new Date(now().getTime() + lookAheadWindowMillis);
                    addPendingEntityInstances(newCheckPointTime);
                } else {
                    LOG.debug("No entities present for sla monitoring.");
                }
            } catch (Throwable e) {
                LOG.error("Feed SLA monitoring failed: ", e);
            }
        }
    }

    private void addPendingInstances(String entityType, Entity entity,
                                     String clusterName,
                                     List<Date> instances) throws FalconException {
        if (instances != null && !instances.isEmpty()) {
            for (Date date : instances) {
                LOG.debug("Adding pending instance ={} for entity= {} in cluster>={} and entityType={}", date,
                        entity.getName(), clusterName, entityType);
                MONITORING_JDBC_STATE_STORE.putPendingInstances(entity.getName(), clusterName, date,
                        entityType);
            }
        }
    }

    private void addPendingEntityInstances(Date checkPointTime) throws FalconException {
        Set<String> currentClusters = DeploymentUtil.getCurrentClusters();
        List<MonitoredEntityBean> entityBeanList = MONITORING_JDBC_STATE_STORE.getAllMonitoredEntities();
        for(MonitoredEntityBean monitoredEntityBean : entityBeanList) {
            String entityName = monitoredEntityBean.getEntityName();
            String entityType = monitoredEntityBean.getEntityType();
            if (EntityType.FEED.name().equalsIgnoreCase(entityType)
                    || isEntityRunning(EntityUtil.getEntity(entityType, entityName))) {
                Date lastMonitoredInstanceTime = monitoredEntityBean.getLastMonitoredTime();
                Date newCheckPointTime = checkPointTime;
                Entity entity = EntityUtil.getEntity(entityType, entityName);
                Set<String> clustersDefined = EntityUtil.getClustersDefined(entity);
                List<org.apache.falcon.entity.v0.cluster.Cluster> clusters = new ArrayList();
                for (String cluster : clustersDefined) {
                    clusters.add(ClusterHelper.getCluster(cluster));
                }
                for (org.apache.falcon.entity.v0.cluster.Cluster entityCluster : clusters) {
                    if (currentClusters.contains(entityCluster.getName())) {
                        Date endTime = EntityUtil.getEndTime(entity, entityCluster.getName());
                        if (endTime.before(now())) {
                            newCheckPointTime = endTime;
                        }
                        List<Date> instances = EntityUtil.getEntityInstanceTimesInBetween(entity,
                                entityCluster.getName(), lastMonitoredInstanceTime, newCheckPointTime);
                        addPendingInstances(entityType, entity, entityCluster.getName(), instances);
                        // update last monitored time with the new checkpoint time
                        MONITORING_JDBC_STATE_STORE.updateLastMonitoredTime(entityName, entityType,
                                new Date(newCheckPointTime.getTime() + MINUTE_DELAY));
                    }
                }
            }
        }
    }


    /**
     * Checks the availability of all the pendingInstances and removes the ones which have become available.
     */
    private void checkPendingInstanceAvailability() throws FalconException {
        List<PendingInstanceBean> pendingInstanceBeans = MONITORING_JDBC_STATE_STORE.getAllPendingInstances();
        if (pendingInstanceBeans.isEmpty()){
            LOG.info("No pending instances to be checked");
            return;
        }
        for(PendingInstanceBean pendingInstanceBean : pendingInstanceBeans){
            boolean status = checkEntityInstanceAvailability(pendingInstanceBean.getEntityName(),
                    pendingInstanceBean.getClusterName(), pendingInstanceBean.getNominalTime(),
                    pendingInstanceBean.getEntityType());
            if (status) {
                MONITORING_JDBC_STATE_STORE.deletePendingInstance(pendingInstanceBean.getEntityName(),
                        pendingInstanceBean.getClusterName(), pendingInstanceBean.getNominalTime(),
                        pendingInstanceBean.getEntityType());
            }
        }
    }

    // checks whether a given entity instance is available or not
    private boolean checkEntityInstanceAvailability(String entityName, String clusterName, Date nominalTime,
                                                    String entityType) throws FalconException {
        Entity entity = EntityUtil.getEntity(entityType, entityName);
        authenticateUser(entity);
        try {
            if (entity.getEntityType() == EntityType.PROCESS){
                LOG.trace("Checking instance availability status for entity:{}, cluster:{}, "
                        + "instanceTime:{}", entity.getName(), clusterName, nominalTime, entityType);
                AbstractWorkflowEngine wfEngine = WorkflowEngineFactory.getWorkflowEngine();
                InstancesResult instancesResult = wfEngine.getStatus(entity, nominalTime,
                        new Date(nominalTime.getTime() + 200), PROCESS_LIFE_CYCLE, false);
                if (instancesResult.getInstances().length > 0) {
                    if (instancesResult.getInstances()[0].status.equals(InstancesResult.WorkflowStatus.SUCCEEDED)){
                        LOG.trace("Entity instance(Process:{}, cluster:{}, instanceTime:{}) is available.",
                                entity.getName(), clusterName, nominalTime);
                        return true;
                    }
                } else if ((System.currentTimeMillis() - nominalTime.getTime())/(1000*60*60*24) >= Integer.parseInt(
                        RuntimeProperties.get().getProperty("workflow.history.expiration.period.days", "7"))) {
                    return true;
                }
                return false;
            }
            if (entity.getEntityType() == EntityType.FEED){
                LOG.trace("Checking instance availability status for feed:{}, cluster:{}, instanceTime:{}",
                        entity.getName(), clusterName, nominalTime);

                FeedInstanceStatus.AvailabilityStatus status = FeedHelper.getFeedInstanceStatus((Feed) entity,
                        clusterName, nominalTime);
                if (status.equals(FeedInstanceStatus.AvailabilityStatus.AVAILABLE)
                        || status.equals(FeedInstanceStatus.AvailabilityStatus.EMPTY)) {
                    LOG.trace("Feed instance(feed:{}, cluster:{}, instanceTime:{}) is available.", entity.getName(),
                            clusterName, nominalTime);
                    return true;
                }
            }
        } catch (Throwable e) {
            LOG.error("Couldn't find status for Entity:{}, cluster:{}, entityType{}, nominalTime{}", entityName,
                    clusterName, entityType, nominalTime, e);
        }
        LOG.debug("Entity instance(entity:{}, cluster:{}, instanceTime:{}) is not available.", entity.getName(),
                clusterName, nominalTime);
        return false;
    }


    /**
     * Returns all the instances between given time range which have missed slaLow or slaHigh for given entity.
     *
     * Only entities which have defined sla in their definition are considered.
     * Only the entity instances between the given time range are considered.
     * Start time and end time are both inclusive.
     * @param start start time, inclusive
     * @param end end time, inclusive
     * @return Set of pending entity instances belonging to the given range which have missed SLA
     * @throws FalconException
     */
    public Set<SchedulableEntityInstance> getEntitySLAMissPendingAlerts(Date start, Date end)
        throws FalconException {
        Set<SchedulableEntityInstance> result = new HashSet<>();
        for(PendingInstanceBean pendingInstanceBean : MONITORING_JDBC_STATE_STORE.getAllPendingInstances()){
            Pair<String, String> entityClusterPair = new Pair<>(pendingInstanceBean.getEntityName(),
                    pendingInstanceBean.getClusterName());
            String entityType = pendingInstanceBean.getEntityType();
            if (entityType.equalsIgnoreCase(EntityType.FEED.toString())){
                Feed feed = EntityUtil.getEntity(entityType, entityClusterPair.first);
                Cluster cluster = FeedHelper.getCluster(feed, entityClusterPair.second);
                Sla sla = FeedHelper.getSLA(cluster, feed);
                if (sla != null) {
                    Set<Pair<Date, String>> slaStatus = getFeedSLAStatus(sla, start, end,
                        MONITORING_JDBC_STATE_STORE.getNominalInstances(entityClusterPair.first,
                            entityClusterPair.second, entityType));
                    for (Pair<Date, String> status : slaStatus) {
                        SchedulableEntityInstance instance = new SchedulableEntityInstance(entityClusterPair.first,
                            entityClusterPair.second, status.first, EntityType.FEED);
                        instance.setTags(status.second);
                        result.add(instance);
                    }
                }
            } else {
                Process process = EntityUtil.getEntity(entityType, entityClusterPair.first);
                org.apache.falcon.entity.v0.process.Cluster cluster = ProcessHelper.getCluster(process,
                        entityClusterPair.second);
                org.apache.falcon.entity.v0.process.Sla sla = ProcessHelper.getSLA(cluster, process);
                if (sla != null && isEntityRunning(process)){
                    Set<Pair<Date, String>> slaStatus = getProcessSLAStatus(sla, start, end,
                            MONITORING_JDBC_STATE_STORE.getNominalInstances(entityClusterPair.first,
                                    entityClusterPair.second, entityType));
                    for (Pair<Date, String> status : slaStatus) {
                        SchedulableEntityInstance instance = new SchedulableEntityInstance(entityClusterPair.first,
                                entityClusterPair.second, status.first, EntityType.PROCESS);
                        instance.setTags(status.second);
                        result.add(instance);
                    }
                }
            }
        }
        return result;
    }

    /**
     * Returns all the instances of a given entity between the given time range
     * which missed sla.Only those instances are included which have missed either slaLow or slaHigh.
     * @param entityName name of the feed
     * @param clusterName cluster name
     * @param start start time, inclusive
     * @param end end time, inclusive
     * @return Pending instances of the given entity which belong to the given time range and have missed SLA.
     * @throws FalconException
     */
    public Set<SchedulableEntityInstance> getEntitySLAMissPendingAlerts(String entityName, String clusterName,
                                          Date start, Date end, String entityType) throws FalconException {
        Set<SchedulableEntityInstance> result = new HashSet<>();
        List<Date> missingInstances = MONITORING_JDBC_STATE_STORE.getNominalInstancesBetweenTimeRange(entityName,
                clusterName, entityType, start, end);
        if (missingInstances == null){
            return result;
        }
        Entity entity = EntityUtil.getEntity(entityType, entityName);

        if (entity.getEntityType() == EntityType.FEED) {
            Sla sla = FeedHelper.getSLA(clusterName, (Feed) entity);

            if (sla != null) {
                Set<Pair<Date, String>> slaStatus = getFeedSLAStatus(sla, start, end, missingInstances);
                for (Pair<Date, String> status : slaStatus){
                    SchedulableEntityInstance instance = new SchedulableEntityInstance(entityName, clusterName,
                                status.first, EntityType.FEED);
                    instance.setTags(status.second);
                    result.add(instance);
                }
            }
            return result;
        } else {
            org.apache.falcon.entity.v0.process.Sla sla = ProcessHelper.getSLA(clusterName, (Process) entity);
            if (sla != null){
                Set<Pair<Date, String>> slaStatus = getProcessSLAStatus(sla, start, end, missingInstances);
                for (Pair<Date, String> status : slaStatus){
                    SchedulableEntityInstance instance = new SchedulableEntityInstance(entityName, clusterName,
                            status.first, EntityType.PROCESS);
                    instance.setTags(status.second);
                    result.add(instance);
                }
            }
        }
        return result;
    }

    Set<Pair<Date, String>> getFeedSLAStatus(Sla sla, Date start, Date end, List<Date> missingInstances)
        throws FalconException {
        Date now = new Date();
        Frequency slaLow = sla.getSlaLow();
        Frequency slaHigh = sla.getSlaHigh();
        Set<Pair<Date, String>> result = new HashSet<>();
        for (Date nominalTime : missingInstances) {
            if (!nominalTime.before(start) && !nominalTime.after(end)) {
                ExpressionHelper.setReferenceDate(nominalTime);
                ExpressionHelper evaluator = ExpressionHelper.get();
                Long slaHighDuration = evaluator.evaluate(slaHigh.toString(), Long.class);
                Long slaLowDuration = evaluator.evaluate(slaLow.toString(), Long.class);
                Date slaCriticalTime = new Date(nominalTime.getTime() + slaHighDuration);
                Date slaWarnTime = new Date(nominalTime.getTime() + slaLowDuration);
                if (slaCriticalTime.before(now)) {
                    result.add(new Pair<>(nominalTime, TAG_CRITICAL));
                } else if (slaWarnTime.before(now)) {
                    result.add(new Pair<>(nominalTime, TAG_WARN));
                }
            }
        }
        return result;
    }

    private Set<Pair<Date, String>> getProcessSLAStatus(org.apache.falcon.entity.v0.process.Sla sla, Date start,
                                                        Date end, List<Date> missingInstances) throws FalconException {
        Date now = new Date();
        Frequency slaHigh = sla.getShouldEndIn();
        Set<Pair<Date, String>> result = new HashSet<>();
        for (Date nominalTime : missingInstances) {
            if (!nominalTime.before(start) && !nominalTime.after(end)) {
                ExpressionHelper.setReferenceDate(nominalTime);
                ExpressionHelper evaluator = ExpressionHelper.get();
                Long slaHighDuration = evaluator.evaluate(slaHigh.toString(), Long.class);
                Date slaCriticalTime = new Date(nominalTime.getTime() + slaHighDuration);
                if (slaCriticalTime.before(now)) {
                    result.add(new Pair<>(nominalTime, TAG_CRITICAL));
                }
            }
        }
        return result;
    }

    @VisibleForTesting
    Date getInitialStartTime(Entity entity, String clusterName, String entityType) throws FalconException {
        if (entity.getEntityType() == EntityType.FEED){
            Sla sla = FeedHelper.getSLA(clusterName, (Feed) entity);
            if (sla == null) {
                throw new IllegalStateException("InitialStartTime can not be determined as the feed: "
                    + entity.getName() + " and cluster: " + clusterName + " does not have any sla");
            }
            Date startTime = FeedHelper.getFeedValidityStart((Feed) entity, clusterName);
            Frequency slaLow = sla.getSlaLow();
            Date slaTime = new Date(now().getTime() - DateUtil.getFrequencyInMillis(slaLow));
            return startTime.before(slaTime) ? startTime : slaTime;
        } else{
            org.apache.falcon.entity.v0.process.Sla sla = ProcessHelper.getSLA(clusterName, (Process)entity);
            if (sla == null) {
                throw new IllegalStateException("InitialStartTime can not be determined as the feed: "
                        + entity.getName() + " and cluster: " + clusterName + " does not have any sla");
            }
            Date startTime = ProcessHelper.getProcessValidityStart((Process) entity, clusterName);
            Frequency slaLow = sla.getShouldEndIn();
            Date slaTime = new Date(now().getTime() - DateUtil.getFrequencyInMillis(slaLow));
            return startTime.before(slaTime) ? startTime : slaTime;
        }
    }

    public void makeProcessInstanceAvailable(String clusterName, String entityName, String date, String entityType) {
        Date nominalTime = null;
        try {
            nominalTime = DateUtil.parseDateFalconTZ(date);
        }catch (ParseException e){
            LOG.error("Exception while translating the date:", e);
        }
        if (nominalTime!= null){
            List<Date> instances = (MONITORING_JDBC_STATE_STORE.getNominalInstances(entityName, clusterName,
                entityType));
            if (!CollectionUtils.isEmpty(instances)){
                MONITORING_JDBC_STATE_STORE.deletePendingInstance(entityName, clusterName, nominalTime,
                    entityType);
            }
        }
    }

    // Authenticate user only if not already authenticated.
    private void authenticateUser(Entity entity){
        if (!CurrentUser.isAuthenticated()) {
            if (StringUtils.isNotBlank(entity.getACL().getOwner())) {
                CurrentUser.authenticate(entity.getACL().getOwner());
            } else {
                CurrentUser.authenticate(System.getProperty("user.name"));
            }
        }
    }


    private boolean isEntityRunning(Entity entity) throws FalconException {
        authenticateUser(entity);
        AbstractWorkflowEngine workflowEngine = WorkflowEngineFactory.getWorkflowEngine();
        return workflowEngine.isActive(entity) && !workflowEngine.isSuspended(entity)
                && !workflowEngine.isCompleted(entity);
    }
}