[8.x] Forbid the removal of the write block if the index is read-only (#120647)

Forbids on 8.x the removal/update of a write block if at least one node of the cluster cannot write the index.

Backport of #120648 for 8.18.

Relates ES-10320

Author: tlrx

server/src/main/java/org/elasticsearch/cluster/block/ClusterBlocks.java

@@ -146,6 +146,10 @@ public boolean hasIndexBlock(String index, ClusterBlock block) {
         return indicesBlocks.containsKey(index) && indicesBlocks.get(index).contains(block);
     }
 
+    public boolean hasIndexBlockLevel(String index, ClusterBlockLevel level) {
+        return blocksForIndex(level, index).isEmpty() == false;
+    }
+
     public boolean hasIndexBlockWithId(String index, int blockId) {
         final Set<ClusterBlock> clusterBlocks = indicesBlocks.get(index);
         if (clusterBlocks != null) {
@@ -398,6 +402,10 @@ public boolean hasIndexBlock(String index, ClusterBlock block) {
             return indices.getOrDefault(index, Set.of()).contains(block);
         }
 
+        public boolean hasIndexBlockLevel(String index, ClusterBlockLevel level) {
+            return indices.getOrDefault(index, Set.of()).stream().anyMatch(clusterBlock -> clusterBlock.contains(level));
+        }
+
         public Builder removeIndexBlock(String index, ClusterBlock block) {
             if (indices.containsKey(index) == false) {
                 return this;

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

@@ -36,6 +36,7 @@
 import org.elasticsearch.common.settings.IndexScopedSettings;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexSettings;
@@ -51,7 +52,9 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.function.BiFunction;
+import java.util.function.Function;
 
+import static org.elasticsearch.cluster.metadata.MetadataIndexStateService.VERIFIED_READ_ONLY_SETTING;
 import static org.elasticsearch.index.IndexSettings.same;
 
 /**
@@ -181,11 +184,14 @@ ClusterState execute(ClusterState currentState) {
 
             RoutingTable.Builder routingTableBuilder = null;
             Metadata.Builder metadataBuilder = Metadata.builder(currentState.metadata());
+            final var minSupportedIndexVersion = currentState.nodes().getMinSupportedIndexVersion();
 
             // allow to change any settings to a closed index, and only allow dynamic settings to be changed
             // on an open index
             Set<Index> openIndices = new HashSet<>();
             Set<Index> closedIndices = new HashSet<>();
+            Set<Index> readOnlyIndices = null;
+
             final String[] actualIndices = new String[request.indices().length];
             for (int i = 0; i < request.indices().length; i++) {
                 Index index = request.indices()[i];
@@ -197,6 +203,12 @@ ClusterState execute(ClusterState currentState) {
                 } else {
                     closedIndices.add(index);
                 }
+                if (metadata.getCompatibilityVersion().before(minSupportedIndexVersion)) {
+                    if (readOnlyIndices == null) {
+                        readOnlyIndices = new HashSet<>();
+                    }
+                    readOnlyIndices.add(index);
+                }
             }
 
             if (skippedSettings.isEmpty() == false && openIndices.isEmpty() == false) {
@@ -327,10 +339,23 @@ ClusterState execute(ClusterState currentState) {
                 }
             }
 
+            // provides the value of VERIFIED_READ_ONLY_SETTING before block changes
+            final Function<String, Boolean> verifiedReadOnly = indexName -> VERIFIED_READ_ONLY_SETTING.get(
+                currentState.metadata().index(indexName).getSettings()
+            );
+
             final ClusterBlocks.Builder blocks = ClusterBlocks.builder().blocks(currentState.blocks());
             boolean changedBlocks = false;
             for (IndexMetadata.APIBlock block : IndexMetadata.APIBlock.values()) {
-                changedBlocks |= maybeUpdateClusterBlock(actualIndices, blocks, block.block, block.setting, openSettings, metadataBuilder);
+                changedBlocks |= maybeUpdateClusterBlock(
+                    actualIndices,
+                    blocks,
+                    block.block,
+                    block.setting,
+                    openSettings,
+                    metadataBuilder,
+                    verifiedReadOnly
+                );
             }
             changed |= changedBlocks;
 
@@ -359,6 +384,7 @@ ClusterState execute(ClusterState currentState) {
                     // This step is mandatory since we allow to update non-dynamic settings on closed indices.
                     indicesService.verifyIndexMetadata(updatedMetadata, updatedMetadata);
                 }
+                verifyReadOnlyIndices(readOnlyIndices, updatedState.blocks());
             } catch (IOException ex) {
                 throw ExceptionsHelper.convertToElastic(ex);
             }
@@ -417,6 +443,18 @@ public static void updateIndexSettings(
         }
     }
 
+    private static void verifyReadOnlyIndices(@Nullable Set<Index> readOnlyIndices, ClusterBlocks blocks) {
+        if (readOnlyIndices != null) {
+            for (Index readOnlyIndex : readOnlyIndices) {
+                if (blocks.hasIndexBlockLevel(readOnlyIndex.getName(), ClusterBlockLevel.WRITE) == false) {
+                    throw new IllegalArgumentException(
+                        String.format(Locale.ROOT, "Can't remove the write block on read-only compatible index %s", readOnlyIndex)
+                    );
+                }
+            }
+        }
+    }
+
     /**
      * Updates the cluster block only iff the setting exists in the given settings
      */
@@ -426,7 +464,8 @@ private static boolean maybeUpdateClusterBlock(
         ClusterBlock block,
         Setting<Boolean> setting,
         Settings openSettings,
-        Metadata.Builder metadataBuilder
+        Metadata.Builder metadataBuilder,
+        Function<String, Boolean> verifiedReadOnlyBeforeBlockChanges
     ) {
         boolean changed = false;
         if (setting.exists(openSettings)) {
@@ -436,16 +475,32 @@ private static boolean maybeUpdateClusterBlock(
                     if (blocks.hasIndexBlock(index, block) == false) {
                         blocks.addIndexBlock(index, block);
                         changed = true;
+                        if (block.contains(ClusterBlockLevel.WRITE)) {
+                            var isVerifiedReadOnly = verifiedReadOnlyBeforeBlockChanges.apply(index);
+                            if (isVerifiedReadOnly) {
+                                var indexMetadata = metadataBuilder.get(index);
+                                metadataBuilder.put(
+                                    IndexMetadata.builder(indexMetadata)
+                                        .settings(
+                                            Settings.builder()
+                                                .put(indexMetadata.getSettings())
+                                                .put(VERIFIED_READ_ONLY_SETTING.getKey(), true)
+                                        )
+                                );
+                            }
+                        }
                     }
                 } else {
                     if (blocks.hasIndexBlock(index, block)) {
                         blocks.removeIndexBlock(index, block);
                         changed = true;
                         if (block.contains(ClusterBlockLevel.WRITE)) {
-                            IndexMetadata indexMetadata = metadataBuilder.get(index);
-                            Settings.Builder indexSettings = Settings.builder().put(indexMetadata.getSettings());
-                            indexSettings.remove(MetadataIndexStateService.VERIFIED_READ_ONLY_SETTING.getKey());
-                            metadataBuilder.put(IndexMetadata.builder(indexMetadata).settings(indexSettings));
+                            if (blocks.hasIndexBlockLevel(index, ClusterBlockLevel.WRITE) == false) {
+                                var indexMetadata = metadataBuilder.get(index);
+                                var indexSettings = Settings.builder().put(indexMetadata.getSettings());
+                                indexSettings.remove(VERIFIED_READ_ONLY_SETTING.getKey());
+                                metadataBuilder.put(IndexMetadata.builder(indexMetadata).settings(indexSettings));
+                            }
                         }
                     }
                 }