use index mapping and define default index metadata for top n queries

ansjcy · ansjcy · commit 03355c84d877 · 2025-01-23T18:17:59.000-08:00
Signed-off-by: Chenyang Ji &lt;cyji@amazon.com&gt;
diff --git a/src/main/java/org/opensearch/plugin/insights/core/exporter/LocalIndexExporter.java b/src/main/java/org/opensearch/plugin/insights/core/exporter/LocalIndexExporter.java
@@ -11,21 +11,30 @@
 import static org.opensearch.plugin.insights.core.service.TopQueriesService.isTopQueriesIndex;
 import static org.opensearch.plugin.insights.settings.QueryInsightsSettings.DEFAULT_DELETE_AFTER_VALUE;
 
+import java.io.IOException;
+import java.nio.charset.Charset;
 import java.time.Instant;
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.opensearch.ResourceAlreadyExistsException;
+import org.opensearch.action.admin.indices.create.CreateIndexRequest;
+import org.opensearch.action.admin.indices.create.CreateIndexResponse;
 import org.opensearch.action.bulk.BulkRequestBuilder;
 import org.opensearch.action.bulk.BulkResponse;
 import org.opensearch.action.index.IndexRequest;
 import org.opensearch.client.Client;
+import org.opensearch.cluster.ClusterState;
 import org.opensearch.cluster.metadata.IndexMetadata;
+import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.xcontent.XContentFactory;
 import org.opensearch.core.action.ActionListener;
@@ -44,19 +53,38 @@ public final class LocalIndexExporter implements QueryInsightsExporter {
      */
     private final Logger logger = LogManager.getLogger();
     private final Client client;
+    private final ClusterService clusterService;
+    private final String indexMapping;
     private DateTimeFormatter indexPattern;
     private int deleteAfter;
     private final String id;
+    private static final int DEFAULT_NUMBER_OF_REPLICA = 1;
+    private static final int DEFAULT_NUMBER_OF_SHARDS = 1;
+    private static final List<String> DEFAULT_SORTED_FIELDS = List.of(
+        "measurements.latency.number",
+        "measurements.cpu.number",
+        "measurements.memory.number"
+    );
+    private static final List<String> DEFAULT_SORTED_ORDERS  = List.of(
+        "desc",
+        "desc",
+        "desc"
+    );
 
     /**
      * Constructor of LocalIndexExporter
      *
      * @param client OS client
+     * @param clusterService cluster service
      * @param indexPattern the pattern of index to export to
+     * @param indexMapping the index mapping file
+     * @param id id of the exporter
      */
-    public LocalIndexExporter(final Client client, final DateTimeFormatter indexPattern, final String id) {
+    public LocalIndexExporter(final Client client, final ClusterService clusterService, final DateTimeFormatter indexPattern, final String indexMapping, final String id) {
         this.indexPattern = indexPattern;
         this.client = client;
+        this.clusterService = clusterService;
+        this.indexMapping = indexMapping;
         this.deleteAfter = DEFAULT_DELETE_AFTER_VALUE;
         this.id = id;
     }
@@ -96,28 +124,72 @@ public void export(final List<SearchQueryRecord> records) {
         }
         try {
             final String indexName = buildLocalIndexName();
-            final BulkRequestBuilder bulkRequestBuilder = client.prepareBulk().setTimeout(TimeValue.timeValueMinutes(1));
-            for (SearchQueryRecord record : records) {
-                bulkRequestBuilder.add(
-                    new IndexRequest(indexName).source(record.toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS))
+            if (!checkIndexExists(indexName)) {
+                CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
+
+                createIndexRequest.settings(Settings.builder()
+                    .putList("index.sort.field", DEFAULT_SORTED_FIELDS)
+                    .putList("index.sort.order", DEFAULT_SORTED_ORDERS)
+                    .put("index.number_of_shards", DEFAULT_NUMBER_OF_SHARDS)
+                    .put("index.number_of_replicas", DEFAULT_NUMBER_OF_REPLICA)
                 );
+                createIndexRequest.mapping(readIndexMappings());
+
+                client.admin().indices().create(createIndexRequest, new ActionListener<>() {
+                    @Override
+                    public void onResponse(CreateIndexResponse createIndexResponse) {
+                        if (createIndexResponse.isAcknowledged()) {
+                            try {
+                                bulk(indexName, records);
+                            } catch (IOException e) {
+                                OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.LOCAL_INDEX_EXPORTER_EXCEPTIONS);
+                                logger.error("Unable to index query insights data: ", e);
+                            }
+                        }
+                    }
+                    @Override
+                    public void onFailure(Exception e) {
+                        if (e instanceof ResourceAlreadyExistsException) {
+                            try {
+                                bulk(indexName, records);
+                            } catch (IOException ex) {
+                                OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.LOCAL_INDEX_EXPORTER_EXCEPTIONS);
+                                logger.error("Unable to index query insights data: ", e);
+                            }
+                        } else {
+                            OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.LOCAL_INDEX_EXPORTER_EXCEPTIONS);
+                            logger.error("Unable to create query insights index: ", e);
+                        }
+                    }
+                });
+            } else {
+                bulk(indexName, records);
             }
-            bulkRequestBuilder.execute(new ActionListener<BulkResponse>() {
-                @Override
-                public void onResponse(BulkResponse bulkItemResponses) {}
-
-                @Override
-                public void onFailure(Exception e) {
-                    OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.LOCAL_INDEX_EXPORTER_BULK_FAILURES);
-                    logger.error("Failed to execute bulk operation for query insights data: ", e);
-                }
-            });
         } catch (final Exception e) {
             OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.LOCAL_INDEX_EXPORTER_EXCEPTIONS);
             logger.error("Unable to index query insights data: ", e);
         }
     }
 
+    private void bulk(final String indexName, final List<SearchQueryRecord> records) throws IOException {
+        final BulkRequestBuilder bulkRequestBuilder = client.prepareBulk().setTimeout(TimeValue.timeValueMinutes(1));
+        for (SearchQueryRecord record : records) {
+            bulkRequestBuilder.add(
+                new IndexRequest(indexName).id(record.getId()).source(record.toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS))
+            );
+        }
+        bulkRequestBuilder.execute(new ActionListener<BulkResponse>() {
+            @Override
+            public void onResponse(BulkResponse bulkItemResponses) {}
+
+            @Override
+            public void onFailure(Exception e) {
+                OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.LOCAL_INDEX_EXPORTER_BULK_FAILURES);
+                logger.error("Failed to execute bulk operation for query insights data: ", e);
+            }
+        });
+    }
+
     /**
      * Close the exporter sink
      */
@@ -174,4 +246,27 @@ public static String generateLocalIndexDateHash() {
         // Generate a 5-digit numeric hash from the date's hashCode
         return String.format(Locale.ROOT, "%05d", (currentDate.hashCode() % 100000 + 100000) % 100000);
     }
+
+    /**
+     * check if index exists
+     * @return boolean
+     */
+    private boolean checkIndexExists(String indexName) {
+        ClusterState clusterState = clusterService.state();
+        return clusterState.getRoutingTable().hasIndex(indexName);
+    }
+
+    /**
+     * get correlation rule index mappings
+     * @return mappings of correlation rule index
+     * @throws IOException IOException
+     */
+    private String readIndexMappings() throws IOException {
+        return new String(
+            Objects.requireNonNull(LocalIndexExporter.class.getClassLoader().getResourceAsStream(indexMapping))
+                .readAllBytes(),
+            Charset.defaultCharset()
+        );
+    }
+
 }
diff --git a/src/main/java/org/opensearch/plugin/insights/core/exporter/QueryInsightsExporterFactory.java b/src/main/java/org/opensearch/plugin/insights/core/exporter/QueryInsightsExporterFactory.java
@@ -16,6 +16,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.opensearch.client.Client;
+import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.plugin.insights.core.metrics.OperationalMetric;
 import org.opensearch.plugin.insights.core.metrics.OperationalMetricsCounter;
 
@@ -28,15 +29,18 @@ public class QueryInsightsExporterFactory {
      */
     private final Logger logger = LogManager.getLogger();
     final private Client client;
+    final private ClusterService clusterService;
     final private Map<String, QueryInsightsExporter> exporters;
 
     /**
      * Constructor of QueryInsightsExporterFactory
      *
      * @param client OS client
+     * @param clusterService cluster service
      */
-    public QueryInsightsExporterFactory(final Client client) {
+    public QueryInsightsExporterFactory(final Client client, final ClusterService clusterService) {
         this.client = client;
+        this.clusterService = clusterService;
         this.exporters = new HashMap<>();
     }
 
@@ -64,13 +68,15 @@ public void validateExporterType(final String exporterType) throws IllegalArgume
     /**
      * Create an exporter based on provided parameters
      *
+     * @param id id of the exporter
      * @param type The type of exporter to create
-     * @param indexPattern the index pattern if creating a index exporter
+     * @param indexPattern the index pattern if creating an index exporter
+     * @param indexMapping index mapping file
      * @return QueryInsightsExporter the created exporter sink
      */
-    public QueryInsightsExporter createExporter(String id, SinkType type, String indexPattern) {
+    public QueryInsightsExporter createExporter(String id, SinkType type, String indexPattern, String indexMapping) {
         if (SinkType.LOCAL_INDEX.equals(type)) {
-            QueryInsightsExporter exporter = new LocalIndexExporter(client, DateTimeFormatter.ofPattern(indexPattern, Locale.ROOT), id);
+            QueryInsightsExporter exporter = new LocalIndexExporter(client, clusterService, DateTimeFormatter.ofPattern(indexPattern, Locale.ROOT), indexMapping, id);
             this.exporters.put(id, exporter);
             return exporter;
         }
diff --git a/src/main/java/org/opensearch/plugin/insights/core/service/QueryInsightsService.java b/src/main/java/org/opensearch/plugin/insights/core/service/QueryInsightsService.java
@@ -444,7 +444,8 @@ private void setExporterAndReader(final SinkType sinkType, final Map<String, Ind
                 queryInsightsExporterFactory.createExporter(
                     TOP_QUERIES_LOCAL_INDEX_EXPORTER_ID,
                     sinkType,
-                    DEFAULT_TOP_N_QUERIES_INDEX_PATTERN
+                    DEFAULT_TOP_N_QUERIES_INDEX_PATTERN,
+                    "mappings/top-queries-record.json"
                 );
             }
         } else {
diff --git a/src/main/resources/mappings/top-queries-record.json b/src/main/resources/mappings/top-queries-record.json
@@ -0,0 +1,127 @@
+{
+  "dynamic": true,
+  "_meta": {
+    "schema_version": 1,
+    "query_insights_feature_space": "top_n_queries"
+  },
+  "properties": {
+    "id" : {
+      "type" : "text",
+      "fields" : {
+        "keyword" : {
+          "type" : "keyword",
+          "ignore_above" : 256
+        }
+      }
+    },
+    "node_id" : {
+      "type" : "text",
+      "fields" : {
+        "keyword" : {
+          "type" : "keyword",
+          "ignore_above" : 256
+        }
+      }
+    },
+    "timestamp" : {
+      "type" : "long"
+    },
+    "total_shards" : {
+      "type" : "long"
+    },
+    "group_by" : {
+      "type" : "text",
+      "fields" : {
+        "keyword" : {
+          "type" : "keyword",
+          "ignore_above" : 256
+        }
+      }
+    },
+    "phase_latency_map" : {
+      "properties" : {
+        "expand" : {
+          "type" : "long"
+        },
+        "fetch" : {
+          "type" : "long"
+        },
+        "query" : {
+          "type" : "long"
+        }
+      }
+    },
+    "search_type" : {
+      "type" : "text",
+      "fields" : {
+        "keyword" : {
+          "type" : "keyword",
+          "ignore_above" : 256
+        }
+      }
+    },
+    "task_resource_usages" : {
+      "properties" : {
+        "action" : {
+          "type" : "text",
+          "fields" : {
+            "keyword" : {
+              "type" : "keyword",
+              "ignore_above" : 256
+            }
+          }
+        },
+        "nodeId" : {
+          "type" : "text",
+          "fields" : {
+            "keyword" : {
+              "type" : "keyword",
+              "ignore_above" : 256
+            }
+          }
+        },
+        "parentTaskId" : {
+          "type" : "long"
+        },
+        "taskId" : {
+          "type" : "long"
+        },
+        "taskResourceUsage" : {
+          "properties" : {
+            "cpu_time_in_nanos" : {
+              "type" : "long"
+            },
+            "memory_in_bytes" : {
+              "type" : "long"
+            }
+          }
+        }
+      }
+    },
+    "measurements": {
+      "properties": {
+        "latency": {
+          "properties": {
+            "number": {
+              "type": "double"
+            }
+          }
+        },
+        "cpu": {
+          "properties": {
+            "number": {
+              "type": "double"
+            }
+          }
+        },
+        "memory": {
+          "properties": {
+            "number": {
+              "type": "double"
+            }
+          }
+        }
+      }
+    }
+  }
+}
