Add metrics for active worker and active coordinator counts

Author: gertjanal

core/trino-main/src/main/java/io/trino/node/AllNodes.java

@@ -25,7 +25,8 @@ public record AllNodes(
         Set<InternalNode> drainingNodes,
         Set<InternalNode> drainedNodes,
         Set<InternalNode> shuttingDownNodes,
-        Set<InternalNode> activeCoordinators)
+        Set<InternalNode> activeCoordinators,
+        Set<InternalNode> activeWorkers)
 {
     public AllNodes
     {
@@ -35,5 +36,6 @@ public record AllNodes(
         drainingNodes = ImmutableSet.copyOf(requireNonNull(drainingNodes, "drainingNodes is null"));
         shuttingDownNodes = ImmutableSet.copyOf(requireNonNull(shuttingDownNodes, "shuttingDownNodes is null"));
         activeCoordinators = ImmutableSet.copyOf(requireNonNull(activeCoordinators, "activeCoordinators is null"));
+        activeWorkers = ImmutableSet.copyOf(requireNonNull(activeWorkers, "activeWorkers is null"));
     }
 }

core/trino-main/src/main/java/io/trino/node/CoordinatorNodeManager.java

@@ -236,7 +236,11 @@ private synchronized void refreshNodesInternal()
                 .filter(InternalNode::isCoordinator)
                 .collect(toImmutableSet());
 
-        AllNodes allNodes = new AllNodes(activeNodes, inactiveNodes, drainingNodes, drainedNodes, shuttingDownNodes, coordinators);
+        Set<InternalNode> workers = activeNodes.stream()
+                .filter(InternalNode::isWorker)
+                .collect(toImmutableSet());
+
+        AllNodes allNodes = new AllNodes(activeNodes, inactiveNodes, drainingNodes, drainedNodes, shuttingDownNodes, coordinators, workers);
         // only update if all nodes actually changed (note: this does not include the connectors registered with the nodes)
         if (!allNodes.equals(this.allNodes)) {
             // assign allNodes to a local variable for use in the callback below
@@ -284,6 +288,18 @@ public int getShuttingDownNodeCount()
         return getAllNodes().shuttingDownNodes().size();
     }
 
+    @Managed
+    public int getActiveCoordinatorCount()
+    {
+        return getAllNodes().activeCoordinators().size();
+    }
+
+    @Managed
+    public int getActiveWorkerCount()
+    {
+        return getAllNodes().activeWorkers().size();
+    }
+
     @VisibleForTesting
     synchronized Set<InternalNode> getInvalidNodes()
     {

core/trino-main/src/main/java/io/trino/node/InternalNode.java

@@ -41,15 +41,22 @@ public class InternalNode
     private final URI internalUri;
     private final NodeVersion nodeVersion;
     private final boolean coordinator;
+    private final boolean worker;
     private final long longHashCode;
 
     public InternalNode(String nodeIdentifier, URI internalUri, NodeVersion nodeVersion, boolean coordinator)
+    {
+        this(nodeIdentifier, internalUri, nodeVersion, coordinator, !coordinator);
+    }
+
+    public InternalNode(String nodeIdentifier, URI internalUri, NodeVersion nodeVersion, boolean coordinator, boolean worker)
     {
         nodeIdentifier = emptyToNull(nullToEmpty(nodeIdentifier).trim());
         this.nodeIdentifier = requireNonNull(nodeIdentifier, "nodeIdentifier is null or empty");
         this.internalUri = requireNonNull(internalUri, "internalUri is null");
         this.nodeVersion = requireNonNull(nodeVersion, "nodeVersion is null");
         this.coordinator = coordinator;
+        this.worker = worker;
         this.longHashCode = new XxHash64(coordinator ? 1 : 0)
                 .update(nodeIdentifier.getBytes(UTF_8))
                 .update(internalUri.toString().getBytes(UTF_8))
@@ -102,6 +109,12 @@ public boolean isCoordinator()
         return coordinator;
     }
 
+    @Override
+    public boolean isWorker()
+    {
+        return worker;
+    }
+
     public NodeVersion getNodeVersion()
     {
         return nodeVersion;

core/trino-main/src/main/java/io/trino/node/TestingInternalNodeManager.java

@@ -40,7 +40,7 @@
 public class TestingInternalNodeManager
         implements InternalNodeManager
 {
-    public static final InternalNode CURRENT_NODE = new InternalNode("local", URI.create("local://127.0.0.1:8080"), NodeVersion.UNKNOWN, true);
+    public static final InternalNode CURRENT_NODE = new InternalNode("local", URI.create("local://127.0.0.1:8080"), NodeVersion.UNKNOWN, true, true);
 
     private final InternalNode currentNode;
     private final ExecutorService nodeStateEventExecutor;
@@ -63,6 +63,7 @@ public TestingInternalNodeManager(InternalNode currentNode)
                 ImmutableSet.of(),
                 ImmutableSet.of(),
                 ImmutableSet.of(),
+                ImmutableSet.of(currentNode),
                 ImmutableSet.of(currentNode));
         this.nodeStateEventExecutor = newSingleThreadExecutor(daemonThreadsNamed("node-state-events-%s"));
     }
@@ -100,6 +101,9 @@ public synchronized void addNodes(Collection<InternalNode> internalNodes)
                 ImmutableSet.of(),
                 newActiveNodes.stream()
                         .filter(InternalNode::isCoordinator)
+                        .collect(toImmutableSet()),
+                newActiveNodes.stream()
+                        .filter(InternalNode::isWorker)
                         .collect(toImmutableSet())));
     }
 
@@ -119,6 +123,9 @@ public synchronized void removeNode(InternalNode internalNode)
                 ImmutableSet.of(),
                 newActiveNodes.stream()
                         .filter(InternalNode::isCoordinator)
+                        .collect(toImmutableSet()),
+                newActiveNodes.stream()
+                        .filter(InternalNode::isWorker)
                         .collect(toImmutableSet())));
     }
 

core/trino-main/src/main/java/io/trino/server/ServerMainModule.java

@@ -486,13 +486,15 @@ public static InternalNode currentInternalNode(
             HttpServerInfo httpServerInfo,
             NodeVersion nodeVersion,
             ServerConfig serverConfig,
+            NodeSchedulerConfig nodeSchedulerConfig,
             InternalCommunicationConfig internalCommunicationConfig)
     {
         return new InternalNode(
                 nodeInfo.getNodeId(),
                 internalCommunicationConfig.isHttpsRequired() ? httpServerInfo.getHttpsUri() : httpServerInfo.getHttpUri(),
                 nodeVersion,
-                serverConfig.isCoordinator());
+                serverConfig.isCoordinator(),
+                !serverConfig.isCoordinator() || nodeSchedulerConfig.isIncludeCoordinator());
     }
 
     private static class RegisterFunctionBundles

core/trino-spi/pom.xml

@@ -291,6 +291,12 @@
                                     <new>method void io.trino.spi.block.ShortArrayBlockEncoding::&lt;init&gt;(boolean)</new>
                                     <justification>ShortArrayBlockEncoding need to accept a parameter to enable SIMD support</justification>
                                 </item>
+                                <item>
+                                    <ignore>true</ignore>
+                                    <code>java.method.addedToInterface</code>
+                                    <new>method boolean io.trino.spi.Node::isWorker()</new>
+                                    <justification>Added the boolean method isWorker</justification>
+                                </item>
                             </differences>
                         </revapi.differences>
                     </analysisConfiguration>

core/trino-spi/src/main/java/io/trino/spi/Node.java

@@ -24,4 +24,6 @@ public interface Node
     String getVersion();
 
     boolean isCoordinator();
+
+    boolean isWorker();
 }