Address comments

Signed-off-by: Lakshya Taragi <[email protected]>

CHANGELOG.md

@@ -12,6 +12,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Implement WithFieldName interface in ValuesSourceAggregationBuilder & FieldSortBuilder ([#15916](https://github.com/opensearch-project/OpenSearch/pull/15916))
 - Add successfulSearchShardIndices in searchRequestContext ([#15967](https://github.com/opensearch-project/OpenSearch/pull/15967))
 - Remove identity-related feature flagged code from the RestController ([#15430](https://github.com/opensearch-project/OpenSearch/pull/15430))
+- Add new metric REMOTE_STORE to NodeStats API response ([#15611](https://github.com/opensearch-project/OpenSearch/pull/15611))
 
 ### Dependencies
 - Bump `com.azure:azure-identity` from 1.13.0 to 1.13.2 ([#15578](https://github.com/opensearch-project/OpenSearch/pull/15578))

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

@@ -21,9 +21,8 @@
 
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
 
-import static org.opensearch.action.admin.cluster.node.stats.NodesStatsRequest.Metric.REMOTE_STORE_NODE_STATS;
+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 {
@@ -195,26 +194,31 @@ public void testLastSuccessfulFetchOfPinnedTimestampsPresentInNodeStats() throws
             .put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_SCHEDULER_INTERVAL.getKey(), "1m")
             .build();
         internalCluster().startClusterManagerOnlyNode(pinnedTimestampEnabledSettings);
-        internalCluster().startDataOnlyNodes(2, pinnedTimestampEnabledSettings);
-        ensureStableCluster(3);
+        String remoteNodeName = internalCluster().startDataOnlyNodes(1, pinnedTimestampEnabledSettings).get(0);
+        ensureStableCluster(2);
+        RemoteStorePinnedTimestampService remoteStorePinnedTimestampService = internalCluster().getInstance(
+            RemoteStorePinnedTimestampService.class,
+            remoteNodeName
+        );
 
-        logger.info("Sleeping for 70 seconds to wait for fetching of pinned timestamps");
-        Thread.sleep(70000);
+        remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueSeconds(1));
 
-        long lastSuccessfulFetchOfPinnedTimestamps = RemoteStorePinnedTimestampService.getPinnedTimestamps().v1();
-        assertTrue(lastSuccessfulFetchOfPinnedTimestamps > 0L);
         assertBusy(() -> {
+            long lastSuccessfulFetchOfPinnedTimestamps = RemoteStorePinnedTimestampService.getPinnedTimestamps().v1();
+            assertTrue(lastSuccessfulFetchOfPinnedTimestamps > 0L);
             NodesStatsResponse nodesStatsResponse = internalCluster().client()
                 .admin()
                 .cluster()
                 .prepareNodesStats()
-                .addMetric(REMOTE_STORE_NODE_STATS.metricName())
+                .addMetric(REMOTE_STORE.metricName())
                 .execute()
                 .actionGet();
             for (NodeStats nodeStats : nodesStatsResponse.getNodes()) {
                 long lastRecordedFetch = nodeStats.getRemoteStoreNodeStats().getLastSuccessfulFetchOfPinnedTimestamps();
                 assertTrue(lastRecordedFetch >= lastSuccessfulFetchOfPinnedTimestamps);
             }
-        }, 1, TimeUnit.MINUTES);
+        });
+
+        remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueMinutes(3));
     }
 }

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

@@ -221,7 +221,7 @@ public enum Metric {
         REPOSITORIES("repositories"),
         ADMISSION_CONTROL("admission_control"),
         CACHE_STATS("caches"),
-        REMOTE_STORE_NODE_STATS("remote_store_node_stats");
+        REMOTE_STORE("remote_store");
 
         private String metricName;
 

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

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

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

@@ -1329,8 +1329,7 @@ protected Node(
                 segmentReplicationStatsTracker,
                 repositoryService,
                 admissionControlService,
-                cacheService,
-                remoteStoreNodeService
+                cacheService
             );
 
             final SearchService searchService = newSearchService(

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

@@ -54,7 +54,8 @@
 import org.opensearch.indices.IndicesService;
 import org.opensearch.ingest.IngestService;
 import org.opensearch.monitor.MonitorService;
-import org.opensearch.node.remotestore.RemoteStoreNodeService;
+import org.opensearch.node.remotestore.RemoteStoreNodeStats;
+import org.opensearch.node.remotestore.RemoteStorePinnedTimestampService;
 import org.opensearch.plugins.PluginsService;
 import org.opensearch.ratelimitting.admissioncontrol.AdmissionControlService;
 import org.opensearch.repositories.RepositoriesService;
@@ -102,7 +103,6 @@ public class NodeService implements Closeable {
     private final AdmissionControlService admissionControlService;
     private final SegmentReplicationStatsTracker segmentReplicationStatsTracker;
     private final CacheService cacheService;
-    private final RemoteStoreNodeService remoteStoreNodeService;
 
     NodeService(
         Settings settings,
@@ -130,8 +130,7 @@ public class NodeService implements Closeable {
         SegmentReplicationStatsTracker segmentReplicationStatsTracker,
         RepositoriesService repositoriesService,
         AdmissionControlService admissionControlService,
-        CacheService cacheService,
-        RemoteStoreNodeService remoteStoreNodeService
+        CacheService cacheService
     ) {
         this.settings = settings;
         this.threadPool = threadPool;
@@ -161,7 +160,6 @@ public class NodeService implements Closeable {
         clusterService.addStateApplier(searchPipelineService);
         this.segmentReplicationStatsTracker = segmentReplicationStatsTracker;
         this.cacheService = cacheService;
-        this.remoteStoreNodeService = remoteStoreNodeService;
     }
 
     public NodeInfo info(
@@ -246,7 +244,7 @@ public NodeStats stats(
         boolean repositoriesStats,
         boolean admissionControl,
         boolean cacheService,
-        boolean remoteStoreNodeService
+        boolean remoteStore
     ) {
         // 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)
@@ -280,7 +278,7 @@ public NodeStats stats(
             repositoriesStats ? this.repositoriesService.getRepositoriesStats() : null,
             admissionControl ? this.admissionControlService.stats() : null,
             cacheService ? this.cacheService.stats(indices) : null,
-            remoteStoreNodeService ? this.remoteStoreNodeService.getRemoteStoreNodeStats() : null
+            remoteStore ? new RemoteStoreNodeStats(RemoteStorePinnedTimestampService.getPinnedTimestamps().v1()) : null
         );
     }
 

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

@@ -18,7 +18,6 @@
 import org.opensearch.common.settings.ClusterSettings;
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.util.FeatureFlags;
-import org.opensearch.indices.RemoteStoreSettings;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.repositories.Repository;
 import org.opensearch.repositories.RepositoryException;
@@ -250,12 +249,4 @@ public static boolean isMigratingToRemoteStore(Metadata metadata) {
 
         return (isMixedMode && isRemoteStoreMigrationDirection);
     }
-
-    public RemoteStoreNodeStats getRemoteStoreNodeStats() {
-        long lastSuccessfulFetchOfPinnedTimestamps = 0;
-        if (RemoteStoreSettings.isPinnedTimestampsEnabled()) {
-            lastSuccessfulFetchOfPinnedTimestamps = RemoteStorePinnedTimestampService.getPinnedTimestamps().v1();
-        }
-        return new RemoteStoreNodeStats(lastSuccessfulFetchOfPinnedTimestamps);
-    }
 }

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

@@ -8,23 +8,22 @@
 
 package org.opensearch.node.remotestore;
 
-import org.opensearch.Version;
 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 org.opensearch.indices.RemoteStoreSettings;
 
 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_node_stats";
+    public static final String STATS_NAME = "remote_store";
     public static final String LAST_SUCCESSFUL_FETCH_OF_PINNED_TIMESTAMPS = "last_successful_fetch_of_pinned_timestamps";
 
     /**
@@ -41,18 +40,12 @@ public long getLastSuccessfulFetchOfPinnedTimestamps() {
     }
 
     public RemoteStoreNodeStats(StreamInput in) throws IOException {
-        // TODO: change version to V_2_18_0
-        if (in.getVersion().onOrAfter(Version.CURRENT) && RemoteStoreSettings.isPinnedTimestampsEnabled()) {
-            this.lastSuccessfulFetchOfPinnedTimestamps = in.readOptionalLong();
-        }
+        this.lastSuccessfulFetchOfPinnedTimestamps = in.readLong();
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
-        // TODO: change version to V_2_18_0
-        if (out.getVersion().onOrAfter(Version.CURRENT) && RemoteStoreSettings.isPinnedTimestampsEnabled()) {
-            out.writeOptionalLong(this.lastSuccessfulFetchOfPinnedTimestamps);
-        }
+        out.writeLong(this.lastSuccessfulFetchOfPinnedTimestamps);
     }
 
     @Override
@@ -66,4 +59,21 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
     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);
+                }
             }
         }
     }
@@ -996,6 +1005,12 @@ public void apply(String action, AdmissionControlActionType admissionControlActi
             nodeCacheStats = new NodeCacheStats(cacheStatsMap, flags);
         }
 
+        RemoteStoreNodeStats remoteStoreNodeStats = null;
+        if (frequently()) {
+            long lastSuccessfulFetchOfPinnedTimestamps = randomNonNegativeLong();
+            remoteStoreNodeStats = new RemoteStoreNodeStats(lastSuccessfulFetchOfPinnedTimestamps);
+        }
+
         // 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(
@@ -1028,7 +1043,7 @@ public void apply(String action, AdmissionControlActionType admissionControlActi
             null,
             admissionControlStats,
             nodeCacheStats,
-            null
+            remoteStoreNodeStats
         );
     }