Skip to content

Commit 4be5440

Browse files
talevytalevy
authored
[7.x] Add ingest info to Cluster Stats (#48485) (#48661)
* Add ingest info to Cluster Stats (#48485) This commit enhances the ClusterStatsNodes response to include global processor usage stats on a per-processor basis. example output: ``` ... "processor_stats": { "gsub": { "count": 0, "failed": 0 "current": 0 "time_in_millis": 0 }, "script": { "count": 0, "failed": 0 "current": 0, "time_in_millis": 0 } } ... ``` The purpose for this enhancement is to make it easier to collect stats on how specific processors are being used across the cluster beyond the current per-node usage statistics that currently exist in node stats. Closes #46146. * fix BWC of ingest stats The introduction of processor types into IngestStats had a bug. It was set to `null` and set as the key to the map. This would throw a NPE. This commit resolves this by setting all the processor types from previous versions that are not serializing it out to `_NOT_AVAILABLE`.
1 parent d0ead68 commit 4be5440

File tree

9 files changed

+181
-15
lines changed

9 files changed

+181
-15
lines changed

‎docs/reference/cluster/stats.asciidoc‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -227,6 +227,12 @@ The API returns the following response:
227227
},
228228
...
229229
],
230+
"ingest": {
231+
"number_of_pipelines" : 1,
232+
"processor_stats": {
233+
...
234+
}
235+
},
230236
"network_types": {
231237
...
232238
},
@@ -244,6 +250,7 @@ The API returns the following response:
244250
// TESTRESPONSE[s/"plugins": \[[^\]]*\]/"plugins": $body.$_path/]
245251
// TESTRESPONSE[s/"network_types": \{[^\}]*\}/"network_types": $body.$_path/]
246252
// TESTRESPONSE[s/"discovery_types": \{[^\}]*\}/"discovery_types": $body.$_path/]
253+
// TESTRESPONSE[s/"processor_stats": \{[^\}]*\}/"processor_stats": $body.$_path/]
247254
// TESTRESPONSE[s/"count": \{[^\}]*\}/"count": $body.$_path/]
248255
// TESTRESPONSE[s/"packaging_types": \[[^\]]*\]/"packaging_types": $body.$_path/]
249256
// TESTRESPONSE[s/: true|false/: $body.$_path/]

‎server/src/main/java/org/elasticsearch/action/admin/cluster/stats/ClusterStatsNodes.java‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,9 @@
4949
import java.util.List;
5050
import java.util.Map;
5151
import java.util.Set;
52+
import java.util.SortedMap;
5253
import java.util.TreeMap;
54+
import java.util.concurrent.TimeUnit;
5355
import java.util.concurrent.atomic.AtomicInteger;
5456

5557
public class ClusterStatsNodes implements ToXContentFragment {
@@ -64,6 +66,7 @@ public class ClusterStatsNodes implements ToXContentFragment {
6466
private final NetworkTypes networkTypes;
6567
private final DiscoveryTypes discoveryTypes;
6668
private final PackagingTypes packagingTypes;
69+
private final IngestStats ingestStats;
6770

6871
ClusterStatsNodes(List<ClusterStatsNodeResponse> nodeResponses) {
6972
this.versions = new HashSet<>();
@@ -97,6 +100,7 @@ public class ClusterStatsNodes implements ToXContentFragment {
97100
this.networkTypes = new NetworkTypes(nodeInfos);
98101
this.discoveryTypes = new DiscoveryTypes(nodeInfos);
99102
this.packagingTypes = new PackagingTypes(nodeInfos);
103+
this.ingestStats = new IngestStats(nodeStats);
100104
}
101105

102106
public Counts getCounts() {
@@ -178,6 +182,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
178182
discoveryTypes.toXContent(builder, params);
179183

180184
packagingTypes.toXContent(builder, params);
185+
186+
ingestStats.toXContent(builder, params);
187+
181188
return builder;
182189
}
183190

@@ -690,4 +697,68 @@ public XContentBuilder toXContent(final XContentBuilder builder, final Params pa
690697

691698
}
692699

700+
static class IngestStats implements ToXContentFragment {
701+
702+
final int pipelineCount;
703+
final SortedMap<String, long[]> stats;
704+
705+
IngestStats(final List<NodeStats> nodeStats) {
706+
Set<String> pipelineIds = new HashSet<>();
707+
SortedMap<String, long[]> stats = new TreeMap<>();
708+
for (NodeStats nodeStat : nodeStats) {
709+
if (nodeStat.getIngestStats() != null) {
710+
for (Map.Entry<String,
711+
List<org.elasticsearch.ingest.IngestStats.ProcessorStat>> processorStats : nodeStat.getIngestStats()
712+
.getProcessorStats().entrySet()) {
713+
pipelineIds.add(processorStats.getKey());
714+
for (org.elasticsearch.ingest.IngestStats.ProcessorStat stat : processorStats.getValue()) {
715+
stats.compute(stat.getType(), (k, v) -> {
716+
org.elasticsearch.ingest.IngestStats.Stats nodeIngestStats = stat.getStats();
717+
if (v == null) {
718+
return new long[] {
719+
nodeIngestStats.getIngestCount(),
720+
nodeIngestStats.getIngestFailedCount(),
721+
nodeIngestStats.getIngestCurrent(),
722+
nodeIngestStats.getIngestTimeInMillis()
723+
};
724+
} else {
725+
v[0] += nodeIngestStats.getIngestCount();
726+
v[1] += nodeIngestStats.getIngestFailedCount();
727+
v[2] += nodeIngestStats.getIngestCurrent();
728+
v[3] += nodeIngestStats.getIngestTimeInMillis();
729+
return v;
730+
}
731+
});
732+
}
733+
}
734+
}
735+
}
736+
this.pipelineCount = pipelineIds.size();
737+
this.stats = Collections.unmodifiableSortedMap(stats);
738+
}
739+
740+
@Override
741+
public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
742+
builder.startObject("ingest");
743+
{
744+
builder.field("number_of_pipelines", pipelineCount);
745+
builder.startObject("processor_stats");
746+
for (Map.Entry<String, long[]> stat : stats.entrySet()) {
747+
long[] statValues = stat.getValue();
748+
builder.startObject(stat.getKey());
749+
builder.field("count", statValues[0]);
750+
builder.field("failed", statValues[1]);
751+
builder.field("current", statValues[2]);
752+
builder.humanReadableField("time_in_millis", "time",
753+
new TimeValue(statValues[3], TimeUnit.MILLISECONDS));
754+
builder.endObject();
755+
}
756+
builder.endObject();
757+
}
758+
builder.endObject();
759+
return builder;
760+
}
761+
762+
}
763+
693764
}

‎server/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -94,7 +94,7 @@ protected ClusterStatsNodeResponse newNodeResponse(StreamInput in) throws IOExce
9494
protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeRequest) {
9595
NodeInfo nodeInfo = nodeService.info(true, true, false, true, false, true, false, true, false, false);
9696
NodeStats nodeStats = nodeService.stats(CommonStatsFlags.NONE,
97-
true, true, true, false, true, false, false, false, false, false, false, false);
97+
true, true, true, false, true, false, false, false, false, false, true, false);
9898
List<ShardStats> shardsStats = new ArrayList<>();
9999
for (IndexService indexService : indicesService) {
100100
for (IndexShard indexShard : indexService) {

‎server/src/main/java/org/elasticsearch/ingest/IngestService.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -413,7 +413,7 @@ public IngestStats stats() {
413413
processorMetrics.forEach(t -> {
414414
Processor processor = t.v1();
415415
IngestMetric processorMetric = t.v2();
416-
statsBuilder.addProcessorMetrics(id, getProcessorName(processor), processorMetric);
416+
statsBuilder.addProcessorMetrics(id, getProcessorName(processor), processor.getType(), processorMetric);
417417
});
418418
});
419419
return statsBuilder.build();

‎server/src/main/java/org/elasticsearch/ingest/IngestStats.java‎

Lines changed: 20 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -69,8 +69,12 @@ public IngestStats(StreamInput in) throws IOException {
6969
List<ProcessorStat> processorStatsPerPipeline = new ArrayList<>(processorsSize);
7070
for (int j = 0; j < processorsSize; j++) {
7171
String processorName = in.readString();
72+
String processorType = "_NOT_AVAILABLE";
73+
if (in.getVersion().onOrAfter(Version.V_7_6_0)) {
74+
processorType = in.readString();
75+
}
7276
Stats processorStat = new Stats(in);
73-
processorStatsPerPipeline.add(new ProcessorStat(processorName, processorStat));
77+
processorStatsPerPipeline.add(new ProcessorStat(processorName, processorType, processorStat));
7478
}
7579
this.processorStats.put(pipelineId, processorStatsPerPipeline);
7680
}
@@ -92,6 +96,9 @@ public void writeTo(StreamOutput out) throws IOException {
9296
out.writeVInt(processorStatsForPipeline.size());
9397
for (ProcessorStat processorStat : processorStatsForPipeline) {
9498
out.writeString(processorStat.getName());
99+
if (out.getVersion().onOrAfter(Version.V_7_6_0)) {
100+
out.writeString(processorStat.getType());
101+
}
95102
processorStat.getStats().writeTo(out);
96103
}
97104
}
@@ -115,9 +122,12 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
115122
for (ProcessorStat processorStat : processorStatsForPipeline) {
116123
builder.startObject();
117124
builder.startObject(processorStat.getName());
125+
builder.field("type", processorStat.getType());
126+
builder.startObject("stats");
118127
processorStat.getStats().toXContent(builder, params);
119128
builder.endObject();
120129
builder.endObject();
130+
builder.endObject();
121131
}
122132
}
123133
builder.endArray();
@@ -229,9 +239,9 @@ Builder addPipelineMetrics(String pipelineId, IngestMetric pipelineMetric) {
229239
return this;
230240
}
231241

232-
Builder addProcessorMetrics(String pipelineId, String processorName, IngestMetric metric) {
242+
Builder addProcessorMetrics(String pipelineId, String processorName, String processorType, IngestMetric metric) {
233243
this.processorStats.computeIfAbsent(pipelineId, k -> new ArrayList<>())
234-
.add(new ProcessorStat(processorName, metric.createStats()));
244+
.add(new ProcessorStat(processorName, processorType, metric.createStats()));
235245
return this;
236246
}
237247

@@ -267,17 +277,23 @@ public Stats getStats() {
267277
*/
268278
public static class ProcessorStat {
269279
private final String name;
280+
private final String type;
270281
private final Stats stats;
271282

272-
public ProcessorStat(String name, Stats stats) {
283+
public ProcessorStat(String name, String type, Stats stats) {
273284
this.name = name;
285+
this.type = type;
274286
this.stats = stats;
275287
}
276288

277289
public String getName() {
278290
return name;
279291
}
280292

293+
public String getType() {
294+
return type;
295+
}
296+
281297
public Stats getStats() {
282298
return stats;
283299
}

‎server/src/test/java/org/elasticsearch/action/admin/cluster/node/stats/NodeStatsTests.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -315,7 +315,7 @@ public void testSerialization() throws IOException {
315315
}
316316
}
317317

318-
private static NodeStats createNodeStats() {
318+
public static NodeStats createNodeStats() {
319319
DiscoveryNode node = new DiscoveryNode("test_node", buildNewFakeTransportAddress(),
320320
emptyMap(), emptySet(), VersionUtils.randomVersion(random()));
321321
OsStats osStats = null;
@@ -456,7 +456,8 @@ private static NodeStats createNodeStats() {
456456
for (int j =0; j < numProcessors;j++) {
457457
IngestStats.Stats processorStats = new IngestStats.Stats
458458
(randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong());
459-
processorPerPipeline.add(new IngestStats.ProcessorStat(randomAlphaOfLengthBetween(3, 10), processorStats));
459+
processorPerPipeline.add(new IngestStats.ProcessorStat(randomAlphaOfLengthBetween(3, 10),
460+
randomAlphaOfLengthBetween(3, 10), processorStats));
460461
}
461462
ingestProcessorStats.put(pipelineId,processorPerPipeline);
462463
}

‎server/src/test/java/org/elasticsearch/action/admin/cluster/stats/ClusterStatsNodesTests.java‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,18 +20,26 @@
2020
package org.elasticsearch.action.admin.cluster.stats;
2121

2222
import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
23+
import org.elasticsearch.action.admin.cluster.node.stats.NodeStats;
24+
import org.elasticsearch.action.admin.cluster.node.stats.NodeStatsTests;
2325
import org.elasticsearch.cluster.node.DiscoveryNode;
2426
import org.elasticsearch.common.network.NetworkModule;
2527
import org.elasticsearch.common.settings.Settings;
2628
import org.elasticsearch.common.xcontent.XContentType;
2729
import org.elasticsearch.test.ESTestCase;
2830

2931
import java.util.Arrays;
32+
import java.util.Collections;
33+
import java.util.Iterator;
3034
import java.util.List;
35+
import java.util.Map;
36+
import java.util.SortedMap;
37+
import java.util.TreeMap;
3138

3239
import static java.util.Collections.emptyList;
3340
import static java.util.Collections.singletonList;
3441
import static org.elasticsearch.common.xcontent.XContentHelper.toXContent;
42+
import static org.hamcrest.Matchers.equalTo;
3543

3644
public class ClusterStatsNodesTests extends ESTestCase {
3745

@@ -59,6 +67,41 @@ public void testNetworkTypesToXContent() throws Exception {
5967
+ "}", toXContent(stats, XContentType.JSON, randomBoolean()).utf8ToString());
6068
}
6169

70+
public void testIngestStats() throws Exception {
71+
NodeStats nodeStats = randomValueOtherThanMany(n -> n.getIngestStats() == null, NodeStatsTests::createNodeStats);
72+
73+
SortedMap<String, long[]> processorStats = new TreeMap<>();
74+
nodeStats.getIngestStats().getProcessorStats().values().forEach(l -> l.forEach(s -> processorStats.put(s.getType(),
75+
new long[] { s.getStats().getIngestCount(), s.getStats().getIngestFailedCount(),
76+
s.getStats().getIngestCurrent(), s.getStats().getIngestTimeInMillis()})));
77+
ClusterStatsNodes.IngestStats stats = new ClusterStatsNodes.IngestStats(Collections.singletonList(nodeStats));
78+
assertThat(stats.pipelineCount, equalTo(nodeStats.getIngestStats().getProcessorStats().size()));
79+
String processorStatsString = "{";
80+
Iterator<Map.Entry<String, long[]>> iter = processorStats.entrySet().iterator();
81+
while (iter.hasNext()) {
82+
Map.Entry<String, long[]> entry = iter.next();
83+
long[] statValues = entry.getValue();
84+
long count = statValues[0];
85+
long failedCount = statValues[1];
86+
long current = statValues[2];
87+
long timeInMillis = statValues[3];
88+
processorStatsString += "\"" + entry.getKey() + "\":{\"count\":" + count
89+
+ ",\"failed\":" + failedCount
90+
+ ",\"current\":" + current
91+
+ ",\"time_in_millis\":" + timeInMillis
92+
+ "}";
93+
if (iter.hasNext()) {
94+
processorStatsString += ",";
95+
}
96+
}
97+
processorStatsString += "}";
98+
assertThat(toXContent(stats, XContentType.JSON, false).utf8ToString(), equalTo(
99+
"{\"ingest\":{"
100+
+ "\"number_of_pipelines\":" + stats.pipelineCount + ","
101+
+ "\"processor_stats\":" + processorStatsString
102+
+ "}}"));
103+
}
104+
62105
private static NodeInfo createNodeInfo(String nodeId, String transportType, String httpType) {
63106
Settings.Builder settings = Settings.builder();
64107
if (transportType != null) {

‎server/src/test/java/org/elasticsearch/ingest/IngestStatsTests.java‎

Lines changed: 30 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,7 @@ public void testSerialization() throws IOException {
4242
Map<String, List<IngestStats.ProcessorStat>> processorStats = createProcessorStats(pipelineStats);
4343
IngestStats ingestStats = new IngestStats(totalStats, pipelineStats, processorStats);
4444
IngestStats serializedStats = serialize(ingestStats);
45-
assertIngestStats(ingestStats, serializedStats, true);
45+
assertIngestStats(ingestStats, serializedStats, true, true);
4646
}
4747

4848
public void testReadLegacyStream() throws IOException {
@@ -63,7 +63,24 @@ public void testReadLegacyStream() throws IOException {
6363
in.setVersion(VersionUtils.getPreviousVersion(Version.V_6_5_0));
6464
IngestStats serializedStats = new IngestStats(in);
6565
IngestStats expectedStats = new IngestStats(totalStats, pipelineStats, Collections.emptyMap());
66-
assertIngestStats(expectedStats, serializedStats, false);
66+
assertIngestStats(expectedStats, serializedStats, false, true);
67+
}
68+
69+
public void testBWCIngestProcessorTypeStats() throws IOException {
70+
IngestStats.Stats totalStats = new IngestStats.Stats(50, 100, 200, 300);
71+
List<IngestStats.PipelineStat> pipelineStats = createPipelineStats();
72+
Map<String, List<IngestStats.ProcessorStat>> processorStats = createProcessorStats(pipelineStats);
73+
IngestStats expectedIngestStats = new IngestStats(totalStats, pipelineStats, processorStats);
74+
75+
//legacy output logic
76+
BytesStreamOutput out = new BytesStreamOutput();
77+
out.setVersion(VersionUtils.getPreviousVersion(Version.V_7_6_0));
78+
expectedIngestStats.writeTo(out);
79+
80+
StreamInput in = out.bytes().streamInput();
81+
in.setVersion(VersionUtils.getPreviousVersion(Version.V_7_6_0));
82+
IngestStats serializedStats = new IngestStats(in);
83+
assertIngestStats(expectedIngestStats, serializedStats, true, false);
6784
}
6885

6986
private List<IngestStats.PipelineStat> createPipelineStats() {
@@ -75,9 +92,10 @@ private List<IngestStats.PipelineStat> createPipelineStats() {
7592

7693
private Map<String, List<IngestStats.ProcessorStat>> createProcessorStats(List<IngestStats.PipelineStat> pipelineStats){
7794
assert(pipelineStats.size() >= 2);
78-
IngestStats.ProcessorStat processor1Stat = new IngestStats.ProcessorStat("processor1", new IngestStats.Stats(1, 1, 1, 1));
79-
IngestStats.ProcessorStat processor2Stat = new IngestStats.ProcessorStat("processor2", new IngestStats.Stats(2, 2, 2, 2));
80-
IngestStats.ProcessorStat processor3Stat = new IngestStats.ProcessorStat("processor3", new IngestStats.Stats(47, 97, 197, 297));
95+
IngestStats.ProcessorStat processor1Stat = new IngestStats.ProcessorStat("processor1", "type", new IngestStats.Stats(1, 1, 1, 1));
96+
IngestStats.ProcessorStat processor2Stat = new IngestStats.ProcessorStat("processor2", "type", new IngestStats.Stats(2, 2, 2, 2));
97+
IngestStats.ProcessorStat processor3Stat = new IngestStats.ProcessorStat("processor3", "type",
98+
new IngestStats.Stats(47, 97, 197, 297));
8199
//pipeline1 -> processor1,processor2; pipeline2 -> processor3
82100
return MapBuilder.<String, List<IngestStats.ProcessorStat>>newMapBuilder()
83101
.put(pipelineStats.get(0).getPipelineId(), Stream.of(processor1Stat, processor2Stat).collect(Collectors.toList()))
@@ -92,7 +110,8 @@ private IngestStats serialize(IngestStats stats) throws IOException {
92110
return new IngestStats(in);
93111
}
94112

95-
private void assertIngestStats(IngestStats ingestStats, IngestStats serializedStats, boolean expectProcessors){
113+
private void assertIngestStats(IngestStats ingestStats, IngestStats serializedStats, boolean expectProcessors,
114+
boolean expectProcessorTypes){
96115
assertNotSame(ingestStats, serializedStats);
97116
assertNotSame(ingestStats.getTotalStats(), serializedStats.getTotalStats());
98117
assertNotSame(ingestStats.getPipelineStats(), serializedStats.getPipelineStats());
@@ -114,6 +133,11 @@ private void assertIngestStats(IngestStats ingestStats, IngestStats serializedSt
114133
for (IngestStats.ProcessorStat serializedProcessorStat : serializedProcessorStats) {
115134
IngestStats.ProcessorStat ps = it.next();
116135
assertEquals(ps.getName(), serializedProcessorStat.getName());
136+
if (expectProcessorTypes) {
137+
assertEquals(ps.getType(), serializedProcessorStat.getType());
138+
} else {
139+
assertEquals("_NOT_AVAILABLE", serializedProcessorStat.getType());
140+
}
117141
assertStats(ps.getStats(), serializedProcessorStat.getStats());
118142
}
119143
assertFalse(it.hasNext());

‎x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/collector/cluster/ClusterStatsMonitoringDocTests.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -535,7 +535,11 @@ public void testToXContent() throws IOException {
535535
+ "\"type\":\"docker\","
536536
+ "\"count\":1"
537537
+ "}"
538-
+ "]"
538+
+ "],"
539+
+ "\"ingest\":{"
540+
+ "\"number_of_pipelines\":0,"
541+
+ "\"processor_stats\":{}"
542+
+ "}"
539543
+ "}"
540544
+ "},"
541545
+ "\"cluster_state\":{"

0 commit comments

Comments
 (0)