Add per-node search-cluster stats to the indexer adapter

graylog-storage-elasticsearch7/src/main/java/org/graylog/storage/elasticsearch7/ClusterAdapterES7.java

@@ -49,6 +49,7 @@
 import org.graylog2.system.stats.elasticsearch.IndicesStats;
 import org.graylog2.system.stats.elasticsearch.NodeInfo;
 import org.graylog2.system.stats.elasticsearch.NodeOSInfo;
+import org.graylog2.system.stats.elasticsearch.NodeStats;
 import org.graylog2.system.stats.elasticsearch.NodesStats;
 import org.graylog2.system.stats.elasticsearch.ShardStats;
 import org.slf4j.Logger;
@@ -307,6 +308,24 @@ private NodeOSInfo createNodeHostInfo(JsonNode nodesOsJson) {
         );
     }
 
+    @Override
+    public Map<String, NodeStats> nodesStats() {
+        final Request request = new Request("GET", "/_nodes/stats/os,jvm");
+        final JsonNode nodesJson = jsonApi.perform(request, "Couldn't read Elasticsearch nodes stats data!");
+
+        final JsonNode nodes = nodesJson.at("/nodes");
+        return toStream(nodes.fieldNames())
+                .collect(Collectors.toMap(name -> name, name -> createNodeStats(nodes.get(name))));
+    }
+
+    private NodeStats createNodeStats(JsonNode nodeJson) {
+        return new NodeStats(
+                nodeJson.at("/name").asText(),
+                nodeJson.at("/os/cpu/percent").asDouble(-1),
+                nodeJson.at("/jvm/mem/heap_used_percent").asDouble(-1)
+        );
+    }
+
     public <T> Stream<T> toStream(Iterator<T> iterator) {
         return StreamSupport.stream(((Iterable<T>) () -> iterator).spliterator(), false);
     }

graylog-storage-elasticsearch7/src/test/java/org/graylog/storage/elasticsearch7/ClusterAdapterES7Test.java

@@ -29,6 +29,7 @@
 import org.graylog2.indexer.cluster.health.SIUnitParser;
 import org.graylog2.indexer.indices.HealthStatus;
 import org.graylog2.shared.bindings.providers.ObjectMapperProvider;
+import org.graylog2.system.stats.elasticsearch.NodeStats;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -174,6 +175,20 @@ void testDeflectorHealth() {
         assertThat(clusterAdapter.deflectorHealth(Set.of("foo_deflector", "bar_deflector", "baz_deflector"))).contains(HealthStatus.Red);
     }
 
+    @Test
+    void nodesStatsParsesPerNodeCpuAndHeapPercent() throws IOException {
+        when(jsonApi.perform(any(), anyString())).thenReturn(objectMapper.readTree("""
+                {"nodes":{
+                  "nodeId1":{"name":"es01","os":{"cpu":{"percent":42}},"jvm":{"mem":{"heap_used_percent":73}}}
+                }}"""));
+
+        final NodeStats stats = clusterAdapter.nodesStats().get("nodeId1");
+
+        assertThat(stats.name()).isEqualTo("es01");
+        assertThat(stats.cpuPercent()).isEqualTo(42.0);
+        assertThat(stats.jvmHeapUsedPercent()).isEqualTo(73.0);
+    }
+
     private void mockNodesResponse() throws IOException {
         when(jsonApi.perform(any(), anyString()))
                 .thenReturn(objectMapper.readTree(Resources.getResource("nodes-response-without-host-field.json")));

graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ClusterAdapterOS2.java

@@ -50,6 +50,7 @@
 import org.graylog2.system.stats.elasticsearch.IndicesStats;
 import org.graylog2.system.stats.elasticsearch.NodeInfo;
 import org.graylog2.system.stats.elasticsearch.NodeOSInfo;
+import org.graylog2.system.stats.elasticsearch.NodeStats;
 import org.graylog2.system.stats.elasticsearch.NodesStats;
 import org.graylog2.system.stats.elasticsearch.ShardStats;
 import org.slf4j.Logger;
@@ -316,6 +317,24 @@ private NodeOSInfo createNodeHostInfo(JsonNode nodesOsJson) {
         );
     }
 
+    @Override
+    public Map<String, NodeStats> nodesStats() {
+        final Request request = new Request("GET", "/_nodes/stats/os,jvm");
+        final JsonNode nodesJson = jsonApi.perform(request, "Couldn't read Opensearch nodes stats data!");
+
+        final JsonNode nodes = nodesJson.at("/nodes");
+        return toStream(nodes.fieldNames())
+                .collect(Collectors.toMap(name -> name, name -> createNodeStats(nodes.get(name))));
+    }
+
+    private NodeStats createNodeStats(JsonNode nodeJson) {
+        return new NodeStats(
+                nodeJson.at("/name").asText(),
+                nodeJson.at("/os/cpu/percent").asDouble(-1),
+                nodeJson.at("/jvm/mem/heap_used_percent").asDouble(-1)
+        );
+    }
+
     public <T> Stream<T> toStream(Iterator<T> iterator) {
         return StreamSupport.stream(((Iterable<T>) () -> iterator).spliterator(), false);
     }

graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/ClusterAdapterOS2Test.java

@@ -34,6 +34,7 @@
 import org.graylog2.indexer.cluster.health.SIUnitParser;
 import org.graylog2.indexer.indices.HealthStatus;
 import org.graylog2.shared.bindings.providers.ObjectMapperProvider;
+import org.graylog2.system.stats.elasticsearch.NodeStats;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -209,6 +210,34 @@ void testClusterShardAllocation() {
 
 
 
+    @Test
+    void nodesStatsParsesPerNodeCpuAndHeapPercent() throws IOException {
+        when(jsonApi.perform(any(), anyString())).thenReturn(objectMapper.readTree("""
+                {"nodes":{
+                  "nodeId1":{"name":"os01","os":{"cpu":{"percent":42}},"jvm":{"mem":{"heap_used_percent":73}}},
+                  "nodeId2":{"name":"os02","os":{"cpu":{"percent":5}},"jvm":{"mem":{"heap_used_percent":18}}}
+                }}"""));
+
+        final Map<String, NodeStats> stats = clusterAdapter.nodesStats();
+
+        assertThat(stats).hasSize(2);
+        assertThat(stats.get("nodeId1").name()).isEqualTo("os01");
+        assertThat(stats.get("nodeId1").cpuPercent()).isEqualTo(42.0);
+        assertThat(stats.get("nodeId1").jvmHeapUsedPercent()).isEqualTo(73.0);
+        assertThat(stats.get("nodeId2").name()).isEqualTo("os02");
+    }
+
+    @Test
+    void nodesStatsReportsMinusOneForAbsentFields() throws IOException {
+        when(jsonApi.perform(any(), anyString())).thenReturn(objectMapper.readTree("""
+                {"nodes":{"nodeId1":{"name":"os01"}}}"""));
+
+        final NodeStats stats = clusterAdapter.nodesStats().get("nodeId1");
+
+        assertThat(stats.cpuPercent()).isEqualTo(-1.0);
+        assertThat(stats.jvmHeapUsedPercent()).isEqualTo(-1.0);
+    }
+
     private void mockNodesResponse() throws IOException {
         when(jsonApi.perform(any(), anyString()))
                 .thenReturn(objectMapper.readTree(Resources.getResource("nodes-response-without-host-field.json")));

graylog-storage-opensearch3/src/main/java/org/graylog/storage/opensearch3/ClusterAdapterOS.java

@@ -37,6 +37,7 @@
 import org.graylog2.system.stats.elasticsearch.ClusterStats;
 import org.graylog2.system.stats.elasticsearch.IndicesStats;
 import org.graylog2.system.stats.elasticsearch.NodeOSInfo;
+import org.graylog2.system.stats.elasticsearch.NodeStats;
 import org.graylog2.system.stats.elasticsearch.NodesStats;
 import org.graylog2.system.stats.elasticsearch.ShardStats;
 import org.opensearch.client.json.JsonData;
@@ -340,6 +341,26 @@ private NodeOSInfo createNodeHostInfo(JsonNode nodesOsJson) {
         );
     }
 
+    @Override
+    public Map<String, NodeStats> nodesStats() {
+        Request request = Requests.builder()
+                .endpoint("/_nodes/stats/os,jvm")
+                .method("GET")
+                .build();
+        JsonNode json = opensearchClient.performRequest(request, "Couldn't read Opensearch nodes stats data!");
+        JsonNode nodes = json.at("/nodes");
+        return toStream(nodes.fieldNames())
+                .collect(Collectors.toMap(name -> name, name -> createNodeStats(nodes.get(name))));
+    }
+
+    private NodeStats createNodeStats(JsonNode nodeJson) {
+        return new NodeStats(
+                nodeJson.at("/name").asText(),
+                nodeJson.at("/os/cpu/percent").asDouble(-1),
+                nodeJson.at("/jvm/mem/heap_used_percent").asDouble(-1)
+        );
+    }
+
     public <T> Stream<T> toStream(Iterator<T> iterator) {
         return StreamSupport.stream(((Iterable<T>) () -> iterator).spliterator(), false);
     }

graylog-storage-opensearch3/src/test/java/org/graylog/storage/opensearch3/ClusterAdapterOSTest.java

@@ -26,6 +26,7 @@
 import org.graylog2.indexer.cluster.health.NodeShardAllocation;
 import org.graylog2.indexer.cluster.health.SIUnitParser;
 import org.graylog2.indexer.indices.HealthStatus;
+import org.graylog2.system.stats.elasticsearch.NodeStats;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -138,4 +139,21 @@ void testClusterShardAllocation() {
                 .extracting(NodeShardAllocation::shards)
                 .containsExactly(15, 16);
     }
+
+    @Test
+    void nodesStatsParsesPerNodeCpuAndHeapPercent() {
+        final OfficialOpensearchClient statsClient = ServerlessOpenSearchClient.builder()
+                .stubResponse("GET", "/_nodes/stats/os,jvm", """
+                        {"nodes":{
+                          "nodeId1":{"name":"os01","os":{"cpu":{"percent":42}},"jvm":{"mem":{"heap_used_percent":73}}}
+                        }}""")
+                .build();
+        final ClusterAdapterOS adapter = new ClusterAdapterOS(statsClient, Duration.seconds(1));
+
+        final NodeStats stats = adapter.nodesStats().get("nodeId1");
+
+        assertThat(stats.name()).isEqualTo("os01");
+        assertThat(stats.cpuPercent()).isEqualTo(42.0);
+        assertThat(stats.jvmHeapUsedPercent()).isEqualTo(73.0);
+    }
 }

graylog2-server/src/main/java/org/graylog2/indexer/cluster/Cluster.java

@@ -28,11 +28,13 @@
 import org.graylog2.indexer.indices.HealthStatus;
 import org.graylog2.rest.models.system.indexer.responses.ClusterHealth;
 import org.graylog2.system.stats.elasticsearch.ElasticsearchStats;
+import org.graylog2.system.stats.elasticsearch.NodeStats;
 import org.graylog2.system.stats.elasticsearch.ShardStats;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.Arrays;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
@@ -90,6 +92,10 @@ public Set<NodeDiskUsageStats> getDiskUsageStats() {
         return clusterAdapter.diskUsageStats();
     }
 
+    public Map<String, NodeStats> getNodesStats() {
+        return clusterAdapter.nodesStats();
+    }
+
     public ClusterAllocationDiskSettings getClusterAllocationDiskSettings() {
         return clusterAdapter.clusterAllocationDiskSettings();
     }

graylog2-server/src/main/java/org/graylog2/indexer/cluster/ClusterAdapter.java

@@ -26,6 +26,7 @@
 import org.graylog2.system.stats.elasticsearch.ClusterStats;
 import org.graylog2.system.stats.elasticsearch.NodeInfo;
 import org.graylog2.system.stats.elasticsearch.NodeOSInfo;
+import org.graylog2.system.stats.elasticsearch.NodeStats;
 import org.graylog2.system.stats.elasticsearch.ShardStats;
 
 import java.util.Collection;
@@ -64,6 +65,12 @@ public interface ClusterAdapter {
 
     Map<String, NodeOSInfo> nodesHostInfo();
 
+    /**
+     * Live per-node runtime utilization ({@code _nodes/stats/os,jvm}): CPU percent and JVM heap-used percent, keyed
+     * by node id. A single bounded round-trip; the search-cluster health reporters sample and window this on the leader.
+     */
+    Map<String, NodeStats> nodesStats();
+
     ShardStats shardStats();
 
     Optional<HealthStatus> deflectorHealth(Collection<String> indices);

graylog2-server/src/main/java/org/graylog2/system/stats/elasticsearch/NodeStats.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog2.system.stats.elasticsearch;
+
+/**
+ * Live per-node runtime stats of a search-cluster node, read from {@code _nodes/stats/os,jvm}. Distinct from
+ * {@link NodeOSInfo} (static OS facts): these are the sampled utilization percentages the health reporters window.
+ *
+ * @param name                the node's display name (from {@code /name}).
+ * @param cpuPercent          OS CPU utilization {@code 0..100}, or {@code -1} when the source did not report it.
+ * @param jvmHeapUsedPercent  JVM heap used {@code 0..100} (OpenSearch reports the percentage directly), or {@code -1}.
+ */
+public record NodeStats(String name, double cpuPercent, double jvmHeapUsedPercent) {
+}