| /** |
| * 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 java.text.ParseException; |
| import java.util.HashSet; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Date; |
| 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.Pair; |
| import org.apache.falcon.entity.FeedInstanceStatus; |
| import org.apache.falcon.entity.ClusterHelper; |
| import org.apache.falcon.entity.EntityUtil; |
| import org.apache.falcon.entity.FeedHelper; |
| 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.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.APIResult; |
| 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.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.apache.falcon.entity.v0.process.Process; |
| |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| |
| /** |
| * Service to monitor Feed SLAs. |
| */ |
| public final class EntitySLAMonitoringService implements ConfigurationChangeListener, FalconService { |
| private static final Logger LOG = LoggerFactory.getLogger("FeedSLA"); |
| |
| private static final MonitoringJdbcStateStore MONITORING_JDBC_STATE_STORE = new MonitoringJdbcStateStore(); |
| |
| private static final int ONE_MS = 1; |
| |
| private static final EntitySLAMonitoringService SERVICE = new EntitySLAMonitoringService(); |
| |
| public static final String TAG_CRITICAL = "Missed-SLA-High"; |
| public static final String TAG_WARN = "Missed-SLA-Low"; |
| |
| 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 feed instances. |
| */ |
| private int statusCheckFrequencySeconds; // 10 minutes |
| |
| |
| /** |
| * Time Duration (in milliseconds) in future for generating pending feed instances. |
| * |
| * In every cycle pending feed 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 feed 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 { |
| Set<String> currentClusters = DeploymentUtil.getCurrentClusters(); |
| if (entity.getEntityType() == EntityType.FEED) { |
| Feed feed = (Feed) entity; |
| // currently sla service is enabled only for fileSystemStorage |
| if (feed.getLocations() != null || feed.getSla() != null || checkFeedClusterSLA(feed)) { |
| for (Cluster cluster : feed.getClusters().getClusters()) { |
| if (currentClusters.contains(cluster.getName())) { |
| if (FeedHelper.getSLA(cluster, feed) != null) { |
| LOG.debug("Adding feed:{} for monitoring", feed.getName()); |
| MONITORING_JDBC_STATE_STORE.putMonitoredEntity(feed.getName(), EntityType.FEED.toString()); |
| break; |
| } |
| } |
| } |
| } |
| } |
| if (entity.getEntityType() == EntityType.PROCESS){ |
| Process process = (Process) entity; |
| if (process.getSla() != null || checkProcessClusterSLA(process)){ |
| for (org.apache.falcon.entity.v0.process.Cluster cluster : process.getClusters().getClusters()) { |
| if (currentClusters.contains(cluster.getName())) { |
| LOG.debug("Adding process:{} for monitoring", process.getName()); |
| MONITORING_JDBC_STATE_STORE.putMonitoredEntity(process.getName(), |
| EntityType.PROCESS.toString()); |
| break; |
| } |
| } |
| } |
| } |
| } |
| |
| public Boolean checkFeedClusterSLA(Feed feed){ |
| for(Cluster cluster : feed.getClusters().getClusters()){ |
| Sla sla = FeedHelper.getSLA(cluster, feed); |
| if (sla != null){ |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| |
| public 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.getLocations() != null) { |
| for (Cluster cluster : feed.getClusters().getClusters()) { |
| if (currentClusters.contains(cluster.getName()) && FeedHelper.getSLA(cluster, feed) != null) { |
| MONITORING_JDBC_STATE_STORE.deleteMonitoringEntity(feed.getName(), EntityType.FEED.toString()); |
| MONITORING_JDBC_STATE_STORE.deletePendingInstances(feed.getName(), cluster.getName(), |
| EntityType.FEED.toString()); |
| } |
| } |
| } |
| } |
| 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())) { |
| 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(oldProcess)){ |
| onRemove(newProcess); |
| } else if (!isSLAMonitoringEnabledInCurrentColo(newProcess)){ |
| 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()); |
| } |
| } |
| } |
| |
| 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 { |
| onAdd(entity); |
| } |
| |
| @Override |
| public String getName() { |
| return EntitySLAMonitoringService.class.getSimpleName(); |
| } |
| |
| @Override |
| public void init() throws FalconException { |
| String uri = StartupProperties.get().getProperty("feed.sla.service.store.uri"); |
| storePath = new Path(uri); |
| filePath = new Path(storePath, "feedSLAMonitoringService"); |
| fileSystem = initializeFileSystem(); |
| |
| String freq = StartupProperties.get().getProperty("feed.sla.statusCheck.frequency.seconds", "600"); |
| statusCheckFrequencySeconds = Integer.parseInt(freq); |
| |
| freq = StartupProperties.get().getProperty("feed.sla.lookAheadWindow.millis", "900000"); |
| lookAheadWindowMillis = Integer.parseInt(freq); |
| LOG.debug("No old state exists at: {}, Initializing a clean state.", filePath.toString()); |
| ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1); |
| executor.scheduleWithFixedDelay(new Monitor(), 0, statusCheckFrequencySeconds, TimeUnit.SECONDS); |
| } |
| |
| public void makeFeedInstanceAvailable(String feedName, String clusterName, Date nominalTime) |
| throws FalconException { |
| LOG.info("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 feed 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 feed 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.getAllMonitoredEntity().size() > 0) { |
| checkPendingInstanceAvailability(EntityType.FEED.toString()); |
| checkPendingInstanceAvailability(EntityType.PROCESS.toString()); |
| |
| // add Instances from last checked time to 10 minutes from now(some buffer for status check) |
| Date now = new Date(); |
| Date newCheckPoint = new Date(now.getTime() + lookAheadWindowMillis); |
| addNewPendingFeedInstances(newCheckPoint, EntityType.FEED.toString()); |
| addNewPendingFeedInstances(newCheckPoint, EntityType.PROCESS.toString()); |
| } |
| } catch (Throwable e) { |
| LOG.error("Feed SLA monitoring failed: ", e); |
| } |
| } |
| } |
| |
| |
| void addNewPendingFeedInstances(Date to, String entityType) throws FalconException { |
| Set<String> currentClusters = DeploymentUtil.getCurrentClusters(); |
| List<MonitoredEntityBean> entityBeanList = MONITORING_JDBC_STATE_STORE. |
| getAllMonitoredEntityForEntity(entityType); |
| for(MonitoredEntityBean monitoredEntityBean : entityBeanList) { |
| String entityName = monitoredEntityBean.getFeedName(); |
| Entity entity = EntityUtil.getEntity(entityType, entityName); |
| LOG.debug("entityName:"+ entityName+"entity:"+entity); |
| Set<String> clusters = EntityUtil.getClustersDefined(entity); |
| List<org.apache.falcon.entity.v0.cluster.Cluster> cluster = new ArrayList(); |
| for(String string : clusters){ |
| cluster.add(ClusterHelper.getCluster(string)); |
| } |
| for (org.apache.falcon.entity.v0.cluster.Cluster entityCluster : cluster) { |
| if (currentClusters.contains(entityCluster.getName())) { |
| // get start of instances from the database |
| Date nextInstanceTime = MONITORING_JDBC_STATE_STORE.getLastInstanceTime(entityName, |
| entityType); |
| Pair<String, String> key = new Pair<>(entity.getName(), entityCluster.getName()); |
| if (nextInstanceTime == null) { |
| nextInstanceTime = getInitialStartTime(entity, entityCluster.getName(), entityType); |
| } else { |
| nextInstanceTime = new Date(nextInstanceTime.getTime() + ONE_MS); |
| } |
| |
| Set<Date> instances = new HashSet<>(); |
| org.apache.falcon.entity.v0.cluster.Cluster currentCluster = |
| EntityUtil.getEntity(EntityType.CLUSTER, entityCluster.getName()); |
| nextInstanceTime = EntityUtil.getNextStartTime(entity, currentCluster, nextInstanceTime); |
| LOG.info("nextInstanceTime:"+ nextInstanceTime + "entityName:"+entityName); |
| Date endDate; |
| if (entityType.equals(EntityType.FEED.toString())){ |
| endDate = FeedHelper.getClusterValidity((Feed) entity, currentCluster.getName()).getEnd(); |
| }else { |
| endDate = ProcessHelper.getClusterValidity((Process) entity, |
| currentCluster.getName()).getEnd(); |
| } |
| while (nextInstanceTime.before(to) && nextInstanceTime.before(endDate)) { |
| LOG.info("Adding instance={} for <entity,cluster>={}", nextInstanceTime, key); |
| instances.add(nextInstanceTime); |
| nextInstanceTime = new Date(nextInstanceTime.getTime() + ONE_MS); |
| nextInstanceTime = EntityUtil.getNextStartTime(entity, currentCluster, nextInstanceTime); |
| } |
| |
| for(Date date:instances){ |
| MONITORING_JDBC_STATE_STORE.putPendingInstances(entity.getName(), entityCluster.getName(), date, |
| entityType); |
| } |
| } |
| } |
| } |
| } |
| |
| |
| /** |
| * Checks the availability of all the pendingInstances and removes the ones which have become available. |
| */ |
| private void checkPendingInstanceAvailability(String entityType) throws FalconException { |
| LOG.debug("Size "+MONITORING_JDBC_STATE_STORE.getAllMonitoredEntity().size()); |
| if (MONITORING_JDBC_STATE_STORE.getAllPendingInstances() == null){ |
| LOG.info("Returning as size of pending instance is zero"); |
| return; |
| } |
| for(PendingInstanceBean pendingInstanceBean : MONITORING_JDBC_STATE_STORE.getAllPendingInstances()){ |
| for (Date date : MONITORING_JDBC_STATE_STORE.getNominalInstances(pendingInstanceBean.getEntityName(), |
| pendingInstanceBean.getClusterName(), entityType)) { |
| boolean status = checkEntityInstanceAvailability(pendingInstanceBean.getEntityName(), |
| pendingInstanceBean.getClusterName(), date, entityType); |
| if (status) { |
| MONITORING_JDBC_STATE_STORE.deletePendingInstance(pendingInstanceBean.getEntityName(), |
| pendingInstanceBean.getClusterName(), date, EntityType.FEED.toString()); |
| } |
| } |
| } |
| } |
| |
| // checks whether a given feed 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 (entityType.equals(EntityType.PROCESS.toString())){ |
| LOG.debug("Checking instance availability status for entity:{}, cluster:{}, " |
| + "instanceTime:{}", entity.getName(), clusterName, nominalTime, entityType); |
| AbstractWorkflowEngine wfEngine = WorkflowEngineFactory.getWorkflowEngine(); |
| InstancesResult instancesResult = wfEngine.getStatus(entity, nominalTime, nominalTime, null, null); |
| if (instancesResult.getStatus().equals(APIResult.Status.SUCCEEDED)){ |
| LOG.debug("Entity instance(feed:{}, cluster:{}, instanceTime:{}) is available.", entity.getName(), |
| clusterName, nominalTime); |
| return true; |
| } |
| return false; |
| } |
| if (entityType.equals(EntityType.FEED.toString())){ |
| LOG.debug("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.debug("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 {@link org.apache.falcon.entity.v0.feed.Feed} instances between given time range which have missed |
| * slaLow or slaHigh. |
| * |
| * Only feeds which have defined sla in their definition are considered. |
| * Only the feed 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 feed 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> feedClusterPair = new Pair<>(pendingInstanceBean.getEntityName(), |
| pendingInstanceBean.getClusterName()); |
| Feed feed = EntityUtil.getEntity(EntityType.FEED, feedClusterPair.first); |
| Cluster cluster = FeedHelper.getCluster(feed, feedClusterPair.second); |
| Sla sla = FeedHelper.getSLA(cluster, feed); |
| if (sla != null) { |
| Set<Pair<Date, String>> slaStatus = getFeedSLAStatus(sla, start, end, |
| MONITORING_JDBC_STATE_STORE.getNominalInstances(pendingInstanceBean.getEntityName(), |
| pendingInstanceBean.getClusterName(), EntityType.FEED.toString())); |
| for (Pair<Date, String> status : slaStatus) { |
| SchedulableEntityInstance instance = new SchedulableEntityInstance(feedClusterPair.first, |
| feedClusterPair.second, status.first, EntityType.FEED); |
| instance.setTags(status.second); |
| result.add(instance); |
| } |
| } |
| } |
| return result; |
| } |
| |
| /** |
| * Returns all {@link org.apache.falcon.entity.v0.feed.Feed} instances of a given feed 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 feed instances of the given feed 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.getNominalInstances(entityName, clusterName, |
| entityType); |
| if (missingInstances == null || !Arrays.asList(EntityType.FEED.toString(), |
| EntityType.PROCESS.toString()).contains(entityType)){ |
| return result; |
| } |
| Entity entity = EntityUtil.getEntity(entityType, entityName); |
| |
| if (entityType.equals(EntityType.FEED.toString())) { |
| 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; |
| } |
| |
| 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 (entityType.equals(EntityType.FEED.toString())){ |
| 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(DateUtil.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(DateUtil.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); |
| } |
| } |
| } |
| |
| private void authenticateUser(Entity entity){ |
| if (StringUtils.isNotBlank(entity.getACL().getOwner())) { |
| CurrentUser.authenticate(entity.getACL().getOwner()); |
| } else { |
| CurrentUser.authenticate(System.getProperty("user.name")); |
| } |
| } |
| } |
