Introduce separate HistogramWithSum for situations where we want an accurate sum

and don't mind the additional fields, while avoiding this overhead for situations
where we don't need the sum.

Also fixes a bug where the histogram sum wasn't reset on 'clear'.

Author: mcculls

dd-java-agent/agent-otel/otel-bootstrap/src/main/java/datadog/trace/bootstrap/otel/metrics/data/OtelHistogramSketch.java

@@ -1,21 +1,20 @@
 package datadog.trace.bootstrap.otel.metrics.data;
 
 import datadog.metrics.api.Histogram;
+import datadog.metrics.api.HistogramWithSum;
 import java.util.List;
 
 final class OtelHistogramSketch extends OtelAggregator {
-  private final Histogram histogram;
-  private volatile double totalSum;
+  private final HistogramWithSum histogram;
 
   OtelHistogramSketch(List<Double> bucketBoundaries) {
-    this.histogram = Histogram.newHistogram(bucketBoundaries);
+    this.histogram = Histogram.newHistogramWithSum(bucketBoundaries);
   }
 
   @Override
   void doRecordDouble(double value) {
     synchronized (histogram) {
       histogram.accept(value);
-      totalSum += value;
     }
   }
 
@@ -34,10 +33,9 @@ OtelPoint doCollect(boolean reset) {
       count = histogram.getCount();
       binBoundaries = histogram.getBinBoundaries();
       binCounts = histogram.getBinCounts();
-      sum = totalSum;
+      sum = histogram.getSum();
       if (reset) {
         histogram.clear();
-        totalSum = 0;
       }
     }
     return new OtelHistogramPoint(count, binBoundaries, binCounts, sum);

dd-java-agent/instrumentation/spark/spark-common/src/main/java/datadog/trace/instrumentation/spark/SparkAggregatedTaskMetrics.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.core.JsonGenerator;
 import datadog.metrics.api.Histogram;
+import datadog.metrics.api.HistogramWithSum;
 import datadog.trace.api.Config;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
 import java.io.IOException;
@@ -67,7 +68,7 @@ class SparkAggregatedTaskMetrics {
   private Histogram diskBytesSpilledHistogram;
 
   // Used for Spark SQL Plan metrics ONLY, don't put in regular span for now
-  private Map<Long, Histogram> externalAccumulableHistograms;
+  private Map<Long, HistogramWithSum> externalAccumulableHistograms;
 
   public SparkAggregatedTaskMetrics() {}
 
@@ -148,10 +149,11 @@ public void addTaskMetrics(
 
           externalAccumulators.forEach(
               acc -> {
-                Histogram hist = externalAccumulableHistograms.get(acc.id());
+                HistogramWithSum hist = externalAccumulableHistograms.get(acc.id());
                 if (hist == null) {
                   hist =
-                      Histogram.newHistogram(HISTOGRAM_RELATIVE_ACCURACY, HISTOGRAM_MAX_NUM_BINS);
+                      Histogram.newHistogramWithSum(
+                          HISTOGRAM_RELATIVE_ACCURACY, HISTOGRAM_MAX_NUM_BINS);
                 }
 
                 try {
@@ -315,7 +317,7 @@ private Histogram lazyHistogramAccept(Histogram hist, double value) {
   // Used to put external accum metrics to JSON for Spark SQL plans
   public void externalAccumToJson(JsonGenerator generator, SQLMetricInfo info) throws IOException {
     if (externalAccumulableHistograms != null) {
-      Histogram hist = externalAccumulableHistograms.get(info.accumulatorId());
+      HistogramWithSum hist = externalAccumulableHistograms.get(info.accumulatorId());
       String name = info.name();
 
       if (name != null && hist != null) {

products/metrics/metrics-api/src/main/java/datadog/metrics/api/Histogram.java

@@ -5,8 +5,6 @@
 
 public interface Histogram {
 
-  double getSum();
-
   double getCount();
 
   boolean isEmpty();
@@ -41,7 +39,11 @@ static Histogram newHistogram(double relativeAccuracy, int maxNumBins) {
     return Histograms.factory.newHistogram(relativeAccuracy, maxNumBins);
   }
 
-  static Histogram newHistogram(List<Double> binBoundaries) {
-    return Histograms.factory.newHistogram(binBoundaries);
+  static HistogramWithSum newHistogramWithSum(double relativeAccuracy, int maxNumBins) {
+    return Histograms.factory.newHistogramWithSum(relativeAccuracy, maxNumBins);
+  }
+
+  static HistogramWithSum newHistogramWithSum(List<Double> binBoundaries) {
+    return Histograms.factory.newHistogramWithSum(binBoundaries);
   }
 }

products/metrics/metrics-api/src/main/java/datadog/metrics/api/HistogramWithSum.java

@@ -0,0 +1,6 @@
+package datadog.metrics.api;
+
+/** Adds exact summary statistics to the histogram */
+public interface HistogramWithSum extends Histogram {
+  double getSum();
+}

products/metrics/metrics-api/src/main/java/datadog/metrics/api/Histograms.java

@@ -23,6 +23,8 @@ public interface Factory {
 
     Histogram newHistogram(double relativeAccuracy, int maxNumBins);
 
-    Histogram newHistogram(List<Double> binBoundaries);
+    HistogramWithSum newHistogramWithSum(double relativeAccuracy, int maxNumBins);
+
+    HistogramWithSum newHistogramWithSum(List<Double> binBoundaries);
   }
 }

products/metrics/metrics-api/src/main/java/datadog/metrics/api/NoOpHistogram.java

@@ -4,8 +4,8 @@
 import java.util.Collections;
 import java.util.List;
 
-class NoOpHistogram implements Histogram {
-  public static final Histogram INSTANCE = new NoOpHistogram();
+class NoOpHistogram implements HistogramWithSum {
+  public static final HistogramWithSum INSTANCE = new NoOpHistogram();
 
   @Override
   public double getCount() {

products/metrics/metrics-api/src/main/java/datadog/metrics/api/NoOpHistogramsFactory.java

@@ -19,7 +19,12 @@ public Histogram newHistogram(double relativeAccuracy, int maxNumBins) {
   }
 
   @Override
-  public Histogram newHistogram(List<Double> binBoundaries) {
+  public HistogramWithSum newHistogramWithSum(double relativeAccuracy, int maxNumBins) {
+    return NoOpHistogram.INSTANCE;
+  }
+
+  @Override
+  public HistogramWithSum newHistogramWithSum(List<Double> binBoundaries) {
     return NoOpHistogram.INSTANCE;
   }
 }

products/metrics/metrics-lib/src/main/java/datadog/metrics/impl/DDSketchHistogram.java

@@ -12,71 +12,48 @@
 /** Wrapper around the DDSketch library so that it can be used in an instrumentation */
 public class DDSketchHistogram implements Histogram {
   private final DDSketch sketch;
-  private double sum;
-  // We use a compensated sum to avoid accumulating rounding errors.
-  // See https://en.wikipedia.org/wiki/Kahan_summation_algorithm.
-  private double sumCompensation; // Low order bits of sum
-  private double simpleSum; // Used to compute right sum for non-finite inputs
 
   public DDSketchHistogram(DDSketch sketch) {
     this.sketch = sketch;
-    this.sum = 0;
-    this.simpleSum = 0;
-    this.sumCompensation = 0;
   }
 
   @Override
-  public double getCount() {
+  public final double getCount() {
     return sketch.getCount();
   }
 
   @Override
-  public double getSum() {
-    // Better error bounds to add both terms as the final sum
-    final double tmp = sum + sumCompensation;
-    if (Double.isNaN(tmp) && Double.isInfinite(simpleSum)) {
-      // If the compensated sum is spuriously NaN from accumulating one or more same-signed infinite
-      // values, return the correctly-signed infinity stored in simpleSum.
-      return simpleSum;
-    } else {
-      return tmp;
-    }
-  }
-
-  @Override
-  public boolean isEmpty() {
+  public final boolean isEmpty() {
     return sketch.isEmpty();
   }
 
   @Override
   public void accept(double value) {
     sketch.accept(value);
-    updateSum(value);
   }
 
   @Override
   public void accept(double value, double count) {
     sketch.accept(value, count);
-    updateSum(value * count);
   }
 
   @Override
-  public double getValueAtQuantile(double quantile) {
+  public final double getValueAtQuantile(double quantile) {
     return sketch.getValueAtQuantile(quantile);
   }
 
   @Override
-  public double getMinValue() {
+  public final double getMinValue() {
     return sketch.getMinValue();
   }
 
   @Override
-  public double getMaxValue() {
+  public final double getMaxValue() {
     return sketch.getMaxValue();
   }
 
   @Override
-  public List<Double> getBinBoundaries() {
+  public final List<Double> getBinBoundaries() {
     List<Double> boundaries = new ArrayList<>();
     // Bin boundaries are defined as a sequence of upper bounds. When our sketch contains gaps we
     // must introduce extra bounds to represent the lower bound of any bins after a gap; otherwise
@@ -98,7 +75,7 @@ public List<Double> getBinBoundaries() {
   }
 
   @Override
-  public List<Double> getBinCounts() {
+  public final List<Double> getBinCounts() {
     List<Double> counts = new ArrayList<>();
     // to maintain alignment with getBoundaries we must introduce zero counts for
     // boundaries inserted to represent the lower bound of any bins after a gap.
@@ -123,19 +100,7 @@ public void clear() {
   }
 
   @Override
-  public ByteBuffer serialize() {
+  public final ByteBuffer serialize() {
     return sketch.serialize();
   }
-
-  private void updateSum(double value) {
-    simpleSum += value;
-    sumWithCompensation(value);
-  }
-
-  private void sumWithCompensation(double value) {
-    final double tmp = value - sumCompensation;
-    final double velvel = sum + tmp; // Little wolf of rounding error
-    sumCompensation = (velvel - sum) - tmp;
-    sum = velvel;
-  }
 }

products/metrics/metrics-lib/src/main/java/datadog/metrics/impl/DDSketchHistogramWithSum.java

@@ -0,0 +1,62 @@
+package datadog.metrics.impl;
+
+import com.datadoghq.sketch.ddsketch.DDSketch;
+import datadog.metrics.api.HistogramWithSum;
+
+/** Adds exact summary statistics to the DDSketch wrapper */
+public final class DDSketchHistogramWithSum extends DDSketchHistogram implements HistogramWithSum {
+  private double sum;
+  // We use a compensated sum to avoid accumulating rounding errors.
+  // See https://en.wikipedia.org/wiki/Kahan_summation_algorithm.
+  private double sumCompensation; // Low order bits of sum
+  private double simpleSum; // Used to compute right sum for non-finite inputs
+
+  public DDSketchHistogramWithSum(DDSketch sketch) {
+    super(sketch);
+  }
+
+  @Override
+  public double getSum() {
+    // Better error bounds to add both terms as the final sum
+    final double tmp = sum + sumCompensation;
+    if (Double.isNaN(tmp) && Double.isInfinite(simpleSum)) {
+      // If the compensated sum is spuriously NaN from accumulating one or more same-signed infinite
+      // values, return the correctly-signed infinity stored in simpleSum.
+      return simpleSum;
+    } else {
+      return tmp;
+    }
+  }
+
+  @Override
+  public void accept(double value) {
+    super.accept(value);
+    updateSum(value);
+  }
+
+  @Override
+  public void accept(double value, double count) {
+    super.accept(value, count);
+    updateSum(value * count);
+  }
+
+  @Override
+  public void clear() {
+    super.clear();
+    sum = 0;
+    simpleSum = 0;
+    sumCompensation = 0;
+  }
+
+  private void updateSum(double value) {
+    simpleSum += value;
+    sumWithCompensation(value);
+  }
+
+  private void sumWithCompensation(double value) {
+    final double tmp = value - sumCompensation;
+    final double velvel = sum + tmp; // Little wolf of rounding error
+    sumCompensation = (velvel - sum) - tmp;
+    sum = velvel;
+  }
+}

products/metrics/metrics-lib/src/main/java/datadog/metrics/impl/DDSketchHistograms.java

@@ -11,6 +11,7 @@
 import com.datadoghq.sketch.ddsketch.mapping.LogarithmicMapping;
 import com.datadoghq.sketch.ddsketch.store.CollapsingLowestDenseStore;
 import datadog.metrics.api.Histogram;
+import datadog.metrics.api.HistogramWithSum;
 import datadog.metrics.api.Histograms;
 import java.util.List;
 
@@ -44,15 +45,21 @@ public Histogram newHistogram(double relativeAccuracy, int maxNumBins) {
   }
 
   @Override
-  public Histogram newHistogram(List<Double> binBoundaries) {
+  public HistogramWithSum newHistogramWithSum(double relativeAccuracy, int maxNumBins) {
+    DDSketch sketch = DDSketches.logarithmicCollapsingLowestDense(relativeAccuracy, maxNumBins);
+    return new DDSketchHistogramWithSum(sketch);
+  }
+
+  @Override
+  public HistogramWithSum newHistogramWithSum(List<Double> binBoundaries) {
     validateBoundaries(binBoundaries);
     DDSketch sketch =
         new DDSketch(
             new ExplicitBoundaries(binBoundaries),
             () -> new CollapsingLowestDenseStore(0), // negative store not used
             () -> new CollapsingLowestDenseStore(binBoundaries.size() + 1),
             Double.NEGATIVE_INFINITY); // assign all negative/zero values to first bin
-    return new DDSketchHistogram(sketch);
+    return new DDSketchHistogramWithSum(sketch);
   }
 
   static void validateBoundaries(List<Double> boundaries) {