Add current_application_duration_ms to cluster state download stats in node stats API; add UT for test coverage (#20922)

Signed-off-by: Ayushi Arya <ayuaryak@amazon.com>

Author: Ayushiarya246

server/src/main/java/org/opensearch/cluster/service/ClusterApplier.java

@@ -64,6 +64,14 @@ public interface ClusterApplier {
      */
     void onNewClusterState(String source, Supplier<ClusterState> clusterStateSupplier, ClusterApplyListener listener);
 
+    /**
+     * Returns the duration in milliseconds of the currently running cluster state application,
+     * or 0 if no application is in progress
+     */
+    default long getCurrentApplicationDurationMs() {
+        return 0;
+    }
+
     /**
      * Listener for results of cluster state application
      *

server/src/main/java/org/opensearch/cluster/service/ClusterApplierService.java

@@ -129,6 +129,10 @@ public class ClusterApplierService extends AbstractLifecycleComponent implements
 
     private final ClusterManagerMetrics clusterManagerMetrics;
 
+    // application duration tracking
+    private static final long NOT_RUNNING = -1L;
+    private volatile long applicationStartTimeNanos = NOT_RUNNING;
+
     public ClusterApplierService(String nodeName, Settings settings, ClusterSettings clusterSettings, ThreadPool threadPool) {
         this(nodeName, settings, clusterSettings, threadPool, new ClusterManagerMetrics(NoopMetricsRegistry.INSTANCE));
     }
@@ -468,7 +472,7 @@ private void runTask(UpdateTask task) {
             logger.debug("processing [{}]: ignoring, cluster applier service not started", task.source);
             return;
         }
-
+        this.applicationStartTimeNanos = System.nanoTime();
         logger.debug("processing [{}]: execute", task.source);
         final ClusterState previousClusterState = state.get();
 
@@ -492,6 +496,7 @@ private void runTask(UpdateTask task) {
                 e
             );
             warnAboutSlowTaskIfNeeded(executionTime, task.source, stopWatch);
+            this.applicationStartTimeNanos = NOT_RUNNING;
             task.listener.onFailure(task.source, e);
             return;
         }
@@ -500,6 +505,7 @@ private void runTask(UpdateTask task) {
             TimeValue executionTime = TimeValue.timeValueMillis(Math.max(0, currentTimeInMillis() - startTimeMS));
             logger.debug("processing [{}]: took [{}] no change in cluster state", task.source, executionTime);
             warnAboutSlowTaskIfNeeded(executionTime, task.source, stopWatch);
+            this.applicationStartTimeNanos = NOT_RUNNING;
             task.listener.onSuccess(task.source);
         } else {
             if (logger.isTraceEnabled()) {
@@ -524,6 +530,7 @@ private void runTask(UpdateTask task) {
                     newClusterState.stateUUID()
                 );
                 warnAboutSlowTaskIfNeeded(executionTime, task.source, stopWatch);
+                this.applicationStartTimeNanos = NOT_RUNNING;
                 // Then we call the ClusterApplyListener of the task
                 task.listener.onSuccess(task.source);
             } catch (Exception e) {
@@ -555,6 +562,7 @@ private void runTask(UpdateTask task) {
                 // failing to apply a cluster state with an exception indicates a bug in validation or in one of the appliers; if we
                 // continue we will retry with the same cluster state but that might not help.
                 assert applicationMayFail();
+                this.applicationStartTimeNanos = NOT_RUNNING;
                 task.listener.onFailure(task.source, e);
             }
         }
@@ -785,6 +793,19 @@ protected boolean applicationMayFail() {
         return false;
     }
 
+    /**
+     * Returns the duration in milliseconds of the currently running cluster state application,
+     * or 0 if no application is in progress.
+     */
+    @Override
+    public long getCurrentApplicationDurationMs() {
+        long startNanos = this.applicationStartTimeNanos;
+        if (startNanos == NOT_RUNNING) {
+            return 0;
+        }
+        return TimeValue.nsecToMSec(System.nanoTime() - startNanos);
+    }
+
     /**
      * Pre-commit State of the cluster-applier
      * @return ClusterState

server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java

@@ -224,6 +224,7 @@ public static RemoteClusterStateValidationMode parseString(String mode) {
     private final Supplier<RepositoriesService> repositoriesService;
     private final Settings settings;
     private final LongSupplier relativeTimeNanosSupplier;
+    private final LongSupplier applicationDurationMsSupplier;
     private final ThreadPool threadpool;
     private final List<IndexMetadataUploadListener> indexMetadataUploadListeners;
     private BlobStoreRepository blobStoreRepository;
@@ -269,13 +270,15 @@ public RemoteClusterStateService(
         LongSupplier relativeTimeNanosSupplier,
         ThreadPool threadPool,
         List<IndexMetadataUploadListener> indexMetadataUploadListeners,
-        NamedWriteableRegistry namedWriteableRegistry
+        NamedWriteableRegistry namedWriteableRegistry,
+        LongSupplier applicationDurationMsSupplier
     ) {
         assert isRemoteClusterStateConfigured(settings) : "Remote cluster state is not configured";
         this.nodeId = nodeId;
         this.repositoriesService = repositoriesService;
         this.settings = settings;
         this.relativeTimeNanosSupplier = relativeTimeNanosSupplier;
+        this.applicationDurationMsSupplier = applicationDurationMsSupplier;
         this.threadpool = threadPool;
         clusterSettings = clusterService.getClusterSettings();
         this.slowWriteLoggingThreshold = clusterSettings.get(SLOW_WRITE_LOGGING_THRESHOLD);
@@ -2115,11 +2118,15 @@ public PersistedStateStats getUploadStats() {
     }
 
     public PersistedStateStats getFullDownloadStats() {
-        return remoteStateStats.getRemoteFullDownloadStats();
+        RemoteDownloadStats stats = (RemoteDownloadStats) remoteStateStats.getRemoteFullDownloadStats();
+        stats.setCurrentApplicationDurationMs(applicationDurationMsSupplier.getAsLong());
+        return stats;
     }
 
     public PersistedStateStats getDiffDownloadStats() {
-        return remoteStateStats.getRemoteDiffDownloadStats();
+        RemoteDownloadStats stats = (RemoteDownloadStats) remoteStateStats.getRemoteDiffDownloadStats();
+        stats.setCurrentApplicationDurationMs(applicationDurationMsSupplier.getAsLong());
+        return stats;
     }
 
     public void fullDownloadFailed() {

server/src/main/java/org/opensearch/gateway/remote/RemoteDownloadStats.java

@@ -22,11 +22,14 @@ public class RemoteDownloadStats extends PersistedStateStats {
     private AtomicLong checksumValidationFailedCount = new AtomicLong(0);
     public static final String INCOMING_PUBLICATION_FAILED_COUNT = "incoming_publication_failed_count";
     private AtomicLong incomingPublicationFailedCount = new AtomicLong(0);
+    static final String CURRENT_APPLICATION_DURATION_MS = "current_application_duration_ms";
+    private AtomicLong currentApplicationDurationMs = new AtomicLong(0);
 
     public RemoteDownloadStats(String statsName) {
         super(statsName);
         addToExtendedFields(CHECKSUM_VALIDATION_FAILED_COUNT, checksumValidationFailedCount);
         addToExtendedFields(INCOMING_PUBLICATION_FAILED_COUNT, incomingPublicationFailedCount);
+        addToExtendedFields(CURRENT_APPLICATION_DURATION_MS, currentApplicationDurationMs);
     }
 
     public void checksumValidationFailedCount() {
@@ -44,4 +47,12 @@ public void incomingPublicationFailedCount() {
     public long getIncomingPublicationFailedCount() {
         return incomingPublicationFailedCount.get();
     }
+
+    public void setCurrentApplicationDurationMs(long durationMs) {
+        currentApplicationDurationMs.set(durationMs);
+    }
+
+    public long getCurrentApplicationDurationMs() {
+        return currentApplicationDurationMs.get();
+    }
 }

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

@@ -891,7 +891,9 @@ protected Node(final Environment initialEnvironment, Collection<PluginInfo> clas
                     threadPool::preciseRelativeTimeInNanos,
                     threadPool,
                     List.of(remoteIndexPathUploader),
-                    namedWriteableRegistry
+                    namedWriteableRegistry,
+                    // Supplier for current cluster state application duration in ms (0 if idle)
+                    () -> clusterService.getClusterApplierService().getCurrentApplicationDurationMs()
                 );
                 remoteClusterStateCleanupManager = remoteClusterStateService.getCleanupManager();
             } else {

server/src/test/java/org/opensearch/cluster/coordination/PersistedStateStatsTests.java

@@ -8,9 +8,12 @@
 
 package org.opensearch.cluster.coordination;
 
+import org.opensearch.common.io.stream.BytesStreamOutput;
+import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.test.OpenSearchTestCase;
 import org.junit.Before;
 
+import java.io.IOException;
 import java.util.concurrent.atomic.AtomicLong;
 
 public class PersistedStateStatsTests extends OpenSearchTestCase {
@@ -59,4 +62,63 @@ public void testAddMultipleFields() {
         assertEquals(42, persistedStateStats.getExtendedFields().get(fieldName1).get());
         assertEquals(84, persistedStateStats.getExtendedFields().get(fieldName2).get());
     }
+
+    public void testBwcSerializationWithExtraExtendedField() throws IOException {
+        // Simulate NEW node writing stats with 3 extended fields
+        PersistedStateStats newNodeStats = new PersistedStateStats("test_download");
+        newNodeStats.addToExtendedFields("checksum_validation_failed_count", new AtomicLong(0));
+        newNodeStats.addToExtendedFields("incoming_publication_failed_count", new AtomicLong(1));
+        newNodeStats.addToExtendedFields("current_application_duration_ms", new AtomicLong(42));
+
+        // Serialize
+        BytesStreamOutput out = new BytesStreamOutput();
+        newNodeStats.writeTo(out);
+
+        // Deserialize as generic PersistedStateStats (simulates OLD node reading)
+        StreamInput in = out.bytes().streamInput();
+        PersistedStateStats oldNodeDeserialized = new PersistedStateStats(in);
+
+        // Old node reads all 3 fields successfully — no bytes left unread
+        assertEquals(3, oldNodeDeserialized.getExtendedFields().size());
+        assertEquals(42, oldNodeDeserialized.getExtendedFields().get("current_application_duration_ms").get());
+
+        // Simulate OLD node writing stats with only 2 extended fields
+        PersistedStateStats oldNodeStats = new PersistedStateStats("test_download");
+        oldNodeStats.addToExtendedFields("checksum_validation_failed_count", new AtomicLong(0));
+        oldNodeStats.addToExtendedFields("incoming_publication_failed_count", new AtomicLong(1));
+        // Note: NO current_application_duration_ms
+
+        BytesStreamOutput out2 = new BytesStreamOutput();
+        oldNodeStats.writeTo(out2);
+
+        StreamInput in2 = out2.bytes().streamInput();
+        PersistedStateStats newNodeDeserialized = new PersistedStateStats(in2);
+
+        // New node reads only 2 fields — current_application_duration_ms simply absent
+        assertEquals(2, newNodeDeserialized.getExtendedFields().size());
+        assertFalse(newNodeDeserialized.getExtendedFields().containsKey("current_application_duration_ms"));
+    }
+
+    // serialization with extendedFields
+    public void testSerializationRoundTripWithExtendedFields() throws IOException {
+        PersistedStateStats original = new PersistedStateStats("test_download");
+        original.stateSucceeded();
+        original.stateTook(100);
+        original.addToExtendedFields("current_application_duration_ms", new AtomicLong(5000));
+
+        // Serialize
+        BytesStreamOutput out = new BytesStreamOutput();
+        original.writeTo(out);
+
+        // Deserialize
+        StreamInput in = out.bytes().streamInput();
+        PersistedStateStats deserialized = new PersistedStateStats(in);
+
+        assertEquals("test_download", deserialized.getStatsName());
+        assertEquals(1, deserialized.getSuccessCount());
+        assertEquals(100, deserialized.getTotalTimeInMillis());
+        assertTrue(deserialized.getExtendedFields().containsKey("current_application_duration_ms"));
+        assertEquals(5000, deserialized.getExtendedFields().get("current_application_duration_ms").get());
+    }
+
 }

server/src/test/java/org/opensearch/cluster/service/ClusterApplierServiceTests.java

@@ -677,6 +677,68 @@ public void onFailure(String source, Exception e) {
         verifyNoInteractions(listenerslatencyHistogram);
     }
 
+    // testing for application duration tracking
+
+    public void testGetCurrentApplicationDurationMsReturnsZeroWhenIdle() {
+        // No task running → should return 0
+        assertEquals(0, clusterApplierService.getCurrentApplicationDurationMs());
+    }
+
+    public void testGetCurrentApplicationDurationMsDuringApplication() throws Exception {
+
+        CountDownLatch taskStarted = new CountDownLatch(1);
+        CountDownLatch taskCanFinish = new CountDownLatch(1);
+        CountDownLatch taskDone = new CountDownLatch(1);
+
+        clusterApplierService.onNewClusterState("test", () -> {
+            taskStarted.countDown();
+            try {
+                taskCanFinish.await();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+            return ClusterState.builder(clusterApplierService.state()).build();
+        }, new ClusterApplier.ClusterApplyListener() {
+            @Override
+            public void onSuccess(String source) {
+                taskDone.countDown();
+            }
+
+            @Override
+            public void onFailure(String source, Exception e) {
+                taskDone.countDown();
+            }
+        });
+
+        taskStarted.await();  // Wait for task to start
+        long duration = clusterApplierService.getCurrentApplicationDurationMs();
+        assertTrue("Duration should be >= 0 during application, got: " + duration, duration >= 0);
+        taskCanFinish.countDown();  // Let task finish
+        taskDone.await();  // Wait for task to fully complete to avoid interference with other tests
+    }
+
+    public void testGetCurrentApplicationDurationMsResetsAfterCompletion() throws Exception {
+        CountDownLatch taskDone = new CountDownLatch(1);
+
+        clusterApplierService.onNewClusterState(
+            "test",
+            () -> ClusterState.builder(clusterApplierService.state()).build(),
+            new ClusterApplier.ClusterApplyListener() {
+                @Override
+                public void onSuccess(String source) {
+                    taskDone.countDown();
+                }
+
+                @Override
+                public void onFailure(String source, Exception e) {
+                    taskDone.countDown();
+                }
+            }
+        );
+        taskDone.await();
+        assertEquals(0, clusterApplierService.getCurrentApplicationDurationMs());
+    }
+
     static class TimedClusterApplierService extends ClusterApplierService {
 
         final ClusterSettings clusterSettings;

server/src/test/java/org/opensearch/gateway/GatewayMetaStatePersistedStateTests.java

@@ -524,7 +524,8 @@ public void testDataOnlyNodePersistence() throws Exception {
                                 DefaultRemoteStoreSettings.INSTANCE
                             )
                         ),
-                        writableRegistry()
+                        writableRegistry(),
+                        () -> 0L
                     );
                 } else {
                     return null;

server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java

@@ -263,7 +263,8 @@ public void setup() {
                     DefaultRemoteStoreSettings.INSTANCE
                 )
             ),
-            namedWriteableRegistry
+            namedWriteableRegistry,
+            () -> 0L
         );
     }
 
@@ -305,7 +306,8 @@ public void testFailInitializationWhenRemoteStateDisabled() {
                         DefaultRemoteStoreSettings.INSTANCE
                     )
                 ),
-                writableRegistry()
+                writableRegistry(),
+                () -> 0L
             )
         );
     }
@@ -383,7 +385,8 @@ public void testWriteFullMetadataSuccessPublicationEnabled() throws IOException
                     DefaultRemoteStoreSettings.INSTANCE
                 )
             ),
-            writableRegistry()
+            writableRegistry(),
+            () -> 0L
         );
         assertTrue(remoteClusterStateService.isRemotePublicationEnabled());
         final ClusterState clusterState = generateClusterStateWithOneIndex().nodes(nodesWithLocalNodeClusterManager())
@@ -757,7 +760,8 @@ public void testWriteIncrementalMetadataSuccessWhenPublicationEnabled() throws I
                     DefaultRemoteStoreSettings.INSTANCE
                 )
             ),
-            writableRegistry()
+            writableRegistry(),
+            () -> 0L
         );
         assertTrue(remoteClusterStateService.isRemotePublicationEnabled());
         final ClusterState clusterState = generateClusterStateWithOneIndex().nodes(nodesWithLocalNodeClusterManager()).build();
@@ -2798,7 +2802,8 @@ public void testRemoteRoutingTableInitializedWhenEnabled() {
                     DefaultRemoteStoreSettings.INSTANCE
                 )
             ),
-            writableRegistry()
+            writableRegistry(),
+            () -> 0L
         );
         assertTrue(remoteClusterStateService.getRemoteRoutingTableService() instanceof InternalRemoteRoutingTableService);
     }
@@ -3070,7 +3075,8 @@ private void initializeRoutingTable() {
                     DefaultRemoteStoreSettings.INSTANCE
                 )
             ),
-            writableRegistry()
+            writableRegistry(),
+            () -> 0L
         );
     }
 
@@ -3100,7 +3106,8 @@ private void initializeWithChecksumEnabled(RemoteClusterStateService.RemoteClust
                     DefaultRemoteStoreSettings.INSTANCE
                 )
             ),
-            writableRegistry()
+            writableRegistry(),
+            () -> 0L
         );
     }