Create a new monitor for node-level write load (#131560)

Plugs a basic WriteLoadConstraintMonitor
into the code and sets it up with access to
information it will need to implement ES-11992.

Closes ES-11991

Author: DiannaHohensee

server/src/main/java/org/elasticsearch/cluster/routing/allocation/WriteLoadConstraintMonitor.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.cluster.routing.allocation;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.cluster.ClusterInfo;
+import org.elasticsearch.cluster.ClusterInfoService;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.routing.RerouteService;
+import org.elasticsearch.common.Priority;
+import org.elasticsearch.common.settings.ClusterSettings;
+import org.elasticsearch.gateway.GatewayService;
+
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+/**
+ * Monitors the node-level write thread pool usage across the cluster and initiates (coming soon) a rebalancing round (via
+ * {@link RerouteService#reroute}) whenever a node crosses the node-level write load thresholds.
+ *
+ * TODO (ES-11992): implement
+ */
+public class WriteLoadConstraintMonitor {
+    private static final Logger logger = LogManager.getLogger(WriteLoadConstraintMonitor.class);
+    private final WriteLoadConstraintSettings writeLoadConstraintSettings;
+    private final Supplier<ClusterState> clusterStateSupplier;
+    private final LongSupplier currentTimeMillisSupplier;
+    private final RerouteService rerouteService;
+
+    public WriteLoadConstraintMonitor(
+        ClusterSettings clusterSettings,
+        LongSupplier currentTimeMillisSupplier,
+        Supplier<ClusterState> clusterStateSupplier,
+        RerouteService rerouteService
+    ) {
+        this.writeLoadConstraintSettings = new WriteLoadConstraintSettings(clusterSettings);
+        this.clusterStateSupplier = clusterStateSupplier;
+        this.currentTimeMillisSupplier = currentTimeMillisSupplier;
+        this.rerouteService = rerouteService;
+    }
+
+    /**
+     * Receives a copy of the latest {@link ClusterInfo} whenever the {@link ClusterInfoService} collects it. Processes the new
+     * {@link org.elasticsearch.cluster.NodeUsageStatsForThreadPools} and initiates rebalancing, via reroute, if a node in the cluster
+     * exceeds thread pool usage thresholds.
+     */
+    public void onNewInfo(ClusterInfo clusterInfo) {
+        final ClusterState state = clusterStateSupplier.get();
+        if (state.blocks().hasGlobalBlock(GatewayService.STATE_NOT_RECOVERED_BLOCK)) {
+            logger.debug("skipping monitor as the cluster state is not recovered yet");
+            return;
+        }
+
+        if (writeLoadConstraintSettings.getWriteLoadConstraintEnabled() == WriteLoadConstraintSettings.WriteLoadDeciderStatus.DISABLED) {
+            logger.trace("skipping monitor because the write load decider is disabled");
+            return;
+        }
+
+        logger.trace("processing new cluster info");
+
+        boolean reroute = false;
+        String explanation = "";
+        final long currentTimeMillis = currentTimeMillisSupplier.getAsLong();
+
+        // TODO (ES-11992): implement
+
+        if (reroute) {
+            logger.debug("rerouting shards: [{}]", explanation);
+            rerouteService.reroute("disk threshold monitor", Priority.NORMAL, ActionListener.wrap(ignored -> {
+                final var reroutedClusterState = clusterStateSupplier.get();
+
+                // TODO (ES-11992): implement
+
+            }, e -> logger.debug("reroute failed", e)));
+        } else {
+            logger.trace("no reroute required");
+        }
+    }
+}

server/src/main/java/org/elasticsearch/cluster/routing/allocation/WriteLoadConstraintSettings.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.routing.RerouteService;
 import org.elasticsearch.common.Priority;
+import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.unit.RatioValue;
 import org.elasticsearch.core.TimeValue;
@@ -98,4 +99,28 @@ public enum WriteLoadDeciderStatus {
         Setting.Property.Dynamic,
         Setting.Property.NodeScope
     );
+
+    WriteLoadDeciderStatus writeLoadDeciderStatus;
+    TimeValue writeLoadDeciderRerouteIntervalSetting;
+
+    WriteLoadConstraintSettings(ClusterSettings clusterSettings) {
+        clusterSettings.initializeAndWatch(WRITE_LOAD_DECIDER_ENABLED_SETTING, this::setWriteLoadConstraintEnabled);
+        clusterSettings.initializeAndWatch(WRITE_LOAD_DECIDER_REROUTE_INTERVAL_SETTING, this::setWriteLoadDeciderRerouteIntervalSetting);
+    };
+
+    private void setWriteLoadConstraintEnabled(WriteLoadDeciderStatus status) {
+        this.writeLoadDeciderStatus = status;
+    }
+
+    public WriteLoadDeciderStatus getWriteLoadConstraintEnabled() {
+        return this.writeLoadDeciderStatus;
+    }
+
+    public TimeValue getWriteLoadDeciderRerouteIntervalSetting() {
+        return this.writeLoadDeciderRerouteIntervalSetting;
+    }
+
+    private void setWriteLoadDeciderRerouteIntervalSetting(TimeValue timeValue) {
+        this.writeLoadDeciderRerouteIntervalSetting = timeValue;
+    }
 }

server/src/main/java/org/elasticsearch/node/NodeConstruction.java

@@ -60,6 +60,7 @@
 import org.elasticsearch.cluster.routing.RerouteService;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.routing.allocation.DiskThresholdMonitor;
+import org.elasticsearch.cluster.routing.allocation.WriteLoadConstraintMonitor;
 import org.elasticsearch.cluster.routing.allocation.WriteLoadForecaster;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.cluster.version.CompatibilityVersions;
@@ -785,6 +786,15 @@ private void construct(
             )::onNewInfo
         );
 
+        clusterInfoService.addListener(
+            new WriteLoadConstraintMonitor(
+                clusterService.getClusterSettings(),
+                threadPool.relativeTimeInMillisSupplier(),
+                clusterService::state,
+                rerouteService
+            )::onNewInfo
+        );
+
         IndicesModule indicesModule = new IndicesModule(pluginsService.filterPlugins(MapperPlugin.class).toList());
         modules.add(indicesModule);
 

server/src/test/java/org/elasticsearch/cluster/InternalClusterInfoServiceSchedulingTests.java

@@ -21,11 +21,14 @@
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeUtils;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.cluster.routing.RerouteService;
+import org.elasticsearch.cluster.routing.allocation.WriteLoadConstraintMonitor;
 import org.elasticsearch.cluster.routing.allocation.WriteLoadConstraintSettings;
 import org.elasticsearch.cluster.service.ClusterApplierService;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.cluster.service.FakeThreadPoolMasterService;
 import org.elasticsearch.cluster.service.MasterService;
+import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.DeterministicTaskQueue;
@@ -95,6 +98,18 @@ protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
             mockEstimatedHeapUsageCollector,
             mockNodeUsageStatsForThreadPoolsCollector
         );
+        final WriteLoadConstraintMonitor usageMonitor = spy(
+            new WriteLoadConstraintMonitor(
+                clusterService.getClusterSettings(),
+                threadPool.relativeTimeInMillisSupplier(),
+                clusterService::state,
+                new RerouteService() {
+                    @Override
+                    public void reroute(String reason, Priority priority, ActionListener<Void> listener) {}
+                }
+            )
+        );
+        clusterInfoService.addListener(usageMonitor::onNewInfo);
         clusterService.addListener(clusterInfoService);
         clusterInfoService.addListener(ignored -> {});
 
@@ -132,13 +147,15 @@ protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
         for (int i = 0; i < 3; i++) {
             Mockito.clearInvocations(mockEstimatedHeapUsageCollector);
             Mockito.clearInvocations(mockNodeUsageStatsForThreadPoolsCollector);
+            Mockito.clearInvocations(usageMonitor);
             final int initialRequestCount = client.requestCount;
             final long duration = INTERNAL_CLUSTER_INFO_UPDATE_INTERVAL_SETTING.get(settings).millis();
             runFor(deterministicTaskQueue, duration);
             deterministicTaskQueue.runAllRunnableTasks();
             assertThat(client.requestCount, equalTo(initialRequestCount + 2)); // should have run two client requests per interval
             verify(mockEstimatedHeapUsageCollector).collectClusterHeapUsage(any()); // Should poll for heap usage once per interval
             verify(mockNodeUsageStatsForThreadPoolsCollector).collectUsageStats(any());
+            verify(usageMonitor).onNewInfo(any());
         }
 
         final AtomicBoolean failMaster2 = new AtomicBoolean();