First idea in adding histogram metric

andsel · andsel · commit 3f17f116a887 · 2025-08-06T15:03:27.000+02:00
diff --git a/logstash-core/build.gradle b/logstash-core/build.gradle
@@ -245,6 +245,7 @@ dependencies {
         exclude group: 'com.google.guava', module: 'guava'
     }
     implementation 'org.javassist:javassist:3.30.2-GA'
+    implementation 'org.hdrhistogram:HdrHistogram:2.2.2'
     testImplementation "org.apache.logging.log4j:log4j-core:${log4jVersion}:tests"
     testImplementation 'org.hamcrest:hamcrest:2.2'
     testImplementation 'org.hamcrest:hamcrest-library:2.2'
diff --git a/logstash-core/lib/logstash/instrument/metric_type.rb b/logstash-core/lib/logstash/instrument/metric_type.rb
@@ -29,6 +29,7 @@ def self.create(type, namespaces, key)
       when :gauge   then return org.logstash.instrument.metrics.gauge.LazyDelegatingGauge.new(key.to_s)
       when :uptime  then return org.logstash.instrument.metrics.UptimeMetric.new(key.to_s)
       when :timer   then return org.logstash.instrument.metrics.timer.TimerMetric::create(key.to_s)
+      when :histogram then return org.logstash.instrument.metrics.histogram.HistogramMetric.new(key.to_s)
       else fail NameError, "Unknown Metric Type `#{type}`"
       end
     end
diff --git a/logstash-core/src/main/java/org/logstash/execution/AbstractPipelineExt.java b/logstash-core/src/main/java/org/logstash/execution/AbstractPipelineExt.java
@@ -92,6 +92,7 @@
 import org.logstash.instrument.metrics.MetricType;
 import org.logstash.instrument.metrics.NullMetricExt;
 import org.logstash.instrument.metrics.UpScaledMetric;
+import org.logstash.instrument.metrics.histogram.HistogramMetric;
 import org.logstash.instrument.metrics.timer.TimerMetric;
 import org.logstash.instrument.metrics.UptimeMetric;
 import org.logstash.instrument.metrics.counter.LongCounter;
@@ -286,6 +287,12 @@ private AbstractPipelineExt initialize(final ThreadContext context,
         } catch (InvalidIRException iirex) {
             throw new IllegalArgumentException(iirex);
         }
+
+
+        // init histogram sample
+        final RubySymbol[] eventsNamespace = buildNamespace(EVENTS_KEY);
+        initOrGetHistogramMetric(context, eventsNamespace, BATCH_SIZE_KEY);
+
         return this;
     }
 
@@ -640,6 +647,16 @@ private TimerMetric initOrGetTimerMetric(final ThreadContext context,
         return retrievedMetric.toJava(TimerMetric.class);
     }
 
+    private HistogramMetric initOrGetHistogramMetric(final ThreadContext context,
+                                                    final RubySymbol[] subPipelineNamespacePath,
+                                                    final RubySymbol metricName) {
+        final IRubyObject collector = this.metric.collector(context);
+        final IRubyObject fullNamespace = pipelineNamespacedPath(subPipelineNamespacePath);
+
+        final IRubyObject retrievedMetric = collector.callMethod(context, "get", new IRubyObject[]{fullNamespace, metricName, context.runtime.newSymbol("histogram")});
+        return retrievedMetric.toJava(HistogramMetric.class);
+    }
+
     private Optional<NumberGauge> initOrGetNumberGaugeMetric(final ThreadContext context,
                                                              final RubySymbol[] subPipelineNamespacePath,
                                                              final RubySymbol metricName) {
diff --git a/logstash-core/src/main/java/org/logstash/execution/QueueReadClientBase.java b/logstash-core/src/main/java/org/logstash/execution/QueueReadClientBase.java
@@ -34,6 +34,7 @@
 import org.logstash.RubyUtil;
 import org.logstash.instrument.metrics.AbstractNamespacedMetricExt;
 import org.logstash.instrument.metrics.MetricKeys;
+import org.logstash.instrument.metrics.histogram.HistogramMetric;
 import org.logstash.instrument.metrics.timer.TimerMetric;
 import org.logstash.instrument.metrics.counter.LongCounter;
 
@@ -60,6 +61,7 @@ public abstract class QueueReadClientBase extends RubyObject implements QueueRea
     private transient LongCounter pipelineMetricOut;
     private transient LongCounter pipelineMetricFiltered;
     private transient TimerMetric pipelineMetricTime;
+    private transient HistogramMetric pipelineMetricBatch;
 
     protected QueueReadClientBase(final Ruby runtime, final RubyClass metaClass) {
         super(runtime, metaClass);
@@ -90,6 +92,7 @@ public IRubyObject setPipelineMetric(final IRubyObject metric) {
             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);
         }
         return this;
     }
@@ -193,6 +196,7 @@ public void startMetrics(QueueBatch batch) {
         // JTODO getId has been deprecated in JDK 19, when JDK 21 is the target version use threadId() instead
         long threadId = Thread.currentThread().getId();
         inflightBatches.put(threadId, batch);
+        pipelineMetricBatch.update(batch.filteredSize());
     }
 
     @Override
diff --git a/logstash-core/src/main/java/org/logstash/instrument/metrics/AbstractNamespacedMetricExt.java b/logstash-core/src/main/java/org/logstash/instrument/metrics/AbstractNamespacedMetricExt.java
@@ -53,6 +53,11 @@ public IRubyObject timer(final ThreadContext context, final IRubyObject key) {
         return getTimer(context, key);
     }
 
+    @JRubyMethod
+    public IRubyObject histogram(final ThreadContext context, final IRubyObject key) {
+        return getHistogram(context, key);
+    }
+
     @JRubyMethod(required = 1, optional = 1)
     public IRubyObject increment(final ThreadContext context, final IRubyObject[] args) {
         return doIncrement(context, args);
@@ -95,6 +100,8 @@ protected abstract IRubyObject getGauge(ThreadContext context, IRubyObject key,
 
     protected abstract IRubyObject getTimer(ThreadContext context, IRubyObject key);
 
+    protected abstract IRubyObject getHistogram(ThreadContext context, IRubyObject key);
+
     protected abstract IRubyObject doTime(ThreadContext context, IRubyObject key, Block block);
 
     protected abstract IRubyObject doReportTime(ThreadContext context,
diff --git a/logstash-core/src/main/java/org/logstash/instrument/metrics/AbstractSimpleMetricExt.java b/logstash-core/src/main/java/org/logstash/instrument/metrics/AbstractSimpleMetricExt.java
@@ -60,6 +60,13 @@ public IRubyObject timer(final ThreadContext context,
         return getTimer(context, namespace, key);
     }
 
+    @JRubyMethod
+    public IRubyObject histogram(final ThreadContext context,
+                                 final IRubyObject namespace,
+                                 final IRubyObject key) {
+        return getHistogram(context, namespace, key);
+    }
+
     @JRubyMethod(name = "report_time")
     public IRubyObject reportTime(final ThreadContext context, final IRubyObject namespace,
         final IRubyObject key, final IRubyObject duration) {
@@ -83,6 +90,8 @@ protected abstract IRubyObject getGauge(ThreadContext context, IRubyObject names
 
     protected abstract IRubyObject getTimer(ThreadContext context, IRubyObject namespace, IRubyObject key);
 
+    protected abstract IRubyObject getHistogram(ThreadContext context, IRubyObject namespace, IRubyObject key);
+
     protected abstract IRubyObject doReportTime(ThreadContext context, IRubyObject namespace,
         IRubyObject key, IRubyObject duration);
 
diff --git a/logstash-core/src/main/java/org/logstash/instrument/metrics/MetricExt.java b/logstash-core/src/main/java/org/logstash/instrument/metrics/MetricExt.java
@@ -54,6 +54,7 @@ public final class MetricExt extends AbstractSimpleMetricExt {
 
     private static final RubySymbol GAUGE = RubyUtil.RUBY.newSymbol("gauge");
     private static final RubySymbol TIMER = RubyUtil.RUBY.newSymbol("timer");
+    private static final RubySymbol HISTOGRAM = RubyUtil.RUBY.newSymbol("histogram");
     private static final RubySymbol SET = RubyUtil.RUBY.newSymbol("set");
     private static final RubySymbol GET = RubyUtil.RUBY.newSymbol("get");
 
@@ -153,6 +154,16 @@ protected IRubyObject getTimer(final ThreadContext context,
                 );
     }
 
+    @Override
+    protected IRubyObject getHistogram(final ThreadContext context,
+                                       final IRubyObject namespace,
+                                       final IRubyObject key) {
+        MetricExt.validateKey(context, null, key);
+        return collector.callMethod(context,
+                "get", new IRubyObject[]{normalizeNamespace(namespace), key, HISTOGRAM}
+        );
+    }
+
     @Override
     protected IRubyObject doReportTime(final ThreadContext context, final IRubyObject namespace,
         final IRubyObject key, final IRubyObject duration) {
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,4 +116,5 @@ 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");
 }
diff --git a/logstash-core/src/main/java/org/logstash/instrument/metrics/MetricType.java b/logstash-core/src/main/java/org/logstash/instrument/metrics/MetricType.java
@@ -77,6 +77,11 @@ public enum MetricType {
      * A flow-rate {@link FlowMetric}, instantiated with one or more backing {@link Metric}{@code <Number>}.
      */
     FLOW_RATE("flow/rate"),
+
+    /**
+     * A histogram metric to record values and obtain percentiles measurements.
+     * */
+    HISTOGRAM_LONG("histogram/long"),
     ;
 
     private final String type;
diff --git a/logstash-core/src/main/java/org/logstash/instrument/metrics/NamespacedMetricExt.java b/logstash-core/src/main/java/org/logstash/instrument/metrics/NamespacedMetricExt.java
@@ -84,6 +84,11 @@ protected IRubyObject getTimer(ThreadContext context, IRubyObject key) {
         return metric.timer(context, namespaceName, key);
     }
 
+    @Override
+    protected IRubyObject getHistogram(ThreadContext context, IRubyObject key) {
+        return metric.histogram(context, namespaceName, key);
+    }
+
     @Override
     protected IRubyObject doIncrement(final ThreadContext context, final IRubyObject[] args) {
         if (args.length == 1) {
diff --git a/logstash-core/src/main/java/org/logstash/instrument/metrics/NullMetricExt.java b/logstash-core/src/main/java/org/logstash/instrument/metrics/NullMetricExt.java
@@ -94,6 +94,14 @@ protected IRubyObject getTimer(final ThreadContext context,
         return NULL_TIMER_METRIC;
     }
 
+    @Override
+    protected IRubyObject getHistogram(final ThreadContext context,
+                                       final IRubyObject namespace,
+                                       final IRubyObject key) {
+        MetricExt.validateKey(context, null, key);
+        return context.nil;
+    }
+
     @Override
     protected IRubyObject doReportTime(final ThreadContext context, final IRubyObject namespace,
         final IRubyObject key, final IRubyObject duration) {
diff --git a/logstash-core/src/main/java/org/logstash/instrument/metrics/NullNamespacedMetricExt.java b/logstash-core/src/main/java/org/logstash/instrument/metrics/NullNamespacedMetricExt.java
@@ -90,6 +90,11 @@ protected IRubyObject getTimer(ThreadContext context, IRubyObject key) {
         return this.metric.getTimer(context, namespaceName, key);
     }
 
+    @Override
+    protected IRubyObject getHistogram(ThreadContext context, IRubyObject key) {
+        return this.metric.getHistogram(context, namespaceName, key);
+    }
+
     @Override
     protected IRubyObject doIncrement(final ThreadContext context, final IRubyObject[] args) {
         return context.nil;
diff --git a/logstash-core/src/main/java/org/logstash/instrument/metrics/histogram/HistogramMetric.java b/logstash-core/src/main/java/org/logstash/instrument/metrics/histogram/HistogramMetric.java
@@ -0,0 +1,62 @@
+package org.logstash.instrument.metrics.histogram;
+
+import org.HdrHistogram.Histogram;
+import org.HdrHistogram.Recorder;
+import org.jruby.RubySymbol;
+import org.jruby.runtime.ThreadContext;
+import org.jruby.runtime.builtin.IRubyObject;
+import org.logstash.RubyUtil;
+import org.logstash.instrument.metrics.AbstractMetric;
+import org.logstash.instrument.metrics.AbstractNamespacedMetricExt;
+import org.logstash.instrument.metrics.MetricType;
+
+public class HistogramMetric extends AbstractMetric<HistogramSnapshot> {
+
+    private final Recorder histogramRecorder;
+
+    private final Histogram lifetimeHistogram;
+
+    /**
+     * Constructor
+     *
+     *
+     * @param name The name of this metric. This value may be used for display purposes.
+     */
+    public HistogramMetric(String name) {
+        super(name);
+        histogramRecorder = new Recorder(1_000_000, 3);
+        lifetimeHistogram = histogramRecorder.getIntervalHistogram();
+    }
+
+    public static HistogramMetric fromRubyBase(AbstractNamespacedMetricExt metric, RubySymbol key) {
+            final ThreadContext context = RubyUtil.RUBY.getCurrentContext();
+            final IRubyObject histogram = metric.histogram(context, key);
+            final HistogramMetric javaTimer;
+            if (HistogramMetric.class.isAssignableFrom(histogram.getJavaClass())) {
+                javaTimer = histogram.toJava(HistogramMetric.class);
+            } else {
+                throw new IllegalArgumentException("Metric " + key + " is not a histogram");
+            }
+            return javaTimer;
+    }
+
+    @Override
+    public MetricType getType() {
+        return MetricType.HISTOGRAM_LONG;
+    }
+
+    @Override
+    public HistogramSnapshot getValue() {
+        lifetimeHistogram.add(histogramRecorder.getIntervalHistogram());
+        return new HistogramSnapshot(lifetimeHistogram);
+    }
+
+    /**
+     * Adds a recorded value.
+     *
+     * @param value the value to record
+     */
+    public void update(long value) {
+        histogramRecorder.recordValue(value);
+    }
+}
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
@@ -0,0 +1,33 @@
+package org.logstash.instrument.metrics.histogram;
+
+import org.HdrHistogram.Histogram;
+
+/**
+ * Class to expose percentiles retrieved from an HdrHistogram.
+ * */
+public final class HistogramSnapshot {
+
+    private final long percentile75;
+    private final long percentile90;
+
+    public HistogramSnapshot(Histogram hdrHistogram) {
+        percentile75 = hdrHistogram.getValueAtPercentile(75);
+        percentile90 = hdrHistogram.getValueAtPercentile(90);
+    }
+
+    public double get75Percentile() {
+        return percentile75;
+    }
+
+    public double get90Percentile() {
+        return percentile90;
+    }
+
+    @Override
+    public String toString() {
+        return "HistogramSnapshot{" +
+                "percentile75=" + percentile75 +
+                ", percentile90=" + percentile90 +
+                '}';
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -245,6 +245,7 @@ dependencies {`
`245`	`245`	`exclude group: 'com.google.guava', module: 'guava'`
`246`	`246`	`}`
`247`	`247`	`implementation 'org.javassist:javassist:3.30.2-GA'`
	`248`	`+ implementation 'org.hdrhistogram:HdrHistogram:2.2.2'`
`248`	`249`	`testImplementation "org.apache.logging.log4j:log4j-core:${log4jVersion}:tests"`
`249`	`250`	`testImplementation 'org.hamcrest:hamcrest:2.2'`
`250`	`251`	`testImplementation 'org.hamcrest:hamcrest-library:2.2'`
Original file line number	Diff line number	Diff line change
`@@ -116,4 +116,5 @@ 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`	`120`	`}`