-
Notifications
You must be signed in to change notification settings - Fork 123
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
13 changed files
with
308 additions
and
152 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
147 changes: 147 additions & 0 deletions
147
src/main/java/emissary/core/sentinel/protocols/Protocol.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,147 @@ | ||
| package emissary.core.sentinel.protocols; | ||
|
|
||
| import emissary.config.ConfigUtil; | ||
| import emissary.config.Configurator; | ||
| import emissary.core.Factory; | ||
| import emissary.core.Namespace; | ||
| import emissary.core.NamespaceException; | ||
| import emissary.core.sentinel.Sentinel; | ||
| import emissary.core.sentinel.protocols.actions.Action; | ||
| import emissary.core.sentinel.protocols.actions.Notify; | ||
| import emissary.core.sentinel.protocols.rules.AllMaxTime; | ||
| import emissary.core.sentinel.protocols.rules.Rule; | ||
| import emissary.directory.DirectoryPlace; | ||
| import emissary.directory.KeyManipulator; | ||
|
|
||
| import org.apache.commons.lang3.StringUtils; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.io.IOException; | ||
| import java.lang.invoke.MethodHandles; | ||
| import java.util.Collections; | ||
| import java.util.Map; | ||
| import java.util.StringJoiner; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
|
|
||
| public class Protocol { | ||
|
|
||
| private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); | ||
| protected static final String DEFAULT_RULE = "DEFAULT"; | ||
| private Configurator config; | ||
| private boolean enabled = false; | ||
| private Action action; | ||
| private final Map<String, Rule> rules = new ConcurrentHashMap<>(); // key: place, value: rule | ||
| private final Map<String, Integer> placeAgentCounts = new ConcurrentHashMap<>(); // key: place, value: number of agents in place | ||
|
|
||
| public Protocol(String conf) { | ||
| configure(conf); | ||
| } | ||
|
|
||
| public boolean isEnabled() { | ||
| return enabled; | ||
| } | ||
|
|
||
| public Map<String, Rule> getRules() { | ||
| return Collections.unmodifiableMap(rules); | ||
| } | ||
|
|
||
| public Map<String, Integer> getPlaceAgentCounts() { | ||
| return Collections.unmodifiableMap(placeAgentCounts); | ||
| } | ||
|
|
||
| public Action getAction() { | ||
| return action; | ||
| } | ||
|
|
||
| /** | ||
| * Run the configured rules over the watched mobile-agents | ||
| * | ||
| * @throws NamespaceException if there is a problem looking up resources in the {@link Namespace} | ||
| */ | ||
| public void run(Map<String, Sentinel.Tracker> trackers) { | ||
| placeAgentCounts.clear(); | ||
| for (Sentinel.Tracker tracker : trackers.values()) { | ||
| String placeKey = getPlaceKey(tracker); | ||
| if (StringUtils.isNotBlank(placeKey)) { | ||
| placeAgentCounts.put(placeKey, placeAgentCounts.getOrDefault(placeKey, 0) + 1); | ||
| } | ||
| } | ||
| logger.debug("Checking agents {}", placeAgentCounts); | ||
| for (Map.Entry<String, Integer> item : placeAgentCounts.entrySet()) { | ||
| Rule rule = rules.getOrDefault(item.getKey(), rules.get(DEFAULT_RULE)); | ||
| logger.trace("Found {} for {}", rule, item.getKey()); | ||
| check(trackers, item.getKey(), item.getValue()); | ||
| } | ||
| } | ||
|
|
||
| public String getPlaceKey(Sentinel.Tracker tracker) { | ||
| return getPlaceSimpleName(tracker.getPlaceName()); | ||
| } | ||
|
|
||
| public static String getPlaceSimpleName(String place) { | ||
| return StringUtils.substringAfterLast(place, "/"); | ||
| } | ||
|
|
||
| protected void configure(String conf) { | ||
| try { | ||
| this.config = ConfigUtil.getConfigInfo(conf); | ||
| init(); | ||
| } catch (IOException e) { | ||
| logger.warn("Cannot read " + conf + ", skipping"); | ||
| } | ||
| } | ||
|
|
||
| protected void init() { | ||
| this.enabled = config.findBooleanEntry("ENABLED", false); | ||
| if (enabled) { | ||
| logger.trace("Loading rules..."); | ||
| for (String ruleId : config.findEntries("RULE_ID")) { | ||
| try { | ||
| validate(ruleId); | ||
| Map<String, String> map = config.findStringMatchMap(ruleId + "_"); | ||
| String rule = map.getOrDefault("RULE", AllMaxTime.class.getName()); | ||
| Rule ruleImpl = (Rule) Factory.create(rule, ruleId, map.get("TIME_LIMIT_MINUTES"), map.get("THRESHOLD")); | ||
| logger.debug("Sentinel loaded {}", ruleImpl); | ||
| this.rules.put(ruleId, ruleImpl); | ||
| } catch (Exception e) { | ||
| logger.warn("Unable to configure Sentinel for {}: {}", ruleId, e.getMessage()); | ||
| } | ||
| } | ||
|
|
||
| // if no rules then disable protocol | ||
| if (this.rules.isEmpty()) { | ||
| this.enabled = false; | ||
| return; | ||
| } | ||
|
|
||
| String action = config.findStringEntry("ACTION", Notify.class.getName()); | ||
| this.action = (Action) Factory.create(action); | ||
| } | ||
| } | ||
|
|
||
| protected void validate(String place) throws NamespaceException { | ||
| // validate that the place exists | ||
| if (!DEFAULT_RULE.equalsIgnoreCase(place)) { | ||
| DirectoryPlace directoryPlace = Namespace.lookup(DirectoryPlace.class).iterator().next(); | ||
| if (directoryPlace.getEntries().stream() | ||
| .noneMatch(entry -> place.equalsIgnoreCase(KeyManipulator.getServiceClassname(entry.getFullKey())))) { | ||
| throw new IllegalStateException("Place not found in the directory"); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| protected void check(Map<String, Sentinel.Tracker> tracker, String placeSimpleName, Integer count) { | ||
| if (getRules().values().stream().allMatch(rule -> rule.condition(tracker, placeSimpleName, count))) { | ||
| getAction().trigger(tracker, placeSimpleName, count); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return new StringJoiner(", ", Protocol.class.getSimpleName() + "[", "]") | ||
| .add("rules=" + rules) | ||
| .add("action=" + action) | ||
| .toString(); | ||
| } | ||
| } |
28 changes: 28 additions & 0 deletions
28
src/main/java/emissary/core/sentinel/protocols/actions/Action.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,28 @@ | ||
| package emissary.core.sentinel.protocols.actions; | ||
|
|
||
| import emissary.core.sentinel.Sentinel; | ||
|
|
||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.lang.invoke.MethodHandles; | ||
| import java.util.Map; | ||
|
|
||
| public abstract class Action { | ||
|
|
||
| protected static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); | ||
|
|
||
| /** | ||
| * Try to terminate the JVM | ||
| * | ||
| * @param trackers the listing of agents, places, and filenames that's currently processing | ||
| * @param placeSimpleName the place name currently processing on one or more mobile agents | ||
| * @param count number of mobile agents stuck on the place | ||
| */ | ||
| public abstract void trigger(Map<String, Sentinel.Tracker> trackers, String placeSimpleName, Integer count); | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return getClass().getName(); | ||
| } | ||
| } |
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
Oops, something went wrong.