From f7b9cc73f8ee48cfec615cccd603563f4e10d20c Mon Sep 17 00:00:00 2001 From: patrickmann Date: Wed, 8 Jul 2026 13:44:21 +0200 Subject: [PATCH] Add per-node search-cluster stats to the indexer adapter Adds ClusterAdapter.nodesStats() which reads _nodes/stats/os,jvm in a single round-trip and exposes per-node os.cpu.percent and jvm.mem.heap_used_percent through a new NodeStats value object. Implemented across the OpenSearch 2/3 and Elasticsearch 7 adapters, with a Cluster.getNodesStats() facade. Supports the Enterprise System Overview search-cluster health reporters (Graylog2/graylog-plugin-enterprise#14661). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../elasticsearch7/ClusterAdapterES7.java | 19 ++++++++++++ .../elasticsearch7/ClusterAdapterES7Test.java | 15 ++++++++++ .../opensearch2/ClusterAdapterOS2.java | 19 ++++++++++++ .../opensearch2/ClusterAdapterOS2Test.java | 29 +++++++++++++++++++ .../storage/opensearch3/ClusterAdapterOS.java | 21 ++++++++++++++ .../opensearch3/ClusterAdapterOSTest.java | 18 ++++++++++++ .../org/graylog2/indexer/cluster/Cluster.java | 6 ++++ .../indexer/cluster/ClusterAdapter.java | 7 +++++ .../system/stats/elasticsearch/NodeStats.java | 28 ++++++++++++++++++ 9 files changed, 162 insertions(+) create mode 100644 graylog2-server/src/main/java/org/graylog2/system/stats/elasticsearch/NodeStats.java diff --git a/graylog-storage-elasticsearch7/src/main/java/org/graylog/storage/elasticsearch7/ClusterAdapterES7.java b/graylog-storage-elasticsearch7/src/main/java/org/graylog/storage/elasticsearch7/ClusterAdapterES7.java index 634e68f36b9f..915445d3b8e2 100644 --- a/graylog-storage-elasticsearch7/src/main/java/org/graylog/storage/elasticsearch7/ClusterAdapterES7.java +++ b/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 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 Stream toStream(Iterator iterator) { return StreamSupport.stream(((Iterable) () -> iterator).spliterator(), false); } diff --git a/graylog-storage-elasticsearch7/src/test/java/org/graylog/storage/elasticsearch7/ClusterAdapterES7Test.java b/graylog-storage-elasticsearch7/src/test/java/org/graylog/storage/elasticsearch7/ClusterAdapterES7Test.java index 6510fd969e31..b0f498a6db4f 100644 --- a/graylog-storage-elasticsearch7/src/test/java/org/graylog/storage/elasticsearch7/ClusterAdapterES7Test.java +++ b/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"))); diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ClusterAdapterOS2.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ClusterAdapterOS2.java index 710da232dec6..34e1e0f3fccb 100644 --- a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ClusterAdapterOS2.java +++ b/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 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 Stream toStream(Iterator iterator) { return StreamSupport.stream(((Iterable) () -> iterator).spliterator(), false); } diff --git a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/ClusterAdapterOS2Test.java b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/ClusterAdapterOS2Test.java index 1117545866c6..92fe3e6445fa 100644 --- a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/ClusterAdapterOS2Test.java +++ b/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 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"))); diff --git a/graylog-storage-opensearch3/src/main/java/org/graylog/storage/opensearch3/ClusterAdapterOS.java b/graylog-storage-opensearch3/src/main/java/org/graylog/storage/opensearch3/ClusterAdapterOS.java index 683f0ddc656d..a1a68f4e4bb4 100644 --- a/graylog-storage-opensearch3/src/main/java/org/graylog/storage/opensearch3/ClusterAdapterOS.java +++ b/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 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 Stream toStream(Iterator iterator) { return StreamSupport.stream(((Iterable) () -> iterator).spliterator(), false); } diff --git a/graylog-storage-opensearch3/src/test/java/org/graylog/storage/opensearch3/ClusterAdapterOSTest.java b/graylog-storage-opensearch3/src/test/java/org/graylog/storage/opensearch3/ClusterAdapterOSTest.java index 007e404b32ad..9a04e0cc0782 100644 --- a/graylog-storage-opensearch3/src/test/java/org/graylog/storage/opensearch3/ClusterAdapterOSTest.java +++ b/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); + } } diff --git a/graylog2-server/src/main/java/org/graylog2/indexer/cluster/Cluster.java b/graylog2-server/src/main/java/org/graylog2/indexer/cluster/Cluster.java index 94781c2e8106..89fe1dc8c9b3 100644 --- a/graylog2-server/src/main/java/org/graylog2/indexer/cluster/Cluster.java +++ b/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 getDiskUsageStats() { return clusterAdapter.diskUsageStats(); } + public Map getNodesStats() { + return clusterAdapter.nodesStats(); + } + public ClusterAllocationDiskSettings getClusterAllocationDiskSettings() { return clusterAdapter.clusterAllocationDiskSettings(); } diff --git a/graylog2-server/src/main/java/org/graylog2/indexer/cluster/ClusterAdapter.java b/graylog2-server/src/main/java/org/graylog2/indexer/cluster/ClusterAdapter.java index 11c923dcc7aa..b7357304c3e5 100644 --- a/graylog2-server/src/main/java/org/graylog2/indexer/cluster/ClusterAdapter.java +++ b/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 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 nodesStats(); + ShardStats shardStats(); Optional deflectorHealth(Collection indices); diff --git a/graylog2-server/src/main/java/org/graylog2/system/stats/elasticsearch/NodeStats.java b/graylog2-server/src/main/java/org/graylog2/system/stats/elasticsearch/NodeStats.java new file mode 100644 index 000000000000..16aa49473fa6 --- /dev/null +++ b/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 + * . + */ +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) { +}