Add Integ tests

Signed-off-by: Ashish Singh <ssashish@amazon.com>

server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationBaseIT.java

@@ -95,10 +95,6 @@ protected void assertDocCounts(int expectedDocCount, String... nodeNames) {
         }
     }
 
-    protected ClusterState getClusterState() {
-        return client(internalCluster().getClusterManagerName()).admin().cluster().prepareState().get().getState();
-    }
-
     protected DiscoveryNode getNodeContainingPrimaryShard() {
         final ClusterState state = getClusterState();
         final ShardRouting primaryShard = state.routingTable().index(INDEX_NAME).shard(0).primaryShard();

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

@@ -16,26 +16,17 @@
 import org.opensearch.index.IndexModule;
 import org.opensearch.index.IndexSettings;
 import org.opensearch.indices.replication.common.ReplicationType;
-import org.opensearch.plugins.Plugin;
 import org.opensearch.test.OpenSearchIntegTestCase;
-import org.opensearch.test.transport.MockTransportService;
 
 import java.nio.file.Path;
-import java.util.Collection;
 
-import static java.util.Arrays.asList;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 
 public class RemoteStoreBaseIntegTestCase extends OpenSearchIntegTestCase {
     protected static final String REPOSITORY_NAME = "test-remore-store-repo";
     protected static final int SHARD_COUNT = 1;
     protected static final int REPLICA_COUNT = 1;
 
-    @Override
-    protected Collection<Class<? extends Plugin>> nodePlugins() {
-        return asList(MockTransportService.TestPlugin.class);
-    }
-
     @Override
     protected boolean addMockInternalEngine() {
         return false;
@@ -47,6 +38,10 @@ protected Settings featureFlagSettings() {
     }
 
     public Settings indexSettings() {
+        return defaultIndexSettings();
+    }
+
+    private Settings defaultIndexSettings() {
         return Settings.builder()
             .put(super.indexSettings())
             .put(IndexModule.INDEX_QUERY_CACHE_ENABLED_SETTING.getKey(), false)
@@ -59,6 +54,22 @@ public Settings indexSettings() {
             .build();
     }
 
+    protected Settings remoteStoreIndexSettings(int numberOfReplicas) {
+        return Settings.builder()
+            .put(defaultIndexSettings())
+            .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
+            .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, numberOfReplicas)
+            .build();
+    }
+
+    protected Settings remoteTranslogIndexSettings(int numberOfReplicas) {
+        return Settings.builder()
+            .put(remoteStoreIndexSettings(numberOfReplicas))
+            .put(IndexMetadata.SETTING_REMOTE_TRANSLOG_STORE_ENABLED, true)
+            .put(IndexMetadata.SETTING_REMOTE_TRANSLOG_STORE_REPOSITORY, REPOSITORY_NAME)
+            .build();
+    }
+
     @Before
     public void setup() {
         internalCluster().startClusterManagerOnlyNode();

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

@@ -52,22 +52,6 @@ public Settings indexSettings() {
         return remoteStoreIndexSettings(0);
     }
 
-    private Settings remoteStoreIndexSettings(int numberOfReplicas) {
-        return Settings.builder()
-            .put(super.indexSettings())
-            .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
-            .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, numberOfReplicas)
-            .build();
-    }
-
-    private Settings remoteTranslogIndexSettings(int numberOfReplicas) {
-        return Settings.builder()
-            .put(remoteStoreIndexSettings(numberOfReplicas))
-            .put(IndexMetadata.SETTING_REMOTE_TRANSLOG_STORE_ENABLED, true)
-            .put(IndexMetadata.SETTING_REMOTE_TRANSLOG_STORE_REPOSITORY, REPOSITORY_NAME)
-            .build();
-    }
-
     private IndexResponse indexSingleDoc() {
         return client().prepareIndex(INDEX_NAME)
             .setId(UUIDs.randomBase64UUID())

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

@@ -0,0 +1,93 @@
+/*
+ * 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.remotestore;
+
+import org.opensearch.action.admin.cluster.remotestore.stats.RemoteStoreStats;
+import org.opensearch.action.admin.cluster.remotestore.stats.RemoteStoreStatsResponse;
+import org.opensearch.action.index.IndexResponse;
+import org.opensearch.cluster.ClusterState;
+import org.opensearch.cluster.node.DiscoveryNode;
+import org.opensearch.common.UUIDs;
+import org.opensearch.index.remote.RemoteRefreshSegmentTracker;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.stream.Collectors;
+
+public class RemoteStoreStatsIT extends RemoteStoreBaseIntegTestCase {
+
+    private static final String INDEX_NAME = "remote-store-test-idx-1";
+
+    public void testStatsResponseFromAllNodes() {
+
+        // Step 1 - We create cluster, create an index, and then index documents into. We also do multiple refreshes/flushes
+        // during this time frame. This ensures that the segment upload has started.
+        internalCluster().startDataOnlyNodes(3);
+        if (randomBoolean()) {
+            createIndex(INDEX_NAME, remoteTranslogIndexSettings(0));
+        } else {
+            createIndex(INDEX_NAME, remoteStoreIndexSettings(0));
+        }
+        ensureYellowAndNoInitializingShards(INDEX_NAME);
+        ensureGreen(INDEX_NAME);
+
+        // Indexing documents along with refreshes and flushes.
+        for (int i = 0; i < randomIntBetween(5, 10); i++) {
+            if (randomBoolean()) {
+                flush(INDEX_NAME);
+            } else {
+                refresh(INDEX_NAME);
+            }
+            int numberOfOperations = randomIntBetween(20, 50);
+            for (int j = 0; j < numberOfOperations; j++) {
+                indexSingleDoc();
+            }
+        }
+
+        // Step 2 - We find all the nodes that are present in the cluster. We make the remote store stats api call from
+        // each of the node in the cluster and check that the response is coming as expected.
+        ClusterState state = getClusterState();
+        List<String> nodes = state.nodes().getNodes().values().stream().map(DiscoveryNode::getName).collect(Collectors.toList());
+        String shardId = "0";
+        for (String node : nodes) {
+            RemoteStoreStatsResponse response = client(node).admin().cluster().prepareRemoteStoreStats(INDEX_NAME, shardId).get();
+            assertTrue(response.getSuccessfulShards() > 0);
+            assertTrue(response.getShards() != null && response.getShards().length != 0);
+            final String indexShardId = String.format(Locale.ROOT, "[%s][%s]", INDEX_NAME, shardId);
+            List<RemoteStoreStats> matches = Arrays.stream(response.getShards())
+                .filter(stat -> indexShardId.equals(stat.getStats().shardId.toString()))
+                .collect(Collectors.toList());
+            assertEquals(1, matches.size());
+            RemoteRefreshSegmentTracker.Stats stats = matches.get(0).getStats();
+            assertEquals(0, stats.refreshTimeLagMs);
+            assertEquals(stats.localRefreshNumber, stats.remoteRefreshNumber);
+            assertTrue(stats.uploadBytesStarted > 0);
+            assertEquals(0, stats.uploadBytesFailed);
+            assertTrue(stats.uploadBytesSucceeded > 0);
+            assertTrue(stats.totalUploadsStarted > 0);
+            assertEquals(0, stats.totalUploadsFailed);
+            assertTrue(stats.totalUploadsSucceeded > 0);
+            assertEquals(0, stats.rejectionCount);
+            assertEquals(0, stats.consecutiveFailuresCount);
+            assertEquals(0, stats.bytesLag);
+            assertTrue(stats.uploadBytesMovingAverage > 0);
+            assertTrue(stats.uploadBytesPerSecMovingAverage > 0);
+            assertTrue(stats.uploadTimeMovingAverage > 0);
+        }
+    }
+
+    private IndexResponse indexSingleDoc() {
+        return client().prepareIndex(INDEX_NAME)
+            .setId(UUIDs.randomBase64UUID())
+            .setSource(randomAlphaOfLength(5), randomAlphaOfLength(5))
+            .get();
+    }
+
+}

test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java

@@ -2474,4 +2474,8 @@ protected String replicaNodeName(String indexName) {
         return clusterState.getRoutingNodes().node(nodeId).node().getName();
     }
 
+    protected ClusterState getClusterState() {
+        return client(internalCluster().getClusterManagerName()).admin().cluster().prepareState().get().getState();
+    }
+
 }