[Backport 2.x] Otel counters for error metrics

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

@@ -29,6 +29,7 @@
 import org.opensearch.env.Environment;
 import org.opensearch.env.NodeEnvironment;
 import org.opensearch.plugin.insights.core.listener.QueryInsightsListener;
+import org.opensearch.plugin.insights.core.metrics.OperationalMetricsCounter;
 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;
@@ -74,6 +75,8 @@ public Collection<Object> createComponents(
         final Tracer tracer,
         final MetricsRegistry metricsRegistry
     ) {
+        // initialize operational metrics counters
+        OperationalMetricsCounter.initialize(clusterService.getClusterName().toString(), metricsRegistry);
         // create top n queries service
         final QueryInsightsService queryInsightsService = new QueryInsightsService(
             clusterService.getClusterSettings(),

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

@@ -22,6 +22,8 @@
 import org.opensearch.common.xcontent.XContentFactory;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.xcontent.ToXContent;
+import org.opensearch.plugin.insights.core.metrics.OperationalMetric;
+import org.opensearch.plugin.insights.core.metrics.OperationalMetricsCounter;
 import org.opensearch.plugin.insights.rules.model.SearchQueryRecord;
 
 /**
@@ -90,10 +92,12 @@ 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);
         }
     }

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

@@ -22,6 +22,8 @@
 import org.joda.time.format.DateTimeFormat;
 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;
 
 /**
  * Factory class for validating and creating exporters based on provided settings
@@ -59,6 +61,7 @@ public void validateExporterConfig(final Settings settings) throws IllegalArgume
         try {
             type = SinkType.parse(settings.get(EXPORTER_TYPE, DEFAULT_TOP_QUERIES_EXPORTER_TYPE));
         } catch (IllegalArgumentException e) {
+            OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.INVALID_EXPORTER_TYPE_FAILURES);
             throw new IllegalArgumentException(
                 String.format(
                     Locale.ROOT,
@@ -77,6 +80,7 @@ public void validateExporterConfig(final Settings settings) throws IllegalArgume
                 try {
                     DateTimeFormat.forPattern(indexPattern);
                 } catch (Exception e) {
+                    OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.INVALID_INDEX_PATTERN_EXCEPTIONS);
                     throw new IllegalArgumentException(
                         String.format(Locale.ROOT, "Invalid index pattern [%s] configured for the exporter", indexPattern)
                     );

src/main/java/org/opensearch/plugin/insights/core/listener/QueryInsightsListener.java

@@ -32,6 +32,8 @@
 import org.opensearch.common.inject.Inject;
 import org.opensearch.core.tasks.resourcetracker.TaskResourceInfo;
 import org.opensearch.core.xcontent.ToXContent;
+import org.opensearch.plugin.insights.core.metrics.OperationalMetric;
+import org.opensearch.plugin.insights.core.metrics.OperationalMetricsCounter;
 import org.opensearch.plugin.insights.core.service.QueryInsightsService;
 import org.opensearch.plugin.insights.core.service.categorizer.QueryShapeGenerator;
 import org.opensearch.plugin.insights.rules.model.Attribute;
@@ -261,6 +263,7 @@ private void constructSearchQueryRecord(final SearchPhaseContext context, final
             SearchQueryRecord record = new SearchQueryRecord(request.getOrCreateAbsoluteStartMillis(), measurements, attributes);
             queryInsightsService.addRecord(record);
         } catch (Exception e) {
+            OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.DATA_INGEST_EXCEPTIONS);
             log.error(String.format(Locale.ROOT, "fail to ingest query insight data, error: %s", e));
         }
     }

src/main/java/org/opensearch/plugin/insights/core/metrics/OperationalMetric.java

@@ -0,0 +1,37 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.plugin.insights.core.metrics;
+
+import java.util.Locale;
+
+public enum OperationalMetric {
+    LOCAL_INDEX_READER_PARSING_EXCEPTIONS("Number of errors when parsing with LocalIndexReader"),
+    LOCAL_INDEX_EXPORTER_BULK_FAILURES("Number of failures when ingesting Query Insights data to local indices"),
+    LOCAL_INDEX_EXPORTER_EXCEPTIONS("Number of exceptions in Query Insights LocalIndexExporter"),
+    INVALID_EXPORTER_TYPE_FAILURES("Number of invalid exporter type failures"),
+    INVALID_INDEX_PATTERN_EXCEPTIONS("Number of invalid index pattern exceptions"),
+    DATA_INGEST_EXCEPTIONS("Number of exceptions during data ingest in Query Insights"),
+    QUERY_CATEGORIZE_EXCEPTIONS("Number of exceptions when categorizing the queries"),
+    EXPORTER_FAIL_TO_CLOSE_EXCEPTION("Number of failures when closing the exporter");
+
+    private final String description;
+
+    OperationalMetric(String description) {
+        this.description = description;
+    }
+
+    public String getDescription() {
+        return description;
+    }
+
+    @Override
+    public String toString() {
+        return String.format(Locale.ROOT, "%s (%s)", name(), description);
+    }
+}

src/main/java/org/opensearch/plugin/insights/core/metrics/OperationalMetricsCounter.java

@@ -0,0 +1,95 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.plugin.insights.core.metrics;
+
+import java.util.Locale;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Stream;
+import org.opensearch.telemetry.metrics.Counter;
+import org.opensearch.telemetry.metrics.MetricsRegistry;
+import org.opensearch.telemetry.metrics.tags.Tags;
+
+/**
+ * Class contains all the Counters related to search query types.
+ */
+public final class OperationalMetricsCounter {
+    private static final String PREFIX = "search.insights.";
+    private static final String CLUSTER_NAME_TAG = "cluster_name";
+    private static final String UNIT = "1";
+
+    private final String clusterName;
+    private final MetricsRegistry metricsRegistry;
+    private final ConcurrentHashMap<OperationalMetric, Counter> metricCounterMap;
+
+    private static OperationalMetricsCounter instance;
+
+    /**
+     * Constructor of OperationalMetricsCounter
+     * @param metricsRegistry the OTel metrics registry
+     */
+    private OperationalMetricsCounter(String clusterName, MetricsRegistry metricsRegistry) {
+        this.clusterName = clusterName;
+        this.metricsRegistry = metricsRegistry;
+        this.metricCounterMap = new ConcurrentHashMap<>();
+        Stream.of(OperationalMetric.values()).forEach(name -> metricCounterMap.computeIfAbsent(name, this::createMetricCounter));
+    }
+
+    /**
+     * Initializes the singleton instance of OperationalMetricsCounter.
+     * This method must be called once before accessing the instance.
+     *
+     * @param clusterName the name of the cluster
+     * @param metricsRegistry the OTel metrics registry
+     */
+    public static synchronized void initialize(String clusterName, MetricsRegistry metricsRegistry) {
+        instance = new OperationalMetricsCounter(clusterName, metricsRegistry);
+    }
+
+    /**
+     * Get the singleton instance of OperationalMetricsCounter.
+     *
+     * @return the singleton instance
+     * @throws IllegalStateException if the instance is not yet initialized
+     */
+    public static synchronized OperationalMetricsCounter getInstance() {
+        if (instance == null) {
+            throw new IllegalStateException("OperationalMetricsCounter is not initialized. Call initialize() first.");
+        }
+        return instance;
+    }
+
+    /**
+     * Increment the operational metrics counter, attaching custom tags
+     *
+     * @param metricName name of the metric
+     * @param customTags custom tags of this metric
+     */
+    public void incrementCounter(OperationalMetric metricName, Tags customTags) {
+        Counter counter = metricCounterMap.computeIfAbsent(metricName, this::createMetricCounter);
+        Tags metricsTags = (customTags == null ? Tags.create() : customTags).addTag(CLUSTER_NAME_TAG, clusterName);
+        counter.add(1, metricsTags);
+    }
+
+    /**
+     * Increment the operational metrics counter
+     *
+     * @param metricName name of the metric
+     */
+    public void incrementCounter(OperationalMetric metricName) {
+        this.incrementCounter(metricName, null);
+    }
+
+    private Counter createMetricCounter(OperationalMetric metricName) {
+        return metricsRegistry.createCounter(
+            PREFIX + metricName.toString().toLowerCase(Locale.ROOT) + ".count",
+            metricName.getDescription(),
+            UNIT
+        );
+    }
+}

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

@@ -27,6 +27,8 @@
 import org.opensearch.index.query.QueryBuilder;
 import org.opensearch.index.query.QueryBuilders;
 import org.opensearch.index.query.RangeQueryBuilder;
+import org.opensearch.plugin.insights.core.metrics.OperationalMetric;
+import org.opensearch.plugin.insights.core.metrics.OperationalMetricsCounter;
 import org.opensearch.plugin.insights.rules.model.SearchQueryRecord;
 import org.opensearch.search.SearchHit;
 import org.opensearch.search.builder.SearchSourceBuilder;
@@ -113,6 +115,7 @@ public List<SearchQueryRecord> read(final String from, final String to) {
                     records.add(record);
                 }
             } catch (IndexNotFoundException ignored) {} catch (Exception e) {
+                OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.LOCAL_INDEX_READER_PARSING_EXCEPTIONS);
                 logger.error("Unable to parse search hit: ", e);
             }
             curr = curr.plusDays(1);

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

@@ -21,6 +21,8 @@
 import org.opensearch.client.Client;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.core.xcontent.NamedXContentRegistry;
+import org.opensearch.plugin.insights.core.metrics.OperationalMetric;
+import org.opensearch.plugin.insights.core.metrics.OperationalMetricsCounter;
 
 /**
  * Factory class for validating and creating Readers based on provided settings
@@ -57,6 +59,7 @@ public void validateReaderConfig(final Settings settings) throws IllegalArgument
         try {
             DateTimeFormat.forPattern(indexPattern);
         } catch (Exception e) {
+            OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.INVALID_INDEX_PATTERN_EXCEPTIONS);
             throw new IllegalArgumentException(
                 String.format(Locale.ROOT, "Invalid index pattern [%s] configured for the Reader", indexPattern)
             );

src/main/java/org/opensearch/plugin/insights/core/service/QueryInsightsService.java

@@ -30,6 +30,8 @@
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.xcontent.NamedXContentRegistry;
 import org.opensearch.plugin.insights.core.exporter.QueryInsightsExporterFactory;
+import org.opensearch.plugin.insights.core.metrics.OperationalMetric;
+import org.opensearch.plugin.insights.core.metrics.OperationalMetricsCounter;
 import org.opensearch.plugin.insights.core.reader.QueryInsightsReaderFactory;
 import org.opensearch.plugin.insights.core.service.categorizer.SearchQueryCategorizer;
 import org.opensearch.plugin.insights.rules.model.GroupingType;
@@ -188,6 +190,7 @@ public void drainRecords() {
             try {
                 searchQueryCategorizer.consumeRecords(records);
             } catch (Exception e) {
+                OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.QUERY_CATEGORIZE_EXCEPTIONS);
                 logger.error("Error while trying to categorize the queries.", e);
             }
         }

src/main/java/org/opensearch/plugin/insights/core/service/TopQueriesService.java

@@ -40,6 +40,8 @@
 import org.opensearch.plugin.insights.core.exporter.QueryInsightsExporter;
 import org.opensearch.plugin.insights.core.exporter.QueryInsightsExporterFactory;
 import org.opensearch.plugin.insights.core.exporter.SinkType;
+import org.opensearch.plugin.insights.core.metrics.OperationalMetric;
+import org.opensearch.plugin.insights.core.metrics.OperationalMetricsCounter;
 import org.opensearch.plugin.insights.core.reader.QueryInsightsReader;
 import org.opensearch.plugin.insights.core.reader.QueryInsightsReaderFactory;
 import org.opensearch.plugin.insights.core.service.grouper.MinMaxHeapQueryGrouper;
@@ -265,6 +267,7 @@ public void setExporter(final Settings settings) {
                 try {
                     queryInsightsExporterFactory.closeExporter(this.exporter);
                 } catch (IOException e) {
+                    OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.EXPORTER_FAIL_TO_CLOSE_EXCEPTION);
                     logger.error("Fail to close the current exporter when updating exporter, error: ", e);
                 }
                 this.exporter = queryInsightsExporterFactory.createExporter(
@@ -278,6 +281,7 @@ public void setExporter(final Settings settings) {
                 queryInsightsExporterFactory.closeExporter(this.exporter);
                 this.exporter = null;
             } catch (IOException e) {
+                OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.EXPORTER_FAIL_TO_CLOSE_EXCEPTION);
                 logger.error("Fail to close the current exporter when disabling exporter, error: ", e);
             }
         }

src/test/java/org/opensearch/plugin/insights/core/exporter/QueryInsightsExporterFactoryTests.java

@@ -8,7 +8,9 @@
 
 package org.opensearch.plugin.insights.core.exporter;
 
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 import static org.opensearch.plugin.insights.settings.QueryInsightsSettings.DEFAULT_TOP_QUERIES_EXPORTER_TYPE;
 import static org.opensearch.plugin.insights.settings.QueryInsightsSettings.EXPORTER_TYPE;
 import static org.opensearch.plugin.insights.settings.QueryInsightsSettings.EXPORT_INDEX;
@@ -17,6 +19,9 @@
 import org.junit.Before;
 import org.opensearch.client.Client;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.plugin.insights.core.metrics.OperationalMetricsCounter;
+import org.opensearch.telemetry.metrics.Counter;
+import org.opensearch.telemetry.metrics.MetricsRegistry;
 import org.opensearch.test.OpenSearchTestCase;
 
 /**
@@ -27,10 +32,16 @@ public class QueryInsightsExporterFactoryTests extends OpenSearchTestCase {
 
     private final Client client = mock(Client.class);
     private QueryInsightsExporterFactory queryInsightsExporterFactory;
+    private MetricsRegistry metricsRegistry;
 
     @Before
     public void setup() {
         queryInsightsExporterFactory = new QueryInsightsExporterFactory(client);
+        metricsRegistry = mock(MetricsRegistry.class);
+        when(metricsRegistry.createCounter(any(String.class), any(String.class), any(String.class))).thenAnswer(
+            invocation -> mock(Counter.class)
+        );
+        OperationalMetricsCounter.initialize("cluster", metricsRegistry);
     }
 
     public void testValidateConfigWhenResetExporter() {

src/test/java/org/opensearch/plugin/insights/core/metrics/OperationalMetricsCounterTests.java

@@ -0,0 +1,45 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.plugin.insights.core.metrics;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.mockito.ArgumentCaptor;
+import org.opensearch.telemetry.metrics.Counter;
+import org.opensearch.telemetry.metrics.MetricsRegistry;
+import org.opensearch.telemetry.metrics.tags.Tags;
+import org.opensearch.test.OpenSearchTestCase;
+
+/**
+ * Unit tests for the {@link OperationalMetricsCounter} class.
+ */
+public class OperationalMetricsCounterTests extends OpenSearchTestCase {
+    private static final String CLUSTER_NAME = "test-cluster";
+
+    public void testSingletonInitializationAndIncrement() {
+        Counter mockCounter = mock(Counter.class);
+        MetricsRegistry metricsRegistry = mock(MetricsRegistry.class);
+        // Stub the createCounter method to return the mockCounter
+        when(metricsRegistry.createCounter(any(), any(), any())).thenReturn(mockCounter);
+        OperationalMetricsCounter.initialize(CLUSTER_NAME, metricsRegistry);
+        OperationalMetricsCounter instance = OperationalMetricsCounter.getInstance();
+        ArgumentCaptor<String> nameCaptor = ArgumentCaptor.forClass(String.class);
+        verify(metricsRegistry, times(8)).createCounter(nameCaptor.capture(), any(), eq("1"));
+        assertNotNull(instance);
+        instance.incrementCounter(OperationalMetric.LOCAL_INDEX_READER_PARSING_EXCEPTIONS);
+        instance.incrementCounter(OperationalMetric.LOCAL_INDEX_READER_PARSING_EXCEPTIONS);
+        instance.incrementCounter(OperationalMetric.LOCAL_INDEX_READER_PARSING_EXCEPTIONS);
+        verify(mockCounter, times(3)).add(eq(1.0), any(Tags.class));
+    }
+}