Add capability to restrict async durability mode for remote indexes

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

server/src/main/java/org/opensearch/cluster/metadata/MetadataCreateIndexService.java

@@ -88,6 +88,7 @@
 import org.opensearch.index.mapper.MapperService.MergeReason;
 import org.opensearch.index.query.QueryShardContext;
 import org.opensearch.index.shard.IndexSettingProvider;
+import org.opensearch.index.translog.Translog;
 import org.opensearch.indices.IndexCreationException;
 import org.opensearch.indices.IndicesService;
 import org.opensearch.indices.InvalidIndexNameException;
@@ -922,6 +923,9 @@ static Settings aggregateIndexSettings(
         validateTranslogRetentionSettings(indexSettings);
         validateStoreTypeSettings(indexSettings);
         validateRefreshIntervalSettings(request.settings(), clusterSettings);
+        if (isRemoteStoreAttributePresent(settings)) {
+            validateTranslogDurabilitySettings(request.settings(), clusterSettings);
+        }
 
         return indexSettings;
     }
@@ -1491,7 +1495,7 @@ public static void validateTranslogRetentionSettings(Settings indexSettings) {
     /**
      * Validates {@code index.refresh_interval} is equal or below the {@code cluster.minimum.index.refresh_interval}.
      *
-     * @param requestSettings settings passed in during index create request
+     * @param requestSettings settings passed in during index create/update request
      * @param clusterSettings cluster setting
      */
     static void validateRefreshIntervalSettings(Settings requestSettings, ClusterSettings clusterSettings) {
@@ -1510,4 +1514,26 @@ static void validateRefreshIntervalSettings(Settings requestSettings, ClusterSet
             );
         }
     }
+
+    /**
+     * Validates {@code index.translog.durability} is not async if the {@code cluster.remote_store.index.restrict.async-durability} is set to true.
+     *
+     * @param requestSettings settings passed in during index create/update request
+     * @param clusterSettings cluster setting
+     */
+    static void validateTranslogDurabilitySettings(Settings requestSettings, ClusterSettings clusterSettings) {
+        if (IndexSettings.INDEX_TRANSLOG_DURABILITY_SETTING.exists(requestSettings) == false
+            || clusterSettings.get(IndicesService.CLUSTER_REMOTE_INDEX_RESTRICT_ASYNC_DURABILITY_SETTING) == false) {
+            return;
+        }
+        Translog.Durability durability = IndexSettings.INDEX_TRANSLOG_DURABILITY_SETTING.get(requestSettings);
+        if (durability.equals(Translog.Durability.ASYNC)) {
+            throw new IllegalArgumentException(
+                "index setting [index.translog.durability=async] is not allowed as cluster setting ["
+                    + IndicesService.CLUSTER_REMOTE_INDEX_RESTRICT_ASYNC_DURABILITY_SETTING.getKey()
+                    + "=true]"
+            );
+        }
+
+    }
 }

server/src/main/java/org/opensearch/cluster/metadata/MetadataUpdateSettingsService.java

@@ -73,8 +73,11 @@
 import java.util.Set;
 
 import static org.opensearch.action.support.ContextPreservingActionListener.wrapPreservingContext;
+import static org.opensearch.cluster.metadata.MetadataCreateIndexService.validateRefreshIntervalSettings;
+import static org.opensearch.cluster.metadata.MetadataCreateIndexService.validateTranslogDurabilitySettings;
 import static org.opensearch.common.settings.AbstractScopedSettings.ARCHIVED_SETTINGS_PREFIX;
 import static org.opensearch.index.IndexSettings.same;
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteStoreAttributePresent;
 
 /**
  * Service responsible for submitting update index settings requests
@@ -127,7 +130,10 @@ public void updateSettings(
             .normalizePrefix(IndexMetadata.INDEX_SETTING_PREFIX)
             .build();
 
-        MetadataCreateIndexService.validateRefreshIntervalSettings(normalizedSettings, clusterService.getClusterSettings());
+        validateRefreshIntervalSettings(normalizedSettings, clusterService.getClusterSettings());
+        if (isRemoteStoreAttributePresent(clusterService.getSettings())) {
+            validateTranslogDurabilitySettings(normalizedSettings, clusterService.getClusterSettings());
+        }
 
         Settings.Builder settingsForClosedIndices = Settings.builder();
         Settings.Builder settingsForOpenIndices = Settings.builder();

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

@@ -672,7 +672,8 @@ public void apply(Settings value, Settings current, Settings previous) {
                 // Remote cluster state settings
                 RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING,
                 RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING,
-                IndicesService.CLUSTER_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING
+                IndicesService.CLUSTER_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING,
+                IndicesService.CLUSTER_REMOTE_INDEX_RESTRICT_ASYNC_DURABILITY_SETTING
             )
         )
     );

server/src/main/java/org/opensearch/indices/IndicesService.java

@@ -285,6 +285,18 @@ public class IndicesService extends AbstractLifecycleComponent
         Property.Dynamic
     );
 
+    /**
+     * This setting is used to restrict creation or updation of index where the `index.translog.durability` index setting
+     * is set as ASYNC if enabled. If disabled, any of the durability mode can be used and switched at any later time from
+     * one to another.
+     */
+    public static final Setting<Boolean> CLUSTER_REMOTE_INDEX_RESTRICT_ASYNC_DURABILITY_SETTING = Setting.boolSetting(
+        "cluster.remote_store.index.restrict.async-durability",
+        false,
+        Property.NodeScope,
+        Property.Final
+    );
+
     /**
      * The node's settings.
      */

server/src/test/java/org/opensearch/cluster/metadata/MetadataCreateIndexServiceTests.java

@@ -70,6 +70,8 @@
 import org.opensearch.index.IndexSettings;
 import org.opensearch.index.mapper.MapperService;
 import org.opensearch.index.query.QueryShardContext;
+import org.opensearch.index.translog.Translog;
+import org.opensearch.indices.IndicesService;
 import org.opensearch.indices.InvalidAliasNameException;
 import org.opensearch.indices.InvalidIndexNameException;
 import org.opensearch.indices.ShardLimitValidator;
@@ -132,8 +134,10 @@
 import static org.opensearch.index.IndexSettings.INDEX_REFRESH_INTERVAL_SETTING;
 import static org.opensearch.index.IndexSettings.INDEX_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING;
 import static org.opensearch.index.IndexSettings.INDEX_SOFT_DELETES_SETTING;
+import static org.opensearch.index.IndexSettings.INDEX_TRANSLOG_DURABILITY_SETTING;
 import static org.opensearch.indices.IndicesService.CLUSTER_DEFAULT_INDEX_REFRESH_INTERVAL_SETTING;
 import static org.opensearch.indices.IndicesService.CLUSTER_MINIMUM_INDEX_REFRESH_INTERVAL_SETTING;
+import static org.opensearch.indices.IndicesService.CLUSTER_REMOTE_INDEX_RESTRICT_ASYNC_DURABILITY_SETTING;
 import static org.opensearch.indices.IndicesService.CLUSTER_REPLICATION_TYPE_SETTING;
 import static org.opensearch.indices.ShardLimitValidatorTests.createTestShardLimitService;
 import static org.opensearch.node.Node.NODE_ATTRIBUTES;
@@ -1725,6 +1729,87 @@ public void testRefreshIntervalValidationFailureWithIndexSetting() {
         );
     }
 
+    public void testAnyTranslogDurabilityWhenRestrictSettingFalse() {
+        // This checks that aggregateIndexSettings works for the case when the cluster setting
+        // cluster.remote_store.index.restrict.async-durability is false or not set, it allows all types of durability modes
+        request = new CreateIndexClusterStateUpdateRequest("create index", "test", "test");
+        final Settings.Builder requestSettings = Settings.builder();
+        Translog.Durability durability = randomFrom(Translog.Durability.REQUEST, Translog.Durability.ASYNC);
+        requestSettings.put(INDEX_TRANSLOG_DURABILITY_SETTING.getKey(), durability);
+        request.settings(requestSettings.build());
+        if (randomBoolean()) {
+            Settings settings = Settings.builder().put(CLUSTER_REMOTE_INDEX_RESTRICT_ASYNC_DURABILITY_SETTING.getKey(), false).build();
+            clusterSettings = new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
+        }
+        Settings indexSettings = aggregateIndexSettings(
+            ClusterState.EMPTY_STATE,
+            request,
+            Settings.EMPTY,
+            null,
+            Settings.EMPTY,
+            IndexScopedSettings.DEFAULT_SCOPED_SETTINGS,
+            randomShardLimitService(),
+            Collections.emptySet(),
+            clusterSettings
+        );
+        assertFalse(clusterSettings.get(IndicesService.CLUSTER_REMOTE_INDEX_RESTRICT_ASYNC_DURABILITY_SETTING));
+        assertEquals(durability, INDEX_TRANSLOG_DURABILITY_SETTING.get(indexSettings));
+    }
+
+    public void testAsyncDurabilityThrowsExceptionWhenRestrictSettingTrue() {
+        // This checks that aggregateIndexSettings works for the case when the cluster setting
+        // cluster.remote_store.index.restrict.async-durability is false or not set, it allows all types of durability modes
+        request = new CreateIndexClusterStateUpdateRequest("create index", "test", "test");
+        final Settings.Builder requestSettings = Settings.builder();
+        requestSettings.put(INDEX_TRANSLOG_DURABILITY_SETTING.getKey(), Translog.Durability.ASYNC);
+        request.settings(requestSettings.build());
+        Settings settings = Settings.builder().put(CLUSTER_REMOTE_INDEX_RESTRICT_ASYNC_DURABILITY_SETTING.getKey(), true).build();
+        clusterSettings = new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
+        IllegalArgumentException exception = assertThrows(
+            IllegalArgumentException.class,
+            () -> aggregateIndexSettings(
+                ClusterState.EMPTY_STATE,
+                request,
+                Settings.EMPTY,
+                null,
+                Settings.builder().put("node.attr.remote_store.setting", "test").build(),
+                IndexScopedSettings.DEFAULT_SCOPED_SETTINGS,
+                randomShardLimitService(),
+                Collections.emptySet(),
+                clusterSettings
+            )
+        );
+        // verify that the message is as expected
+        assertEquals(
+            "index setting [index.translog.durability=async] is not allowed as cluster setting [cluster.remote_store.index.restrict.async-durability=true]",
+            exception.getMessage()
+        );
+    }
+
+    public void testRequestDurabilityWhenRestrictSettingTrue() {
+        // This checks that aggregateIndexSettings works for the case when the cluster setting
+        // cluster.remote_store.index.restrict.async-durability is false or not set, it allows all types of durability modes
+        request = new CreateIndexClusterStateUpdateRequest("create index", "test", "test");
+        final Settings.Builder requestSettings = Settings.builder();
+        requestSettings.put(INDEX_TRANSLOG_DURABILITY_SETTING.getKey(), Translog.Durability.REQUEST);
+        request.settings(requestSettings.build());
+        Settings settings = Settings.builder().put(CLUSTER_REMOTE_INDEX_RESTRICT_ASYNC_DURABILITY_SETTING.getKey(), true).build();
+        clusterSettings = new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
+        Settings indexSettings = aggregateIndexSettings(
+            ClusterState.EMPTY_STATE,
+            request,
+            Settings.EMPTY,
+            null,
+            Settings.EMPTY,
+            IndexScopedSettings.DEFAULT_SCOPED_SETTINGS,
+            randomShardLimitService(),
+            Collections.emptySet(),
+            clusterSettings
+        );
+        assertTrue(clusterSettings.get(IndicesService.CLUSTER_REMOTE_INDEX_RESTRICT_ASYNC_DURABILITY_SETTING));
+        assertEquals(Translog.Durability.REQUEST, INDEX_TRANSLOG_DURABILITY_SETTING.get(indexSettings));
+    }
+
     private IndexTemplateMetadata addMatchingTemplate(Consumer<IndexTemplateMetadata.Builder> configurator) {
         IndexTemplateMetadata.Builder builder = templateMetadataBuilder("template1", "te*");
         configurator.accept(builder);