Use single directory for metadata (#50639)

ywelsch · DaveCTurner · web-flow · commit 0830b7dc1216 · 2020-01-08T14:06:15.000+01:00
Earlier PRs for #48701 introduced a separate directory for the cluster state. This is not needed though, and introduces an additional unnecessary cognitive burden to the users. Co-Authored-By: David Turner <david.turner@elastic.co>
diff --git a/server/src/main/java/org/elasticsearch/env/NodeEnvironment.java b/server/src/main/java/org/elasticsearch/env/NodeEnvironment.java
@@ -50,7 +50,6 @@
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.core.internal.io.IOUtils;
-import org.elasticsearch.gateway.PersistedClusterStateService;
 import org.elasticsearch.gateway.MetaDataStateFormat;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexSettings;
@@ -384,12 +383,9 @@ private static boolean upgradeLegacyNodeFolders(Logger logger, Settings settings
                 final Set<String> folderNames = new HashSet<>();
                 final Set<String> expectedFolderNames = new HashSet<>(Arrays.asList(
 
-                    // node state directory, also containing MetaDataStateFormat-based global metadata
+                    // node state directory, containing MetaDataStateFormat-based node metadata as well as cluster state
                     MetaDataStateFormat.STATE_DIR_NAME,
 
-                    // Lucene-based metadata folder
-                    PersistedClusterStateService.METADATA_DIRECTORY_NAME,
-
                     // indices
                     INDICES_FOLDER));
 
diff --git a/server/src/main/java/org/elasticsearch/env/NodeRepurposeCommand.java b/server/src/main/java/org/elasticsearch/env/NodeRepurposeCommand.java
@@ -31,8 +31,8 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.internal.io.IOUtils;
-import org.elasticsearch.gateway.PersistedClusterStateService;
 import org.elasticsearch.gateway.MetaDataStateFormat;
+import org.elasticsearch.gateway.PersistedClusterStateService;
 
 import java.io.IOException;
 import java.nio.file.Path;
@@ -117,8 +117,6 @@ private void processNoMasterNoDataNode(Terminal terminal, Path[] dataPaths, Envi
 
         removePaths(terminal, indexPaths); // clean-up shard dirs
         // clean-up all metadata dirs
-        IOUtils.rm(Stream.of(dataPaths).map(path -> path.resolve(PersistedClusterStateService.METADATA_DIRECTORY_NAME))
-            .toArray(Path[]::new));
         MetaDataStateFormat.deleteMetaState(dataPaths);
         IOUtils.rm(Stream.of(dataPaths).map(path -> path.resolve(INDICES_FOLDER)).toArray(Path[]::new));
 
diff --git a/server/src/main/java/org/elasticsearch/gateway/PersistedClusterStateService.java b/server/src/main/java/org/elasticsearch/gateway/PersistedClusterStateService.java
@@ -54,6 +54,7 @@
 import org.elasticsearch.common.io.stream.ReleasableBytesStreamOutput;
 import org.elasticsearch.common.lease.Releasable;
 import org.elasticsearch.common.logging.Loggers;
+import org.elasticsearch.common.lucene.Lucene;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
@@ -117,7 +118,7 @@ public class PersistedClusterStateService {
     private static final String INDEX_UUID_FIELD_NAME = "index_uuid";
     private static final int COMMIT_DATA_SIZE = 4;
 
-    public static final String METADATA_DIRECTORY_NAME = "_metadata";
+    public static final String METADATA_DIRECTORY_NAME = MetaDataStateFormat.STATE_DIR_NAME;
 
     private final Path[] dataPaths;
     private final String nodeId;
@@ -177,6 +178,16 @@ public Writer createWriter() throws IOException {
         return new Writer(metaDataIndexWriters, nodeId, bigArrays);
     }
 
+    /**
+     * Remove all persisted cluster states from the given data paths, for use in tests. Should only be called when there is no open
+     * {@link Writer} on these paths.
+     */
+    public static void deleteAll(Path[] dataPaths) throws IOException {
+        for (Path dataPath : dataPaths) {
+            Lucene.cleanLuceneIndex(new SimpleFSDirectory(dataPath.resolve(METADATA_DIRECTORY_NAME)));
+        }
+    }
+
     // exposed for tests
     Directory createDirectory(Path path) throws IOException {
         // it is possible to disable the use of MMapDirectory for indices, and it may be surprising to users that have done so if we still
diff --git a/server/src/test/java/org/elasticsearch/cluster/coordination/CoordinatorTests.java b/server/src/test/java/org/elasticsearch/cluster/coordination/CoordinatorTests.java
@@ -991,17 +991,18 @@ public void testCannotJoinClusterWithDifferentUUID() throws IllegalAccessExcepti
             cluster1.runRandomly();
             cluster1.stabilise();
 
-            final ClusterNode newNode;
+            final ClusterNode nodeInOtherCluster;
             try (Cluster cluster2 = new Cluster(3)) {
                 cluster2.runRandomly();
                 cluster2.stabilise();
 
-                final ClusterNode nodeInOtherCluster = randomFrom(cluster2.clusterNodes);
-                newNode = cluster1.new ClusterNode(nextNodeIndex.getAndIncrement(),
-                    nodeInOtherCluster.getLocalNode(), n -> cluster1.new MockPersistedState(n, nodeInOtherCluster.persistedState,
-                    Function.identity(), Function.identity()), nodeInOtherCluster.nodeSettings);
+                nodeInOtherCluster = randomFrom(cluster2.clusterNodes);
             }
 
+            final ClusterNode newNode = cluster1.new ClusterNode(nextNodeIndex.getAndIncrement(),
+                nodeInOtherCluster.getLocalNode(), n -> cluster1.new MockPersistedState(n, nodeInOtherCluster.persistedState,
+                Function.identity(), Function.identity()), nodeInOtherCluster.nodeSettings);
+
             cluster1.clusterNodes.add(newNode);
 
             MockLogAppender mockAppender = new MockLogAppender();
diff --git a/server/src/test/java/org/elasticsearch/cluster/coordination/UnsafeBootstrapAndDetachCommandIT.java b/server/src/test/java/org/elasticsearch/cluster/coordination/UnsafeBootstrapAndDetachCommandIT.java
@@ -25,7 +25,6 @@
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.env.NodeMetaData;
@@ -36,11 +35,9 @@
 import org.elasticsearch.test.InternalTestCluster;
 
 import java.io.IOException;
-import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
-import java.util.stream.Stream;
 
 import static org.elasticsearch.action.support.WriteRequest.RefreshPolicy.IMMEDIATE;
 import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
@@ -175,9 +172,7 @@ public void testBootstrapNoClusterState() throws IOException {
         internalCluster().stopRandomDataNode();
         Environment environment = TestEnvironment.newEnvironment(
             Settings.builder().put(internalCluster().getDefaultSettings()).put(dataPathSettings).build());
-        IOUtils.rm(Stream.of(nodeEnvironment.nodeDataPaths())
-            .map(path -> path.resolve(PersistedClusterStateService.METADATA_DIRECTORY_NAME))
-            .toArray(Path[]::new));
+        PersistedClusterStateService.deleteAll(nodeEnvironment.nodeDataPaths());
 
         expectThrows(() -> unsafeBootstrap(environment), ElasticsearchNodeCommand.CS_MISSING_MSG);
     }
@@ -191,9 +186,7 @@ public void testDetachNoClusterState() throws IOException {
         internalCluster().stopRandomDataNode();
         Environment environment = TestEnvironment.newEnvironment(
             Settings.builder().put(internalCluster().getDefaultSettings()).put(dataPathSettings).build());
-        IOUtils.rm(Stream.of(nodeEnvironment.nodeDataPaths())
-            .map(path -> path.resolve(PersistedClusterStateService.METADATA_DIRECTORY_NAME))
-            .toArray(Path[]::new));
+        PersistedClusterStateService.deleteAll(nodeEnvironment.nodeDataPaths());
 
         expectThrows(() -> detachCluster(environment), ElasticsearchNodeCommand.CS_MISSING_MSG);
     }
diff --git a/server/src/test/java/org/elasticsearch/gateway/PersistedClusterStateServiceTests.java b/server/src/test/java/org/elasticsearch/gateway/PersistedClusterStateServiceTests.java
@@ -40,6 +40,7 @@
 import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
+import org.elasticsearch.env.NodeMetaData;
 import org.elasticsearch.gateway.PersistedClusterStateService.Writer;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
@@ -201,9 +202,7 @@ public void testFailsOnMismatchedNodeIds() throws IOException {
             }
         }
 
-        for (Path dataPath : dataPaths2) {
-            IOUtils.rm(dataPath.resolve(MetaDataStateFormat.STATE_DIR_NAME));
-        }
+        NodeMetaData.FORMAT.cleanupOldFiles(Long.MAX_VALUE, dataPaths2);
 
         final Path[] combinedPaths = Stream.concat(Arrays.stream(dataPaths1), Arrays.stream(dataPaths2)).toArray(Path[]::new);
 
diff --git a/test/framework/src/main/java/org/elasticsearch/cluster/coordination/AbstractCoordinatorTestCase.java b/test/framework/src/main/java/org/elasticsearch/cluster/coordination/AbstractCoordinatorTestCase.java
@@ -56,13 +56,15 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor;
 import org.elasticsearch.discovery.DiscoveryModule;
 import org.elasticsearch.discovery.SeedHostsProvider;
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.gateway.ClusterStateUpdaters;
 import org.elasticsearch.gateway.GatewayService;
 import org.elasticsearch.gateway.MockGatewayMetaState;
+import org.elasticsearch.gateway.PersistedClusterStateService;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.disruption.DisruptableMockTransport;
@@ -740,12 +742,14 @@ class MockPersistedState implements CoordinationState.PersistedState {
                     if (oldState.nodeEnvironment != null) {
                         nodeEnvironment = oldState.nodeEnvironment;
                         final MetaData updatedMetaData = adaptGlobalMetaData.apply(oldState.getLastAcceptedState().metaData());
-                        if (updatedMetaData != oldState.getLastAcceptedState().metaData()) {
-                            throw new AssertionError("TODO adapting persistent metadata is not supported yet");
-                        }
                         final long updatedTerm = adaptCurrentTerm.apply(oldState.getCurrentTerm());
-                        if (updatedTerm != oldState.getCurrentTerm()) {
-                            throw new AssertionError("TODO adapting persistent current term is not supported yet");
+                        if (updatedMetaData != oldState.getLastAcceptedState().metaData() || updatedTerm != oldState.getCurrentTerm()) {
+                            try (PersistedClusterStateService.Writer writer =
+                                     new PersistedClusterStateService(nodeEnvironment, xContentRegistry(), BigArrays.NON_RECYCLING_INSTANCE)
+                                         .createWriter()) {
+                                writer.writeFullStateAndCommit(updatedTerm,
+                                    ClusterState.builder(oldState.getLastAcceptedState()).metaData(updatedMetaData).build());
+                            }
                         }
                         final MockGatewayMetaState gatewayMetaState = new MockGatewayMetaState(newLocalNode);
                         gatewayMetaState.start(Settings.EMPTY, nodeEnvironment, xContentRegistry());
