Move metadata storage to Lucene (elastic#50907)

Today we split the on-disk cluster metadata across many files: one file for the metadata of each
index, plus one file for the global metadata and another for the manifest. Most metadata updates
only touch a few of these files, but some must write them all. If a node holds a large number of
indices then it's possible its disks are not fast enough to process a complete metadata update before timing out. In severe cases affecting master-eligible nodes this can prevent an election
from succeeding.

This commit uses Lucene as a metadata storage for the cluster state, and is a squashed version
of the following PRs that were targeting a feature branch:


* Introduce Lucene-based metadata persistence (elastic#48733)

This commit introduces `LucenePersistedState` which master-eligible nodes
can use to persist the cluster metadata in a Lucene index rather than in
many separate files.

Relates elastic#48701

* Remove per-index metadata without assigned shards (elastic#49234)

Today on master-eligible nodes we maintain per-index metadata files for every
index. However, we also keep this metadata in the `LucenePersistedState`, and
only use the per-index metadata files for importing dangling indices. However
there is no point in importing a dangling index without any shard data, so we
do not need to maintain these extra files any more.

This commit removes per-index metadata files from nodes which do not hold any
shards of those indices.

Relates elastic#48701

* Use Lucene exclusively for metadata storage (elastic#50144)

This moves metadata persistence to Lucene for all node types. It also reenables BWC and adds
an interoperability layer for upgrades from prior versions.

This commit disables a number of tests related to dangling indices and command-line tools.
Those will be addressed in follow-ups.

Relates elastic#48701

* Add command-line tool support for Lucene-based metadata storage (elastic#50179)

Adds command-line tool support (unsafe-bootstrap, detach-cluster, repurpose, & shard
commands) for the Lucene-based metadata storage.

Relates elastic#48701

* Use single directory for metadata (elastic#50639)

Earlier PRs for elastic#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>

* Add async dangling indices support (elastic#50642)

Adds support for writing out dangling indices in an asynchronous way. Also provides an option to
avoid writing out dangling indices at all.

Relates elastic#48701

* Fold node metadata into new node storage (elastic#50741)

Moves node metadata to uses the new storage mechanism (see elastic#48701) as the authoritative source.

* Write CS asynchronously on data-only nodes (elastic#50782)

Writes cluster states out asynchronously on data-only nodes. The main reason for writing out
the cluster state at all is so that the data-only nodes can snap into a cluster, that they can do a
bit of bootstrap validation and so that the shard recovery tools work.
Cluster states that are written asynchronously have their voting configuration adapted to a non
existing configuration so that these nodes cannot mistakenly become master even if their node
role is changed back and forth.

Relates elastic#48701

* Remove persistent cluster settings tool (elastic#50694)

Adds the elasticsearch-node remove-settings tool to remove persistent settings from the on
disk cluster state in case where it contains incompatible settings that prevent the cluster from
forming.

Relates elastic#48701

* Make cluster state writer resilient to disk issues (elastic#50805)

Adds handling to make the cluster state writer resilient to disk issues. Relates to elastic#48701

* Omit writing global metadata if no change (elastic#50901)

Uses the same optimization for the new cluster state storage layer as the old one, writing global
metadata only when changed. Avoids writing out the global metadata if none of the persistent
fields changed. Speeds up server:integTest by ~10%.

Relates elastic#48701

* DanglingIndicesIT should ensure node removed first (elastic#50896)

These tests occasionally failed because the deletion was submitted before the
restarting node was removed from the cluster, causing the deletion not to be
fully acked. This commit fixes this by checking the restarting node has been
removed from the cluster.

Co-authored-by: David Turner <david.turner@elastic.co>

docs/reference/commands/node-tool.asciidoc

@@ -3,9 +3,9 @@
 
 The `elasticsearch-node` command enables you to perform certain unsafe
 operations on a node that are only possible while it is shut down. This command
-allows you to adjust the <<modules-node,role>> of a node and may be able to
-recover some data after a disaster or start a node even if it is incompatible
-with the data on disk.
+allows you to adjust the <<modules-node,role>> of a node, unsafely edit cluster
+settings and may be able to recover some data after a disaster or start a node
+even if it is incompatible with the data on disk.
 
 [float]
 === Synopsis
@@ -20,13 +20,17 @@ bin/elasticsearch-node repurpose|unsafe-bootstrap|detach-cluster|override-versio
 [float]
 === Description
 
-This tool has four modes:
+This tool has five modes:
 
 * `elasticsearch-node repurpose` can be used to delete unwanted data from a
   node if it used to be a <<data-node,data node>> or a
   <<master-node,master-eligible node>> but has been repurposed not to have one
   or other of these roles.
 
+* `elasticsearch-node remove-settings` can be used to remove persistent settings
+   from the cluster state in case where it contains incompatible settings that
+   prevent the cluster from forming.
+
 * `elasticsearch-node unsafe-bootstrap` can be used to perform _unsafe cluster
   bootstrapping_.  It forces one of the nodes to form a brand-new cluster on
   its own, using its local copy of the cluster metadata.
@@ -76,6 +80,26 @@ The tool provides a summary of the data to be deleted and asks for confirmation
 before making any changes. You can get detailed information about the affected
 indices and shards by passing the verbose (`-v`) option.
 
+[float]
+==== Removing persistent cluster settings
+
+There may be situations where a node contains persistent cluster
+settings that prevent the cluster from forming. Since the cluster cannot form,
+it is not possible to remove these settings using the
+<<cluster-update-settings>> API.
+
+The `elasticsearch-node remove-settings` tool allows you to forcefully remove
+those persistent settings from the on-disk cluster state. The tool takes a
+list of settings as parameters that should be removed, and also supports
+wildcard patterns.
+
+The intended use is:
+
+* Stop the node
+* Run `elasticsearch-node remove-settings name-of-setting-to-remove` on the node
+* Repeat for all other master-eligible nodes
+* Start the nodes
+
 [float]
 ==== Recovering data after a disaster
 
@@ -143,9 +167,9 @@ If there is at least one remaining master-eligible node, but it is not possible
 to restart a majority of them, then the `elasticsearch-node unsafe-bootstrap`
 command will unsafely override the cluster's <<modules-discovery-voting,voting
 configuration>> as if performing another
-<<modules-discovery-bootstrap-cluster,cluster bootstrapping process>>. 
+<<modules-discovery-bootstrap-cluster,cluster bootstrapping process>>.
 The target node can then form a new cluster on its own by using
-the cluster metadata held locally on the target node. 
+the cluster metadata held locally on the target node.
 
 [WARNING]
 These steps can lead to arbitrary data loss since the target node may not hold the latest cluster
@@ -290,6 +314,9 @@ it can join a different cluster.
 `override-version`:: Overwrites the version number stored in the data path so
 that a node can start despite being incompatible with the on-disk data.
 
+`remove-settings`:: Forcefully removes the provided persistent cluster settings
+from the on-disk cluster state.
+
 `-E <KeyValuePair>`:: Configures a setting.
 
 `-h, --help`:: Returns all of the command parameters.
@@ -346,6 +373,40 @@ Confirm [y/N] y
 Node successfully repurposed to no-master and no-data.
 ----
 
+[float]
+==== Removing persistent cluster settings
+
+If your nodes contain persistent cluster settings that prevent the cluster
+from forming, i.e., can't be removed using the <<cluster-update-settings>> API,
+you can run the following commands to remove one or more cluster settings.
+
+[source,txt]
+----
+node$ ./bin/elasticsearch-node remove-settings xpack.monitoring.exporters.my_exporter.host
+
+    WARNING: Elasticsearch MUST be stopped before running this tool.
+
+The following settings will be removed:
+xpack.monitoring.exporters.my_exporter.host: "10.1.2.3"
+
+You should only run this tool if you have incompatible settings in the
+cluster state that prevent the cluster from forming.
+This tool can cause data loss and its use should be your last resort.
+
+Do you want to proceed?
+
+Confirm [y/N] y
+
+Settings were successfully removed from the cluster state
+----
+
+You can also use wildcards to remove multiple settings, for example using
+
+[source,txt]
+----
+node$ ./bin/elasticsearch-node remove-settings xpack.monitoring.*
+----
+
 [float]
 ==== Unsafe cluster bootstrapping
 

server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationState.java

@@ -25,6 +25,8 @@
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 
+import java.io.Closeable;
+import java.io.IOException;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
@@ -432,15 +434,14 @@ public void invariant() {
         assert publishVotes.isEmpty() || electionWon();
     }
 
-    public void close() {
+    public void close() throws IOException {
         persistedState.close();
     }
 
     /**
      * Pluggable persistence layer for {@link CoordinationState}.
-     *
      */
-    public interface PersistedState {
+    public interface PersistedState extends Closeable {
 
         /**
          * Returns the current term
@@ -497,7 +498,8 @@ default void markLastAcceptedStateAsCommitted() {
             }
         }
 
-        default void close() {}
+        default void close() throws IOException {
+        }
     }
 
     /**

server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java

@@ -72,6 +72,7 @@
 import org.elasticsearch.transport.TransportResponse.Empty;
 import org.elasticsearch.transport.TransportService;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -702,7 +703,7 @@ protected void doStop() {
     }
 
     @Override
-    protected void doClose() {
+    protected void doClose() throws IOException {
         final CoordinationState coordinationState = this.coordinationState.get();
         if (coordinationState != null) {
             // This looks like a race that might leak an unclosed CoordinationState if it's created while execution is here, but this method

server/src/main/java/org/elasticsearch/cluster/coordination/DetachClusterCommand.java

@@ -18,11 +18,12 @@
  */
 package org.elasticsearch.cluster.coordination;
 
+import joptsimple.OptionSet;
 import org.elasticsearch.cli.Terminal;
-import org.elasticsearch.cluster.metadata.Manifest;
+import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.env.Environment;
+import org.elasticsearch.gateway.PersistedClusterStateService;
 
 import java.io.IOException;
 import java.nio.file.Path;
@@ -48,14 +49,21 @@ public DetachClusterCommand() {
 
 
     @Override
-    protected void processNodePaths(Terminal terminal, Path[] dataPaths, Environment env) throws IOException {
-        final Tuple<Manifest, MetaData> manifestMetaDataTuple = loadMetaData(terminal, dataPaths);
-        final Manifest manifest = manifestMetaDataTuple.v1();
-        final MetaData metaData = manifestMetaDataTuple.v2();
+    protected void processNodePaths(Terminal terminal, Path[] dataPaths, OptionSet options, Environment env) throws IOException {
+        final PersistedClusterStateService persistedClusterStateService = createPersistedClusterStateService(dataPaths);
+
+        terminal.println(Terminal.Verbosity.VERBOSE, "Loading cluster state");
+        final ClusterState oldClusterState = loadTermAndClusterState(persistedClusterStateService, env).v2();
+        final ClusterState newClusterState = ClusterState.builder(oldClusterState)
+            .metaData(updateMetaData(oldClusterState.metaData())).build();
+        terminal.println(Terminal.Verbosity.VERBOSE,
+            "[old cluster state = " + oldClusterState + ", new cluster state = " + newClusterState + "]");
 
         confirm(terminal, CONFIRMATION_MSG);
 
-        writeNewMetaData(terminal, manifest, updateCurrentTerm(), metaData, updateMetaData(metaData), dataPaths);
+        try (PersistedClusterStateService.Writer writer = persistedClusterStateService.createWriter()) {
+            writer.writeFullStateAndCommit(updateCurrentTerm(), newClusterState);
+        }
 
         terminal.println(NODE_DETACHED_MSG);
     }

server/src/main/java/org/elasticsearch/cluster/coordination/ElasticsearchNodeCommand.java

@@ -26,75 +26,92 @@
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.cli.EnvironmentAwareCommand;
 import org.elasticsearch.cli.Terminal;
-import org.elasticsearch.cluster.metadata.Manifest;
-import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.cli.UserException;
+import org.elasticsearch.cluster.ClusterModule;
+import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
+import org.elasticsearch.env.NodeMetaData;
+import org.elasticsearch.gateway.PersistedClusterStateService;
+import org.elasticsearch.indices.IndicesModule;
 
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.Objects;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 public abstract class ElasticsearchNodeCommand extends EnvironmentAwareCommand {
     private static final Logger logger = LogManager.getLogger(ElasticsearchNodeCommand.class);
     protected static final String DELIMITER = "------------------------------------------------------------------------\n";
-
     static final String STOP_WARNING_MSG =
             DELIMITER +
                     "\n" +
                     "    WARNING: Elasticsearch MUST be stopped before running this tool." +
                     "\n";
     protected static final String FAILED_TO_OBTAIN_NODE_LOCK_MSG = "failed to lock node's directory, is Elasticsearch still running?";
-    static final String NO_NODE_FOLDER_FOUND_MSG = "no node folder is found in data folder(s), node has not been started yet?";
-    static final String NO_MANIFEST_FILE_FOUND_MSG = "no manifest file is found, do you run pre 7.0 Elasticsearch?";
-    protected static final String GLOBAL_GENERATION_MISSING_MSG =
-        "no metadata is referenced from the manifest file, cluster has never been bootstrapped?";
-    static final String NO_GLOBAL_METADATA_MSG = "failed to find global metadata, metadata corrupted?";
-    static final String WRITE_METADATA_EXCEPTION_MSG = "exception occurred when writing new metadata to disk";
     protected static final String ABORTED_BY_USER_MSG = "aborted by user";
+    static final String NO_NODE_FOLDER_FOUND_MSG = "no node folder is found in data folder(s), node has not been started yet?";
+    static final String NO_NODE_METADATA_FOUND_MSG = "no node meta data is found, node has not been started yet?";
+    protected static final String CS_MISSING_MSG =
+        "cluster state is empty, cluster has never been bootstrapped?";
+
+    protected static final NamedXContentRegistry namedXContentRegistry = new NamedXContentRegistry(
+        Stream.of(ClusterModule.getNamedXWriteables().stream(), IndicesModule.getNamedXContents().stream())
+            .flatMap(Function.identity())
+            .collect(Collectors.toList()));
 
     public ElasticsearchNodeCommand(String description) {
         super(description);
     }
 
-    protected void processNodePaths(Terminal terminal, OptionSet options, Environment env) throws IOException {
+    public static PersistedClusterStateService createPersistedClusterStateService(Path[] dataPaths) throws IOException {
+        final NodeMetaData nodeMetaData = PersistedClusterStateService.nodeMetaData(dataPaths);
+        if (nodeMetaData == null) {
+            throw new ElasticsearchException(NO_NODE_METADATA_FOUND_MSG);
+        }
+
+        String nodeId = nodeMetaData.nodeId();
+        return new PersistedClusterStateService(dataPaths, nodeId, namedXContentRegistry, BigArrays.NON_RECYCLING_INSTANCE, true);
+    }
+
+    public static ClusterState clusterState(Environment environment, PersistedClusterStateService.OnDiskState onDiskState) {
+        return ClusterState.builder(ClusterName.CLUSTER_NAME_SETTING.get(environment.settings()))
+            .version(onDiskState.lastAcceptedVersion)
+            .metaData(onDiskState.metaData)
+            .build();
+    }
+
+    public static Tuple<Long, ClusterState> loadTermAndClusterState(PersistedClusterStateService psf,
+                                                                    Environment env) throws IOException {
+        final PersistedClusterStateService.OnDiskState bestOnDiskState = psf.loadBestOnDiskState();
+        if (bestOnDiskState.empty()) {
+            throw new ElasticsearchException(CS_MISSING_MSG);
+        }
+        return Tuple.tuple(bestOnDiskState.currentTerm, clusterState(env, bestOnDiskState));
+    }
+
+    protected void processNodePaths(Terminal terminal, OptionSet options, Environment env) throws IOException, UserException {
         terminal.println(Terminal.Verbosity.VERBOSE, "Obtaining lock for node");
         try (NodeEnvironment.NodeLock lock = new NodeEnvironment.NodeLock(logger, env, Files::exists)) {
             final Path[] dataPaths =
                     Arrays.stream(lock.getNodePaths()).filter(Objects::nonNull).map(p -> p.path).toArray(Path[]::new);
             if (dataPaths.length == 0) {
                 throw new ElasticsearchException(NO_NODE_FOLDER_FOUND_MSG);
             }
-            processNodePaths(terminal, dataPaths, env);
+            processNodePaths(terminal, dataPaths, options, env);
         } catch (LockObtainFailedException e) {
             throw new ElasticsearchException(FAILED_TO_OBTAIN_NODE_LOCK_MSG, e);
         }
     }
 
-    protected Tuple<Manifest, MetaData> loadMetaData(Terminal terminal, Path[] dataPaths) throws IOException {
-        terminal.println(Terminal.Verbosity.VERBOSE, "Loading manifest file");
-        final Manifest manifest = Manifest.FORMAT.loadLatestState(logger, NamedXContentRegistry.EMPTY, dataPaths);
-
-        if (manifest == null) {
-            throw new ElasticsearchException(NO_MANIFEST_FILE_FOUND_MSG);
-        }
-        if (manifest.isGlobalGenerationMissing()) {
-            throw new ElasticsearchException(GLOBAL_GENERATION_MISSING_MSG);
-        }
-        terminal.println(Terminal.Verbosity.VERBOSE, "Loading global metadata file");
-        final MetaData metaData = MetaData.FORMAT_PRESERVE_CUSTOMS.loadGeneration(
-            logger, NamedXContentRegistry.EMPTY, manifest.getGlobalGeneration(), dataPaths);
-        if (metaData == null) {
-            throw new ElasticsearchException(NO_GLOBAL_METADATA_MSG + " [generation = " + manifest.getGlobalGeneration() + "]");
-        }
-
-        return Tuple.tuple(manifest, metaData);
-    }
-
     protected void confirm(Terminal terminal, String msg) {
         terminal.println(msg);
         String text = terminal.readText("Confirm [y/N] ");
@@ -104,7 +121,7 @@ protected void confirm(Terminal terminal, String msg) {
     }
 
     @Override
-    protected final void execute(Terminal terminal, OptionSet options, Environment env) throws Exception {
+    public final void execute(Terminal terminal, OptionSet options, Environment env) throws Exception {
         terminal.println(STOP_WARNING_MSG);
         if (validateBeforeLock(terminal, env)) {
             processNodePaths(terminal, options, env);
@@ -126,44 +143,11 @@ protected boolean validateBeforeLock(Terminal terminal, Environment env) {
      * Process the paths. Locks for the paths is held during this method invocation.
      * @param terminal the terminal to use for messages
      * @param dataPaths the paths of the node to process
+     * @param options the command line options
      * @param env the env of the node to process
      */
-    protected abstract void processNodePaths(Terminal terminal, Path[] dataPaths, Environment env) throws IOException;
-
-
-    protected void writeNewMetaData(Terminal terminal, Manifest oldManifest, long newCurrentTerm,
-                                    MetaData oldMetaData, MetaData newMetaData, Path[] dataPaths) {
-        long newGeneration;
-        try {
-            terminal.println(Terminal.Verbosity.VERBOSE,
-                    "[clusterUUID = " + oldMetaData.clusterUUID() + ", committed = " + oldMetaData.clusterUUIDCommitted() + "] => " +
-                         "[clusterUUID = " + newMetaData.clusterUUID() + ", committed = " + newMetaData.clusterUUIDCommitted() + "]");
-            terminal.println(Terminal.Verbosity.VERBOSE, "New coordination metadata is " + newMetaData.coordinationMetaData());
-            terminal.println(Terminal.Verbosity.VERBOSE, "Writing new global metadata to disk");
-            newGeneration = MetaData.FORMAT.write(newMetaData, dataPaths);
-            Manifest newManifest = new Manifest(newCurrentTerm, oldManifest.getClusterStateVersion(), newGeneration,
-                    oldManifest.getIndexGenerations());
-            terminal.println(Terminal.Verbosity.VERBOSE, "New manifest is " + newManifest);
-            terminal.println(Terminal.Verbosity.VERBOSE, "Writing new manifest file to disk");
-            Manifest.FORMAT.writeAndCleanup(newManifest, dataPaths);
-        } catch (Exception e) {
-            terminal.println(Terminal.Verbosity.VERBOSE, "Cleaning up new metadata");
-            MetaData.FORMAT.cleanupOldFiles(oldManifest.getGlobalGeneration(), dataPaths);
-            throw new ElasticsearchException(WRITE_METADATA_EXCEPTION_MSG, e);
-        }
-        // if cleaning old files fail, we still succeeded.
-        try {
-            cleanUpOldMetaData(terminal, dataPaths, newGeneration);
-        } catch (Exception e) {
-            terminal.println(Terminal.Verbosity.SILENT,
-                "Warning: Cleaning up old metadata failed, but operation was otherwise successful (message: " + e.getMessage() + ")");
-        }
-    }
-
-    protected void cleanUpOldMetaData(Terminal terminal, Path[] dataPaths, long newGeneration) {
-        terminal.println(Terminal.Verbosity.VERBOSE, "Cleaning up old metadata");
-        MetaData.FORMAT.cleanupOldFiles(newGeneration, dataPaths);
-    }
+    protected abstract void processNodePaths(Terminal terminal, Path[] dataPaths, OptionSet options, Environment env)
+        throws IOException, UserException;
 
     protected NodeEnvironment.NodePath[] toNodePaths(Path[] dataPaths) {
         return Arrays.stream(dataPaths).map(ElasticsearchNodeCommand::createNodePath).toArray(NodeEnvironment.NodePath[]::new);