Merge branch 'main' into pluginize-zstd

Signed-off-by: Andrew Ross <[email protected]>
opensearch-project · Sep 6, 2023 · 5e92890 · 5e92890
2 parents 443f37a + 1b32913
commit 5e92890
Show file tree

Hide file tree

Showing 69 changed files with 2,719 additions and 822 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -103,6 +103,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add average concurrency metric for concurrent segment search ([#9670](https://github.com/opensearch-project/OpenSearch/issues/9670))
 - [Remote state] Integrate remote cluster state in publish/commit flow ([#9665](https://github.com/opensearch-project/OpenSearch/pull/9665))
 - [Segment Replication] Adding segment replication statistics rolled up at index, node and cluster level ([#9709](https://github.com/opensearch-project/OpenSearch/pull/9709))
+- [Remote Store] Changes to introduce repository registration during bootstrap via node attributes. ([#9105](https://github.com/opensearch-project/OpenSearch/pull/9105))
 
 ### Dependencies
 - Bump `org.apache.logging.log4j:log4j-core` from 2.17.1 to 2.20.0 ([#8307](https://github.com/opensearch-project/OpenSearch/pull/8307))
@@ -183,6 +184,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add support for query profiler with concurrent aggregation ([#9248](https://github.com/opensearch-project/OpenSearch/pull/9248))
 - Cleanup Unreferenced file on segment merge failure ([#9503](https://github.com/opensearch-project/OpenSearch/pull/9503))
 - Move ZStd to a plugin ([#9658](https://github.com/opensearch-project/OpenSearch/pull/9658))
+- [Remote Store] Add support for Remote Translog Store upload stats in `_nodes/stats/` API ([#8908](https://github.com/opensearch-project/OpenSearch/pull/8908))
 
 ### Deprecated
 

diff --git a/distribution/src/config/opensearch.yml b/distribution/src/config/opensearch.yml
@@ -92,10 +92,10 @@ ${path.logs}
 # cluster.remote_store.enabled: true
 #
 # Repository to use for segment upload while enforcing remote store for an index
-# cluster.remote_store.segment.repository: my-repo-1
+# node.attr.remote_store.segment.repository: my-repo-1
 #
 # Repository to use for translog upload while enforcing remote store for an index
-# cluster.remote_store.translog.repository: my-repo-1
+# node.attr.remote_store.translog.repository: my-repo-1
 #
 # ---------------------------------- Experimental Features -----------------------------------
 #

diff --git a/...rc/internalClusterTest/java/org/opensearch/indices/recovery/IndexPrimaryRelocationIT.java b/...rc/internalClusterTest/java/org/opensearch/indices/recovery/IndexPrimaryRelocationIT.java
@@ -56,15 +56,12 @@ public class IndexPrimaryRelocationIT extends OpenSearchIntegTestCase {
 
     private static final int RELOCATION_COUNT = 15;
 
-    public void setup() {}
-
     public Settings indexSettings() {
         return Settings.builder().put("index.number_of_shards", 1).put("index.number_of_replicas", 0).build();
     }
 
     public void testPrimaryRelocationWhileIndexing() throws Exception {
         internalCluster().ensureAtLeastNumDataNodes(randomIntBetween(2, 3));
-        setup();
         client().admin().indices().prepareCreate("test").setSettings(indexSettings()).setMapping("field", "type=text").get();
         ensureGreen("test");
         AtomicInteger numAutoGenDocs = new AtomicInteger();

diff --git a/server/src/internalClusterTest/java/org/opensearch/indices/stats/IndexStatsIT.java b/server/src/internalClusterTest/java/org/opensearch/indices/stats/IndexStatsIT.java
@@ -71,6 +71,7 @@
 import org.opensearch.index.query.QueryBuilders;
 import org.opensearch.index.remote.RemoteSegmentStats;
 import org.opensearch.index.shard.IndexShard;
+import org.opensearch.index.translog.RemoteTranslogStats;
 import org.opensearch.index.translog.Translog;
 import org.opensearch.indices.IndicesQueryCache;
 import org.opensearch.indices.IndicesRequestCache;
@@ -1436,30 +1437,35 @@ public void testZeroRemoteStoreStatsOnNonRemoteStoreIndex() {
                 .get()
                 .status()
         );
-        ShardStats shard = client().admin().indices().prepareStats(indexName).setSegments(true).get().getShards()[0];
+        ShardStats shard = client().admin().indices().prepareStats(indexName).setSegments(true).setTranslog(true).get().getShards()[0];
         RemoteSegmentStats remoteSegmentStatsFromIndexStats = shard.getStats().getSegments().getRemoteSegmentStats();
         assertZeroRemoteSegmentStats(remoteSegmentStatsFromIndexStats);
+        RemoteTranslogStats remoteTranslogStatsFromIndexStats = shard.getStats().getTranslog().getRemoteTranslogStats();
+        assertZeroRemoteTranslogStats(remoteTranslogStatsFromIndexStats);
+
         NodesStatsResponse nodesStatsResponse = client().admin().cluster().prepareNodesStats(primaryNodeName(indexName)).get();
         RemoteSegmentStats remoteSegmentStatsFromNodesStats = nodesStatsResponse.getNodes()
             .get(0)
             .getIndices()
             .getSegments()
             .getRemoteSegmentStats();
         assertZeroRemoteSegmentStats(remoteSegmentStatsFromNodesStats);
+        RemoteTranslogStats remoteTranslogStatsFromNodesStats = nodesStatsResponse.getNodes()
+            .get(0)
+            .getIndices()
+            .getTranslog()
+            .getRemoteTranslogStats();
+        assertZeroRemoteTranslogStats(remoteTranslogStatsFromNodesStats);
     }
 
     private void assertZeroRemoteSegmentStats(RemoteSegmentStats remoteSegmentStats) {
-        assertEquals(0, remoteSegmentStats.getUploadBytesStarted());
-        assertEquals(0, remoteSegmentStats.getUploadBytesSucceeded());
-        assertEquals(0, remoteSegmentStats.getUploadBytesFailed());
-        assertEquals(0, remoteSegmentStats.getDownloadBytesStarted());
-        assertEquals(0, remoteSegmentStats.getDownloadBytesSucceeded());
-        assertEquals(0, remoteSegmentStats.getDownloadBytesFailed());
-        assertEquals(0, remoteSegmentStats.getTotalRefreshBytesLag());
-        assertEquals(0, remoteSegmentStats.getMaxRefreshBytesLag());
-        assertEquals(0, remoteSegmentStats.getMaxRefreshTimeLag());
-        assertEquals(0, remoteSegmentStats.getTotalUploadTime());
-        assertEquals(0, remoteSegmentStats.getTotalDownloadTime());
+        // Compare with fresh object because all values default to 0 in default fresh object
+        assertEquals(new RemoteSegmentStats(), remoteSegmentStats);
+    }
+
+    private void assertZeroRemoteTranslogStats(RemoteTranslogStats remoteTranslogStats) {
+        // Compare with fresh object because all values default to 0 in default fresh object
+        assertEquals(new RemoteTranslogStats(), remoteTranslogStats);
     }
 
     /**

diff --git a/...rTest/java/org/opensearch/remotestore/AbstractRemoteStoreMockRepositoryIntegTestCase.java b/...rTest/java/org/opensearch/remotestore/AbstractRemoteStoreMockRepositoryIntegTestCase.java
@@ -26,10 +26,15 @@
 import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Locale;
 import java.util.Set;
 import java.util.stream.Collectors;
 
-import static org.opensearch.remotestore.RemoteStoreBaseIntegTestCase.remoteStoreClusterSettings;
+import static org.opensearch.indices.IndicesService.CLUSTER_REPLICATION_TYPE_SETTING;
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_REPOSITORY_SETTINGS_ATTRIBUTE_KEY_PREFIX;
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_REPOSITORY_TYPE_ATTRIBUTE_KEY_FORMAT;
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_SEGMENT_REPOSITORY_NAME_ATTRIBUTE_KEY;
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_TRANSLOG_REPOSITORY_NAME_ATTRIBUTE_KEY;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 
 public abstract class AbstractRemoteStoreMockRepositoryIntegTestCase extends AbstractSnapshotIntegTestCase {
@@ -47,7 +52,6 @@ protected Settings featureFlagSettings() {
     public void setup() {
         FeatureFlagSetter.set(FeatureFlags.REMOTE_STORE);
         FeatureFlagSetter.set(FeatureFlags.SEGMENT_REPLICATION_EXPERIMENTAL);
-        internalCluster().startClusterManagerOnlyNode(remoteStoreClusterSettings(REPOSITORY_NAME, TRANSLOG_REPOSITORY_NAME));
     }
 
     @Override
@@ -66,6 +70,43 @@ protected Settings remoteStoreIndexSettings(int numberOfReplicas) {
             .build();
     }
 
+    public Settings buildRemoteStoreNodeAttributes(Path repoLocation, double ioFailureRate, String skipExceptionBlobList, long maxFailure) {
+        String segmentRepoTypeAttributeKey = String.format(
+            Locale.getDefault(),
+            "node.attr." + REMOTE_STORE_REPOSITORY_TYPE_ATTRIBUTE_KEY_FORMAT,
+            REPOSITORY_NAME
+        );
+        String translogRepoTypeAttributeKey = String.format(
+            Locale.getDefault(),
+            "node.attr." + REMOTE_STORE_REPOSITORY_TYPE_ATTRIBUTE_KEY_FORMAT,
+            TRANSLOG_REPOSITORY_NAME
+        );
+        String segmentRepoSettingsAttributeKeyPrefix = String.format(
+            Locale.getDefault(),
+            "node.attr." + REMOTE_STORE_REPOSITORY_SETTINGS_ATTRIBUTE_KEY_PREFIX,
+            REPOSITORY_NAME
+        );
+        String translogRepoSettingsAttributeKeyPrefix = String.format(
+            Locale.getDefault(),
+            "node.attr." + REMOTE_STORE_REPOSITORY_SETTINGS_ATTRIBUTE_KEY_PREFIX,
+            TRANSLOG_REPOSITORY_NAME
+        );
+
+        return Settings.builder()
+            .put("node.attr." + REMOTE_STORE_SEGMENT_REPOSITORY_NAME_ATTRIBUTE_KEY, REPOSITORY_NAME)
+            .put(segmentRepoTypeAttributeKey, "mock")
+            .put(segmentRepoSettingsAttributeKeyPrefix + "location", repoLocation)
+            .put(segmentRepoSettingsAttributeKeyPrefix + "random_control_io_exception_rate", ioFailureRate)
+            .put(segmentRepoSettingsAttributeKeyPrefix + "skip_exception_on_verification_file", true)
+            .put(segmentRepoSettingsAttributeKeyPrefix + "skip_exception_on_list_blobs", true)
+            .put(segmentRepoSettingsAttributeKeyPrefix + "skip_exception_on_blobs", skipExceptionBlobList)
+            .put(segmentRepoSettingsAttributeKeyPrefix + "max_failure_number", maxFailure)
+            .put("node.attr." + REMOTE_STORE_TRANSLOG_REPOSITORY_NAME_ATTRIBUTE_KEY, TRANSLOG_REPOSITORY_NAME)
+            .put(translogRepoTypeAttributeKey, "mock")
+            .put(translogRepoSettingsAttributeKeyPrefix + "location", repoLocation)
+            .build();
+    }
+
     protected void deleteRepo() {
         logger.info("--> Deleting the repository={}", REPOSITORY_NAME);
         assertAcked(clusterAdmin().prepareDeleteRepository(REPOSITORY_NAME));
@@ -74,26 +115,18 @@ protected void deleteRepo() {
     }
 
     protected String setup(Path repoLocation, double ioFailureRate, String skipExceptionBlobList, long maxFailure) {
-        logger.info("--> Creating repository={} at the path={}", REPOSITORY_NAME, repoLocation);
         // The random_control_io_exception_rate setting ensures that 10-25% of all operations to remote store results in
         /// IOException. skip_exception_on_verification_file & skip_exception_on_list_blobs settings ensures that the
         // repository creation can happen without failure.
-        createRepository(
-            REPOSITORY_NAME,
-            "mock",
-            Settings.builder()
-                .put("location", repoLocation)
-                .put("random_control_io_exception_rate", ioFailureRate)
-                .put("skip_exception_on_verification_file", true)
-                .put("skip_exception_on_list_blobs", true)
-                // Skipping is required for metadata as it is part of recovery
-                .put("skip_exception_on_blobs", skipExceptionBlobList)
-                .put("max_failure_number", maxFailure)
-        );
-        logger.info("--> Creating repository={} at the path={}", TRANSLOG_REPOSITORY_NAME, repoLocation);
-        createRepository(TRANSLOG_REPOSITORY_NAME, "mock", Settings.builder().put("location", repoLocation));
+        Settings.Builder settings = Settings.builder()
+            .put(buildRemoteStoreNodeAttributes(repoLocation, ioFailureRate, skipExceptionBlobList, maxFailure));
+
+        if (randomBoolean()) {
+            settings.put(CLUSTER_REPLICATION_TYPE_SETTING.getKey(), ReplicationType.SEGMENT);
+        }
 
-        String dataNodeName = internalCluster().startDataOnlyNodes(1).get(0);
+        internalCluster().startClusterManagerOnlyNode(settings.build());
+        String dataNodeName = internalCluster().startDataOnlyNode(settings.build());
         createIndex(INDEX_NAME);
         logger.info("--> Created index={}", INDEX_NAME);
         ensureYellowAndNoInitializingShards(INDEX_NAME);

diff --git a/...lClusterTest/java/org/opensearch/remotestore/CreateRemoteIndexClusterDefaultDocRepIT.java b/...lClusterTest/java/org/opensearch/remotestore/CreateRemoteIndexClusterDefaultDocRepIT.java
@@ -75,8 +75,8 @@ public void testDefaultRemoteStoreNoUserOverrideExceptReplicationTypeSegment() t
         verifyRemoteStoreIndexSettings(
             indexSettings,
             "true",
-            "my-segment-repo-1",
-            "my-translog-repo-1",
+            REPOSITORY_NAME,
+            REPOSITORY_2_NAME,
             ReplicationType.SEGMENT.toString(),
             IndexSettings.DEFAULT_REMOTE_TRANSLOG_BUFFER_INTERVAL
         );

diff --git a/server/src/internalClusterTest/java/org/opensearch/remotestore/CreateRemoteIndexIT.java b/server/src/internalClusterTest/java/org/opensearch/remotestore/CreateRemoteIndexIT.java
@@ -13,12 +13,9 @@
 import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
-import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.index.IndexSettings;
 import org.opensearch.indices.replication.common.ReplicationType;
-import org.opensearch.test.FeatureFlagSetter;
 import org.opensearch.test.OpenSearchIntegTestCase;
-import org.junit.After;
 import org.junit.Before;
 
 import java.util.Locale;
@@ -28,53 +25,15 @@
 import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_REMOTE_TRANSLOG_STORE_REPOSITORY;
 import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_REPLICATION_TYPE;
 import static org.opensearch.index.IndexSettings.INDEX_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING;
-import static org.opensearch.remotestore.RemoteStoreBaseIntegTestCase.remoteStoreClusterSettings;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 import static org.hamcrest.Matchers.containsString;
 
 @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST)
-public class CreateRemoteIndexIT extends OpenSearchIntegTestCase {
-
-    @After
-    public void teardown() {
-        assertAcked(clusterAdmin().prepareDeleteRepository("my-segment-repo-1"));
-        assertAcked(clusterAdmin().prepareDeleteRepository("my-translog-repo-1"));
-        assertAcked(clusterAdmin().prepareDeleteRepository("my-custom-repo"));
-    }
-
-    @Override
-    protected Settings nodeSettings(int nodeOriginal) {
-        Settings settings = super.nodeSettings(nodeOriginal);
-        Settings.Builder builder = Settings.builder()
-            .put(remoteStoreClusterSettings("my-segment-repo-1", "my-translog-repo-1"))
-            .put(settings);
-        return builder.build();
-    }
-
-    @Override
-    protected Settings featureFlagSettings() {
-        return Settings.builder().put(super.featureFlagSettings()).put(FeatureFlags.REMOTE_STORE, "true").build();
-    }
+public class CreateRemoteIndexIT extends RemoteStoreBaseIntegTestCase {
 
     @Before
-    public void setup() {
-        FeatureFlagSetter.set(FeatureFlags.REMOTE_STORE);
-        internalCluster().startClusterManagerOnlyNode();
-        assertAcked(
-            clusterAdmin().preparePutRepository("my-segment-repo-1")
-                .setType("fs")
-                .setSettings(Settings.builder().put("location", randomRepoPath().toAbsolutePath()))
-        );
-        assertAcked(
-            clusterAdmin().preparePutRepository("my-translog-repo-1")
-                .setType("fs")
-                .setSettings(Settings.builder().put("location", randomRepoPath().toAbsolutePath()))
-        );
-        assertAcked(
-            clusterAdmin().preparePutRepository("my-custom-repo")
-                .setType("fs")
-                .setSettings(Settings.builder().put("location", randomRepoPath().toAbsolutePath()))
-        );
+    public void setup() throws Exception {
+        internalCluster().startNodes(2);
     }
 
     public void testDefaultRemoteStoreNoUserOverride() throws Exception {
@@ -91,8 +50,8 @@ public void testDefaultRemoteStoreNoUserOverride() throws Exception {
         verifyRemoteStoreIndexSettings(
             indexSettings,
             "true",
-            "my-segment-repo-1",
-            "my-translog-repo-1",
+            REPOSITORY_NAME,
+            REPOSITORY_2_NAME,
             ReplicationType.SEGMENT.toString(),
             IndexSettings.DEFAULT_REMOTE_TRANSLOG_BUFFER_INTERVAL
         );

diff --git a/server/src/internalClusterTest/java/org/opensearch/remotestore/PrimaryTermValidationIT.java b/server/src/internalClusterTest/java/org/opensearch/remotestore/PrimaryTermValidationIT.java
@@ -27,30 +27,38 @@
 import org.opensearch.test.OpenSearchIntegTestCase;
 import org.opensearch.test.disruption.NetworkDisruption;
 import org.opensearch.test.transport.MockTransportService;
+import org.junit.Before;
 
+import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertHitCount;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertNoFailures;
 import static org.hamcrest.Matchers.equalTo;
 
 @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0)
-
 public class PrimaryTermValidationIT extends RemoteStoreBaseIntegTestCase {
 
     private static final String INDEX_NAME = "remote-store-test-idx-1";
+    protected Path absolutePath;
+    protected Path absolutePath2;
 
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {
         return Arrays.asList(MockTransportService.TestPlugin.class);
     }
 
+    @Before
+    public void setup() {
+        absolutePath = randomRepoPath().toAbsolutePath();
+        absolutePath2 = randomRepoPath().toAbsolutePath();
+    }
+
     public void testPrimaryTermValidation() throws Exception {
         // Follower checker interval is lower compared to leader checker so that the cluster manager can remove the node
         // with network partition faster. The follower check retry count is also kept 1.
@@ -61,20 +69,12 @@ public void testPrimaryTermValidation() throws Exception {
             .put(FollowersChecker.FOLLOWER_CHECK_TIMEOUT_SETTING.getKey(), "1s")
             .put(FollowersChecker.FOLLOWER_CHECK_INTERVAL_SETTING.getKey(), "1s")
             .put(FollowersChecker.FOLLOWER_CHECK_RETRY_COUNT_SETTING.getKey(), 1)
-            .put(remoteStoreClusterSettings(REPOSITORY_NAME, REPOSITORY_2_NAME, true))
+            .put(remoteStoreClusterSettings(REPOSITORY_NAME, absolutePath, REPOSITORY_2_NAME, absolutePath2))
             .build();
         internalCluster().startClusterManagerOnlyNode(clusterSettings);
-
-        // Create repository
-        absolutePath = randomRepoPath().toAbsolutePath();
-        assertAcked(
-            clusterAdmin().preparePutRepository(REPOSITORY_NAME).setType("fs").setSettings(Settings.builder().put("location", absolutePath))
-        );
-        absolutePath2 = randomRepoPath().toAbsolutePath();
-        putRepository(absolutePath2, REPOSITORY_2_NAME);
-
-        // Start data nodes and create index
         internalCluster().startDataOnlyNodes(2, clusterSettings);
+
+        // Create index
         createIndex(INDEX_NAME, remoteStoreIndexSettings(1));
         ensureYellowAndNoInitializingShards(INDEX_NAME);
         ensureGreen(INDEX_NAME);
@@ -156,6 +156,7 @@ public void testPrimaryTermValidation() throws Exception {
         // received the following exception.
         ShardNotFoundException exception = assertThrows(ShardNotFoundException.class, () -> indexSameDoc(primaryNode, INDEX_NAME));
         assertTrue(exception.getMessage().contains("no such shard"));
+        internalCluster().clearDisruptionScheme();
         ensureStableCluster(3);
         ensureGreen(INDEX_NAME);
     }