Exposed user metric for histrogam flow batch byte size

Author: andsel

logstash-core/lib/logstash/api/commands/stats.rb

@@ -225,7 +225,8 @@ def report(stats, extended_stats = nil, opts = {})
                 :batch_delay => stats.dig(:config, :batch_delay),
               }
             }
-            ret[:batch] = refine_batch_metrics(stats) if stats.include?(:batch)
+            # ret[:batch] = refine_batch_metrics(stats) if stats.include?(:batch)
+            ret[:batch] = stats[:batch] if stats.include?(:batch)
             ret[:dead_letter_queue] = stats[:dlq] if stats.include?(:dlq)
 
             # if extended_stats were provided, enrich the return value

logstash-core/src/main/java/org/logstash/execution/QueueReadClientBatchMetrics.java

@@ -6,6 +6,8 @@
 import org.logstash.ackedqueue.QueueFactoryExt;
 import org.logstash.ext.JrubyEventExtLibrary;
 import org.logstash.instrument.metrics.AbstractNamespacedMetricExt;
+import org.logstash.instrument.metrics.HdrHistogramFlowMetric;
+import org.logstash.instrument.metrics.HistogramFlowMetric;
 import org.logstash.instrument.metrics.counter.LongCounter;
 import org.logstash.instrument.metrics.gauge.LazyDelegatingGauge;
 
@@ -23,6 +25,7 @@ class QueueReadClientBatchMetrics {
     private LongCounter pipelineMetricBatchCount;
     private LongCounter pipelineMetricBatchByteSize;
     private LongCounter pipelineMetricBatchTotalEvents;
+    private HistogramFlowMetric pipelineMetricBatchByteSizeFlowHistogram;
     private final SecureRandom random = new SecureRandom();
     private LazyDelegatingGauge currentBatchDimensions;
 
@@ -39,6 +42,9 @@ public void setupMetrics(AbstractNamespacedMetricExt namespacedMetric) {
             pipelineMetricBatchTotalEvents = LongCounter.fromRubyBase(batchNamespace, BATCH_TOTAL_EVENTS);
             pipelineMetricBatchByteSize = LongCounter.fromRubyBase(batchNamespace, BATCH_TOTAL_BYTES);
             currentBatchDimensions = LazyDelegatingGauge.fromRubyBase(batchNamespace, BATCH_CURRENT_KEY);
+            pipelineMetricBatchByteSizeFlowHistogram = batchNamespace.asApiMetric()
+                    .namespace("batch_byte_size")
+                    .register("histogram", HdrHistogramFlowMetric.FACTORY);
         }
     }
 
@@ -75,6 +81,7 @@ private void updateBatchSizeMetric(QueueBatch batch) {
             pipelineMetricBatchTotalEvents.increment(batch.filteredSize());
             pipelineMetricBatchByteSize.increment(totalByteSize);
             currentBatchDimensions.set(Arrays.asList(batch.filteredSize(), totalByteSize));
+            pipelineMetricBatchByteSizeFlowHistogram.recordValue(totalByteSize);
         } catch (IllegalArgumentException e) {
             LOG.error("Failed to calculate batch byte size for metrics", e);
         }

logstash-core/src/main/java/org/logstash/instrument/metrics/HdrHistogramFlowMetric.java

@@ -19,6 +19,7 @@
 
 package org.logstash.instrument.metrics;
 
+import co.elastic.logstash.api.UserMetric;
 import org.HdrHistogram.Histogram;
 import org.HdrHistogram.Recorder;
 import org.apache.logging.log4j.LogManager;
@@ -114,6 +115,8 @@ HistogramMetricData computeAggregatedHistogramData() {
 
     private static final Logger LOG = LogManager.getLogger(HdrHistogramFlowMetric.class);
 
+    public static UserMetric.Factory<HistogramFlowMetric> FACTORY = HistogramFlowMetric.PROVIDER.getFactory(HdrHistogramFlowMetric::new);
+
     private static final List<FlowMetricRetentionPolicy> SUPPORTED_POLICIES = List.of(
             FlowMetricRetentionPolicy.BuiltInRetentionPolicy.LAST_1_MINUTE,
             FlowMetricRetentionPolicy.BuiltInRetentionPolicy.LAST_5_MINUTES,
@@ -139,7 +142,6 @@ public Map<String, HistogramMetricData> getValue() {
         final long currentTimeNanos = System.nanoTime();
         histogramsWindows.forEach((policy, window) -> {
             window.recordWindow.baseline(currentTimeNanos).ifPresent(baseline -> {
-                LOG.info("getValue computing aggregated histogram");
                 result.put(policy.policyName().toLowerCase(), window.computeAggregatedHistogramData());
             });
         });

logstash-core/src/main/java/org/logstash/instrument/metrics/HistogramFlowMetric.java

@@ -24,7 +24,24 @@
 
 public interface HistogramFlowMetric extends UserMetric<Map<String, HistogramMetricData>>,
         org.logstash.instrument.metrics.Metric<Map<String, HistogramMetricData>> {
-    
+
+    Provider<HistogramFlowMetric> PROVIDER = new Provider<>(HistogramFlowMetric.class, new HistogramFlowMetric() {
+        @Override
+        public Map<String, HistogramMetricData> getValue() {
+            return Map.of();
+        }
+
+        @Override
+        public String getName() {
+            return "NULL";
+        }
+
+        @Override
+        public void recordValue(long totalByteSize) {
+            // no-op
+        }
+    });
+
     void recordValue(long totalByteSize);
 
     @Override