Skip to content

Commit d3ad2b8

Browse files
patrickmannpatrickmann
committed
Add per-node search-cluster stats to the indexer adapter
Adds ClusterAdapter.nodesStats() reading _nodes/stats/os,jvm for per-node CPU and heap usage (new NodeStats DTO), implemented for ES7/OS2/OS3 and exposed via Cluster.getNodesStats(). Backs the search-cluster health reporters in graylog-plugin-enterprise#14799.
1 parent d16a509 commit d3ad2b8

9 files changed

Lines changed: 162 additions & 0 deletions

File tree

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

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@
4949
import org.graylog2.system.stats.elasticsearch.IndicesStats;
5050
import org.graylog2.system.stats.elasticsearch.NodeInfo;
5151
import org.graylog2.system.stats.elasticsearch.NodeOSInfo;
52+
import org.graylog2.system.stats.elasticsearch.NodeStats;
5253
import org.graylog2.system.stats.elasticsearch.NodesStats;
5354
import org.graylog2.system.stats.elasticsearch.ShardStats;
5455
import org.slf4j.Logger;
@@ -307,6 +308,24 @@ private NodeOSInfo createNodeHostInfo(JsonNode nodesOsJson) {
307308
);
308309
}
309310

311+
@Override
312+
public Map<String, NodeStats> nodesStats() {
313+
final Request request = new Request("GET", "/_nodes/stats/os,jvm");
314+
final JsonNode nodesJson = jsonApi.perform(request, "Couldn't read Elasticsearch nodes stats data!");
315+
316+
final JsonNode nodes = nodesJson.at("/nodes");
317+
return toStream(nodes.fieldNames())
318+
.collect(Collectors.toMap(name -> name, name -> createNodeStats(nodes.get(name))));
319+
}
320+
321+
private NodeStats createNodeStats(JsonNode nodeJson) {
322+
return new NodeStats(
323+
nodeJson.at("/name").asText(),
324+
nodeJson.at("/os/cpu/percent").asDouble(-1),
325+
nodeJson.at("/jvm/mem/heap_used_percent").asDouble(-1)
326+
);
327+
}
328+
310329
public <T> Stream<T> toStream(Iterator<T> iterator) {
311330
return StreamSupport.stream(((Iterable<T>) () -> iterator).spliterator(), false);
312331
}

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

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import org.graylog2.indexer.cluster.health.SIUnitParser;
3030
import org.graylog2.indexer.indices.HealthStatus;
3131
import org.graylog2.shared.bindings.providers.ObjectMapperProvider;
32+
import org.graylog2.system.stats.elasticsearch.NodeStats;
3233
import org.junit.jupiter.api.BeforeEach;
3334
import org.junit.jupiter.api.Test;
3435

@@ -174,6 +175,20 @@ void testDeflectorHealth() {
174175
assertThat(clusterAdapter.deflectorHealth(Set.of("foo_deflector", "bar_deflector", "baz_deflector"))).contains(HealthStatus.Red);
175176
}
176177

178+
@Test
179+
void nodesStatsParsesPerNodeCpuAndHeapPercent() throws IOException {
180+
when(jsonApi.perform(any(), anyString())).thenReturn(objectMapper.readTree("""
181+
{"nodes":{
182+
"nodeId1":{"name":"es01","os":{"cpu":{"percent":42}},"jvm":{"mem":{"heap_used_percent":73}}}
183+
}}"""));
184+
185+
final NodeStats stats = clusterAdapter.nodesStats().get("nodeId1");
186+
187+
assertThat(stats.name()).isEqualTo("es01");
188+
assertThat(stats.cpuPercent()).isEqualTo(42.0);
189+
assertThat(stats.jvmHeapUsedPercent()).isEqualTo(73.0);
190+
}
191+
177192
private void mockNodesResponse() throws IOException {
178193
when(jsonApi.perform(any(), anyString()))
179194
.thenReturn(objectMapper.readTree(Resources.getResource("nodes-response-without-host-field.json")));

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

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@
5050
import org.graylog2.system.stats.elasticsearch.IndicesStats;
5151
import org.graylog2.system.stats.elasticsearch.NodeInfo;
5252
import org.graylog2.system.stats.elasticsearch.NodeOSInfo;
53+
import org.graylog2.system.stats.elasticsearch.NodeStats;
5354
import org.graylog2.system.stats.elasticsearch.NodesStats;
5455
import org.graylog2.system.stats.elasticsearch.ShardStats;
5556
import org.slf4j.Logger;
@@ -316,6 +317,24 @@ private NodeOSInfo createNodeHostInfo(JsonNode nodesOsJson) {
316317
);
317318
}
318319

320+
@Override
321+
public Map<String, NodeStats> nodesStats() {
322+
final Request request = new Request("GET", "/_nodes/stats/os,jvm");
323+
final JsonNode nodesJson = jsonApi.perform(request, "Couldn't read Opensearch nodes stats data!");
324+
325+
final JsonNode nodes = nodesJson.at("/nodes");
326+
return toStream(nodes.fieldNames())
327+
.collect(Collectors.toMap(name -> name, name -> createNodeStats(nodes.get(name))));
328+
}
329+
330+
private NodeStats createNodeStats(JsonNode nodeJson) {
331+
return new NodeStats(
332+
nodeJson.at("/name").asText(),
333+
nodeJson.at("/os/cpu/percent").asDouble(-1),
334+
nodeJson.at("/jvm/mem/heap_used_percent").asDouble(-1)
335+
);
336+
}
337+
319338
public <T> Stream<T> toStream(Iterator<T> iterator) {
320339
return StreamSupport.stream(((Iterable<T>) () -> iterator).spliterator(), false);
321340
}

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

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
import org.graylog2.indexer.cluster.health.SIUnitParser;
3535
import org.graylog2.indexer.indices.HealthStatus;
3636
import org.graylog2.shared.bindings.providers.ObjectMapperProvider;
37+
import org.graylog2.system.stats.elasticsearch.NodeStats;
3738
import org.junit.jupiter.api.BeforeEach;
3839
import org.junit.jupiter.api.Test;
3940

@@ -209,6 +210,34 @@ void testClusterShardAllocation() {
209210

210211

211212

213+
@Test
214+
void nodesStatsParsesPerNodeCpuAndHeapPercent() throws IOException {
215+
when(jsonApi.perform(any(), anyString())).thenReturn(objectMapper.readTree("""
216+
{"nodes":{
217+
"nodeId1":{"name":"os01","os":{"cpu":{"percent":42}},"jvm":{"mem":{"heap_used_percent":73}}},
218+
"nodeId2":{"name":"os02","os":{"cpu":{"percent":5}},"jvm":{"mem":{"heap_used_percent":18}}}
219+
}}"""));
220+
221+
final Map<String, NodeStats> stats = clusterAdapter.nodesStats();
222+
223+
assertThat(stats).hasSize(2);
224+
assertThat(stats.get("nodeId1").name()).isEqualTo("os01");
225+
assertThat(stats.get("nodeId1").cpuPercent()).isEqualTo(42.0);
226+
assertThat(stats.get("nodeId1").jvmHeapUsedPercent()).isEqualTo(73.0);
227+
assertThat(stats.get("nodeId2").name()).isEqualTo("os02");
228+
}
229+
230+
@Test
231+
void nodesStatsReportsMinusOneForAbsentFields() throws IOException {
232+
when(jsonApi.perform(any(), anyString())).thenReturn(objectMapper.readTree("""
233+
{"nodes":{"nodeId1":{"name":"os01"}}}"""));
234+
235+
final NodeStats stats = clusterAdapter.nodesStats().get("nodeId1");
236+
237+
assertThat(stats.cpuPercent()).isEqualTo(-1.0);
238+
assertThat(stats.jvmHeapUsedPercent()).isEqualTo(-1.0);
239+
}
240+
212241
private void mockNodesResponse() throws IOException {
213242
when(jsonApi.perform(any(), anyString()))
214243
.thenReturn(objectMapper.readTree(Resources.getResource("nodes-response-without-host-field.json")));

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

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import org.graylog2.system.stats.elasticsearch.ClusterStats;
3838
import org.graylog2.system.stats.elasticsearch.IndicesStats;
3939
import org.graylog2.system.stats.elasticsearch.NodeOSInfo;
40+
import org.graylog2.system.stats.elasticsearch.NodeStats;
4041
import org.graylog2.system.stats.elasticsearch.NodesStats;
4142
import org.graylog2.system.stats.elasticsearch.ShardStats;
4243
import org.opensearch.client.json.JsonData;
@@ -340,6 +341,26 @@ private NodeOSInfo createNodeHostInfo(JsonNode nodesOsJson) {
340341
);
341342
}
342343

344+
@Override
345+
public Map<String, NodeStats> nodesStats() {
346+
Request request = Requests.builder()
347+
.endpoint("/_nodes/stats/os,jvm")
348+
.method("GET")
349+
.build();
350+
JsonNode json = opensearchClient.performRequest(request, "Couldn't read Opensearch nodes stats data!");
351+
JsonNode nodes = json.at("/nodes");
352+
return toStream(nodes.fieldNames())
353+
.collect(Collectors.toMap(name -> name, name -> createNodeStats(nodes.get(name))));
354+
}
355+
356+
private NodeStats createNodeStats(JsonNode nodeJson) {
357+
return new NodeStats(
358+
nodeJson.at("/name").asText(),
359+
nodeJson.at("/os/cpu/percent").asDouble(-1),
360+
nodeJson.at("/jvm/mem/heap_used_percent").asDouble(-1)
361+
);
362+
}
363+
343364
public <T> Stream<T> toStream(Iterator<T> iterator) {
344365
return StreamSupport.stream(((Iterable<T>) () -> iterator).spliterator(), false);
345366
}

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

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.graylog2.indexer.cluster.health.NodeShardAllocation;
2727
import org.graylog2.indexer.cluster.health.SIUnitParser;
2828
import org.graylog2.indexer.indices.HealthStatus;
29+
import org.graylog2.system.stats.elasticsearch.NodeStats;
2930
import org.junit.jupiter.api.BeforeEach;
3031
import org.junit.jupiter.api.Test;
3132

@@ -138,4 +139,21 @@ void testClusterShardAllocation() {
138139
.extracting(NodeShardAllocation::shards)
139140
.containsExactly(15, 16);
140141
}
142+
143+
@Test
144+
void nodesStatsParsesPerNodeCpuAndHeapPercent() {
145+
final OfficialOpensearchClient statsClient = ServerlessOpenSearchClient.builder()
146+
.stubResponse("GET", "/_nodes/stats/os,jvm", """
147+
{"nodes":{
148+
"nodeId1":{"name":"os01","os":{"cpu":{"percent":42}},"jvm":{"mem":{"heap_used_percent":73}}}
149+
}}""")
150+
.build();
151+
final ClusterAdapterOS adapter = new ClusterAdapterOS(statsClient, Duration.seconds(1));
152+
153+
final NodeStats stats = adapter.nodesStats().get("nodeId1");
154+
155+
assertThat(stats.name()).isEqualTo("os01");
156+
assertThat(stats.cpuPercent()).isEqualTo(42.0);
157+
assertThat(stats.jvmHeapUsedPercent()).isEqualTo(73.0);
158+
}
141159
}

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,11 +28,13 @@
2828
import org.graylog2.indexer.indices.HealthStatus;
2929
import org.graylog2.rest.models.system.indexer.responses.ClusterHealth;
3030
import org.graylog2.system.stats.elasticsearch.ElasticsearchStats;
31+
import org.graylog2.system.stats.elasticsearch.NodeStats;
3132
import org.graylog2.system.stats.elasticsearch.ShardStats;
3233
import org.slf4j.Logger;
3334
import org.slf4j.LoggerFactory;
3435

3536
import java.util.Arrays;
37+
import java.util.Map;
3638
import java.util.Optional;
3739
import java.util.Set;
3840
import java.util.concurrent.CountDownLatch;
@@ -90,6 +92,10 @@ public Set<NodeDiskUsageStats> getDiskUsageStats() {
9092
return clusterAdapter.diskUsageStats();
9193
}
9294

95+
public Map<String, NodeStats> getNodesStats() {
96+
return clusterAdapter.nodesStats();
97+
}
98+
9399
public ClusterAllocationDiskSettings getClusterAllocationDiskSettings() {
94100
return clusterAdapter.clusterAllocationDiskSettings();
95101
}

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

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.graylog2.system.stats.elasticsearch.ClusterStats;
2727
import org.graylog2.system.stats.elasticsearch.NodeInfo;
2828
import org.graylog2.system.stats.elasticsearch.NodeOSInfo;
29+
import org.graylog2.system.stats.elasticsearch.NodeStats;
2930
import org.graylog2.system.stats.elasticsearch.ShardStats;
3031

3132
import java.util.Collection;
@@ -64,6 +65,12 @@ public interface ClusterAdapter {
6465

6566
Map<String, NodeOSInfo> nodesHostInfo();
6667

68+
/**
69+
* Live per-node runtime utilization ({@code _nodes/stats/os,jvm}): CPU percent and JVM heap-used percent, keyed
70+
* by node id. A single bounded round-trip; the search-cluster health reporters sample and window this on the leader.
71+
*/
72+
Map<String, NodeStats> nodesStats();
73+
6774
ShardStats shardStats();
6875

6976
Optional<HealthStatus> deflectorHealth(Collection<String> indices);

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

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

0 commit comments

Comments
 (0)