Make number of segment metadata files in remote segment store configurable

server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

@@ -292,7 +292,7 @@ public void apply(Settings value, Settings current, Settings previous) {
                 RecoverySettings.INDICES_RECOVERY_MAX_CONCURRENT_FILE_CHUNKS_SETTING,
                 RecoverySettings.INDICES_RECOVERY_MAX_CONCURRENT_OPERATIONS_SETTING,
                 RecoverySettings.INDICES_RECOVERY_MAX_CONCURRENT_REMOTE_STORE_STREAMS_SETTING,
-                RecoverySettings.CLUSTER_REMOTE_INDEX_MIN_SEGMENT_METADATA_FILES_SETTING,
+                RecoverySettings.CLUSTER_REMOTE_INDEX_SEGMENT_METADATA_RETENTION_MAX_COUNT_SETTING,
                 ThrottlingAllocationDecider.CLUSTER_ROUTING_ALLOCATION_NODE_INITIAL_PRIMARIES_RECOVERIES_SETTING,
                 ThrottlingAllocationDecider.CLUSTER_ROUTING_ALLOCATION_NODE_INITIAL_REPLICAS_RECOVERIES_SETTING,
                 ThrottlingAllocationDecider.CLUSTER_ROUTING_ALLOCATION_NODE_CONCURRENT_INCOMING_RECOVERIES_SETTING,

server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java

@@ -727,8 +727,10 @@ public Map<String, UploadedSegmentMetadata> getSegmentsUploadedToRemoteStore() {
      * @throws IOException in case of I/O error while reading from / writing to remote segment store
      */
     public void deleteStaleSegments(int lastNMetadataFilesToKeep) throws IOException {
-        if (lastNMetadataFilesToKeep <= 0) {
-            logger.info("Stale segment deletion is disabled if cluster.remote_store.index.min.segment-metadata is set to < 1");
+        if (lastNMetadataFilesToKeep == -1) {
+            logger.info(
+                "Stale segment deletion is disabled if cluster.remote_store.index.segment_metadata.retention.max_count is set to -1"
+            );
             return;
         }
         List<String> sortedMetadataFileList = remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder(

server/src/main/java/org/opensearch/indices/recovery/PeerRecoveryTargetService.java

@@ -247,6 +247,10 @@ private void doRecovery(final long recoveryId, final StartRecoveryRequest preExi
                     indexShard.prepareForIndexRecovery();
                     final boolean hasRemoteSegmentStore = indexShard.indexSettings().isRemoteStoreEnabled();
                     if (hasRemoteSegmentStore) {
+                        // ToDo: This is a temporary mitigation to not fail the peer recovery flow in case there is
+                        // an exception while downloading segments from remote store. For remote backed indexes, we
+                        // plan to revamp this flow so that node-node segment copy will not happen.
+                        // GitHub Issue to track the revamp: https://github.com/opensearch-project/OpenSearch/issues/11331
                         try {
                             indexShard.syncSegmentsFromRemoteSegmentStore(false, recoveryTarget::setLastAccessTime);
                         } catch (Exception e) {

server/src/main/java/org/opensearch/indices/recovery/RecoverySettings.java

@@ -161,9 +161,18 @@ public class RecoverySettings {
      * Controls minimum number of metadata files to keep in remote segment store.
      * {@code value < 1} will disable deletion of stale segment metadata files.
      */
-    public static final Setting<Integer> CLUSTER_REMOTE_INDEX_MIN_SEGMENT_METADATA_FILES_SETTING = Setting.intSetting(
-        "cluster.remote_store.index.min.segment-metadata",
+    public static final Setting<Integer> CLUSTER_REMOTE_INDEX_SEGMENT_METADATA_RETENTION_MAX_COUNT_SETTING = Setting.intSetting(
+        "cluster.remote_store.index.segment_metadata.retention.max_count",
         10,
+        -1,
+        100,
+        v -> {
+            if (v == 0) {
+                throw new IllegalArgumentException(
+                    "Value 0 is not allowed for this setting as it would delete all the data from remote segment store"
+                );
+            }
+        },
         Property.NodeScope,
         Property.Dynamic
     );
@@ -224,9 +233,9 @@ public RecoverySettings(Settings settings, ClusterSettings clusterSettings) {
             this::setInternalActionLongTimeout
         );
         clusterSettings.addSettingsUpdateConsumer(INDICES_RECOVERY_ACTIVITY_TIMEOUT_SETTING, this::setActivityTimeout);
-        minRemoteSegmentMetadataFiles = CLUSTER_REMOTE_INDEX_MIN_SEGMENT_METADATA_FILES_SETTING.get(settings);
+        minRemoteSegmentMetadataFiles = CLUSTER_REMOTE_INDEX_SEGMENT_METADATA_RETENTION_MAX_COUNT_SETTING.get(settings);
         clusterSettings.addSettingsUpdateConsumer(
-            CLUSTER_REMOTE_INDEX_MIN_SEGMENT_METADATA_FILES_SETTING,
+            CLUSTER_REMOTE_INDEX_SEGMENT_METADATA_RETENTION_MAX_COUNT_SETTING,
             this::setMinRemoteSegmentMetadataFiles
         );
     }

server/src/test/java/org/opensearch/indices/recovery/RecoverySettingsDynamicUpdateTests.java

@@ -96,4 +96,60 @@ public void testInternalLongActionTimeout() {
         );
         assertEquals(new TimeValue(duration, timeUnit), recoverySettings.internalActionLongTimeout());
     }
+
+    public void testSegmentMetadataRetention() {
+        // Default value
+        assertEquals(10, recoverySettings.getMinRemoteSegmentMetadataFiles());
+
+        // Setting value < default (10)
+        clusterSettings.applySettings(
+            Settings.builder().put(RecoverySettings.CLUSTER_REMOTE_INDEX_SEGMENT_METADATA_RETENTION_MAX_COUNT_SETTING.getKey(), 5).build()
+        );
+        assertEquals(5, recoverySettings.getMinRemoteSegmentMetadataFiles());
+
+        // Setting min value
+        clusterSettings.applySettings(
+            Settings.builder().put(RecoverySettings.CLUSTER_REMOTE_INDEX_SEGMENT_METADATA_RETENTION_MAX_COUNT_SETTING.getKey(), -1).build()
+        );
+        assertEquals(-1, recoverySettings.getMinRemoteSegmentMetadataFiles());
+
+        // Setting value > default (10) and value < max (100)
+        clusterSettings.applySettings(
+            Settings.builder().put(RecoverySettings.CLUSTER_REMOTE_INDEX_SEGMENT_METADATA_RETENTION_MAX_COUNT_SETTING.getKey(), 15).build()
+        );
+        assertEquals(15, recoverySettings.getMinRemoteSegmentMetadataFiles());
+
+        // Setting value to 0 should fail and retain the existing value
+        assertThrows(
+            IllegalArgumentException.class,
+            () -> clusterSettings.applySettings(
+                Settings.builder()
+                    .put(RecoverySettings.CLUSTER_REMOTE_INDEX_SEGMENT_METADATA_RETENTION_MAX_COUNT_SETTING.getKey(), 0)
+                    .build()
+            )
+        );
+        assertEquals(15, recoverySettings.getMinRemoteSegmentMetadataFiles());
+
+        // Setting value < -1 should fail and retain the existing value
+        assertThrows(
+            IllegalArgumentException.class,
+            () -> clusterSettings.applySettings(
+                Settings.builder()
+                    .put(RecoverySettings.CLUSTER_REMOTE_INDEX_SEGMENT_METADATA_RETENTION_MAX_COUNT_SETTING.getKey(), -5)
+                    .build()
+            )
+        );
+        assertEquals(15, recoverySettings.getMinRemoteSegmentMetadataFiles());
+
+        // Setting value > 100 should fail and retain the existing value
+        assertThrows(
+            IllegalArgumentException.class,
+            () -> clusterSettings.applySettings(
+                Settings.builder()
+                    .put(RecoverySettings.CLUSTER_REMOTE_INDEX_SEGMENT_METADATA_RETENTION_MAX_COUNT_SETTING.getKey(), 105)
+                    .build()
+            )
+        );
+        assertEquals(15, recoverySettings.getMinRemoteSegmentMetadataFiles());
+    }
 }