Publish max queue latency metric

Author: nicktindall

server/src/main/java/org/elasticsearch/action/admin/cluster/node/usage/TransportNodeUsageStatsForThreadPoolsAction.java

@@ -9,27 +9,30 @@
 
 package org.elasticsearch.action.admin.cluster.node.usage;
 
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.FailedNodeException;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.nodes.TransportNodesAction;
 import org.elasticsearch.cluster.NodeUsageStatsForThreadPools;
 import org.elasticsearch.cluster.NodeUsageStatsForThreadPools.ThreadPoolUsageStats;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.routing.allocation.AllocationDeciderMetrics;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.util.concurrent.TaskExecutionTimeTrackingEsThreadPoolExecutor;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.tasks.Task;
+import org.elasticsearch.telemetry.metric.LongWithAttributes;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * Collects some thread pool stats from each data node for purposes of shard allocation balancing. The specific stats are defined in
@@ -42,20 +45,21 @@ public class TransportNodeUsageStatsForThreadPoolsAction extends TransportNodesA
     NodeUsageStatsForThreadPoolsAction.NodeResponse,
     Void> {
 
-    private static final Logger logger = LogManager.getLogger(TransportNodeUsageStatsForThreadPoolsAction.class);
-
     public static final String NAME = "internal:monitor/thread_pool/stats";
     public static final ActionType<NodeUsageStatsForThreadPoolsAction.Response> TYPE = new ActionType<>(NAME);
+    private static final int NO_VALUE = -1;
 
     private final ThreadPool threadPool;
     private final ClusterService clusterService;
+    private final AtomicLong lastMaxQueueLatencyMillis = new AtomicLong(NO_VALUE);
 
     @Inject
     public TransportNodeUsageStatsForThreadPoolsAction(
         ThreadPool threadPool,
         ClusterService clusterService,
         TransportService transportService,
-        ActionFilters actionFilters
+        ActionFilters actionFilters,
+        AllocationDeciderMetrics allocationDeciderMetrics
     ) {
         super(
             NAME,
@@ -67,6 +71,7 @@ public TransportNodeUsageStatsForThreadPoolsAction(
         );
         this.threadPool = threadPool;
         this.clusterService = clusterService;
+        allocationDeciderMetrics.registerWriteLoadDeciderMaxLatencyGauge(this::getMaxQueueLatencyMetric);
     }
 
     @Override
@@ -99,15 +104,17 @@ protected NodeUsageStatsForThreadPoolsAction.NodeResponse nodeOperation(
         assert writeExecutor instanceof TaskExecutionTimeTrackingEsThreadPoolExecutor;
         var trackingForWriteExecutor = (TaskExecutionTimeTrackingEsThreadPoolExecutor) writeExecutor;
 
+        long maxQueueLatencyMillis = Math.max(
+            trackingForWriteExecutor.getMaxQueueLatencyMillisSinceLastPollAndReset(),
+            trackingForWriteExecutor.peekMaxQueueLatencyInQueueMillis()
+        );
+        lastMaxQueueLatencyMillis.set(maxQueueLatencyMillis);
         ThreadPoolUsageStats threadPoolUsageStats = new ThreadPoolUsageStats(
             trackingForWriteExecutor.getMaximumPoolSize(),
             (float) trackingForWriteExecutor.pollUtilization(
                 TaskExecutionTimeTrackingEsThreadPoolExecutor.UtilizationTrackingPurpose.ALLOCATION
             ),
-            Math.max(
-                trackingForWriteExecutor.getMaxQueueLatencyMillisSinceLastPollAndReset(),
-                trackingForWriteExecutor.peekMaxQueueLatencyInQueueMillis()
-            )
+            maxQueueLatencyMillis
         );
 
         Map<String, ThreadPoolUsageStats> perThreadPool = new HashMap<>();
@@ -117,4 +124,13 @@ protected NodeUsageStatsForThreadPoolsAction.NodeResponse nodeOperation(
             new NodeUsageStatsForThreadPools(localNode.getId(), perThreadPool)
         );
     }
+
+    private Collection<LongWithAttributes> getMaxQueueLatencyMetric() {
+        long maxQueueLatencyValue = lastMaxQueueLatencyMillis.getAndSet(NO_VALUE);
+        if (maxQueueLatencyValue != NO_VALUE) {
+            return Set.of(new LongWithAttributes(maxQueueLatencyValue));
+        } else {
+            return Set.of();
+        }
+    }
 }

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

@@ -0,0 +1,39 @@
+/*
+ * 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.elasticsearch.telemetry.metric.LongWithAttributes;
+import org.elasticsearch.telemetry.metric.MeterRegistry;
+
+import java.util.Collection;
+import java.util.function.Supplier;
+
+/**
+ * A place where metrics related to allocation deciders can live
+ */
+public class AllocationDeciderMetrics {
+
+    public static final String WRITE_LOAD_DECIDER_MAX_LATENCY_VALUE = "es.allocation.decider.write_load.max_latency_value.current";
+
+    private final MeterRegistry meterRegistry;
+
+    public AllocationDeciderMetrics(MeterRegistry meterRegistry) {
+        this.meterRegistry = meterRegistry;
+    }
+
+    public void registerWriteLoadDeciderMaxLatencyGauge(Supplier<Collection<LongWithAttributes>> maxLatencySupplier) {
+        meterRegistry.registerLongsGauge(
+            WRITE_LOAD_DECIDER_MAX_LATENCY_VALUE,
+            "max latency for write load decider",
+            "unit",
+            maxLatencySupplier
+        );
+    }
+}

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

@@ -58,6 +58,7 @@
 import org.elasticsearch.cluster.project.ProjectResolverFactory;
 import org.elasticsearch.cluster.routing.BatchedRerouteService;
 import org.elasticsearch.cluster.routing.RerouteService;
+import org.elasticsearch.cluster.routing.allocation.AllocationDeciderMetrics;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.routing.allocation.DiskThresholdMonitor;
 import org.elasticsearch.cluster.routing.allocation.WriteLoadConstraintMonitor;
@@ -1234,6 +1235,8 @@ public Map<String, String> queryFields() {
 
         final ShutdownPrepareService shutdownPrepareService = new ShutdownPrepareService(settings, httpServerTransport, terminationHandler);
 
+        final AllocationDeciderMetrics allocationDeciderMetrics = new AllocationDeciderMetrics(telemetryProvider.getMeterRegistry());
+
         modules.add(loadPersistentTasksService(settingsModule, clusterService, threadPool, clusterModule.getIndexNameExpressionResolver()));
 
         modules.add(
@@ -1324,6 +1327,7 @@ public Map<String, String> queryFields() {
             b.bind(ShutdownPrepareService.class).toInstance(shutdownPrepareService);
             b.bind(OnlinePrewarmingService.class).toInstance(onlinePrewarmingService);
             b.bind(MergeMetrics.class).toInstance(mergeMetrics);
+            b.bind(AllocationDeciderMetrics.class).toInstance(allocationDeciderMetrics);
         });
 
         if (ReadinessService.enabled(environment)) {