Write state also on data nodes if not master eligible

src/main/java/org/elasticsearch/cluster/ClusterChangedEvent.java

@@ -105,7 +105,14 @@ public List<String> indicesCreated() {
      * Returns the indices deleted in this event
      */
     public List<String> indicesDeleted() {
-        if (previousState == null) {
+
+        // if the new cluster state has a new master then we cannot know if an index which is not in the cluster state
+        // is actually supposed to be deleted or imported as dangling instead. for example a new master might not have
+        // the index in its cluster state because it was started with an empty data folder and in this case we want to
+        // import as dangling. we check here for new master too to be on the safe side in this case.
+        // norelease because we are not sure this is actually a good solution
+        // See discussion on https://github.com/elastic/elasticsearch/pull/9952
+        if (hasNewMaster() || previousState == null) {
             return ImmutableList.of();
         }
         if (!metaDataChanged()) {
@@ -165,4 +172,23 @@ public boolean nodesAdded() {
     public boolean nodesChanged() {
         return nodesRemoved() || nodesAdded();
     }
+
+    /**
+     * Checks if this cluster state comes from a different master than the previous one.
+     * This is a workaround for the scenario where a node misses a cluster state  that has either
+     * no master block or state not recovered flag set. In this case we must make sure that
+     * if an index is missing from the cluster state is not deleted immediately but instead imported
+     * as dangling. See discussion on https://github.com/elastic/elasticsearch/pull/9952
+     */
+    private boolean hasNewMaster() {
+        String oldMaster = previousState().getNodes().masterNodeId();
+        String newMaster = state().getNodes().masterNodeId();
+        if (oldMaster == null && newMaster == null) {
+            return false;
+        }
+        if (oldMaster == null && newMaster != null) {
+            return true;
+        }
+        return oldMaster.equals(newMaster) == false;
+    }
 }

src/main/java/org/elasticsearch/gateway/GatewayMetaState.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.gateway;
 
+import com.google.common.collect.ImmutableSet;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;
 import org.elasticsearch.cluster.ClusterChangedEvent;
@@ -27,9 +28,7 @@
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
-import org.elasticsearch.cluster.routing.DjbHashFunction;
-import org.elasticsearch.cluster.routing.HashFunction;
-import org.elasticsearch.cluster.routing.SimpleHashFunction;
+import org.elasticsearch.cluster.routing.*;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.inject.Inject;
@@ -43,6 +42,7 @@
 import java.nio.file.DirectoryStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.*;
 
 /**
  *
@@ -57,7 +57,9 @@ public class GatewayMetaState extends AbstractComponent implements ClusterStateL
     private final DanglingIndicesState danglingIndicesState;
 
     @Nullable
-    private volatile MetaData currentMetaData;
+    private volatile MetaData previousMetaData;
+
+    private volatile ImmutableSet<String> previouslyWrittenIndices = ImmutableSet.of();
 
     @Inject
     public GatewayMetaState(Settings settings, NodeEnvironment nodeEnv, MetaStateService metaStateService,
@@ -76,7 +78,7 @@ public GatewayMetaState(Settings settings, NodeEnvironment nodeEnv, MetaStateSer
         if (DiscoveryNode.masterNode(settings) || DiscoveryNode.dataNode(settings)) {
             nodeEnv.ensureAtomicMoveSupported();
         }
-        if (DiscoveryNode.masterNode(settings)) {
+        if (DiscoveryNode.masterNode(settings) || DiscoveryNode.dataNode(settings)) {
             try {
                 ensureNoPre019State();
                 pre20Upgrade();
@@ -96,55 +98,62 @@ public MetaData loadMetaState() throws Exception {
 
     @Override
     public void clusterChanged(ClusterChangedEvent event) {
+        Set<String> relevantIndices = new HashSet<>();
         final ClusterState state = event.state();
         if (state.blocks().disableStatePersistence()) {
             // reset the current metadata, we need to start fresh...
-            this.currentMetaData = null;
+            this.previousMetaData = null;
+            previouslyWrittenIndices = ImmutableSet.of();
             return;
         }
 
         MetaData newMetaData = state.metaData();
         // we don't check if metaData changed, since we might be called several times and we need to check dangling...
 
         boolean success = true;
-        // only applied to master node, writing the global and index level states
-        if (state.nodes().localNode().masterNode()) {
+        // write the state if this node is a master eligible node or if it is a data node and has shards allocated on it
+        if (state.nodes().localNode().masterNode() || state.nodes().localNode().dataNode()) {
+            if (previousMetaData == null) {
+                try {
+                    // we determine if or if not we write meta data on data only nodes by looking at the shard routing
+                    // and only write if a shard of this index is allocated on this node
+                    // however, closed indices do not appear in the shard routing. if the meta data for a closed index is
+                    // updated it will therefore not be written in case the list of previouslyWrittenIndices is empty (because state
+                    // persistence was disabled or the node was restarted), see getRelevantIndicesOnDataOnlyNode().
+                    // we therefore have to check here if we have shards on disk and add their indices to the previouslyWrittenIndices list
+                    if (isDataOnlyNode(state)) {
+                        ImmutableSet.Builder<String> previouslyWrittenIndicesBuilder = ImmutableSet.builder();
+                        for (IndexMetaData indexMetaData : newMetaData) {
+                            IndexMetaData indexMetaDataOnDisk = null;
+                            if (indexMetaData.state().equals(IndexMetaData.State.CLOSE)) {
+                                indexMetaDataOnDisk = metaStateService.loadIndexState(indexMetaData.index());
+                            }
+                            if (indexMetaDataOnDisk != null) {
+                                previouslyWrittenIndicesBuilder.add(indexMetaDataOnDisk.index());
+                            }
+                        }
+                        previouslyWrittenIndices = previouslyWrittenIndicesBuilder.addAll(previouslyWrittenIndices).build();
+                    }
+                } catch (Throwable e) {
+                    success = false;
+                }
+            }
             // check if the global state changed?
-            if (currentMetaData == null || !MetaData.isGlobalStateEquals(currentMetaData, newMetaData)) {
+            if (previousMetaData == null || !MetaData.isGlobalStateEquals(previousMetaData, newMetaData)) {
                 try {
                     metaStateService.writeGlobalState("changed", newMetaData);
                 } catch (Throwable e) {
                     success = false;
                 }
             }
 
+            Iterable<IndexMetaWriteInfo> writeInfo;
+            relevantIndices = getRelevantIndices(event.state(), previouslyWrittenIndices);
+            writeInfo = resolveStatesToBeWritten(previouslyWrittenIndices, relevantIndices, previousMetaData, event.state().metaData());
             // check and write changes in indices
-            for (IndexMetaData indexMetaData : newMetaData) {
-                String writeReason = null;
-                IndexMetaData currentIndexMetaData;
-                if (currentMetaData == null) {
-                    // a new event..., check from the state stored
-                    try {
-                        currentIndexMetaData = metaStateService.loadIndexState(indexMetaData.index());
-                    } catch (IOException ex) {
-                        throw new ElasticsearchException("failed to load index state", ex);
-                    }
-                } else {
-                    currentIndexMetaData = currentMetaData.index(indexMetaData.index());
-                }
-                if (currentIndexMetaData == null) {
-                    writeReason = "freshly created";
-                } else if (currentIndexMetaData.version() != indexMetaData.version()) {
-                    writeReason = "version changed from [" + currentIndexMetaData.version() + "] to [" + indexMetaData.version() + "]";
-                }
-
-                // we update the writeReason only if we really need to write it
-                if (writeReason == null) {
-                    continue;
-                }
-
+            for (IndexMetaWriteInfo indexMetaWrite : writeInfo) {
                 try {
-                    metaStateService.writeIndex(writeReason, indexMetaData, currentIndexMetaData);
+                    metaStateService.writeIndex(indexMetaWrite.reason, indexMetaWrite.newMetaData, indexMetaWrite.previousMetaData);
                 } catch (Throwable e) {
                     success = false;
                 }
@@ -154,8 +163,27 @@ public void clusterChanged(ClusterChangedEvent event) {
         danglingIndicesState.processDanglingIndices(newMetaData);
 
         if (success) {
-            currentMetaData = newMetaData;
+            previousMetaData = newMetaData;
+            ImmutableSet.Builder<String> builder = ImmutableSet.builder();
+            previouslyWrittenIndices = builder.addAll(relevantIndices).build();
+        }
+    }
+
+    public static Set<String> getRelevantIndices(ClusterState state, ImmutableSet<String> previouslyWrittenIndices) {
+        Set<String> relevantIndices;
+        if (isDataOnlyNode(state)) {
+            relevantIndices = getRelevantIndicesOnDataOnlyNode(state, previouslyWrittenIndices);
+        } else if (state.nodes().localNode().masterNode() == true) {
+            relevantIndices = getRelevantIndicesForMasterEligibleNode(state);
+        } else {
+            relevantIndices = Collections.emptySet();
         }
+        return relevantIndices;
+    }
+
+
+    protected static boolean isDataOnlyNode(ClusterState state) {
+        return ((state.nodes().localNode().masterNode() == false) && state.nodes().localNode().dataNode());
     }
 
     /**
@@ -229,7 +257,7 @@ private void pre20Upgrade() throws Exception {
                 }
             }
         }
-        if (hasCustomPre20HashFunction|| pre20UseType != null) {
+        if (hasCustomPre20HashFunction || pre20UseType != null) {
             logger.warn("Settings [{}] and [{}] are deprecated. Index settings from your old indices have been updated to record the fact that they "
                     + "used some custom routing logic, you can now remove these settings from your `elasticsearch.yml` file", DEPRECATED_SETTING_ROUTING_HASH_FUNCTION, DEPRECATED_SETTING_ROUTING_USE_TYPE);
         }
@@ -251,4 +279,82 @@ private void ensureNoPre019ShardState(NodeEnvironment nodeEnv) throws Exception
             }
         }
     }
+
+    /**
+     * Loads the current meta state for each index in the new cluster state and checks if it has to be persisted.
+     * Each index state that should be written to disk will be returned. This is only run for data only nodes.
+     * It will return only the states for indices that actually have a shard allocated on the current node.
+     *
+     * @param previouslyWrittenIndices    A list of indices for which the state was already written before
+     * @param potentiallyUnwrittenIndices The list of indices for which state should potentially be written
+     * @param previousMetaData            The last meta data we know of. meta data for all indices in previouslyWrittenIndices list is persisted now
+     * @param newMetaData                 The new metadata
+     * @return iterable over all indices states that should be written to disk
+     */
+    public static Iterable<GatewayMetaState.IndexMetaWriteInfo> resolveStatesToBeWritten(ImmutableSet<String> previouslyWrittenIndices, Set<String> potentiallyUnwrittenIndices, MetaData previousMetaData, MetaData newMetaData) {
+        List<GatewayMetaState.IndexMetaWriteInfo> indicesToWrite = new ArrayList<>();
+        for (String index : potentiallyUnwrittenIndices) {
+            IndexMetaData newIndexMetaData = newMetaData.index(index);
+            IndexMetaData previousIndexMetaData = previousMetaData == null ? null : previousMetaData.index(index);
+            String writeReason = null;
+            if (previouslyWrittenIndices.contains(index) == false || previousIndexMetaData == null) {
+                writeReason = "freshly created";
+            } else if (previousIndexMetaData.version() != newIndexMetaData.version()) {
+                writeReason = "version changed from [" + previousIndexMetaData.version() + "] to [" + newIndexMetaData.version() + "]";
+            }
+            if (writeReason != null) {
+                indicesToWrite.add(new GatewayMetaState.IndexMetaWriteInfo(newIndexMetaData, previousIndexMetaData, writeReason));
+            }
+        }
+        return indicesToWrite;
+    }
+
+    public static Set<String> getRelevantIndicesOnDataOnlyNode(ClusterState state, ImmutableSet<String> previouslyWrittenIndices) {
+        RoutingNode newRoutingNode = state.getRoutingNodes().node(state.nodes().localNodeId());
+        if (newRoutingNode == null) {
+            throw new IllegalStateException("cluster state does not contain this node - cannot write index meta state");
+        }
+        Set<String> indices = new HashSet<>();
+        for (MutableShardRouting routing : newRoutingNode) {
+            indices.add(routing.index());
+        }
+        // we have to check the meta data also: closed indices will not appear in the routing table, but we must still write the state if we have it written on disk previously
+        for (IndexMetaData indexMetaData : state.metaData()) {
+            if (previouslyWrittenIndices.contains(indexMetaData.getIndex()) && state.metaData().getIndices().get(indexMetaData.getIndex()).state().equals(IndexMetaData.State.CLOSE)) {
+                indices.add(indexMetaData.getIndex());
+            }
+        }
+        return indices;
+    }
+
+    public static Set<String> getRelevantIndicesForMasterEligibleNode(ClusterState state) {
+        Set<String> relevantIndices;
+        relevantIndices = new HashSet<>();
+        // we have to iterate over the metadata to make sure we also capture closed indices
+        for (IndexMetaData indexMetaData : state.metaData()) {
+            relevantIndices.add(indexMetaData.getIndex());
+        }
+        return relevantIndices;
+    }
+
+
+    public static class IndexMetaWriteInfo {
+        final IndexMetaData newMetaData;
+        final String reason;
+        final IndexMetaData previousMetaData;
+
+        public IndexMetaWriteInfo(IndexMetaData newMetaData, IndexMetaData previousMetaData, String reason) {
+            this.newMetaData = newMetaData;
+            this.reason = reason;
+            this.previousMetaData = previousMetaData;
+        }
+
+        public IndexMetaData getNewMetaData() {
+            return newMetaData;
+        }
+
+        public String getReason() {
+            return reason;
+        }
+    }
 }