Add max value for batch size windowed histograms

Author: andsel

config/logstash.yml

@@ -64,6 +64,7 @@
 # Default is "minimal".
 #
 # pipeline.batch.metrics.sampling_mode: "minimal"
+pipeline.batch.metrics.sampling_mode: full
 #
 # Force Logstash to exit during shutdown even if there are still inflight
 # events in memory. By default, logstash will refuse to quit until all

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

@@ -213,6 +213,8 @@ def refine_batch_metrics(stats)
                 byte_size_histogram_value = byte_size_histogram.value
                 if byte_size_histogram_value[window.to_s]
                   reshape_histogram_percentiles_for_window(:byte_size, byte_size_histogram_value, window, result)
+                  result[:byte_size][:max] ||= {}
+                  result[:byte_size][:max][window] = byte_size_histogram_value[window.to_s].max_value.round
                 end
               end
             end

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

@@ -111,10 +111,12 @@ public static class HistogramMetricData implements Serializable {
         private static final long serialVersionUID = 4711735381843512566L;
         private final long percentile50;
         private final long percentile90;
+        private final long maxValue;
 
         public HistogramMetricData(ValueHistogram histogram) {
             percentile50 = histogram.getValueAtPercentile(50);
             percentile90 = histogram.getValueAtPercentile(90);
+            maxValue = histogram.getMaxValue();
         }
 
         @JsonProperty("p50")
@@ -127,11 +129,17 @@ public long get90Percentile() {
             return percentile90;
         }
 
+        @JsonProperty("max")
+        public long getMaxValue() {
+            return maxValue;
+        }
+
         @Override
         public String toString() {
             return "HistogramMetricData{" +
                     "percentile50=" + percentile50 +
                     ", percentile90=" + percentile90 +
+                    ", maxValue=" + maxValue +
                     '}';
         }
     }
@@ -165,8 +173,6 @@ Optional<ValueHistogram> calculateValue() {
                 if (compareCapture == null) { return Optional.empty(); }
                 if (baselineCapture == null) { return Optional.empty(); }
 
-//                var result = compareCapture.getValueHistogram();
-//                result.subtract(baselineCapture.getValueHistogram());
                 var result = compareCapture.getValueHistogram().subtract(baselineCapture.getValueHistogram());
                 return Optional.of(result);
             });

logstash-core/src/main/java/org/logstash/instrument/metrics/histogram/LifetimeHistogramMetric.java

@@ -85,6 +85,10 @@ public long getTotalCount() {
             return delegate.getTotalCount();
         }
 
+        public long getMaxValue() {
+            return delegate.getMaxValue();
+        }
+
         public ValueHistogram subtract(ValueHistogram other) {
             if (Objects.isNull(other) || Objects.isNull(other.delegate) || other.delegate.getTotalCount() <= 0) {
                 return this;

logstash-core/src/test/java/org/logstash/instrument/metrics/histogram/LifetimeHistogramMetricTest.java

@@ -42,6 +42,7 @@ public void givenValueRecordedThen90thPercentileFromGetValueReflectsIt() {
         ValueHistogram h = sut.getValue();
         assertEquals(1, h.getTotalCount());
         assertEquals(1000L, h.getValueAtPercentile(90));
+        assertEquals(1000L, h.getMaxValue());
     }
 
     @Test
@@ -56,14 +57,24 @@ public void givenLifetimeHistogramWhenRecordingSecondLargerValueThenP90Increases
         ValueHistogram afterFirst = sut.getValue();
         long p90AfterFirst = afterFirst.getValueAtPercentile(90);
         assertEquals(100L, p90AfterFirst);
+        assertEquals(100L, afterFirst.getMaxValue());
 
         sut.recordValue(1000);
         ValueHistogram afterSecond = sut.getValue();
         long p90AfterSecond = afterSecond.getValueAtPercentile(90);
         assertTrue("90th percentile should increase after recording a larger value", p90AfterSecond > p90AfterFirst);
         assertEquals(1000L, p90AfterSecond);
+        assertEquals(1000L, afterSecond.getMaxValue());
 
         long p50AfterSecond = afterSecond.getValueAtPercentile(50);
         assertEquals("previous 90th percentile value (100) should now be at 50th percentile", 100L, p50AfterSecond);
     }
+
+    @Test
+    public void givenValuesRecordedThenGetMaxValueReturnsLargestRecordedValue() {
+        sut.recordValue(100);
+        sut.recordValue(1000);
+        ValueHistogram h = sut.getValue();
+        assertEquals(1000L, h.getMaxValue());
+    }
 }