Integration with Ultrawarm - Follow up

src/main/java/com/amazon/opendistroforelasticsearch/ad/AnomalyDetectorPlugin.java

@@ -72,8 +72,8 @@
 import com.amazon.opendistroforelasticsearch.ad.transport.ThresholdResultTransportAction;
 import com.amazon.opendistroforelasticsearch.ad.transport.handler.AnomalyResultHandler;
 import com.amazon.opendistroforelasticsearch.ad.util.ClientUtil;
-import com.amazon.opendistroforelasticsearch.ad.util.ClusterStateUtils;
 import com.amazon.opendistroforelasticsearch.ad.util.ColdStartRunner;
+import com.amazon.opendistroforelasticsearch.ad.util.DiscoveryNodeFilterer;
 import com.amazon.opendistroforelasticsearch.ad.util.IndexUtils;
 import com.amazon.opendistroforelasticsearch.ad.util.Throttler;
 import com.amazon.opendistroforelasticsearch.jobscheduler.spi.JobSchedulerExtension;
@@ -92,7 +92,6 @@
 import org.elasticsearch.cluster.metadata.MetaData.Custom;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.service.ClusterService;
-import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.IndexScopedSettings;
@@ -123,7 +122,6 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Supplier;
@@ -148,7 +146,7 @@ public class AnomalyDetectorPlugin extends Plugin implements ActionPlugin, Scrip
     private ADStats adStats;
     private NamedXContentRegistry xContentRegistry;
     private ClientUtil clientUtil;
-    private ClusterStateUtils clusterStateUtils;
+    private DiscoveryNodeFilterer nodeFilter;
 
     static {
         SpecialPermission.check();
@@ -208,7 +206,7 @@ public List<RestHandler> getRestHandlers(
         RestStatsAnomalyDetectorAction statsAnomalyDetectorAction = new RestStatsAnomalyDetectorAction(
             restController,
             adStats,
-            this.clusterStateUtils
+            this.nodeFilter
         );
         RestAnomalyDetectorJobAction anomalyDetectorJobAction = new RestAnomalyDetectorJobAction(
             settings,
@@ -267,11 +265,10 @@ public Collection<Object> createComponents(
         RandomCutForestSerDe rcfSerde = new RandomCutForestSerDe();
         CheckpointDao checkpoint = new CheckpointDao(client, clientUtil, CommonName.CHECKPOINT_INDEX_NAME);
 
-        HashMap<String, String> ignoredAttributes = new HashMap<>();
-        ignoredAttributes.put(CommonName.BOX_TYPE_KEY, CommonName.WARM_BOX_TYPE);
-        this.clusterStateUtils = new ClusterStateUtils(clusterService, ignoredAttributes);
+        this.nodeFilter = new DiscoveryNodeFilterer(this.clusterService);
+
         ModelManager modelManager = new ModelManager(
-            clusterStateUtils,
+            nodeFilter,
             jvmService,
             rcfSerde,
             checkpoint,
@@ -295,7 +292,7 @@ public Collection<Object> createComponents(
             AnomalyDetectorSettings.SHINGLE_SIZE
         );
 
-        HashRing hashRing = new HashRing(clusterStateUtils, clock, settings);
+        HashRing hashRing = new HashRing(nodeFilter, clock, settings);
         ADStateManager stateManager = new ADStateManager(
             client,
             xContentRegistry,
@@ -364,12 +361,12 @@ public Collection<Object> createComponents(
                 clock,
                 stateManager,
                 runner,
-                new ADClusterEventListener(clusterService, hashRing, modelManager, clusterStateUtils),
+                new ADClusterEventListener(clusterService, hashRing, modelManager, nodeFilter),
                 deleteUtil,
                 adCircuitBreakerService,
                 adStats,
-                new MasterEventListener(clusterService, threadPool, deleteUtil, client, clock, clientUtil, clusterStateUtils),
-                clusterStateUtils
+                new MasterEventListener(clusterService, threadPool, deleteUtil, client, clock, clientUtil, nodeFilter),
+                nodeFilter
             );
     }
 

src/main/java/com/amazon/opendistroforelasticsearch/ad/cluster/ADClusterEventListener.java

@@ -20,7 +20,7 @@
 import java.util.concurrent.Semaphore;
 
 import com.amazon.opendistroforelasticsearch.ad.ml.ModelManager;
-import com.amazon.opendistroforelasticsearch.ad.util.ClusterStateUtils;
+import com.amazon.opendistroforelasticsearch.ad.util.DiscoveryNodeFilterer;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -45,27 +45,27 @@ public class ADClusterEventListener implements ClusterStateListener {
     private HashRing hashRing;
     private ModelManager modelManager;
     private final ClusterService clusterService;
-    private final ClusterStateUtils clusterStateUtils;
+    private final DiscoveryNodeFilterer nodeFilter;
 
     @Inject
     public ADClusterEventListener(
         ClusterService clusterService,
         HashRing hashRing,
         ModelManager modelManager,
-        ClusterStateUtils clusterStateUtils
+        DiscoveryNodeFilterer nodeFilter
     ) {
         this.clusterService = clusterService;
         this.clusterService.addListener(this);
         this.hashRing = hashRing;
         this.modelManager = modelManager;
         this.inProgress = new Semaphore(1);
-        this.clusterStateUtils = clusterStateUtils;
+        this.nodeFilter = nodeFilter;
     }
 
     @Override
     public void clusterChanged(ClusterChangedEvent event) {
 
-        if (clusterStateUtils.isIgnoredNode(event.state().nodes().getLocalNode())) {
+        if (!nodeFilter.isEligibleNode(event.state().nodes().getLocalNode())) {
             LOG.debug(NODE_NOT_APPLIED_MSG);
             return;
         }
@@ -88,7 +88,7 @@ public void clusterChanged(ClusterChangedEvent event) {
             // Check whether it was a data node that was removed
             boolean dataNodeRemoved = false;
             for (DiscoveryNode removedNode : delta.removedNodes()) {
-                if (!clusterStateUtils.isIgnoredNode(removedNode)) {
+                if (nodeFilter.isEligibleNode(removedNode)) {
                     LOG.info(NODE_REMOVED_MSG + " {}", removedNode.getId());
                     dataNodeRemoved = true;
                     break;
@@ -98,7 +98,7 @@ public void clusterChanged(ClusterChangedEvent event) {
             // Check whether it was a data node that was added
             boolean dataNodeAdded = false;
             for (DiscoveryNode addedNode : delta.addedNodes()) {
-                if (!clusterStateUtils.isIgnoredNode(addedNode)) {
+                if (nodeFilter.isEligibleNode(addedNode)) {
                     LOG.info(NODE_ADDED_MSG + " {}", addedNode.getId());
                     dataNodeAdded = true;
                     break;

src/main/java/com/amazon/opendistroforelasticsearch/ad/cluster/HashRing.java

@@ -31,8 +31,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 
-import com.amazon.opendistroforelasticsearch.ad.util.ClusterStateUtils;
-import com.carrotsearch.hppc.cursors.ObjectCursor;
+import com.amazon.opendistroforelasticsearch.ad.util.DiscoveryNodeFilterer;
 
 public class HashRing {
     private static final Logger LOG = LogManager.getLogger(HashRing.class);
@@ -43,7 +42,7 @@ public class HashRing {
     static final String COOLDOWN_MSG = "Hash ring doesn't respond to cluster state change within the cooldown period.";
 
     private final int VIRTUAL_NODE_COUNT = 100;
-    private final ClusterStateUtils clusterStateUtils;
+    private final DiscoveryNodeFilterer nodeFilter;
     private TreeMap<Integer, DiscoveryNode> circle;
     private Semaphore inProgress;
     // the UTC epoch milliseconds of the most recent successful update
@@ -52,9 +51,9 @@ public class HashRing {
     private final Clock clock;
     private AtomicBoolean membershipChangeRequied;
 
-    public HashRing(ClusterStateUtils clusterStateUtils, Clock clock, Settings settings) {
+    public HashRing(DiscoveryNodeFilterer nodeFilter, Clock clock, Settings settings) {
         this.circle = new TreeMap<Integer, DiscoveryNode>();
-        this.clusterStateUtils = clusterStateUtils;
+        this.nodeFilter = nodeFilter;
         this.inProgress = new Semaphore(1);
         this.clock = clock;
         this.coolDownPeriod = COOLDOWN_MINUTES.get(settings);
@@ -93,8 +92,7 @@ public boolean build() {
         TreeMap<Integer, DiscoveryNode> newCircle = new TreeMap<>();
 
         try {
-            for (ObjectCursor<DiscoveryNode> cursor : clusterStateUtils.getEligibleDataNodes().values()) {
-                DiscoveryNode curNode = cursor.value;
+            for (DiscoveryNode curNode : nodeFilter.getEligibleDataNodes()) {
                 for (int i = 0; i < VIRTUAL_NODE_COUNT; i++) {
                     newCircle.put(Murmur3HashFunction.hash(curNode.getId() + i), curNode);
                 }

src/main/java/com/amazon/opendistroforelasticsearch/ad/cluster/HourlyCron.java

@@ -24,24 +24,24 @@
 
 import com.amazon.opendistroforelasticsearch.ad.transport.CronAction;
 import com.amazon.opendistroforelasticsearch.ad.transport.CronRequest;
-import com.amazon.opendistroforelasticsearch.ad.util.ClusterStateUtils;
+import com.amazon.opendistroforelasticsearch.ad.util.DiscoveryNodeFilterer;
 
 public class HourlyCron implements Runnable {
     private static final Logger LOG = LogManager.getLogger(HourlyCron.class);
     static final String SUCCEEDS_LOG_MSG = "Hourly maintenance succeeds";
     static final String NODE_EXCEPTION_LOG_MSG = "Hourly maintenance of node has exception";
     static final String EXCEPTION_LOG_MSG = "Hourly maintenance has exception.";
-    private ClusterStateUtils clusterStateUtils;
+    private DiscoveryNodeFilterer nodeFilter;
     private Client client;
 
-    public HourlyCron(Client client, ClusterStateUtils clusterStateUtils) {
-        this.clusterStateUtils = clusterStateUtils;
+    public HourlyCron(Client client, DiscoveryNodeFilterer nodeFilter) {
+        this.nodeFilter = nodeFilter;
         this.client = client;
     }
 
     @Override
     public void run() {
-        DiscoveryNode[] dataNodes = clusterStateUtils.getEligibleDataNodes().values().toArray(DiscoveryNode.class);
+        DiscoveryNode[] dataNodes = nodeFilter.getEligibleDataNodes();
 
         // we also add the cancel query function here based on query text from the negative cache.
 

src/main/java/com/amazon/opendistroforelasticsearch/ad/cluster/MasterEventListener.java

@@ -26,7 +26,7 @@
 
 import com.amazon.opendistroforelasticsearch.ad.settings.AnomalyDetectorSettings;
 import com.amazon.opendistroforelasticsearch.ad.util.ClientUtil;
-import com.amazon.opendistroforelasticsearch.ad.util.ClusterStateUtils;
+import com.amazon.opendistroforelasticsearch.ad.util.DiscoveryNodeFilterer;
 
 import org.elasticsearch.threadpool.ThreadPool;
 
@@ -40,7 +40,7 @@ public class MasterEventListener implements LocalNodeMasterListener {
     private Client client;
     private Clock clock;
     private ClientUtil clientUtil;
-    private ClusterStateUtils clusterStateUtils;
+    private DiscoveryNodeFilterer nodeFilter;
 
     public MasterEventListener(
         ClusterService clusterService,
@@ -49,7 +49,7 @@ public MasterEventListener(
         Client client,
         Clock clock,
         ClientUtil clientUtil,
-        ClusterStateUtils clusterStateUtils
+        DiscoveryNodeFilterer nodeFilter
     ) {
         this.clusterService = clusterService;
         this.threadPool = threadPool;
@@ -58,14 +58,13 @@ public MasterEventListener(
         this.clusterService.addLocalNodeMasterListener(this);
         this.clock = clock;
         this.clientUtil = clientUtil;
-        this.clusterStateUtils = clusterStateUtils;
+        this.nodeFilter = nodeFilter;
     }
 
     @Override
     public void onMaster() {
         if (hourlyCron == null) {
-            hourlyCron = threadPool
-                .scheduleWithFixedDelay(new HourlyCron(client, clusterStateUtils), TimeValue.timeValueHours(1), executorName());
+            hourlyCron = threadPool.scheduleWithFixedDelay(new HourlyCron(client, nodeFilter), TimeValue.timeValueHours(1), executorName());
             clusterService.addLifecycleListener(new LifecycleListener() {
                 @Override
                 public void beforeStop() {

src/main/java/com/amazon/opendistroforelasticsearch/ad/constant/CommonName.java

@@ -36,6 +36,9 @@ public class CommonName {
     // Ultrawarm node attributes
     // ======================================
 
+    // hot node
+    public static String HOT_BOX_TYPE = "hot";
+
     // warm node
     public static String WARM_BOX_TYPE = "warm";
 

src/main/java/com/amazon/opendistroforelasticsearch/ad/ml/ModelManager.java

@@ -49,7 +49,7 @@
 import com.amazon.opendistroforelasticsearch.ad.constant.CommonErrorMessages;
 import com.amazon.opendistroforelasticsearch.ad.ml.rcf.CombinedRcfResult;
 import com.amazon.opendistroforelasticsearch.ad.model.AnomalyDetector;
-import com.amazon.opendistroforelasticsearch.ad.util.ClusterStateUtils;
+import com.amazon.opendistroforelasticsearch.ad.util.DiscoveryNodeFilterer;
 import com.amazon.randomcutforest.RandomCutForest;
 import com.amazon.randomcutforest.serialize.RandomCutForestSerDe;
 
@@ -103,7 +103,7 @@ public String getName() {
     private final Duration checkpointInterval;
 
     // dependencies
-    private final ClusterStateUtils clusterStateUtils;
+    private final DiscoveryNodeFilterer nodeFilter;
     private final JvmService jvmService;
     private final RandomCutForestSerDe rcfSerde;
     private final CheckpointDao checkpointDao;
@@ -119,7 +119,7 @@ public String getName() {
     /**
      * Constructor.
      *
-     * @param clusterStateUtils cluster info
+     * @param nodeFilter utility class to select nodesr
      * @param jvmService jvm info
      * @param rcfSerde RCF model serialization
      * @param checkpointDao model checkpoint storage
@@ -143,7 +143,7 @@ public String getName() {
      * @param shingleSize size of a shingle
      */
     public ModelManager(
-        ClusterStateUtils clusterStateUtils,
+        DiscoveryNodeFilterer nodeFilter,
         JvmService jvmService,
         RandomCutForestSerDe rcfSerde,
         CheckpointDao checkpointDao,
@@ -167,7 +167,7 @@ public ModelManager(
         int shingleSize
     ) {
 
-        this.clusterStateUtils = clusterStateUtils;
+        this.nodeFilter = nodeFilter;
         this.jvmService = jvmService;
         this.rcfSerde = rcfSerde;
         this.checkpointDao = checkpointDao;
@@ -260,7 +260,7 @@ public Entry<Integer, Integer> getPartitionedForestSizes(RandomCutForest forest,
         int numPartitions = (int) Math.ceil((double) totalSize / (double) partitionSize);
         int forestSize = (int) Math.ceil((double) forest.getNumberOfTrees() / (double) numPartitions);
 
-        int numNodes = clusterStateUtils.getEligibleDataNodes().size();
+        int numNodes = nodeFilter.getEligibleDataNodes().length;
         if (numPartitions > numNodes) {
             // partition by cluster size
             partitionSize = (long) Math.ceil((double) totalSize / (double) numNodes);

src/main/java/com/amazon/opendistroforelasticsearch/ad/rest/RestStatsAnomalyDetectorAction.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License").
  * You may not use this file except in compliance with the License.
@@ -18,7 +18,7 @@
 import com.amazon.opendistroforelasticsearch.ad.stats.ADStats;
 import com.amazon.opendistroforelasticsearch.ad.transport.ADStatsAction;
 import com.amazon.opendistroforelasticsearch.ad.transport.ADStatsRequest;
-import com.amazon.opendistroforelasticsearch.ad.util.ClusterStateUtils;
+import com.amazon.opendistroforelasticsearch.ad.util.DiscoveryNodeFilterer;
 
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -42,22 +42,22 @@ public class RestStatsAnomalyDetectorAction extends BaseRestHandler {
 
     private static final String STATS_ANOMALY_DETECTOR_ACTION = "stats_anomaly_detector";
     private ADStats adStats;
-    private ClusterStateUtils clusterStateUtils;
+    private DiscoveryNodeFilterer nodeFilter;
 
     /**
      * Constructor
      *
      * @param controller Rest Controller
      * @param adStats ADStats object
-     * @param clusterStateUtils util to get eligible data nodes
+     * @param nodeFilter util class to get eligible data nodes
      */
-    public RestStatsAnomalyDetectorAction(RestController controller, ADStats adStats, ClusterStateUtils clusterStateUtils) {
+    public RestStatsAnomalyDetectorAction(RestController controller, ADStats adStats, DiscoveryNodeFilterer nodeFilter) {
         controller.registerHandler(RestRequest.Method.GET, AD_BASE_URI + "/{nodeId}/stats/", this);
         controller.registerHandler(RestRequest.Method.GET, AD_BASE_URI + "/{nodeId}/stats/{stat}", this);
         controller.registerHandler(RestRequest.Method.GET, AD_BASE_URI + "/stats/", this);
         controller.registerHandler(RestRequest.Method.GET, AD_BASE_URI + "/stats/{stat}", this);
         this.adStats = adStats;
-        this.clusterStateUtils = clusterStateUtils;
+        this.nodeFilter = nodeFilter;
     }
 
     @Override
@@ -87,7 +87,7 @@ private ADStatsRequest getRequest(RestRequest request) {
             String[] nodeIdsArr = nodesIdsStr.split(",");
             adStatsRequest = new ADStatsRequest(nodeIdsArr);
         } else {
-            DiscoveryNode[] dataNodes = clusterStateUtils.getEligibleDataNodes().values().toArray(DiscoveryNode.class);
+            DiscoveryNode[] dataNodes = nodeFilter.getEligibleDataNodes();
             adStatsRequest = new ADStatsRequest(dataNodes);
         }