Code changes for action listener plugin

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

@@ -0,0 +1,28 @@
+/*
+ * 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.decisionmaker.actions;
+
+/**
+ * This listener is notified whenever an action suggestion is
+ * published by the decision maker Publisher
+ */
+public interface ActionListener {
+
+  /**
+   * Called when Publisher emits an action
+   */
+  void actionPublished(Action action);
+}

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

@@ -17,10 +17,13 @@
 
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.PerformanceAnalyzerApp;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.decisionmaker.actions.Action;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.decisionmaker.actions.ActionListener;
 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 java.util.ArrayList;
+import java.util.List;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -30,10 +33,12 @@ public class Publisher extends NonLeafNode<EmptyFlowUnit> {
 
   private Collator collator;
   private boolean isMuted = false;
+  private List<ActionListener> actionListeners;
 
   public Publisher(int evalIntervalSeconds, Collator collator) {
     super(0, evalIntervalSeconds);
     this.collator = collator;
+    this.actionListeners = new ArrayList<>();
   }
 
   @Override
@@ -43,8 +48,11 @@ public EmptyFlowUnit operate() {
 
     Decision decision = collator.getFlowUnits().get(0);
     for (Action action : decision.getActions()) {
-      LOG.info("Executing action: [{}]", action.name());
-      action.execute();
+//      LOG.info("Executing action: [{}]", action.name());
+//      action.execute();
+      for (ActionListener listener: actionListeners) {
+        listener.actionPublished(action);
+      }
     }
     return new EmptyFlowUnit(System.currentTimeMillis());
   }
@@ -67,6 +75,15 @@ public void generateFlowUnitListFromLocal(FlowUnitOperationArgWrapper args) {
         RcaGraphMetrics.GRAPH_NODE_OPERATE_CALL, this.name(), duration);
   }
 
+  /**
+   * Register an action listener with Publisher
+   *
+   * The listener is notified whenever an action is published
+   */
+  public void addActionListener(ActionListener listener) {
+    actionListeners.add(listener);
+  }
+
   /**
    * Publisher does not have downstream nodes and does not emit flow units
    */

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/plugins/Plugin.java

@@ -0,0 +1,30 @@
+/*
+ * 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.plugins;
+
+/**
+ * Allows adding custom extensions to the analysis graph.
+ * <p>
+ * RCA framework plugins can be installed to extend the analysis graph through custom
+ * metric nodes, rca nodes, deciders or action listeners. These can subscribe to flow
+ * units from existing nodes to add new functionality, or override existing graph nodes to
+ * customize for specific use cases.
+ */
+public abstract class Plugin {
+
+  public abstract String name();
+
+}

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/plugins/PluginController.java

@@ -0,0 +1,72 @@
+/*
+ * 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.plugins;
+
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.decisionmaker.actions.ActionListener;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.decisionmaker.deciders.Publisher;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+public class PluginController {
+
+  private static final Logger LOG = LogManager.getLogger(PluginController.class);
+  private final Publisher publisher;
+  private List<Plugin> plugins;
+
+  public PluginController(Publisher publisher) {
+    this.publisher = publisher;
+    this.plugins = new ArrayList<>();
+    loadFrameworkPlugins();
+    registerActionListeners();
+  }
+
+  private void loadFrameworkPlugins() {
+    for (Class<?> pluginClass : PluginControllerConfig.getFrameworkPlugins()) {
+      final Constructor<?>[] constructors = pluginClass.getConstructors();
+      if (constructors.length == 0) {
+        throw new IllegalStateException(
+            "no public constructor found for plugin class: [" + pluginClass.getName() + "]");
+      }
+      if (constructors.length > 1) {
+        throw new IllegalStateException(
+            "unique constructor expected for plugin class: [" + pluginClass.getName() + "]");
+      }
+      if (constructors[0].getParameterCount() != 0) {
+        throw new IllegalStateException(
+            "default constructor expected for plugin class: [" + pluginClass.getName() + "]");
+      }
+
+      try {
+        plugins.add((Plugin) constructors[0].newInstance());
+        LOG.info("loaded plugin: [{}]", plugins.get(plugins.size() - 1).name());
+      } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
+        e.printStackTrace();
+      }
+    }
+  }
+
+  private void registerActionListeners() {
+    for (Plugin plugin: plugins) {
+      if (ActionListener.class.isAssignableFrom(plugin.getClass())) {
+        publisher.addActionListener((ActionListener)plugin);
+      }
+    }
+  }
+}

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/plugins/PluginControllerConfig.java

@@ -0,0 +1,31 @@
+/*
+ * 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.plugins;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class PluginControllerConfig {
+
+  /**
+   * Returns a list of entry point classes for internal framework plugins
+   */
+  public static List<Class<? extends Plugin>> getFrameworkPlugins() {
+    List<Class<? extends Plugin>> frameworkPlugins = new ArrayList<>();
+    frameworkPlugins.add(PublisherEventsLogger.class);
+    return frameworkPlugins;
+  }
+}

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/plugins/PublisherEventsLogger.java

@@ -0,0 +1,40 @@
+/*
+ * 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.plugins;
+
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.decisionmaker.actions.Action;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.decisionmaker.actions.ActionListener;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+/**
+ * A simple listener that logs all actions published by the publisher
+ */
+public class PublisherEventsLogger extends Plugin implements ActionListener {
+
+  private static final Logger LOG = LogManager.getLogger(PublisherEventsLogger.class);
+  public static final String NAME = "publisher_events_logger_plugin";
+
+  @Override
+  public void actionPublished(Action action) {
+    LOG.info("Action: [{}] published by decision maker publisher.", action.name());
+  }
+
+  @Override
+  public String name() {
+    return NAME;
+  }
+}

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

@@ -26,6 +26,7 @@
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.decisionmaker.deciders.QueueHealthDecider;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.metrics.AllMetrics.CommonDimension;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.metricsdb.MetricsDB;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.plugins.PluginController;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.api.AnalysisGraph;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.api.Metric;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.api.Rca;
@@ -192,6 +193,9 @@ public void construct() {
     Publisher publisher = new Publisher(EVALUATION_INTERVAL_SECONDS, collator);
     publisher.addTag(TAG_LOCUS, LOCUS_MASTER_NODE);
     publisher.addAllUpstreams(Collections.singletonList(collator));
+
+    // TODO: Refactor using DI to move out of construct method
+    PluginController pluginController = new PluginController(publisher);
   }
 
   private void constructShardResourceUsageGraph() {