Reader changes for dynamic enable/disable of RCA graph components

pa_config/rca.conf

@@ -85,6 +85,8 @@
     "NodeTemperatureRca",
     "ClusterTemperatureRca"
   ],
+  "muted-deciders": [],
+  "muted-actions": [],
 
   "decider-config-settings": {
     // cache decider - Needs to be updated as per the performance test results

pa_config/rca_idle_master.conf

@@ -96,6 +96,8 @@
     "NodeTemperatureRca",
     "ClusterTemperatureRca"
   ],
+  "muted-deciders": [],
+  "muted-actions": [],
 
   "decider-config-settings": {
     // cache decider - Needs to be updated as per the performance test results

pa_config/rca_master.conf

@@ -96,6 +96,8 @@
     "NodeTemperatureRca",
     "ClusterTemperatureRca"
   ],
+  "muted-deciders": [],
+  "muted-actions": [],
 
   "decider-config-settings": {
     // cache decider - Needs to be updated as per the performance test results

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/config/overrides/ConfigOverridesApplier.java

@@ -0,0 +1,121 @@
+/*
+ * 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.config.overrides;
+
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.PerformanceAnalyzerApp;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.core.RcaConf;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.reader.ClusterDetailsEventProcessor;
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.commons.lang3.math.NumberUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+/**
+ * Class the manages applying the various overrides for RCAs, deciders, and action nodes as
+ * written by the NodeDetails collector and read by the {@link ClusterDetailsEventProcessor}
+ */
+public class ConfigOverridesApplier {
+
+  private static final Logger LOG = LogManager.getLogger(ConfigOverridesApplier.class);
+  private volatile long lastAppliedTimestamp;
+
+  public void applyOverride(final String overridesJson, final String lastUpdatedTimestampString) {
+    if (!valid(overridesJson, lastUpdatedTimestampString)) {
+      LOG.warn("Received invalid overrides or timestamp. Overrides json: {}, last updated "
+          + "timestamp: {}", overridesJson, lastAppliedTimestamp);
+      return;
+    }
+
+    try {
+      long lastUpdatedTimestamp = Long.parseLong(lastUpdatedTimestampString);
+      LOG.debug("Last updated(writer): {}, Last applied(reader): {}", lastUpdatedTimestamp,
+          lastAppliedTimestamp);
+      if (lastUpdatedTimestamp > lastAppliedTimestamp) {
+        apply(ConfigOverridesHelper.deserialize(overridesJson));
+      } else {
+        LOG.debug("Not applying override. Last updated timestamp {} is behind last applied "
+            + "timestamp {}", lastUpdatedTimestamp, lastAppliedTimestamp);
+      }
+    } catch (IOException ioe) {
+      LOG.error("Unable to deserialize overrides JSON:" + overridesJson, ioe);
+    }
+  }
+
+  private void apply(final ConfigOverrides overrides) {
+    if (PerformanceAnalyzerApp.getRcaController() != null && PerformanceAnalyzerApp.getRcaController().isRcaEnabled()) {
+      LOG.info("Applying overrides: {}", overrides.getEnable().getRcas());
+      RcaConf rcaConf = PerformanceAnalyzerApp.getRcaController().getRcaConf();
+      if (rcaConf != null) {
+        Set<String> currentMutedRcaSet = new HashSet<>(rcaConf.getMutedRcaList());
+        Set<String> currentMutedDeciderSet = new HashSet<>(rcaConf.getMutedDeciderList());
+        Set<String> currentMutedActionSet = new HashSet<>(rcaConf.getMutedActionList());
+        // check and remove any nodes that are in the disabled list that were enabled just now.
+        if (overrides.getEnable() != null) {
+          if (overrides.getEnable().getRcas() != null) {
+            currentMutedRcaSet.removeAll(overrides.getEnable().getRcas());
+          }
+          if (overrides.getEnable().getDeciders() != null) {
+            currentMutedDeciderSet.removeAll(overrides.getEnable().getDeciders());
+          }
+          if (overrides.getEnable().getActions() != null) {
+            currentMutedActionSet.removeAll(overrides.getEnable().getActions());
+          }
+        }
+
+        // union the remaining already disabled nodes with the new set of disabled nodes.
+        if (overrides.getDisable() != null) {
+          if (overrides.getDisable().getRcas() != null) {
+            currentMutedRcaSet.addAll(overrides.getDisable().getRcas());
+          }
+          if (overrides.getDisable().getDeciders() != null) {
+            currentMutedDeciderSet.addAll(overrides.getDisable().getDeciders());
+          }
+          if (overrides.getDisable().getActions() != null) {
+            currentMutedActionSet.addAll(overrides.getDisable().getActions());
+          }
+        }
+
+        LOG.debug("New set of muted rcas: {}", currentMutedRcaSet);
+        LOG.debug("New set of muted deciders: {}", currentMutedDeciderSet);
+        LOG.debug("New set of muted actions: {}", currentMutedActionSet);
+        rcaConf.updateRcaConf(currentMutedRcaSet, currentMutedDeciderSet, currentMutedActionSet);
+        setLastAppliedTimestamp(System.currentTimeMillis());
+      }
+    }
+  }
+
+  private boolean valid(final String overridesJson, final String timestamp) {
+    if (overridesJson == null || timestamp == null) {
+      return false;
+    }
+
+    if (overridesJson.isEmpty() || timestamp.isEmpty()) {
+      return false;
+    }
+
+    return NumberUtils.isCreatable(timestamp);
+  }
+
+  public long getLastAppliedTimestamp() {
+    return lastAppliedTimestamp;
+  }
+
+  public void setLastAppliedTimestamp(final long lastAppliedTimestamp) {
+    this.lastAppliedTimestamp = lastAppliedTimestamp;
+  }
+}

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/config/overrides/ConfigOverridesWrapper.java

@@ -15,13 +15,31 @@
 
 package com.amazon.opendistro.elasticsearch.performanceanalyzer.config.overrides;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.annotations.VisibleForTesting;
+
 /**
  * Class responsible for holding the latest config overrides across the cluster.
  */
 public class ConfigOverridesWrapper {
 
     private volatile ConfigOverrides currentClusterConfigOverrides;
     private volatile long lastUpdatedTimestamp;
+    private final ObjectMapper mapper;
+
+    public ConfigOverridesWrapper() {
+        this(new ObjectMapper());
+    }
+
+    /**
+     * Ctor used only for unit test purposes.
+     * @param mapper The object mapper instance.
+     */
+    @VisibleForTesting
+    public ConfigOverridesWrapper(final ObjectMapper mapper) {
+        this.currentClusterConfigOverrides = new ConfigOverrides();
+        this.mapper = mapper;
+    }
 
     public ConfigOverrides getCurrentClusterConfigOverrides() {
         return currentClusterConfigOverrides;

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/decisionmaker/actions/Action.java

@@ -15,6 +15,7 @@
 
 package com.amazon.opendistro.elasticsearch.performanceanalyzer.decisionmaker.actions;
 
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.core.Stats;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.store.rca.cluster.NodeKey;
 
 import java.util.List;
@@ -44,4 +45,9 @@ public interface Action {
 
   /** Returns a summary for the configured action */
   String summary();
+
+  /** Returns if this action is explicitly muted through configuration */
+  default boolean isMuted() {
+    return Stats.getInstance().isActionMuted(name());
+  }
 }

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/decisionmaker/actions/ModifyCacheMaxSizeAction.java

@@ -36,7 +36,7 @@
 
 // TODO: Split the cache action into separate actions for different caches.
 
-public class ModifyCacheMaxSizeAction implements Action {
+public class ModifyCacheMaxSizeAction extends SuppressibleAction {
   private static final Logger LOG = LogManager.getLogger(ModifyCacheMaxSizeAction.class);
   public static final String NAME = "modifyCacheCapacity";
   public static final long COOL_OFF_PERIOD_IN_MILLIS = 300 * 1_000;
@@ -82,7 +82,7 @@ public String name() {
   }
 
   @Override
-  public boolean isActionable() {
+  public boolean canUpdate() {
     if (currentCacheMaxSizeInBytes == null) {
       return false;
     }

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/decisionmaker/actions/ModifyQueueCapacityAction.java

@@ -27,7 +27,7 @@
 import java.util.List;
 import java.util.Map;
 
-public class ModifyQueueCapacityAction implements Action {
+public class ModifyQueueCapacityAction extends SuppressibleAction {
 
   public static final String NAME = "modify_queue_capacity";
   public static final long COOL_OFF_PERIOD_IN_MILLIS = 300 * 1_000;
@@ -56,7 +56,7 @@ public String name() {
   }
 
   @Override
-  public boolean isActionable() {
+  public boolean canUpdate() {
     return desiredCapacity != currentCapacity;
   }
 

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/decisionmaker/actions/SuppressibleAction.java

@@ -0,0 +1,35 @@
+package com.amazon.opendistro.elasticsearch.performanceanalyzer.decisionmaker.actions;
+
+/**
+ * Actions that can be suppressed through a config.
+ */
+public abstract class SuppressibleAction implements Action {
+
+  /**
+   * Returns true if the configured action is actionable, false otherwise.
+   * The method is also declared final to enforce checking for muted-ness of an action before
+   * declaring it actionable.
+   *
+   * <p>Actions that want to define the actionable criterion differently should not inherit from
+   * {@link SuppressibleAction}, instead provide their own implementation. This way, we get to
+   * close {@link SuppressibleAction} while still keeping the {@link Action} interface open for
+   * modification.</p>
+   *
+   * <p>Examples of non-actionable actions are invalid actions or actions that are explicitly
+   * disabled in the conf file.
+   */
+  @Override
+  public final boolean isActionable() {
+    return !isMuted() && canUpdate();
+  }
+
+  /**
+   * Returns true if the configured action is valid, false otherwise.
+   *
+   * <p>Examples of invalid actions are resource configurations where limits have been
+   * reached.</p>
+   *
+   * @return true if valid, false otherwise.
+   */
+  public abstract boolean canUpdate();
+}

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/decisionmaker/deciders/Collator.java

@@ -16,7 +16,6 @@
 package com.amazon.opendistro.elasticsearch.performanceanalyzer.decisionmaker.deciders;
 
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.decisionmaker.actions.ImpactVector;
-
 import java.util.Arrays;
 import java.util.List;
 

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/decisionmaker/deciders/Decider.java

@@ -16,6 +16,7 @@
 package com.amazon.opendistro.elasticsearch.performanceanalyzer.decisionmaker.deciders;
 
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.PerformanceAnalyzerApp;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.decisionmaker.actions.Action;
 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;