forked from lmco/streamflow
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
lmco#40: Added Topology Monitor Service that checks running topologie…
…s to ensure they stay up even after a cluster outage.
- Loading branch information
1 parent
393d0cd
commit edd5aef
Showing
13 changed files
with
610 additions
and
347 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -4,4 +4,4 @@ nbactions.xml | |
| .DS_Store | ||
| *.DS_Store | ||
| *.iml | ||
| .idea | ||
| .idea/ | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
233 changes: 119 additions & 114 deletions
233
streamflow-core/streamflow-engine/src/main/java/streamflow/engine/StormEngine.java
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
91 changes: 91 additions & 0 deletions
91
streamflow-core/streamflow-model/src/main/java/streamflow/model/config/MonitorConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| /** | ||
| * Copyright 2014 Lockheed Martin Corporation | ||
| * | ||
| * Licensed 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 streamflow.model.config; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonAnyGetter; | ||
| import com.fasterxml.jackson.annotation.JsonAnySetter; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.io.Serializable; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
|
|
||
| public class MonitorConfig implements Serializable { | ||
|
|
||
| static Logger LOG = LoggerFactory.getLogger(MonitorConfig.class); | ||
|
|
||
| private boolean enabled = false; | ||
|
|
||
| private int pollingInterval = 60; | ||
|
|
||
| public MonitorConfig() { | ||
| } | ||
|
|
||
| public boolean isEnabled() { | ||
| return Boolean.parseBoolean( | ||
| System.getProperty("monitor.enabled", Boolean.toString(enabled))); | ||
| } | ||
|
|
||
| public void setEnabled(boolean enabled) { | ||
| this.enabled = enabled; | ||
| } | ||
|
|
||
| public int getPollingInterval() { | ||
| if (System.getProperty("monitor.pollingInterval") != null) { | ||
| try { | ||
| pollingInterval = Integer.parseInt(System.getProperty("monitor.pollingInterval")); | ||
| } catch (Exception ex) { | ||
| } | ||
| } | ||
| return pollingInterval; | ||
| } | ||
|
|
||
| public void setPollingInterval(int pollingInterval) { | ||
| this.pollingInterval = pollingInterval; | ||
| } | ||
|
|
||
| @Override | ||
| public int hashCode() { | ||
| int hash = 3; | ||
| hash = 43 * hash + (this.enabled ? 1 : 0); | ||
| hash = 43 * hash + this.pollingInterval; | ||
| return hash; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean equals(Object obj) { | ||
| if (obj == null) { | ||
| return false; | ||
| } | ||
| if (getClass() != obj.getClass()) { | ||
| return false; | ||
| } | ||
| final MonitorConfig other = (MonitorConfig) obj; | ||
| if (this.enabled != other.enabled) { | ||
| return false; | ||
| } | ||
| if (this.pollingInterval != other.pollingInterval) { | ||
| return false; | ||
| } | ||
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return "MonitorConfig{" + "enabled=" + enabled + ", pollingInterval=" + pollingInterval + '}'; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
23 changes: 23 additions & 0 deletions
23
...low-core/streamflow-server/src/main/java/streamflow/server/config/GuavaServiceModule.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| package streamflow.server.config; | ||
|
|
||
| import com.google.common.collect.ImmutableSet; | ||
| import com.google.common.util.concurrent.Service; | ||
| import com.google.inject.AbstractModule; | ||
| import com.google.inject.Provides; | ||
| import streamflow.model.config.MonitorConfig; | ||
| import streamflow.model.config.StreamflowConfig; | ||
| import streamflow.server.service.TopologyMonitorService; | ||
| import streamflow.util.config.ConfigLoader; | ||
|
|
||
| import java.util.Set; | ||
|
|
||
| public class GuavaServiceModule extends AbstractModule { | ||
|
|
||
| @Override | ||
| protected void configure() { | ||
| MonitorConfig monitorConfig = ConfigLoader.getConfig().getMonitor(); | ||
| if (monitorConfig.isEnabled()) { | ||
| bind(TopologyMonitorService.class); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
71 changes: 71 additions & 0 deletions
71
...ore/streamflow-server/src/main/java/streamflow/server/service/TopologyMonitorService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| package streamflow.server.service; | ||
|
|
||
| import com.google.common.util.concurrent.AbstractScheduledService; | ||
| import com.google.inject.Inject; | ||
| import com.google.inject.Singleton; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
| import streamflow.model.Topology; | ||
| import streamflow.model.config.MonitorConfig; | ||
| import streamflow.service.TopologyService; | ||
|
|
||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| @Singleton | ||
| public class TopologyMonitorService extends AbstractScheduledService { | ||
|
|
||
| public static final Logger LOG = LoggerFactory.getLogger(TopologyMonitorService.class); | ||
|
|
||
| private TopologyService topologyService; | ||
| private MonitorConfig monitorConfig; | ||
|
|
||
| @Inject | ||
| public TopologyMonitorService(TopologyService topologyService, MonitorConfig monitorConfig) { | ||
| this.topologyService = topologyService; | ||
| this.monitorConfig = monitorConfig; | ||
| } | ||
|
|
||
| @Override | ||
| protected void startUp() throws Exception { | ||
| LOG.info("Topology Status Monitor Started..."); | ||
| } | ||
|
|
||
| @Override | ||
| protected void runOneIteration() throws Exception { | ||
| // Iterate over all of the topologies for each user to check live status | ||
| for (Topology topology : topologyService.listAllTopologies()) { | ||
| try { | ||
| // Get the current live status of the topology | ||
| String topologyStatusDesired = topology.getStatus(); | ||
| String topologyStatusActual = topologyService.getTopology(topology.getId(), topology.getUserId()).getStatus(); | ||
|
|
||
| if (topologyStatusDesired.equalsIgnoreCase("ACTIVE")) { | ||
| // Topology should be submitted, but isn't active so resubmit to desired state | ||
| if (!topologyStatusActual.equalsIgnoreCase("ACTIVE")) { | ||
| LOG.warn("Topology has a desired state of ACTIVE but is not currently deployed. " | ||
| + "Redeploying topology... ID = " + topology.getId() + ", Name = " + topology.getName()); | ||
|
|
||
| // Resubmit the topology using the same settings as originally submitted | ||
| Topology submittedTopology = topologyService.submitTopology( | ||
| topology.getId(), topology.getUserId(), topology.getClusterId(), | ||
| topology.getLogLevel(), topology.getClassLoaderPolicy()); | ||
|
|
||
| if (topology != null && topology.getStatus().equalsIgnoreCase("ACTIVE")) { | ||
| LOG.info("Topology redeploy succeeded: ID = " + topology.getId() + ", Name = " + topology.getName()); | ||
| } else { | ||
| LOG.error("Topology redeploy failed: ID = " + topology.getId() + ", Name = " + topology.getName()); | ||
| } | ||
| } | ||
| } | ||
| } catch (Exception ex) { | ||
| LOG.error("An exception occurred while checking topology status: ID = " | ||
| + topology.getId() + ", Name = " + topology.getName(), ex); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| protected Scheduler scheduler() { | ||
| return Scheduler.newFixedRateSchedule(monitorConfig.getPollingInterval(), monitorConfig.getPollingInterval(), TimeUnit.SECONDS); | ||
| } | ||
| } |
Oops, something went wrong.