[SnapshotV2] Add timestamp of last successful fetch of pinned timesta…

…mps in node stats (#15611)

---------

Signed-off-by: Lakshya Taragi <lakshya.taragi@gmail.com>
(cherry picked from commit be9f942)
Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

CHANGELOG.md

@@ -14,6 +14,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add support for msearch API to pass search pipeline name - ([#15923](https://github.com/opensearch-project/OpenSearch/pull/15923))
 - Add success and failure metrics for async shard fetch ([#15976](https://github.com/opensearch-project/OpenSearch/pull/15976))
 - [S3 Repository] Change default retry mechanism of s3 clients to Standard Mode ([#15978](https://github.com/opensearch-project/OpenSearch/pull/15978))
+- Add new metric REMOTE_STORE to NodeStats API response ([#15611](https://github.com/opensearch-project/OpenSearch/pull/15611))
 
 ### Dependencies
 - Bump `org.apache.logging.log4j:log4j-core` from 2.23.1 to 2.24.0 ([#15858](https://github.com/opensearch-project/OpenSearch/pull/15858))

server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStorePinnedTimestampsIT.java

@@ -9,6 +9,8 @@
 package org.opensearch.remotestore;
 
 import org.opensearch.action.LatchedActionListener;
+import org.opensearch.action.admin.cluster.node.stats.NodeStats;
+import org.opensearch.action.admin.cluster.node.stats.NodesStatsResponse;
 import org.opensearch.common.collect.Tuple;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
@@ -20,6 +22,8 @@
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 
+import static org.opensearch.action.admin.cluster.node.stats.NodesStatsRequest.Metric.REMOTE_STORE;
+
 @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0)
 public class RemoteStorePinnedTimestampsIT extends RemoteStoreBaseIntegTestCase {
     static final String INDEX_NAME = "remote-store-test-idx-1";
@@ -180,4 +184,41 @@ public void onFailure(Exception e) {
         assertBusy(() -> assertEquals(Set.of(timestamp2, timestamp3), RemoteStorePinnedTimestampService.getPinnedTimestamps().v2()));
         remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueMinutes(3));
     }
+
+    public void testLastSuccessfulFetchOfPinnedTimestampsPresentInNodeStats() throws Exception {
+        logger.info("Starting up cluster manager");
+        logger.info("cluster.remote_store.pinned_timestamps.enabled set to true");
+        logger.info("cluster.remote_store.pinned_timestamps.scheduler_interval set to minimum value of 1minute");
+        Settings pinnedTimestampEnabledSettings = Settings.builder()
+            .put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED.getKey(), true)
+            .put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_SCHEDULER_INTERVAL.getKey(), "1m")
+            .build();
+        internalCluster().startClusterManagerOnlyNode(pinnedTimestampEnabledSettings);
+        String remoteNodeName = internalCluster().startDataOnlyNodes(1, pinnedTimestampEnabledSettings).get(0);
+        ensureStableCluster(2);
+        RemoteStorePinnedTimestampService remoteStorePinnedTimestampService = internalCluster().getInstance(
+            RemoteStorePinnedTimestampService.class,
+            remoteNodeName
+        );
+
+        remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueSeconds(1));
+
+        assertBusy(() -> {
+            long lastSuccessfulFetchOfPinnedTimestamps = RemoteStorePinnedTimestampService.getPinnedTimestamps().v1();
+            assertTrue(lastSuccessfulFetchOfPinnedTimestamps > 0L);
+            NodesStatsResponse nodesStatsResponse = internalCluster().client()
+                .admin()
+                .cluster()
+                .prepareNodesStats()
+                .addMetric(REMOTE_STORE.metricName())
+                .execute()
+                .actionGet();
+            for (NodeStats nodeStats : nodesStatsResponse.getNodes()) {
+                long lastRecordedFetch = nodeStats.getRemoteStoreNodeStats().getLastSuccessfulFetchOfPinnedTimestamps();
+                assertTrue(lastRecordedFetch >= lastSuccessfulFetchOfPinnedTimestamps);
+            }
+        });
+
+        remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueMinutes(3));
+    }
 }

server/src/main/java/org/opensearch/action/admin/cluster/node/stats/NodeStats.java

@@ -60,6 +60,7 @@
 import org.opensearch.monitor.process.ProcessStats;
 import org.opensearch.node.AdaptiveSelectionStats;
 import org.opensearch.node.NodesResourceUsageStats;
+import org.opensearch.node.remotestore.RemoteStoreNodeStats;
 import org.opensearch.ratelimitting.admissioncontrol.stats.AdmissionControlStats;
 import org.opensearch.repositories.RepositoriesStats;
 import org.opensearch.script.ScriptCacheStats;
@@ -163,6 +164,9 @@ public class NodeStats extends BaseNodeResponse implements ToXContentFragment {
     @Nullable
     private NodeCacheStats nodeCacheStats;
 
+    @Nullable
+    private RemoteStoreNodeStats remoteStoreNodeStats;
+
     public NodeStats(StreamInput in) throws IOException {
         super(in);
         timestamp = in.readVLong();
@@ -258,6 +262,12 @@ public NodeStats(StreamInput in) throws IOException {
         } else {
             nodeCacheStats = null;
         }
+        // TODO: change version to V_2_18_0
+        if (in.getVersion().onOrAfter(Version.CURRENT)) {
+            remoteStoreNodeStats = in.readOptionalWriteable(RemoteStoreNodeStats::new);
+        } else {
+            remoteStoreNodeStats = null;
+        }
     }
 
     public NodeStats(
@@ -289,7 +299,8 @@ public NodeStats(
         @Nullable SegmentReplicationRejectionStats segmentReplicationRejectionStats,
         @Nullable RepositoriesStats repositoriesStats,
         @Nullable AdmissionControlStats admissionControlStats,
-        @Nullable NodeCacheStats nodeCacheStats
+        @Nullable NodeCacheStats nodeCacheStats,
+        @Nullable RemoteStoreNodeStats remoteStoreNodeStats
     ) {
         super(node);
         this.timestamp = timestamp;
@@ -320,6 +331,7 @@ public NodeStats(
         this.repositoriesStats = repositoriesStats;
         this.admissionControlStats = admissionControlStats;
         this.nodeCacheStats = nodeCacheStats;
+        this.remoteStoreNodeStats = remoteStoreNodeStats;
     }
 
     public long getTimestamp() {
@@ -482,6 +494,11 @@ public NodeCacheStats getNodeCacheStats() {
         return nodeCacheStats;
     }
 
+    @Nullable
+    public RemoteStoreNodeStats getRemoteStoreNodeStats() {
+        return remoteStoreNodeStats;
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
@@ -547,6 +564,10 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(Version.V_2_14_0)) {
             out.writeOptionalWriteable(nodeCacheStats);
         }
+        // TODO: change version to V_2_18_0
+        if (out.getVersion().onOrAfter(Version.CURRENT)) {
+            out.writeOptionalWriteable(remoteStoreNodeStats);
+        }
     }
 
     @Override
@@ -652,6 +673,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         if (getNodeCacheStats() != null) {
             getNodeCacheStats().toXContent(builder, params);
         }
+        if (getRemoteStoreNodeStats() != null) {
+            getRemoteStoreNodeStats().toXContent(builder, params);
+        }
         return builder;
     }
 }

server/src/main/java/org/opensearch/action/admin/cluster/node/stats/NodesStatsRequest.java

@@ -251,7 +251,8 @@ public enum Metric {
         SEGMENT_REPLICATION_BACKPRESSURE("segment_replication_backpressure"),
         REPOSITORIES("repositories"),
         ADMISSION_CONTROL("admission_control"),
-        CACHE_STATS("caches");
+        CACHE_STATS("caches"),
+        REMOTE_STORE("remote_store");
 
         private String metricName;
 

server/src/main/java/org/opensearch/action/admin/cluster/node/stats/TransportNodesStatsAction.java

@@ -129,7 +129,8 @@ protected NodeStats nodeOperation(NodeStatsRequest nodeStatsRequest) {
             NodesStatsRequest.Metric.SEGMENT_REPLICATION_BACKPRESSURE.containedIn(metrics),
             NodesStatsRequest.Metric.REPOSITORIES.containedIn(metrics),
             NodesStatsRequest.Metric.ADMISSION_CONTROL.containedIn(metrics),
-            NodesStatsRequest.Metric.CACHE_STATS.containedIn(metrics)
+            NodesStatsRequest.Metric.CACHE_STATS.containedIn(metrics),
+            NodesStatsRequest.Metric.REMOTE_STORE.containedIn(metrics)
         );
     }
 

server/src/main/java/org/opensearch/action/admin/cluster/stats/TransportClusterStatsAction.java

@@ -174,6 +174,7 @@ protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeReq
             false,
             false,
             false,
+            false,
             false
         );
         List<ShardStats> shardsStats = new ArrayList<>();

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

@@ -54,6 +54,7 @@
 import org.opensearch.indices.IndicesService;
 import org.opensearch.ingest.IngestService;
 import org.opensearch.monitor.MonitorService;
+import org.opensearch.node.remotestore.RemoteStoreNodeStats;
 import org.opensearch.plugins.PluginsService;
 import org.opensearch.ratelimitting.admissioncontrol.AdmissionControlService;
 import org.opensearch.repositories.RepositoriesService;
@@ -241,7 +242,8 @@ public NodeStats stats(
         boolean segmentReplicationTrackerStats,
         boolean repositoriesStats,
         boolean admissionControl,
-        boolean cacheService
+        boolean cacheService,
+        boolean remoteStoreNodeStats
     ) {
         // for indices stats we want to include previous allocated shards stats as well (it will
         // only be applied to the sensible ones to use, like refresh/merge/flush/indexing stats)
@@ -274,7 +276,8 @@ public NodeStats stats(
             segmentReplicationTrackerStats ? this.segmentReplicationStatsTracker.getTotalRejectionStats() : null,
             repositoriesStats ? this.repositoriesService.getRepositoriesStats() : null,
             admissionControl ? this.admissionControlService.stats() : null,
-            cacheService ? this.cacheService.stats(indices) : null
+            cacheService ? this.cacheService.stats(indices) : null,
+            remoteStoreNodeStats ? new RemoteStoreNodeStats() : null
         );
     }
 

server/src/main/java/org/opensearch/node/remotestore/RemoteStoreNodeStats.java

@@ -0,0 +1,79 @@
+/*
+ * 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.node.remotestore;
+
+import org.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.common.io.stream.StreamOutput;
+import org.opensearch.core.common.io.stream.Writeable;
+import org.opensearch.core.xcontent.ToXContentFragment;
+import org.opensearch.core.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * Node level remote store stats
+ * @opensearch.internal
+ */
+public class RemoteStoreNodeStats implements Writeable, ToXContentFragment {
+
+    public static final String STATS_NAME = "remote_store";
+    public static final String LAST_SUCCESSFUL_FETCH_OF_PINNED_TIMESTAMPS = "last_successful_fetch_of_pinned_timestamps";
+
+    /**
+     * Time stamp for the last successful fetch of pinned timestamps by the {@linkplain  RemoteStorePinnedTimestampService}
+     */
+    private final long lastSuccessfulFetchOfPinnedTimestamps;
+
+    public RemoteStoreNodeStats() {
+        this.lastSuccessfulFetchOfPinnedTimestamps = RemoteStorePinnedTimestampService.getPinnedTimestamps().v1();
+    }
+
+    public long getLastSuccessfulFetchOfPinnedTimestamps() {
+        return this.lastSuccessfulFetchOfPinnedTimestamps;
+    }
+
+    public RemoteStoreNodeStats(StreamInput in) throws IOException {
+        this.lastSuccessfulFetchOfPinnedTimestamps = in.readLong();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeLong(this.lastSuccessfulFetchOfPinnedTimestamps);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject(STATS_NAME);
+        builder.field(LAST_SUCCESSFUL_FETCH_OF_PINNED_TIMESTAMPS, this.lastSuccessfulFetchOfPinnedTimestamps);
+        return builder.endObject();
+    }
+
+    @Override
+    public String toString() {
+        return "RemoteStoreNodeStats{ lastSuccessfulFetchOfPinnedTimestamps=" + lastSuccessfulFetchOfPinnedTimestamps + "}";
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null) {
+            return false;
+        }
+        if (o.getClass() != RemoteStoreNodeStats.class) {
+            return false;
+        }
+        RemoteStoreNodeStats other = (RemoteStoreNodeStats) o;
+        return this.lastSuccessfulFetchOfPinnedTimestamps == other.lastSuccessfulFetchOfPinnedTimestamps;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(lastSuccessfulFetchOfPinnedTimestamps);
+    }
+}

server/src/test/java/org/opensearch/action/admin/cluster/node/stats/NodeStatsTests.java

@@ -95,6 +95,7 @@
 import org.opensearch.node.NodeResourceUsageStats;
 import org.opensearch.node.NodesResourceUsageStats;
 import org.opensearch.node.ResponseCollectorService;
+import org.opensearch.node.remotestore.RemoteStoreNodeStats;
 import org.opensearch.ratelimitting.admissioncontrol.controllers.AdmissionController;
 import org.opensearch.ratelimitting.admissioncontrol.controllers.CpuBasedAdmissionController;
 import org.opensearch.ratelimitting.admissioncontrol.enums.AdmissionControlActionType;
@@ -614,6 +615,14 @@ public void testSerialization() throws IOException {
                 } else {
                     assertEquals(nodeCacheStats, deserializedNodeCacheStats);
                 }
+
+                RemoteStoreNodeStats remoteStoreNodeStats = nodeStats.getRemoteStoreNodeStats();
+                RemoteStoreNodeStats deserializedRemoteStoreNodeStats = deserializedNodeStats.getRemoteStoreNodeStats();
+                if (remoteStoreNodeStats == null) {
+                    assertNull(deserializedRemoteStoreNodeStats);
+                } else {
+                    assertEquals(remoteStoreNodeStats, deserializedRemoteStoreNodeStats);
+                }
             }
         }
     }
@@ -995,6 +1004,16 @@ public void apply(String action, AdmissionControlActionType admissionControlActi
             nodeCacheStats = new NodeCacheStats(cacheStatsMap, flags);
         }
 
+        RemoteStoreNodeStats remoteStoreNodeStats = null;
+        if (frequently()) {
+            remoteStoreNodeStats = new RemoteStoreNodeStats() {
+                @Override
+                public long getLastSuccessfulFetchOfPinnedTimestamps() {
+                    return 123456L;
+                }
+            };
+        }
+
         // TODO: Only remote_store based aspects of NodeIndicesStats are being tested here.
         // It is possible to test other metrics in NodeIndicesStats as well since it extends Writeable now
         return new NodeStats(
@@ -1026,7 +1045,8 @@ public void apply(String action, AdmissionControlActionType admissionControlActi
             segmentReplicationRejectionStats,
             null,
             admissionControlStats,
-            nodeCacheStats
+            nodeCacheStats,
+            remoteStoreNodeStats
         );
     }
 

server/src/test/java/org/opensearch/action/admin/cluster/stats/ClusterStatsNodesTests.java

@@ -349,6 +349,7 @@ private ClusterStatsNodeResponse createClusterStatsNodeResponse(
             null,
             null,
             null,
+            null,
             null
         );
         if (defaultBehavior) {

server/src/test/java/org/opensearch/cluster/DiskUsageTests.java

@@ -195,6 +195,7 @@ public void testFillDiskUsage() {
                 null,
                 null,
                 null,
+                null,
                 null
             ),
             new NodeStats(
@@ -226,6 +227,7 @@ public void testFillDiskUsage() {
                 null,
                 null,
                 null,
+                null,
                 null
             ),
             new NodeStats(
@@ -257,6 +259,7 @@ public void testFillDiskUsage() {
                 null,
                 null,
                 null,
+                null,
                 null
             )
         );
@@ -319,6 +322,7 @@ public void testFillDiskUsageSomeInvalidValues() {
                 null,
                 null,
                 null,
+                null,
                 null
             ),
             new NodeStats(
@@ -350,6 +354,7 @@ public void testFillDiskUsageSomeInvalidValues() {
                 null,
                 null,
                 null,
+                null,
                 null
             ),
             new NodeStats(
@@ -381,6 +386,7 @@ public void testFillDiskUsageSomeInvalidValues() {
                 null,
                 null,
                 null,
+                null,
                 null
             )
         );