Add a cluster level collector for node config settings

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/AppContext.java

@@ -17,6 +17,7 @@
 
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.metrics.AllMetrics;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.util.InstanceDetails;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.store.collector.NodeConfigCache;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.reader.ClusterDetailsEventProcessor;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
@@ -33,9 +34,13 @@
  */
 public class AppContext {
   private volatile ClusterDetailsEventProcessor clusterDetailsEventProcessor;
+  // initiate a node config cache within each AppContext space
+  // to store node config settings from ES
+  private final NodeConfigCache nodeConfigCache;
 
   public AppContext() {
     this.clusterDetailsEventProcessor = null;
+    this.nodeConfigCache = new NodeConfigCache();
   }
 
   public void setClusterDetailsEventProcessor(final ClusterDetailsEventProcessor clusterDetailsEventProcessor) {
@@ -104,4 +109,8 @@ public List<String> getPeerInstanceIps() {
             .map(InstanceDetails::getInstanceIp)
             .collect(Collectors.toList()));
   }
+
+  public NodeConfigCache getNodeConfigCache() {
+    return this.nodeConfigCache;
+  }
 }

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/rca/framework/api/flow_units/NodeConfigFlowUnit.java

@@ -24,7 +24,10 @@
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.api.summaries.HotNodeSummary;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.api.summaries.HotResourceSummary;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.store.rca.cluster.NodeKey;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
+import java.util.stream.Collectors;
 
 /**
  * a flowunit type to carry ES node configurations (queue/cache capacities, etc.)
@@ -84,6 +87,14 @@ public double readConfig(Resource resource) {
     return configSummary.getValue();
   }
 
+  /**
+   * get list of config settings that this flowunit contains
+   * @return list of config settings
+   */
+  public List<Resource> getConfigList() {
+    return new ArrayList<>(configMap.keySet());
+  }
+
   @Override
   public boolean isEmpty() {
     return configMap.isEmpty();

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/rca/framework/core/Node.java

@@ -200,6 +200,10 @@ public void setAppContext(final AppContext appContext) {
     this.appContext = appContext;
   }
 
+  protected AppContext getAppContext() {
+    return this.appContext;
+  }
+
   public InstanceDetails getInstanceDetails() {
     InstanceDetails ret = new InstanceDetails(AllMetrics.NodeRole.UNKNOWN);
     if (this.appContext != null) {

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/rca/store/collector/NodeConfigCache.java

@@ -0,0 +1,103 @@
+/*
+ * 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.
+ * A copy of the License is located at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * or in the "license" file accompanying this file. This file 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 com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.store.collector;
+
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.grpc.Resource;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.store.rca.cluster.NodeKey;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+
+/**
+ * we create a thread-safe unbounded cache instance in {@link com.amazon.opendistro.elasticsearch.performanceanalyzer.AppContext}
+ * to store the node config settings from each node. Any RCA vertex in RCA graph can read the node config directly from
+ * this cache instance. The key of this cache is NodeKey + Resource and value is the actual value of the config setting
+ * (i.e. size of write queue capacity)
+ */
+public class NodeConfigCache {
+
+  private static final int CACHE_TTL = 10;
+  private final Cache<NodeConfigKey, Double> nodeConfigCache;
+
+  //unbounded cache with TTL set to 10 mins
+  public NodeConfigCache() {
+    nodeConfigCache =
+        CacheBuilder.newBuilder()
+            .expireAfterWrite(CACHE_TTL, TimeUnit.MINUTES)
+            .build();
+  }
+
+  /**
+   * add config value into cache
+   * @param nodeKey the NodeKey of the node on which this config is collected
+   * @param config the config type
+   * @param value the config value
+   */
+  public void put(NodeKey nodeKey, Resource config, double value) {
+    nodeConfigCache.put(new NodeConfigKey(nodeKey, config), value);
+  }
+
+  /**
+   * returns the config value that is associated with the nodeKey and config
+   * @param nodeKey the NodeKey of the node
+   * @param config the config type
+   * @return the config value
+   * @throws IllegalArgumentException throws an exception if the config does not exist in cache
+   */
+  public double get(NodeKey nodeKey, Resource config) throws IllegalArgumentException {
+    Double ret = nodeConfigCache.getIfPresent(new NodeConfigKey(nodeKey, config));
+    if (ret == null) {
+      throw new IllegalArgumentException();
+    }
+    return ret;
+  }
+
+  private static class NodeConfigKey {
+    private final NodeKey nodeKey;
+    private final Resource resource;
+
+    public NodeConfigKey(final NodeKey nodeKey, final Resource resource) {
+      this.nodeKey = nodeKey;
+      this.resource = resource;
+    }
+
+    public NodeKey getNodeKey() {
+      return this.nodeKey;
+    }
+
+    public Resource getResource() {
+      return this.resource;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if (obj instanceof NodeConfigKey) {
+        NodeConfigKey key = (NodeConfigKey)obj;
+        return nodeKey.equals(key.getNodeKey()) && resource.equals(key.getResource());
+      }
+      return false;
+    }
+
+    @Override
+    public int hashCode() {
+      return new HashCodeBuilder(17, 37)
+          .append(nodeKey.hashCode())
+          .append(resource.hashCode())
+          .toHashCode();
+    }
+  }
+}

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/rca/store/collector/NodeConfigClusterCollector.java

@@ -0,0 +1,109 @@
+/*
+ * 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.
+ * A copy of the License is located at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * or in the "license" file accompanying this file. This file 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 com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.store.collector;
+
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.PerformanceAnalyzerApp;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.decisionmaker.deciders.EmptyFlowUnit;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.api.flow_units.NodeConfigFlowUnit;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.api.summaries.HotNodeSummary;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.core.NonLeafNode;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.metrics.ExceptionsAndErrors;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.metrics.RcaGraphMetrics;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.scheduler.FlowUnitOperationArgWrapper;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.store.rca.cluster.NodeKey;
+import java.util.List;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+/**
+ * Cluster level node config collector that collect node config settings from each node and
+ * store them into the {@link NodeConfigCache}
+ */
+public class NodeConfigClusterCollector extends NonLeafNode<EmptyFlowUnit> {
+
+  private static final Logger LOG = LogManager.getLogger(NodeConfigClusterCollector.class);
+  private final NodeConfigCollector nodeConfigCollector;
+
+  public NodeConfigClusterCollector(final NodeConfigCollector nodeConfigCollector) {
+    super(0, 5);
+    this.nodeConfigCollector = nodeConfigCollector;
+  }
+
+  /**
+   * read and parse the NodeConfigFlowUnit. retrieve the list of configs from the flowunit
+   * and update the cache entries that are associated with the NodeKey + config type
+   */
+  private void addNodeLevelConfigs() {
+    List<NodeConfigFlowUnit> flowUnits = nodeConfigCollector.getFlowUnits();
+    for (NodeConfigFlowUnit flowUnit : flowUnits) {
+      if (flowUnit.isEmpty() || !flowUnit.hasResourceSummary()) {
+        continue;
+      }
+      HotNodeSummary nodeSummary = flowUnit.getSummary();
+      NodeKey nodeKey = new NodeKey(nodeSummary.getNodeID(), nodeSummary.getHostAddress());
+      NodeConfigCache nodeConfigCache = getAppContext().getNodeConfigCache();
+      flowUnit.getConfigList().forEach(resource -> {
+        double value = flowUnit.readConfig(resource);
+        if (!Double.isNaN(value)) {
+          nodeConfigCache.put(nodeKey, resource, value);
+        }
+      });
+    }
+  }
+
+  @Override
+  public EmptyFlowUnit operate() {
+    addNodeLevelConfigs();
+    return new EmptyFlowUnit(System.currentTimeMillis());
+  }
+
+  @Override
+  public void generateFlowUnitListFromLocal(FlowUnitOperationArgWrapper args) {
+    LOG.debug("Collector: Executing fromLocal: {}", name());
+    long startTime = System.currentTimeMillis();
+
+    try {
+      this.operate();
+    } catch (Exception ex) {
+      LOG.error("Collector: Exception in operate", ex);
+      PerformanceAnalyzerApp.ERRORS_AND_EXCEPTIONS_AGGREGATOR.updateStat(
+          ExceptionsAndErrors.EXCEPTION_IN_OPERATE, name(), 1);
+    }
+    long duration = System.currentTimeMillis() - startTime;
+
+    PerformanceAnalyzerApp.RCA_GRAPH_METRICS_AGGREGATOR.updateStat(
+        RcaGraphMetrics.GRAPH_NODE_OPERATE_CALL, this.name(), duration);
+  }
+
+  /**
+   * NodeConfigClusterCollector does not have downstream nodes and does not emit flow units
+   */
+  @Override
+  public void persistFlowUnit(FlowUnitOperationArgWrapper args) {
+    assert true;
+  }
+
+  @Override
+  public void generateFlowUnitListFromWire(FlowUnitOperationArgWrapper args) {
+    assert true;
+  }
+
+  @Override
+  public void handleNodeMuted() {
+    assert true;
+  }
+
+}

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/rca/store/rca/remediation/NodeConfigCollector.java → src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/rca/store/collector/NodeConfigCollector.java

@@ -13,7 +13,7 @@
  *  permissions and limitations under the License.
  */
 
-package com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.store.rca.remediation;
+package com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.store.collector;
 
 import static com.amazon.opendistro.elasticsearch.performanceanalyzer.metrics.AllMetrics.ThreadPoolDimension.THREAD_POOL_TYPE;
 

src/test/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/rca/framework/api/RcaTestHelper.java

@@ -23,6 +23,8 @@
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.api.summaries.HotShardSummary;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.core.GenericSummary;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.scheduler.FlowUnitOperationArgWrapper;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.store.collector.NodeConfigCache;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.store.rca.cluster.NodeKey;
 import java.time.Clock;
 import java.util.Arrays;
 import java.util.Collections;
@@ -59,6 +61,11 @@ public void mockFlowUnits(List<ResourceFlowUnit<T>> flowUnitList) {
     this.flowUnits = flowUnitList;
   }
 
+  public double readConfig(NodeKey nodeKey, Resource resource) throws IllegalArgumentException {
+    NodeConfigCache nodeConfigCache = getAppContext().getNodeConfigCache();
+    return nodeConfigCache.get(nodeKey, resource);
+  }
+
   public void setClock(Clock clock) {
     this.clock = clock;
   }

src/test/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/store/collector/NodeConfigCacheTest.java

@@ -0,0 +1,64 @@
+/*
+ * 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.
+ * A copy of the License is located at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * or in the "license" file accompanying this file. This file 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 com.amazon.opendistro.elasticsearch.performanceanalyzer.store.collector;
+
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.GradleTaskForRca;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.api.summaries.ResourceUtil;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.store.collector.NodeConfigCache;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.store.rca.cluster.NodeKey;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+@Category(GradleTaskForRca.class)
+public class NodeConfigCacheTest {
+
+  private NodeConfigCache nodeConfigCache;
+  private NodeKey nodeKey1;
+  private NodeKey nodeKey2;
+
+  @Before
+  public void init() {
+    this.nodeConfigCache = new NodeConfigCache();
+    this.nodeKey1 = new NodeKey("node1", "127.0.0.1");
+    this.nodeKey2 = new NodeKey("node2", "127.0.0.2");
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testNonExistentKey() {
+    double val = nodeConfigCache.get(nodeKey1, ResourceUtil.WRITE_QUEUE_CAPACITY);
+    Assert.fail();
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testReadWrongKey() {
+    nodeConfigCache.put(nodeKey1, ResourceUtil.WRITE_QUEUE_CAPACITY, 2.0);
+    double val = nodeConfigCache.get(nodeKey1, ResourceUtil.WRITE_QUEUE_REJECTION);
+    Assert.fail();
+  }
+
+  @Test
+  public void testSetAndGetValue() {
+    nodeConfigCache.put(nodeKey1, ResourceUtil.WRITE_QUEUE_CAPACITY, 3.0);
+    double val = nodeConfigCache.get(nodeKey1, ResourceUtil.WRITE_QUEUE_CAPACITY);
+    Assert.assertEquals(3.0, val, 0.01);
+
+    nodeConfigCache.put(nodeKey1, ResourceUtil.WRITE_QUEUE_CAPACITY, 4.0);
+    val = nodeConfigCache.get(nodeKey1, ResourceUtil.WRITE_QUEUE_CAPACITY);
+    Assert.assertEquals(4.0, val, 0.01);
+  }
+}