Add unreferenced file cleanup count to merge stats (opensearch-projec…

…t#10204)

Signed-off-by: Rishav Sagar <rissag@amazon.com>
Co-authored-by: Rishav Sagar <rissag@amazon.com>

CHANGELOG.md

@@ -89,6 +89,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Async blob read support for encrypted containers ([#10131](https://github.com/opensearch-project/OpenSearch/pull/10131))
 - Add capability to restrict async durability mode for remote indexes ([#10189](https://github.com/opensearch-project/OpenSearch/pull/10189))
 - Add Doc Status Counter for Indexing Engine ([#4562](https://github.com/opensearch-project/OpenSearch/issues/4562))
+- Add unreferenced file cleanup count to merge stats ([#10204](https://github.com/opensearch-project/OpenSearch/pull/10204))
 
 ### Dependencies
 - Bump `peter-evans/create-or-update-comment` from 2 to 3 ([#9575](https://github.com/opensearch-project/OpenSearch/pull/9575))

server/src/main/java/org/opensearch/index/engine/Engine.java

@@ -145,6 +145,7 @@ public abstract class Engine implements LifecycleAware, Closeable {
     protected final EngineConfig engineConfig;
     protected final Store store;
     protected final AtomicBoolean isClosed = new AtomicBoolean(false);
+    private final CounterMetric totalUnreferencedFileCleanUpsPerformed = new CounterMetric();
     private final CountDownLatch closedLatch = new CountDownLatch(1);
     protected final EventListener eventListener;
     protected final ReentrantLock failEngineLock = new ReentrantLock();
@@ -267,6 +268,13 @@ protected final DocsStats docsStats(IndexReader indexReader) {
         return new DocsStats(numDocs, numDeletedDocs, sizeInBytes);
     }
 
+    /**
+     * Returns the unreferenced file cleanup count for this engine.
+     */
+    public long unreferencedFileCleanUpsPerformed() {
+        return totalUnreferencedFileCleanUpsPerformed.count();
+    }
+
     /**
      * Performs the pre-closing checks on the {@link Engine}.
      *
@@ -1340,7 +1348,9 @@ private void cleanUpUnreferencedFiles() {
                     .setOpenMode(IndexWriterConfig.OpenMode.APPEND)
             )
         ) {
-            // do nothing and close this will kick off IndexFileDeleter which will remove all unreferenced files.
+            // do nothing except increasing metric count and close this will kick off IndexFileDeleter which will
+            // remove all unreferenced files
+            totalUnreferencedFileCleanUpsPerformed.inc();
         } catch (Exception ex) {
             logger.error("Error while deleting unreferenced file ", ex);
         }

server/src/main/java/org/opensearch/index/merge/MergeStats.java

@@ -32,6 +32,7 @@
 
 package org.opensearch.index.merge;
 
+import org.opensearch.Version;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
@@ -65,9 +66,9 @@ public class MergeStats implements Writeable, ToXContentFragment {
 
     private long totalBytesPerSecAutoThrottle;
 
-    public MergeStats() {
+    private long unreferencedFileCleanUpsPerformed;
 
-    }
+    public MergeStats() {}
 
     public MergeStats(StreamInput in) throws IOException {
         total = in.readVLong();
@@ -81,6 +82,9 @@ public MergeStats(StreamInput in) throws IOException {
         totalStoppedTimeInMillis = in.readVLong();
         totalThrottledTimeInMillis = in.readVLong();
         totalBytesPerSecAutoThrottle = in.readVLong();
+        if (in.getVersion().onOrAfter(Version.V_3_0_0)) {
+            unreferencedFileCleanUpsPerformed = in.readOptionalVLong();
+        }
     }
 
     public void add(
@@ -133,13 +137,22 @@ public void addTotals(MergeStats mergeStats) {
         this.totalSizeInBytes += mergeStats.totalSizeInBytes;
         this.totalStoppedTimeInMillis += mergeStats.totalStoppedTimeInMillis;
         this.totalThrottledTimeInMillis += mergeStats.totalThrottledTimeInMillis;
+        addUnreferencedFileCleanUpStats(mergeStats.unreferencedFileCleanUpsPerformed);
         if (this.totalBytesPerSecAutoThrottle == Long.MAX_VALUE || mergeStats.totalBytesPerSecAutoThrottle == Long.MAX_VALUE) {
             this.totalBytesPerSecAutoThrottle = Long.MAX_VALUE;
         } else {
             this.totalBytesPerSecAutoThrottle += mergeStats.totalBytesPerSecAutoThrottle;
         }
     }
 
+    public void addUnreferencedFileCleanUpStats(long unreferencedFileCleanUpsPerformed) {
+        this.unreferencedFileCleanUpsPerformed += unreferencedFileCleanUpsPerformed;
+    }
+
+    public long getUnreferencedFileCleanUpsPerformed() {
+        return this.unreferencedFileCleanUpsPerformed;
+    }
+
     /**
      * The total number of merges executed.
      */
@@ -240,6 +253,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.field(Fields.TOTAL_THROTTLE_BYTES_PER_SEC).value(new ByteSizeValue(totalBytesPerSecAutoThrottle).toString());
         }
         builder.field(Fields.TOTAL_THROTTLE_BYTES_PER_SEC_IN_BYTES, totalBytesPerSecAutoThrottle);
+        builder.field(Fields.UNREFERENCED_FILE_CLEANUPS_PERFORMED, unreferencedFileCleanUpsPerformed);
         builder.endObject();
         return builder;
     }
@@ -267,6 +281,7 @@ static final class Fields {
         static final String TOTAL_SIZE_IN_BYTES = "total_size_in_bytes";
         static final String TOTAL_THROTTLE_BYTES_PER_SEC_IN_BYTES = "total_auto_throttle_in_bytes";
         static final String TOTAL_THROTTLE_BYTES_PER_SEC = "total_auto_throttle";
+        static final String UNREFERENCED_FILE_CLEANUPS_PERFORMED = "unreferenced_file_cleanups_performed";
     }
 
     @Override
@@ -282,5 +297,8 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeVLong(totalStoppedTimeInMillis);
         out.writeVLong(totalThrottledTimeInMillis);
         out.writeVLong(totalBytesPerSecAutoThrottle);
+        if (out.getVersion().onOrAfter(Version.V_3_0_0)) {
+            out.writeOptionalVLong(unreferencedFileCleanUpsPerformed);
+        }
     }
 }

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -1383,7 +1383,9 @@ public MergeStats mergeStats() {
         if (engine == null) {
             return new MergeStats();
         }
-        return engine.getMergeStats();
+        final MergeStats mergeStats = engine.getMergeStats();
+        mergeStats.addUnreferencedFileCleanUpStats(engine.unreferencedFileCleanUpsPerformed());
+        return mergeStats;
     }
 
     public SegmentsStats segmentStats(boolean includeSegmentFileSizes, boolean includeUnloadedSegments) {

server/src/test/java/org/opensearch/index/engine/InternalEngineTests.java

@@ -233,7 +233,6 @@
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -3334,6 +3333,9 @@ public void onFailedEngine(String reason, Exception e) {
             );
 
             assertTrue(cleanupCompleted.await(10, TimeUnit.SECONDS));
+            // Cleanup count will be incremented whenever cleanup is performed correctly.
+            long unreferencedFileCleanUpsPerformed = engine.unreferencedFileCleanUpsPerformed();
+            assertThat(unreferencedFileCleanUpsPerformed, equalTo(1L));
         } catch (Exception ex) {
             throw new AssertionError(ex);
         }
@@ -3445,6 +3447,9 @@ public void onFailedEngine(String reason, Exception e) {
             );
 
             assertTrue(cleanupCompleted.await(10, TimeUnit.SECONDS));
+            // Cleanup count will not be incremented whenever cleanup is disabled.
+            long unreferencedFileCleanUpsPerformed = engine.unreferencedFileCleanUpsPerformed();
+            assertThat(unreferencedFileCleanUpsPerformed, equalTo(0L));
         } catch (Exception ex) {
             throw new AssertionError(ex);
         }
@@ -3549,6 +3554,9 @@ public void onFailedEngine(String reason, Exception e) {
             );
 
             assertTrue(cleanupCompleted.await(10, TimeUnit.SECONDS));
+            // Cleanup count will not be incremented whenever there is some issue with cleanup.
+            long unreferencedFileCleanUpsPerformed = engine.unreferencedFileCleanUpsPerformed();
+            assertThat(unreferencedFileCleanUpsPerformed, equalTo(0L));
         } catch (Exception ex) {
             throw new AssertionError(ex);
         }