Move exporter config to query insights level

Signed-off-by: Chenyang Ji <[email protected]>

src/main/java/org/opensearch/plugin/insights/QueryInsightsPlugin.java

@@ -131,21 +131,19 @@ public List<Setting<?>> getSettings() {
             QueryInsightsSettings.TOP_N_LATENCY_QUERIES_ENABLED,
             QueryInsightsSettings.TOP_N_LATENCY_QUERIES_SIZE,
             QueryInsightsSettings.TOP_N_LATENCY_QUERIES_WINDOW_SIZE,
-            QueryInsightsSettings.TOP_N_LATENCY_EXPORTER_SETTINGS,
             QueryInsightsSettings.TOP_N_CPU_QUERIES_ENABLED,
             QueryInsightsSettings.TOP_N_CPU_QUERIES_SIZE,
             QueryInsightsSettings.TOP_N_CPU_QUERIES_WINDOW_SIZE,
-            QueryInsightsSettings.TOP_N_CPU_EXPORTER_SETTINGS,
             QueryInsightsSettings.TOP_N_MEMORY_QUERIES_ENABLED,
             QueryInsightsSettings.TOP_N_MEMORY_QUERIES_SIZE,
             QueryInsightsSettings.TOP_N_MEMORY_QUERIES_WINDOW_SIZE,
-            QueryInsightsSettings.TOP_N_MEMORY_EXPORTER_SETTINGS,
             QueryInsightsSettings.TOP_N_QUERIES_GROUP_BY,
             QueryInsightsSettings.TOP_N_QUERIES_MAX_GROUPS_EXCLUDING_N,
             QueryInsightsSettings.TOP_N_QUERIES_GROUPING_FIELD_NAME,
             QueryInsightsSettings.TOP_N_QUERIES_GROUPING_FIELD_TYPE,
             QueryCategorizationSettings.SEARCH_QUERY_METRICS_ENABLED_SETTING,
             QueryInsightsSettings.TOP_N_EXPORTER_DELETE_AFTER,
+            QueryInsightsSettings.TOP_N_EXPORTER_TYPE,
             QueryCategorizationSettings.SEARCH_QUERY_FIELD_TYPE_CACHE_SIZE_KEY
         );
     }

src/main/java/org/opensearch/plugin/insights/core/exporter/DebugExporter.java

@@ -21,12 +21,18 @@ public final class DebugExporter implements QueryInsightsExporter {
      * Logger of the debug exporter
      */
     private final Logger logger = LogManager.getLogger();
+    private static final String EXPORTER_ID = "debug_exporter";
 
     /**
      * Constructor of DebugExporter
      */
     private DebugExporter() {}
 
+    @Override
+    public String getId() {
+        return EXPORTER_ID;
+    }
+
     private static class InstanceHolder {
         private static final DebugExporter INSTANCE = new DebugExporter();
     }

src/main/java/org/opensearch/plugin/insights/core/exporter/LocalIndexExporter.java

@@ -46,17 +46,24 @@ public final class LocalIndexExporter implements QueryInsightsExporter {
     private final Client client;
     private DateTimeFormatter indexPattern;
     private int deleteAfter;
+    private final String id;
 
     /**
      * Constructor of LocalIndexExporter
      *
      * @param client OS client
      * @param indexPattern the pattern of index to export to
      */
-    public LocalIndexExporter(final Client client, final DateTimeFormatter indexPattern) {
+    public LocalIndexExporter(final Client client, final DateTimeFormatter indexPattern, final String id) {
         this.indexPattern = indexPattern;
         this.client = client;
         this.deleteAfter = DEFAULT_DELETE_AFTER_VALUE;
+        this.id = id;
+    }
+
+    @Override
+    public String getId() {
+        return id;
     }
 
     /**

src/main/java/org/opensearch/plugin/insights/core/exporter/QueryInsightsExporter.java

@@ -22,4 +22,6 @@ public interface QueryInsightsExporter extends Closeable {
      * @param records list of {@link SearchQueryRecord}
      */
     void export(final List<SearchQueryRecord> records);
+
+    String getId();
 }

src/main/java/org/opensearch/plugin/insights/core/exporter/QueryInsightsExporterFactory.java

@@ -8,18 +8,14 @@
 
 package org.opensearch.plugin.insights.core.exporter;
 
-import static org.opensearch.plugin.insights.settings.QueryInsightsSettings.DEFAULT_TOP_QUERIES_EXPORTER_TYPE;
-import static org.opensearch.plugin.insights.settings.QueryInsightsSettings.EXPORTER_TYPE;
-
 import java.io.IOException;
 import java.time.format.DateTimeFormatter;
-import java.util.HashSet;
+import java.util.HashMap;
 import java.util.Locale;
-import java.util.Set;
+import java.util.Map;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.opensearch.client.Client;
-import org.opensearch.common.settings.Settings;
 import org.opensearch.plugin.insights.core.metrics.OperationalMetric;
 import org.opensearch.plugin.insights.core.metrics.OperationalMetricsCounter;
 
@@ -32,7 +28,7 @@ public class QueryInsightsExporterFactory {
      */
     private final Logger logger = LogManager.getLogger();
     final private Client client;
-    final private Set<QueryInsightsExporter> exporters;
+    final private Map<String, QueryInsightsExporter> exporters;
 
     /**
      * Constructor of QueryInsightsExporterFactory
@@ -41,32 +37,26 @@ public class QueryInsightsExporterFactory {
      */
     public QueryInsightsExporterFactory(final Client client) {
         this.client = client;
-        this.exporters = new HashSet<>();
+        this.exporters = new HashMap<>();
     }
 
     /**
      * Validate exporter sink config
      *
-     * @param settings exporter sink config {@link Settings}
+     * @param exporterType exporter sink type
      * @throws IllegalArgumentException if provided exporter sink config settings are invalid
      */
-    public void validateExporterConfig(final Settings settings) throws IllegalArgumentException {
+    public void validateExporterType(final String exporterType) throws IllegalArgumentException {
         // Disable exporter if the EXPORTER_TYPE setting is null
-        if (settings.get(EXPORTER_TYPE) == null) {
+        if (exporterType == null) {
             return;
         }
-        SinkType type;
         try {
-            type = SinkType.parse(settings.get(EXPORTER_TYPE, DEFAULT_TOP_QUERIES_EXPORTER_TYPE));
+            SinkType.parse(exporterType);
         } catch (IllegalArgumentException e) {
             OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.INVALID_EXPORTER_TYPE_FAILURES);
             throw new IllegalArgumentException(
-                String.format(
-                    Locale.ROOT,
-                    "Invalid exporter type [%s], type should be one of %s",
-                    settings.get(EXPORTER_TYPE),
-                    SinkType.allSinkTypes()
-                )
+                String.format(Locale.ROOT, "Invalid exporter type [%s], type should be one of %s", exporterType, SinkType.allSinkTypes())
             );
         }
     }
@@ -78,10 +68,10 @@ public void validateExporterConfig(final Settings settings) throws IllegalArgume
      * @param indexPattern the index pattern if creating a index exporter
      * @return QueryInsightsExporter the created exporter sink
      */
-    public QueryInsightsExporter createExporter(SinkType type, String indexPattern) {
+    public QueryInsightsExporter createExporter(String id, SinkType type, String indexPattern) {
         if (SinkType.LOCAL_INDEX.equals(type)) {
-            QueryInsightsExporter exporter = new LocalIndexExporter(client, DateTimeFormatter.ofPattern(indexPattern, Locale.ROOT));
-            this.exporters.add(exporter);
+            QueryInsightsExporter exporter = new LocalIndexExporter(client, DateTimeFormatter.ofPattern(indexPattern, Locale.ROOT), id);
+            this.exporters.put(id, exporter);
             return exporter;
         }
         return DebugExporter.getInstance();
@@ -101,6 +91,15 @@ public QueryInsightsExporter updateExporter(QueryInsightsExporter exporter, Stri
         return exporter;
     }
 
+    /**
+     * Get a exporter by id
+     * @param id The id of the exporter
+     * @return QueryInsightsReader the Reader
+     */
+    public QueryInsightsExporter getExporter(String id) {
+        return this.exporters.get(id);
+    }
+
     /**
      * Close an exporter
      *
@@ -110,7 +109,7 @@ public QueryInsightsExporter updateExporter(QueryInsightsExporter exporter, Stri
     public void closeExporter(QueryInsightsExporter exporter) throws IOException {
         if (exporter != null) {
             exporter.close();
-            this.exporters.remove(exporter);
+            this.exporters.remove(exporter.getId());
         }
     }
 
@@ -119,7 +118,7 @@ public void closeExporter(QueryInsightsExporter exporter) throws IOException {
      *
      */
     public void closeAllExporters() {
-        for (QueryInsightsExporter exporter : exporters) {
+        for (QueryInsightsExporter exporter : exporters.values()) {
             try {
                 closeExporter(exporter);
             } catch (IOException e) {

src/main/java/org/opensearch/plugin/insights/core/exporter/SinkType.java

@@ -17,6 +17,8 @@
  * Type of supported sinks
  */
 public enum SinkType {
+    /** no exporter */
+    NONE("none"),
     /** debug exporter */
     DEBUG("debug"),
     /** local index exporter */
@@ -60,7 +62,9 @@ public static Set<SinkType> allSinkTypes() {
     public static SinkType getSinkTypeFromExporter(QueryInsightsExporter exporter) {
         if (exporter.getClass().equals(LocalIndexExporter.class)) {
             return SinkType.LOCAL_INDEX;
+        } else if (exporter.getClass().equals(DebugExporter.class)) {
+            return SinkType.DEBUG;
         }
-        return SinkType.DEBUG;
+        return SinkType.NONE;
     }
 }

src/main/java/org/opensearch/plugin/insights/core/reader/LocalIndexReader.java

@@ -46,6 +46,7 @@ public final class LocalIndexReader implements QueryInsightsReader {
     private final Client client;
     private DateTimeFormatter indexPattern;
     private final NamedXContentRegistry namedXContentRegistry;
+    private final String id;
 
     /**
      * Constructor of LocalIndexReader
@@ -54,12 +55,23 @@ public final class LocalIndexReader implements QueryInsightsReader {
      * @param indexPattern the pattern of index to read from
      * @param namedXContentRegistry for parsing purposes
      */
-    public LocalIndexReader(final Client client, final DateTimeFormatter indexPattern, final NamedXContentRegistry namedXContentRegistry) {
+    public LocalIndexReader(
+        final Client client,
+        final DateTimeFormatter indexPattern,
+        final NamedXContentRegistry namedXContentRegistry,
+        final String id
+    ) {
         this.indexPattern = indexPattern;
         this.client = client;
+        this.id = id;
         this.namedXContentRegistry = namedXContentRegistry;
     }
 
+    @Override
+    public String getId() {
+        return id;
+    }
+
     /**
      * Getter of indexPattern
      *

src/main/java/org/opensearch/plugin/insights/core/reader/QueryInsightsReader.java

@@ -25,4 +25,6 @@ public interface QueryInsightsReader extends Closeable {
      * @return List of SearchQueryRecord
      */
     List<SearchQueryRecord> read(final String from, final String to, final String id);
+
+    String getId();
 }

src/main/java/org/opensearch/plugin/insights/core/reader/QueryInsightsReaderFactory.java

@@ -10,9 +10,9 @@
 
 import java.io.IOException;
 import java.time.format.DateTimeFormatter;
-import java.util.HashSet;
+import java.util.HashMap;
 import java.util.Locale;
-import java.util.Set;
+import java.util.Map;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.opensearch.client.Client;
@@ -27,7 +27,7 @@ public class QueryInsightsReaderFactory {
      */
     private final Logger logger = LogManager.getLogger();
     final private Client client;
-    final private Set<QueryInsightsReader> Readers;
+    final private Map<String, QueryInsightsReader> readers;
 
     /**
      * Constructor of QueryInsightsReaderFactory
@@ -36,7 +36,7 @@ public class QueryInsightsReaderFactory {
      */
     public QueryInsightsReaderFactory(final Client client) {
         this.client = client;
-        this.Readers = new HashSet<>();
+        this.readers = new HashMap<>();
     }
 
     /**
@@ -46,39 +46,49 @@ public QueryInsightsReaderFactory(final Client client) {
      * @param namedXContentRegistry for parsing purposes
      * @return QueryInsightsReader the created Reader
      */
-    public QueryInsightsReader createReader(String indexPattern, NamedXContentRegistry namedXContentRegistry) {
-        QueryInsightsReader Reader = new LocalIndexReader(
+    public QueryInsightsReader createReader(String id, String indexPattern, NamedXContentRegistry namedXContentRegistry) {
+        QueryInsightsReader reader = new LocalIndexReader(
             client,
             DateTimeFormatter.ofPattern(indexPattern, Locale.ROOT),
-            namedXContentRegistry
+            namedXContentRegistry,
+            id
         );
-        this.Readers.add(Reader);
-        return Reader;
+        this.readers.put(id, reader);
+        return reader;
     }
 
     /**
-     * Update a Reader based on provided parameters
+     * Update a reader based on provided parameters
      *
-     * @param Reader The Reader to update
-     * @param indexPattern the index pattern if creating an index Reader
-     * @return QueryInsightsReader the updated Reader sink
+     * @param reader The reader to update
+     * @param indexPattern the index pattern if creating an index reader
+     * @return QueryInsightsReader the updated reader sink
      */
-    public QueryInsightsReader updateReader(QueryInsightsReader Reader, String indexPattern) {
-        if (Reader.getClass() == LocalIndexReader.class) {
-            ((LocalIndexReader) Reader).setIndexPattern(DateTimeFormatter.ofPattern(indexPattern, Locale.ROOT));
+    public QueryInsightsReader updateReader(QueryInsightsReader reader, String indexPattern) {
+        if (reader.getClass() == LocalIndexReader.class) {
+            ((LocalIndexReader) reader).setIndexPattern(DateTimeFormatter.ofPattern(indexPattern, Locale.ROOT));
         }
-        return Reader;
+        return reader;
+    }
+
+    /**
+     * Get a reader by id
+     * @param id The id of the reader
+     * @return QueryInsightsReader the Reader
+     */
+    public QueryInsightsReader getReader(String id) {
+        return this.readers.get(id);
     }
 
     /**
-     * Close a Reader
+     * Close a reader
      *
-     * @param Reader the Reader to close
+     * @param reader the Reader to close
      */
-    public void closeReader(QueryInsightsReader Reader) throws IOException {
-        if (Reader != null) {
-            Reader.close();
-            this.Readers.remove(Reader);
+    public void closeReader(QueryInsightsReader reader) throws IOException {
+        if (reader != null) {
+            reader.close();
+            this.readers.remove(reader.getId());
         }
     }
 
@@ -87,11 +97,11 @@ public void closeReader(QueryInsightsReader Reader) throws IOException {
      *
      */
     public void closeAllReaders() {
-        for (QueryInsightsReader Reader : Readers) {
+        for (QueryInsightsReader reader : readers.values()) {
             try {
-                closeReader(Reader);
+                closeReader(reader);
             } catch (IOException e) {
-                logger.error("Fail to close query insights Reader, error: ", e);
+                logger.error("Fail to close query insights reader, error: ", e);
             }
         }
     }