Refactor record and service to make them generic

Signed-off-by: Chenyang Ji <cyji@amazon.com>

plugins/query-insights/build.gradle

@@ -8,8 +8,6 @@
  * Modifications Copyright OpenSearch Contributors. See
  * GitHub history for details.
  */
-apply plugin: 'opensearch.java-rest-test'
-apply plugin: 'opensearch.internal-cluster-test'
 
 opensearchplugin {
   description 'OpenSearch Query Insights Plugin.'

plugins/query-insights/src/internalClusterTest/java/org/opensearch/plugin/insights/QueryInsightsPluginTransportIT.java

@@ -20,6 +20,7 @@
 import org.opensearch.plugin.insights.rules.action.top_queries.TopQueriesAction;
 import org.opensearch.plugin.insights.rules.action.top_queries.TopQueriesRequest;
 import org.opensearch.plugin.insights.rules.action.top_queries.TopQueriesResponse;
+import org.opensearch.plugin.insights.rules.model.MetricType;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.plugins.PluginInfo;
 import org.opensearch.test.OpenSearchIntegTestCase;
@@ -74,11 +75,11 @@ public void testQueryInsightPluginInstalled() {
      * Test get top queries when feature disabled
      */
     public void testGetTopQueriesWhenFeatureDisabled() {
-        TopQueriesRequest request = new TopQueriesRequest();
+        TopQueriesRequest request = new TopQueriesRequest(MetricType.LATENCY);
         TopQueriesResponse response = OpenSearchIntegTestCase.client().execute(TopQueriesAction.INSTANCE, request).actionGet();
         Assert.assertNotEquals(0, response.failures().size());
         Assert.assertEquals(
-            "Cannot get query data when query insight feature is not enabled.",
+            "Cannot get query data when query insight feature is not enabled for MetricType [latency].",
             response.failures().get(0).getCause().getCause().getMessage()
         );
     }
@@ -93,7 +94,7 @@ public void testGetTopQueriesWhenFeatureEnabled() {
             .put(TOP_N_LATENCY_QUERIES_WINDOW_SIZE.getKey(), "600s")
             .build();
 
-        logger.info("--> starting 2 nodes for query insight testing");
+        logger.info("--> starting nodes for query insight testing");
         List<String> nodes = internalCluster().startNodes(TOTAL_NUMBER_OF_NODES, Settings.builder().put(commonSettings).build());
 
         logger.info("--> waiting for nodes to form a cluster");
@@ -118,11 +119,11 @@ public void testGetTopQueriesWhenFeatureEnabled() {
             assertEquals(searchResponse.getFailedShards(), 0);
         }
 
-        TopQueriesRequest request = new TopQueriesRequest();
+        TopQueriesRequest request = new TopQueriesRequest(MetricType.LATENCY);
         TopQueriesResponse response = OpenSearchIntegTestCase.client().execute(TopQueriesAction.INSTANCE, request).actionGet();
         Assert.assertEquals(0, response.failures().size());
         Assert.assertEquals(TOTAL_NUMBER_OF_NODES, response.getNodes().size());
-        Assert.assertEquals(TOTAL_SEARCH_REQUESTS, response.getNodes().stream().mapToInt(o -> o.getLatencyRecords().size()).sum());
+        Assert.assertEquals(TOTAL_SEARCH_REQUESTS, response.getNodes().stream().mapToInt(o -> o.getTopQueriesRecord().size()).sum());
 
         internalCluster().stopAllNodes();
     }
@@ -137,7 +138,7 @@ public void testGetTopQueriesWithSmallTopN() {
             .put(TOP_N_LATENCY_QUERIES_WINDOW_SIZE.getKey(), "600s")
             .build();
 
-        logger.info("--> starting 2 nodes for query insight testing");
+        logger.info("--> starting nodes for query insight testing");
         List<String> nodes = internalCluster().startNodes(TOTAL_NUMBER_OF_NODES, Settings.builder().put(commonSettings).build());
 
         logger.info("--> waiting for nodes to form a cluster");
@@ -162,12 +163,11 @@ public void testGetTopQueriesWithSmallTopN() {
             assertEquals(searchResponse.getFailedShards(), 0);
         }
 
-        TopQueriesRequest request = new TopQueriesRequest();
+        TopQueriesRequest request = new TopQueriesRequest(MetricType.LATENCY);
         TopQueriesResponse response = OpenSearchIntegTestCase.client().execute(TopQueriesAction.INSTANCE, request).actionGet();
         Assert.assertEquals(0, response.failures().size());
         Assert.assertEquals(TOTAL_NUMBER_OF_NODES, response.getNodes().size());
-        // TODO: this should be 1 after changing to cluster level top N.
-        Assert.assertEquals(2, response.getNodes().stream().mapToInt(o -> o.getLatencyRecords().size()).sum());
+        Assert.assertEquals(2, response.getNodes().stream().mapToInt(o -> o.getTopQueriesRecord().size()).sum());
 
         internalCluster().stopAllNodes();
     }
@@ -179,10 +179,10 @@ public void testGetTopQueriesWithSmallWindowSize() {
         Settings commonSettings = Settings.builder()
             .put(TOP_N_LATENCY_QUERIES_ENABLED.getKey(), "true")
             .put(TOP_N_LATENCY_QUERIES_SIZE.getKey(), "100")
-            .put(TOP_N_LATENCY_QUERIES_WINDOW_SIZE.getKey(), "0ms")
+            .put(TOP_N_LATENCY_QUERIES_WINDOW_SIZE.getKey(), "1m")
             .build();
 
-        logger.info("--> starting 2 nodes for query insight testing");
+        logger.info("--> starting nodes for query insight testing");
         List<String> nodes = internalCluster().startNodes(TOTAL_NUMBER_OF_NODES, Settings.builder().put(commonSettings).build());
 
         logger.info("--> waiting for nodes to form a cluster");
@@ -207,11 +207,10 @@ public void testGetTopQueriesWithSmallWindowSize() {
             assertEquals(searchResponse.getFailedShards(), 0);
         }
 
-        TopQueriesRequest request = new TopQueriesRequest();
+        TopQueriesRequest request = new TopQueriesRequest(MetricType.LATENCY);
         TopQueriesResponse response = OpenSearchIntegTestCase.client().execute(TopQueriesAction.INSTANCE, request).actionGet();
         Assert.assertEquals(0, response.failures().size());
         Assert.assertEquals(TOTAL_NUMBER_OF_NODES, response.getNodes().size());
-        Assert.assertEquals(0, response.getNodes().stream().mapToInt(o -> o.getLatencyRecords().size()).sum());
 
         internalCluster().stopAllNodes();
     }

plugins/query-insights/src/main/java/org/opensearch/plugin/insights/QueryInsightsPlugin.java

@@ -18,13 +18,15 @@
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.settings.SettingsFilter;
+import org.opensearch.common.unit.TimeValue;
+import org.opensearch.common.util.concurrent.OpenSearchExecutors;
 import org.opensearch.core.action.ActionResponse;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.core.xcontent.NamedXContentRegistry;
 import org.opensearch.env.Environment;
 import org.opensearch.env.NodeEnvironment;
-import org.opensearch.plugin.insights.core.listener.SearchQueryLatencyListener;
-import org.opensearch.plugin.insights.core.service.TopQueriesByLatencyService;
+import org.opensearch.plugin.insights.core.listener.QueryInsightsListener;
+import org.opensearch.plugin.insights.core.service.QueryInsightsService;
 import org.opensearch.plugin.insights.rules.action.top_queries.TopQueriesAction;
 import org.opensearch.plugin.insights.rules.resthandler.top_queries.RestTopQueriesAction;
 import org.opensearch.plugin.insights.rules.transport.top_queries.TransportTopQueriesAction;
@@ -35,9 +37,12 @@
 import org.opensearch.rest.RestController;
 import org.opensearch.rest.RestHandler;
 import org.opensearch.script.ScriptService;
+import org.opensearch.threadpool.ExecutorBuilder;
+import org.opensearch.threadpool.ScalingExecutorBuilder;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.watcher.ResourceWatcherService;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.function.Supplier;
@@ -66,10 +71,24 @@ public Collection<Object> createComponents(
         Supplier<RepositoriesService> repositoriesServiceSupplier
     ) {
         // create top n queries service
-        TopQueriesByLatencyService topQueriesByLatencyService = new TopQueriesByLatencyService(threadPool, clusterService, client);
+        QueryInsightsService queryInsightsService = new QueryInsightsService(threadPool);
         // top n queries listener
-        SearchQueryLatencyListener searchQueryLatencyListener = new SearchQueryLatencyListener(clusterService, topQueriesByLatencyService);
-        return List.of(topQueriesByLatencyService, searchQueryLatencyListener);
+        QueryInsightsListener queryInsightsListener = new QueryInsightsListener(clusterService, queryInsightsService);
+        return List.of(queryInsightsService, queryInsightsListener);
+    }
+
+    @Override
+    public List<ExecutorBuilder<?>> getExecutorBuilders(Settings settings) {
+        List<ExecutorBuilder<?>> executorBuilders = new ArrayList<>();
+        executorBuilders.add(
+            new ScalingExecutorBuilder(
+                QueryInsightsSettings.QUERY_INSIGHTS_EXECUTOR,
+                1,
+                (OpenSearchExecutors.allocatedProcessors(settings) + 1) / 2,
+                TimeValue.timeValueMinutes(5)
+            )
+        );
+        return executorBuilders;
     }
 
     @Override
@@ -96,11 +115,7 @@ public List<Setting<?>> getSettings() {
             // Settings for top N queries
             QueryInsightsSettings.TOP_N_LATENCY_QUERIES_ENABLED,
             QueryInsightsSettings.TOP_N_LATENCY_QUERIES_SIZE,
-            QueryInsightsSettings.TOP_N_LATENCY_QUERIES_WINDOW_SIZE,
-            QueryInsightsSettings.TOP_N_LATENCY_QUERIES_EXPORTER_ENABLED,
-            QueryInsightsSettings.TOP_N_LATENCY_QUERIES_EXPORTER_TYPE,
-            QueryInsightsSettings.TOP_N_LATENCY_QUERIES_EXPORTER_INTERVAL,
-            QueryInsightsSettings.TOP_N_LATENCY_QUERIES_EXPORTER_IDENTIFIER
+            QueryInsightsSettings.TOP_N_LATENCY_QUERIES_WINDOW_SIZE
         );
     }
 }