Warn on slow metadata performance (#50956)

Has the new cluster state storage layer emit warnings in case metadata performance is very
slow.

Relates #48701

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

@@ -51,7 +51,7 @@ public DetachClusterCommand() {
     @Override
     protected void processNodePaths(Terminal terminal, Path[] dataPaths, int nodeLockId, OptionSet options, Environment env)
         throws IOException {
-        final PersistedClusterStateService persistedClusterStateService = createPersistedClusterStateService(dataPaths);
+        final PersistedClusterStateService persistedClusterStateService = createPersistedClusterStateService(env.settings(), dataPaths);
 
         terminal.println(Terminal.Verbosity.VERBOSE, "Loading cluster state");
         final ClusterState oldClusterState = loadTermAndClusterState(persistedClusterStateService, env).v2();

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

@@ -32,6 +32,8 @@
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.settings.ClusterSettings;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.env.Environment;
@@ -76,14 +78,15 @@ public ElasticsearchNodeCommand(String description) {
                 .withRequiredArg().ofType(Integer.class);
     }
 
-    public static PersistedClusterStateService createPersistedClusterStateService(Path[] dataPaths) throws IOException {
+    public static PersistedClusterStateService createPersistedClusterStateService(Settings settings, 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);
+        return new PersistedClusterStateService(dataPaths, nodeId, namedXContentRegistry, BigArrays.NON_RECYCLING_INSTANCE,
+            new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), () -> 0L, true);
     }
 
     public static ClusterState clusterState(Environment environment, PersistedClusterStateService.OnDiskState onDiskState) {

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

@@ -62,7 +62,7 @@ protected void processNodePaths(Terminal terminal, Path[] dataPaths, int nodeLoc
             throw new UserException(ExitCodes.USAGE, "Must supply at least one setting to remove");
         }
 
-        final PersistedClusterStateService persistedClusterStateService = createPersistedClusterStateService(dataPaths);
+        final PersistedClusterStateService persistedClusterStateService = createPersistedClusterStateService(env.settings(), dataPaths);
 
         terminal.println(Terminal.Verbosity.VERBOSE, "Loading cluster state");
         final Tuple<Long, ClusterState> termAndClusterState = loadTermAndClusterState(persistedClusterStateService, env);

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

@@ -79,7 +79,7 @@ protected boolean validateBeforeLock(Terminal terminal, Environment env) {
 
     protected void processNodePaths(Terminal terminal, Path[] dataPaths, int nodeLockId, OptionSet options, Environment env)
         throws IOException {
-        final PersistedClusterStateService persistedClusterStateService = createPersistedClusterStateService(dataPaths);
+        final PersistedClusterStateService persistedClusterStateService = createPersistedClusterStateService(env.settings(), dataPaths);
 
         final Tuple<Long, ClusterState> state = loadTermAndClusterState(persistedClusterStateService, env);
         final ClusterState oldClusterState = state.v2();

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

@@ -79,7 +79,7 @@
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.gateway.DanglingIndicesState;
 import org.elasticsearch.gateway.GatewayService;
-import org.elasticsearch.gateway.IncrementalClusterStateWriter;
+import org.elasticsearch.gateway.PersistedClusterStateService;
 import org.elasticsearch.http.HttpTransportSettings;
 import org.elasticsearch.index.IndexModule;
 import org.elasticsearch.index.IndexSettings;
@@ -109,9 +109,9 @@
 import org.elasticsearch.search.aggregations.MultiBucketConsumerService;
 import org.elasticsearch.search.fetch.subphase.highlight.FastVectorHighlighter;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.ProxyConnectionStrategy;
 import org.elasticsearch.transport.RemoteClusterService;
 import org.elasticsearch.transport.RemoteConnectionStrategy;
-import org.elasticsearch.transport.ProxyConnectionStrategy;
 import org.elasticsearch.transport.SniffConnectionStrategy;
 import org.elasticsearch.transport.TransportSettings;
 import org.elasticsearch.watcher.ResourceWatcherService;
@@ -252,7 +252,7 @@ public void apply(Settings value, Settings current, Settings previous) {
             GatewayService.RECOVER_AFTER_MASTER_NODES_SETTING,
             GatewayService.RECOVER_AFTER_NODES_SETTING,
             GatewayService.RECOVER_AFTER_TIME_SETTING,
-            IncrementalClusterStateWriter.SLOW_WRITE_LOGGING_THRESHOLD,
+            PersistedClusterStateService.SLOW_WRITE_LOGGING_THRESHOLD,
             NetworkModule.HTTP_DEFAULT_TYPE_SETTING,
             NetworkModule.TRANSPORT_DEFAULT_TYPE_SETTING,
             NetworkModule.HTTP_TYPE_SETTING,

server/src/main/java/org/elasticsearch/env/NodeRepurposeCommand.java

@@ -96,7 +96,7 @@ private void processNoMasterNoDataNode(Terminal terminal, Path[] dataPaths, Envi
 
         Set<Path> indexPaths = uniqueParentPaths(shardDataPaths, indexMetaDataPaths);
 
-        final PersistedClusterStateService persistedClusterStateService = createPersistedClusterStateService(dataPaths);
+        final PersistedClusterStateService persistedClusterStateService = createPersistedClusterStateService(env.settings(), dataPaths);
 
         final MetaData metaData = loadClusterState(terminal, env, persistedClusterStateService).metaData();
         if (indexPaths.isEmpty() && metaData.indices().isEmpty()) {
@@ -134,7 +134,7 @@ private void processMasterNoDataNode(Terminal terminal, Path[] dataPaths, Enviro
             return;
         }
 
-        final PersistedClusterStateService persistedClusterStateService = createPersistedClusterStateService(dataPaths);
+        final PersistedClusterStateService persistedClusterStateService = createPersistedClusterStateService(env.settings(), dataPaths);
 
         final MetaData metaData = loadClusterState(terminal, env, persistedClusterStateService).metaData();
 

server/src/main/java/org/elasticsearch/gateway/IncrementalClusterStateWriter.java

@@ -27,7 +27,6 @@
 import org.elasticsearch.cluster.routing.RoutingNode;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.common.settings.ClusterSettings;
-import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.Index;
@@ -47,9 +46,6 @@ public class IncrementalClusterStateWriter {
 
     private static final Logger logger = LogManager.getLogger(IncrementalClusterStateWriter.class);
 
-    public static final Setting<TimeValue> SLOW_WRITE_LOGGING_THRESHOLD = Setting.timeSetting("gateway.slow_write_logging_threshold",
-        TimeValue.timeValueSeconds(10), TimeValue.ZERO, Setting.Property.NodeScope, Setting.Property.Dynamic);
-
     private final MetaStateService metaStateService;
 
     // We call updateClusterState on the (unique) cluster applier thread so there's no need to synchronize access to these fields.
@@ -67,8 +63,9 @@ public class IncrementalClusterStateWriter {
         this.previousClusterState = clusterState;
         this.relativeTimeMillisSupplier = relativeTimeMillisSupplier;
         this.incrementalWrite = false;
-        this.slowWriteLoggingThreshold = SLOW_WRITE_LOGGING_THRESHOLD.get(settings);
-        clusterSettings.addSettingsUpdateConsumer(SLOW_WRITE_LOGGING_THRESHOLD, this::setSlowWriteLoggingThreshold);
+        this.slowWriteLoggingThreshold = PersistedClusterStateService.SLOW_WRITE_LOGGING_THRESHOLD.get(settings);
+        clusterSettings.addSettingsUpdateConsumer(PersistedClusterStateService.SLOW_WRITE_LOGGING_THRESHOLD,
+            this::setSlowWriteLoggingThreshold);
     }
 
     private void setSlowWriteLoggingThreshold(TimeValue slowWriteLoggingThreshold) {