Reshaped batch size metric response from {pipeline-name}->events->batch_size->[75Percentile, 90Percentile] to {pipeline_name}->batch->event_count->[p75, p90]

andsel · andsel · commit d02881648d8d · 2025-08-13T18:16:48.000+02:00
diff --git a/logstash-core/lib/logstash/api/commands/stats.rb b/logstash-core/lib/logstash/api/commands/stats.rb
@@ -174,6 +174,7 @@ def plugin_stats(stats, plugin_type)
 
           def report(stats, extended_stats = nil, opts = {})
             ret = {
+              :batch => stats[:batch],
               :events => stats[:events],
               :flow => stats[:flow],
               :plugins => {
diff --git a/logstash-core/src/main/java/org/logstash/execution/AbstractPipelineExt.java b/logstash-core/src/main/java/org/logstash/execution/AbstractPipelineExt.java
@@ -290,8 +290,8 @@ private AbstractPipelineExt initialize(final ThreadContext context,
 
 
         // init histogram sample
-        final RubySymbol[] eventsNamespace = buildNamespace(EVENTS_KEY);
-        initOrGetHistogramMetric(context, eventsNamespace, BATCH_SIZE_KEY);
+        final RubySymbol[] batchNamespace = buildNamespace(BATCH_KEY);
+        initOrGetHistogramMetric(context, batchNamespace, BATCH_EVENT_COUNT_KEY);
 
         return this;
     }
@@ -324,8 +324,8 @@ public final IRubyObject openQueue(final ThreadContext context) {
                     new IRubyObject[]{
                             STATS_KEY,
                             PIPELINES_KEY,
-                            pipelineId.convertToString().intern(),
-                            EVENTS_KEY
+                            pipelineId.convertToString().intern()/*,
+                            EVENTS_KEY*/
                     }
                 )
             )
diff --git a/logstash-core/src/main/java/org/logstash/execution/QueueReadClientBase.java b/logstash-core/src/main/java/org/logstash/execution/QueueReadClientBase.java
@@ -42,6 +42,9 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 
+import static org.logstash.instrument.metrics.MetricKeys.BATCH_KEY;
+import static org.logstash.instrument.metrics.MetricKeys.EVENTS_KEY;
+
 /**
  * Common code shared by Persistent and In-Memory queues clients implementation
  * */
@@ -101,11 +104,14 @@ public IRubyObject setEventsMetric(final IRubyObject metric) {
     @JRubyMethod(name = "set_pipeline_metric")
     public IRubyObject setPipelineMetric(final IRubyObject metric) {
         final AbstractNamespacedMetricExt namespacedMetric = (AbstractNamespacedMetricExt) metric;
+        ThreadContext context = metric.getRuntime().getCurrentContext();
+        AbstractNamespacedMetricExt eventsNamespace = namespacedMetric.namespace(context, EVENTS_KEY);
+        AbstractNamespacedMetricExt batchNamespace = namespacedMetric.namespace(context, BATCH_KEY);
         synchronized(namespacedMetric.getMetric()) {
-            pipelineMetricOut = LongCounter.fromRubyBase(namespacedMetric, MetricKeys.OUT_KEY);
-            pipelineMetricFiltered = LongCounter.fromRubyBase(namespacedMetric, MetricKeys.FILTERED_KEY);
-            pipelineMetricTime = TimerMetric.fromRubyBase(namespacedMetric, MetricKeys.DURATION_IN_MILLIS_KEY);
-            pipelineMetricBatch = HistogramMetric.fromRubyBase(namespacedMetric, MetricKeys.BATCH_SIZE_KEY);
+            pipelineMetricOut = LongCounter.fromRubyBase(eventsNamespace, MetricKeys.OUT_KEY);
+            pipelineMetricFiltered = LongCounter.fromRubyBase(eventsNamespace, MetricKeys.FILTERED_KEY);
+            pipelineMetricTime = TimerMetric.fromRubyBase(eventsNamespace, MetricKeys.DURATION_IN_MILLIS_KEY);
+            pipelineMetricBatch = HistogramMetric.fromRubyBase(batchNamespace, MetricKeys.BATCH_EVENT_COUNT_KEY);
         }
         return this;
     }
diff --git a/logstash-core/src/main/java/org/logstash/instrument/metrics/MetricKeys.java b/logstash-core/src/main/java/org/logstash/instrument/metrics/MetricKeys.java
@@ -116,5 +116,7 @@ private MetricKeys() {
 
     public static final RubySymbol WRITES_IN_KEY = RubyUtil.RUBY.newSymbol("writes_in");
 
-    public static final RubySymbol BATCH_SIZE_KEY = RubyUtil.RUBY.newSymbol("batch_size");
+    public static final RubySymbol BATCH_EVENT_COUNT_KEY = RubyUtil.RUBY.newSymbol("event_count");
+
+    public static final RubySymbol BATCH_KEY = RubyUtil.RUBY.newSymbol("batch");
 }
diff --git a/logstash-core/src/main/java/org/logstash/instrument/metrics/histogram/HistogramSnapshot.java b/logstash-core/src/main/java/org/logstash/instrument/metrics/histogram/HistogramSnapshot.java
@@ -1,5 +1,6 @@
 package org.logstash.instrument.metrics.histogram;
 
+import com.fasterxml.jackson.annotation.JsonProperty;
 import org.HdrHistogram.Histogram;
 
 import java.io.Serializable;
@@ -19,10 +20,12 @@ public HistogramSnapshot(Histogram hdrHistogram) {
         percentile90 = hdrHistogram.getValueAtPercentile(90);
     }
 
+    @JsonProperty("p75")
     public double get75Percentile() {
         return percentile75;
     }
 
+    @JsonProperty("p90")
     public double get90Percentile() {
         return percentile90;
     }
diff --git a/logstash-core/src/test/java/org/logstash/ext/JrubyMemoryReadClientExtTest.java b/logstash-core/src/test/java/org/logstash/ext/JrubyMemoryReadClientExtTest.java
@@ -40,6 +40,7 @@
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
+import static org.logstash.instrument.metrics.MetricKeys.BATCH_KEY;
 
 /**
  * Tests for {@link JrubyMemoryReadClientExt}.
@@ -80,7 +81,8 @@ public void givenNonEmptyQueueAndEnabledBatchSizeMetricThenHistogramContainsData
         final QueueBatch batch = client.readBatch();
         assertEquals(1, batch.filteredSize());
 
-        HistogramMetric histogram = HistogramMetric.fromRubyBase(metric, MetricKeys.BATCH_SIZE_KEY);
+        ThreadContext context = metric.getRuntime().getCurrentContext();
+        HistogramMetric histogram = HistogramMetric.fromRubyBase(metric.namespace(context, BATCH_KEY), MetricKeys.BATCH_EVENT_COUNT_KEY);
         HistogramSnapshot metricSnapshot = histogram.getValue();
         assertEquals(1.0, metricSnapshot.get75Percentile(), 0.0001);
         assertEquals(1.0, metricSnapshot.get90Percentile(), 0.0001);
@@ -100,7 +102,8 @@ public void givenNonEmptyQueueAndDisabledBatchSizeMetricThenHistogramIsNotPopula
         final QueueBatch batch = client.readBatch();
         assertEquals(1, batch.filteredSize());
 
-        HistogramMetric histogram = HistogramMetric.fromRubyBase(metric, MetricKeys.BATCH_SIZE_KEY);
+        ThreadContext context = metric.getRuntime().getCurrentContext();
+        HistogramMetric histogram = HistogramMetric.fromRubyBase(metric.namespace(context, BATCH_KEY), MetricKeys.BATCH_EVENT_COUNT_KEY);
         HistogramSnapshot metricSnapshot = histogram.getValue();
         assertEquals(0.0, metricSnapshot.get75Percentile(), 0.0001);
         assertEquals(0.0, metricSnapshot.get90Percentile(), 0.0001);
diff --git a/logstash-core/src/test/java/org/logstash/instrument/metrics/MockNamespacedMetric.java b/logstash-core/src/test/java/org/logstash/instrument/metrics/MockNamespacedMetric.java
@@ -93,7 +93,7 @@ public AbstractMetricExt getMetric() {
 
     @Override
     protected AbstractNamespacedMetricExt createNamespaced(ThreadContext context, IRubyObject name) {
-        return null;
+        return this;
     }
 
     @Override
diff --git a/x-pack/spec/monitoring/schemas/monitoring_document_schema.json b/x-pack/spec/monitoring/schemas/monitoring_document_schema.json
@@ -9,22 +9,26 @@
         "successes": { "type": "number" }
       }
     },
+    "batch": {
+      "type": "object",
+      "properties": {
+        "event_count": {
+          "type": "object",
+          "require": ["p75", "p90"],
+          "properties": {
+            "p75": { "type": "number" },
+            "p90": { "type": "number" }
+          }
+        }
+    },
     "events": {
       "type": "object",
       "required": ["filtered", "in", "duration_in_millis", "out"],
       "properties": {
         "filtered": { "type": "number" },
         "in": { "type": "number" },
         "duration_in_millis": { "type": "number" },
-        "out": { "type": "number" },
-        "batch_size": {
-          "type": "object",
-          "require": ["75Percentile", "90Percentile"],
-          "properties": {
-            "75Percentile": { "type": "number" },
-            "90Percentile": { "type": "number" }
-          }
-        }
+        "out": { "type": "number" }
       }
     }
   },

Original file line number	Diff line number	Diff line change
`@@ -290,8 +290,8 @@ private AbstractPipelineExt initialize(final ThreadContext context,`
`290`	`290`
`291`	`291`
`292`	`292`	`// init histogram sample`
`293`		`- final RubySymbol[] eventsNamespace = buildNamespace(EVENTS_KEY);`
`294`		`- initOrGetHistogramMetric(context, eventsNamespace, BATCH_SIZE_KEY);`
	`293`	`+ final RubySymbol[] batchNamespace = buildNamespace(BATCH_KEY);`
	`294`	`+ initOrGetHistogramMetric(context, batchNamespace, BATCH_EVENT_COUNT_KEY);`
`295`	`295`
`296`	`296`	`return this;`
`297`	`297`	`}`
`@@ -324,8 +324,8 @@ public final IRubyObject openQueue(final ThreadContext context) {`
`324`	`324`	`new IRubyObject[]{`
`325`	`325`	`STATS_KEY,`
`326`	`326`	`PIPELINES_KEY,`
`327`		`- pipelineId.convertToString().intern(),`
`328`		`- EVENTS_KEY`
	`327`	`+ pipelineId.convertToString().intern()/*,`
	`328`	`+ EVENTS_KEY*/`
`329`	`329`	`}`
`330`	`330`	`)`
`331`	`331`	`)`
Original file line number	Diff line number	Diff line change
`@@ -116,5 +116,7 @@ private MetricKeys() {`
`116`	`116`
`117`	`117`	`public static final RubySymbol WRITES_IN_KEY = RubyUtil.RUBY.newSymbol("writes_in");`
`118`	`118`
`119`		`- public static final RubySymbol BATCH_SIZE_KEY = RubyUtil.RUBY.newSymbol("batch_size");`
	`119`	`+ public static final RubySymbol BATCH_EVENT_COUNT_KEY = RubyUtil.RUBY.newSymbol("event_count");`
	`120`	`+`
	`121`	`+ public static final RubySymbol BATCH_KEY = RubyUtil.RUBY.newSymbol("batch");`
`120`	`122`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,6 @@`
`1`	`1`	`package org.logstash.instrument.metrics.histogram;`
`2`	`2`
	`3`	`+import com.fasterxml.jackson.annotation.JsonProperty;`
`3`	`4`	`import org.HdrHistogram.Histogram;`
`4`	`5`
`5`	`6`	`import java.io.Serializable;`
`@@ -19,10 +20,12 @@ public HistogramSnapshot(Histogram hdrHistogram) {`
`19`	`20`	`percentile90 = hdrHistogram.getValueAtPercentile(90);`
`20`	`21`	`}`
`21`	`22`
	`23`	`+ @JsonProperty("p75")`
`22`	`24`	`public double get75Percentile() {`
`23`	`25`	`return percentile75;`
`24`	`26`	`}`
`25`	`27`
	`28`	`+ @JsonProperty("p90")`
`26`	`29`	`public double get90Percentile() {`
`27`	`30`	`return percentile90;`
`28`	`31`	`}`
Original file line number	Diff line number	Diff line change
`@@ -93,7 +93,7 @@ public AbstractMetricExt getMetric() {`
`93`	`93`
`94`	`94`	`@Override`
`95`	`95`	`protected AbstractNamespacedMetricExt createNamespaced(ThreadContext context, IRubyObject name) {`
`96`		`- return null;`
	`96`	`+ return this;`
`97`	`97`	`}`
`98`	`98`
`99`	`99`	`@Override`