Staggered merges metrics

Author: Sukriti Sinha

server/src/main/java/org/opensearch/index/autoforcemerge/AutoForceMergeManager.java

@@ -67,6 +67,8 @@ public class AutoForceMergeManager extends AbstractLifecycleComponent {
     private NodeValidator nodeValidator;
     private ShardValidator shardValidator;
     private Integer allocatedProcessors;
+    private String nodeId;
+    private final AutoForceMergeMetrics autoForceMergeMetrics;
     private ResourceTrackerProvider.ResourceTrackers resourceTrackers;
     private final ForceMergeManagerSettings forceMergeManagerSettings;
     private final CommonStatsFlags flags = new CommonStatsFlags(CommonStatsFlags.Flag.Segments, CommonStatsFlags.Flag.Translog);
@@ -78,14 +80,16 @@ public AutoForceMergeManager(
         ThreadPool threadPool,
         MonitorService monitorService,
         IndicesService indicesService,
-        ClusterService clusterService
+        ClusterService clusterService,
+        AutoForceMergeMetrics autoForceMergeMetrics
     ) {
         this.threadPool = threadPool;
         this.osService = monitorService.osService();
         this.fsService = monitorService.fsService();
         this.jvmService = monitorService.jvmService();
         this.clusterService = clusterService;
         this.indicesService = indicesService;
+        this.autoForceMergeMetrics = autoForceMergeMetrics;
         this.forceMergeManagerSettings = new ForceMergeManagerSettings(clusterService, this::modifySchedulerInterval);
         this.task = new AsyncForceMergeTask();
         this.mergingShards = new HashSet<>();
@@ -98,6 +102,7 @@ protected void doStart() {
         this.shardValidator = new ShardValidator();
         this.allocatedProcessors = OpenSearchExecutors.allocatedProcessors(clusterService.getSettings());
         this.resourceTrackers = ResourceTrackerProvider.create(threadPool);
+        this.nodeId = clusterService.localNode().getId(); 
     }
 
     @Override
@@ -119,20 +124,44 @@ private void modifySchedulerInterval(TimeValue schedulerInterval) {
     }
 
     private void triggerForceMerge() {
-        if (isValidForForceMerge() == false) {
-            return;
+        long startTime = System.currentTimeMillis();
+        try {
+            if (isValidForForceMerge() == false) {
+                return;
+            }
+            executeForceMergeOnShards();
+        } finally {
+            autoForceMergeMetrics.recordInHistogram(
+                autoForceMergeMetrics.totalSchedulerExecutionTime,
+                (double) System.currentTimeMillis() - startTime,
+                autoForceMergeMetrics.getTags(nodeId, null)
+            );
         }
-        executeForceMergeOnShards();
     }
 
     private boolean isValidForForceMerge() {
         if (configurationValidator.hasWarmNodes() == false) {
             resourceTrackers.stop();
             logger.debug("No warm nodes found. Skipping Auto Force merge.");
+            autoForceMergeMetrics.incrementCounter(
+                autoForceMergeMetrics.totalMergesSkipped,
+                1.0,
+                autoForceMergeMetrics.getTags(nodeId, null)
+            );
             return false;
         }
         if (nodeValidator.validate().isAllowed() == false) {
             logger.debug("Node capacity constraints are not allowing to trigger auto ForceMerge");
+            autoForceMergeMetrics.incrementCounter(
+                autoForceMergeMetrics.skipsFromNodeValidator,
+                1.0,
+                autoForceMergeMetrics.getTags(nodeId, null)
+            );
+            autoForceMergeMetrics.incrementCounter(
+                autoForceMergeMetrics.totalMergesSkipped,
+                1.0,
+                autoForceMergeMetrics.getTags(nodeId, null)
+            );
             return false;
         }
         return true;
@@ -157,14 +186,47 @@ private void executeForceMergeOnShards() {
 
     private void executeForceMergeForShard(IndexShard shard) {
         CompletableFuture.runAsync(() -> {
+            long startTime = System.currentTimeMillis();
+            String shardId = String.valueOf(shard.shardId().getId());
             try {
                 mergingShards.add(shard.shardId().getId());
+                autoForceMergeMetrics.incrementCounter(
+                    autoForceMergeMetrics.totalMergesTriggered,
+                    1.0,
+                    autoForceMergeMetrics.getTags(nodeId, null)
+                );
+
+                CommonStats preStats = new CommonStats(indicesService.getIndicesQueryCache(), shard, flags);
+                if (preStats.getSegments() != null) {
+                    autoForceMergeMetrics.incrementCounter(
+                        autoForceMergeMetrics.segmentCount,
+                        (double) preStats.getSegments().getCount(),
+                        autoForceMergeMetrics.getTags(nodeId, shardId)
+                    );
+                    autoForceMergeMetrics.incrementCounter(
+                        autoForceMergeMetrics.shardSize,
+                        (double) preStats.getStore().getSizeInBytes(),
+                        autoForceMergeMetrics.getTags(nodeId, shardId)
+                    );
+                }
+
                 shard.forceMerge(new ForceMergeRequest().maxNumSegments(forceMergeManagerSettings.getSegmentCount()));
                 logger.debug("Merging is completed successfully for the shard {}", shard.shardId());
+
             } catch (Exception e) {
                 logger.error("Error during force merge for shard {}\nException: {}", shard.shardId(), e);
+                autoForceMergeMetrics.incrementCounter(
+                    autoForceMergeMetrics.totalMergesFailed,
+                    1.0,
+                    autoForceMergeMetrics.getTags(nodeId, null)
+                );
             } finally {
                 mergingShards.remove(shard.shardId().getId());
+                autoForceMergeMetrics.recordInHistogram(
+                    autoForceMergeMetrics.shardForceMergeLatency,
+                    (double) System.currentTimeMillis() - startTime,
+                    autoForceMergeMetrics.getTags(nodeId, shardId)
+                );
             }
         }, threadPool.executor(ThreadPool.Names.FORCE_MERGE));
         logger.info("Successfully triggered force merge for shard {}", shard.shardId());

server/src/main/java/org/opensearch/index/autoforcemerge/AutoForceMergeMetrics.java

@@ -0,0 +1,121 @@
+/*
+ * 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.index.autoforcemerge;
+
+import org.opensearch.telemetry.metrics.Counter;
+import org.opensearch.telemetry.metrics.Histogram;
+import org.opensearch.telemetry.metrics.MetricsRegistry;
+import org.opensearch.telemetry.metrics.tags.Tags;
+
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Class containing metrics (counters/latency) specific to Auto Force merges.
+ *
+ * @opensearch.internal
+ */
+public class AutoForceMergeMetrics {
+
+    private static final String LATENCY_METRIC_UNIT_MS = "ms";
+    private static final String COUNTER_METRICS_UNIT = "1";
+    private static final String SIZE_METRIC_UNIT = "bytes";
+
+    public static final String NODE_ID = "NodeId";
+    public static final String SHARD_ID = "ShardId";
+    public static final String INDEX_NAME = "IndexName";
+
+    public final Histogram totalSchedulerExecutionTime;
+    public final Counter totalMergesTriggered;
+    public final Counter totalMergesSkipped;
+    public final Counter skipsFromNodeValidator;
+    public final Counter totalMergesFailed;
+
+    // Shard specific metrics
+    public final Histogram shardForceMergeLatency;
+    public final Counter shardSize;
+    public final Counter segmentCount;
+
+    public AutoForceMergeMetrics(MetricsRegistry metricsRegistry) {
+        totalSchedulerExecutionTime = metricsRegistry.createHistogram(
+            "auto_force_merge.scheduler.execution_time",
+            "Histogram for tracking total scheduler execution time.",
+            LATENCY_METRIC_UNIT_MS
+        );
+
+        totalMergesTriggered = metricsRegistry.createCounter(
+            "auto_force_merge.merges.triggered",
+            "Counter for number of force merges triggered.",
+            COUNTER_METRICS_UNIT
+        );
+
+        totalMergesSkipped = metricsRegistry.createCounter(
+            "auto_force_merge.merges.skipped.total",
+            "Counter for number of force merges skipped.",
+            COUNTER_METRICS_UNIT
+        );
+
+        skipsFromNodeValidator = metricsRegistry.createCounter(
+            "auto_force_merge.merges.skipped.node_validator",
+            "Counter for number of force merges skipped due to Node Validator.",
+            COUNTER_METRICS_UNIT
+        );
+
+        totalMergesFailed = metricsRegistry.createCounter(
+            "auto_force_merge.merges.failed",
+            "Counter for number of force merges failed.",
+            COUNTER_METRICS_UNIT
+        );
+
+        shardForceMergeLatency = metricsRegistry.createHistogram(
+            "auto_force_merge.shard.merge_latency",
+            "Histogram for tracking time taken by force merge on individual shards.",
+            LATENCY_METRIC_UNIT_MS
+        );
+
+        shardSize = metricsRegistry.createCounter(
+            "auto_force_merge.shard.size",
+            "Counter for tracking shard size during force merge operations.",
+            SIZE_METRIC_UNIT
+        );
+
+        segmentCount = metricsRegistry.createCounter(
+            "auto_force_merge.shard.segment_count",
+            "Counter for tracking segment count during force merge operations.",
+            COUNTER_METRICS_UNIT
+        );
+    }
+
+    public void recordInHistogram(Histogram histogram, Double value, Optional<Tags> tags) {
+        if (Objects.isNull(tags) || tags.isEmpty()) {
+            histogram.record(value);
+            return;
+        }
+        histogram.record(value, tags.get());
+    }
+
+    public void incrementCounter(Counter counter, Double value, Optional<Tags> tags) {
+        if (Objects.isNull(tags) || tags.isEmpty()) {
+            counter.add(value);
+            return;
+        }
+        counter.add(value, tags.get());
+    }
+
+    public Optional<Tags> getTags(String nodeId, String shardId) {
+        Optional<Tags> tags = Optional.of(Tags.create());
+        if (nodeId != null) {
+            tags.get().addTag(NODE_ID, nodeId);
+        }
+        if (shardId != null) {
+            tags.get().addTag(SHARD_ID, shardId);
+        }
+        return tags;
+    }
+}

server/src/main/java/org/opensearch/node/Node.java

@@ -161,6 +161,7 @@
 import org.opensearch.index.SegmentReplicationStatsTracker;
 import org.opensearch.index.analysis.AnalysisRegistry;
 import org.opensearch.index.autoforcemerge.AutoForceMergeManager;
+import org.opensearch.index.autoforcemerge.AutoForceMergeMetrics;
 import org.opensearch.index.compositeindex.CompositeIndexSettings;
 import org.opensearch.index.engine.EngineFactory;
 import org.opensearch.index.engine.MergedSegmentWarmerFactory;
@@ -1208,7 +1209,14 @@ protected Node(final Environment initialEnvironment, Collection<PluginInfo> clas
                 workloadGroupService
             );
 
-            this.autoForceMergeManager = new AutoForceMergeManager(threadPool, monitorService, indicesService, clusterService);
+            final AutoForceMergeMetrics autoForceMergeMetrics = new AutoForceMergeMetrics(metricsRegistry);
+            this.autoForceMergeManager = new AutoForceMergeManager(
+                threadPool,
+                monitorService,
+                indicesService,
+                clusterService,
+                autoForceMergeMetrics
+            );
 
             final Collection<SecureSettingsFactory> secureSettingsFactories = pluginsService.filterPlugins(Plugin.class)
                 .stream()