From e675fd1bd88ad09f25981bacd6c87a18692f56bb Mon Sep 17 00:00:00 2001 From: beanliu Date: Wed, 24 Feb 2021 14:17:39 +1100 Subject: [PATCH 01/16] add histogram aggregator --- .../aggregator/DoubleHistogramBenchmark.java | 74 +++++++++ .../metrics/aggregator/AggregatorFactory.java | 12 ++ .../aggregator/DoubleHistogramAggregator.java | 153 ++++++++++++++++++ .../aggregator/HistogramAccumulation.java | 38 +++++ .../HistogramAggregatorFactory.java | 54 +++++++ .../aggregator/ImmutableDoubleArray.java | 100 ++++++++++++ .../aggregator/ImmutableLongArray.java | 113 +++++++++++++ .../aggregator/AggregatorFactoryTest.java | 70 ++++++++ .../DoubleHistogramAggregatorTest.java | 150 +++++++++++++++++ 9 files changed, 764 insertions(+) create mode 100644 sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java create mode 100644 sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java create mode 100644 sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAccumulation.java create mode 100644 sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java create mode 100644 sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableDoubleArray.java create mode 100644 sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableLongArray.java create mode 100644 sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java diff --git a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java new file mode 100644 index 00000000000..f70ce64f656 --- /dev/null +++ b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java @@ -0,0 +1,74 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.metrics.aggregator; + +import io.opentelemetry.sdk.common.InstrumentationLibraryInfo; +import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor; +import io.opentelemetry.sdk.metrics.common.InstrumentType; +import io.opentelemetry.sdk.metrics.common.InstrumentValueType; +import io.opentelemetry.sdk.resources.Resource; +import java.util.concurrent.TimeUnit; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.annotations.Warmup; + +@State(Scope.Benchmark) +public class DoubleHistogramBenchmark { + private static final Aggregator aggregator = + AggregatorFactory.histogram(new double[] {10, 100, 1_000}, /* stateful= */ false) + .create( + Resource.getDefault(), + InstrumentationLibraryInfo.empty(), + InstrumentDescriptor.create( + "name", + "description", + "1", + InstrumentType.VALUE_RECORDER, + InstrumentValueType.DOUBLE)); + private AggregatorHandle aggregatorHandle; + + @Setup(Level.Trial) + public final void setup() { + aggregatorHandle = aggregator.createHandle(); + } + + @Benchmark + @Fork(1) + @Warmup(iterations = 5, time = 1) + @Measurement(iterations = 10, time = 1) + @OutputTimeUnit(TimeUnit.MILLISECONDS) + @Threads(value = 10) + public void aggregate_10Threads() { + aggregatorHandle.recordDouble(100.0056); + } + + @Benchmark + @Fork(1) + @Warmup(iterations = 5, time = 1) + @Measurement(iterations = 10, time = 1) + @OutputTimeUnit(TimeUnit.MILLISECONDS) + @Threads(value = 5) + public void aggregate_5Threads() { + aggregatorHandle.recordDouble(100.0056); + } + + @Benchmark + @Fork(1) + @Warmup(iterations = 5, time = 1) + @Measurement(iterations = 10, time = 1) + @OutputTimeUnit(TimeUnit.MILLISECONDS) + @Threads(value = 1) + public void aggregate_1Threads() { + aggregatorHandle.recordDouble(100.0056); + } +} diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactory.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactory.java index 05887a3fd20..1c4a627d374 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactory.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactory.java @@ -77,6 +77,18 @@ static AggregatorFactory minMaxSumCount() { return MinMaxSumCountAggregatorFactory.INSTANCE; } + /** + * Returns an {@code AggregatorFactory} that calculates an approximation of the distribution of + * the measurements taken. + * + * @param stateful configures if the aggregator is stateful. + * @param boundaries configures the fixed bucket boundaries. + * @return an {@code AggregationFactory} that calculates histogram of recorded measurements. + */ + static AggregatorFactory histogram(double[] boundaries, boolean stateful) { + return new HistogramAggregatorFactory(boundaries, stateful); + } + /** * Returns a new {@link Aggregator}. * diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java new file mode 100644 index 00000000000..f516fcba250 --- /dev/null +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java @@ -0,0 +1,153 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.metrics.aggregator; + +import io.opentelemetry.api.metrics.common.Labels; +import io.opentelemetry.sdk.common.InstrumentationLibraryInfo; +import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor; +import io.opentelemetry.sdk.metrics.data.DoubleGaugeData; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.resources.Resource; +import java.util.Arrays; +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.locks.ReentrantLock; +import javax.annotation.concurrent.GuardedBy; + +final class DoubleHistogramAggregator extends AbstractAggregator { + private final ImmutableDoubleArray boundaries; + + DoubleHistogramAggregator( + Resource resource, + InstrumentationLibraryInfo instrumentationLibraryInfo, + InstrumentDescriptor instrumentDescriptor, + ImmutableDoubleArray boundaries, + boolean stateful) { + super(resource, instrumentationLibraryInfo, instrumentDescriptor, stateful); + this.boundaries = boundaries; + } + + @Override + public AggregatorHandle createHandle() { + return new Handle(this.boundaries); + } + + /** + * Return the result of the merge of two histogram accumulations. As long as one Aggregator + * instance produces all Accumulations with constant boundaries we don't need to worry about + * merging accumulations with different boundaries. + */ + @Override + public final HistogramAccumulation merge(HistogramAccumulation x, HistogramAccumulation y) { + long[] mergedCounts = new long[x.getCounts().length()]; + for (int i = 0; i < x.getCounts().length(); ++i) { + mergedCounts[i] = x.getCounts().get(i) + y.getCounts().get(i); + } + return HistogramAccumulation.create( + x.getSum() + y.getSum(), ImmutableLongArray.copyOf(mergedCounts)); + } + + @Override + public final MetricData toMetricData( + Map accumulationByLabels, + long startEpochNanos, + long lastCollectionEpoch, + long epochNanos) { + // effectively no-op, will convert to histogram data in other PRs + return MetricData.createDoubleGauge( + getResource(), + getInstrumentationLibraryInfo(), + getInstrumentDescriptor().getName(), + getInstrumentDescriptor().getDescription(), + getInstrumentDescriptor().getUnit(), + DoubleGaugeData.create(Collections.emptyList())); + } + + @Override + public HistogramAccumulation accumulateDouble(double value) { + return HistogramAccumulation.create(value, ImmutableLongArray.of(1)); + } + + @Override + public HistogramAccumulation accumulateLong(long value) { + return HistogramAccumulation.create(value, ImmutableLongArray.of(1)); + } + + static final class Handle extends AggregatorHandle { + private final ImmutableDoubleArray boundaries; + + private final ReentrantLock lock = new ReentrantLock(); + + @GuardedBy("lock") + private final State current; + + Handle(ImmutableDoubleArray boundaries) { + this.boundaries = boundaries; + this.current = new State(this.boundaries.length() + 1); + } + + // Benchmark shows that linear search performs better than binary search with ordinary + // buckets. + private int findBucketIndex(double value) { + for (int i = 0; i < boundaries.length(); ++i) { + if (Double.compare(value, boundaries.get(i)) <= 0) { + return i; + } + } + return boundaries.length(); + } + + @Override + protected HistogramAccumulation doAccumulateThenReset() { + lock.lock(); + try { + HistogramAccumulation acc = + HistogramAccumulation.create(current.sum, ImmutableLongArray.copyOf(current.counts)); + current.reset(); + return acc; + } finally { + lock.unlock(); + } + } + + @Override + protected void doRecordDouble(double value) { + int bucketIndex = findBucketIndex(value); + + lock.lock(); + try { + current.record(bucketIndex, value); + } finally { + lock.unlock(); + } + } + + @Override + protected void doRecordLong(long value) { + doRecordDouble((double) value); + } + + private static final class State { + private double sum; + private final long[] counts; + + public State(int bucketSize) { + this.counts = new long[bucketSize]; + reset(); + } + + private void reset() { + this.sum = 0; + Arrays.fill(this.counts, 0); + } + + private void record(int bucketIndex, double value) { + this.sum += value; + this.counts[bucketIndex]++; + } + } + } +} diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAccumulation.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAccumulation.java new file mode 100644 index 00000000000..dd2aad43608 --- /dev/null +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAccumulation.java @@ -0,0 +1,38 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.metrics.aggregator; + +import com.google.auto.value.AutoValue; +import javax.annotation.concurrent.Immutable; + +@Immutable +@AutoValue +abstract class HistogramAccumulation { + /** + * Creates a new {@link HistogramAccumulation} with the given values. + * + * @return a new {@link HistogramAccumulation} with the given values. + */ + static HistogramAccumulation create(double sum, ImmutableLongArray counts) { + return new AutoValue_HistogramAccumulation(sum, counts); + } + + HistogramAccumulation() {} + + /** + * The sum of all measurements recorded. + * + * @return the sum of recorded measurements. + */ + abstract double getSum(); + + /** + * The counts in each bucket. + * + * @return the counts in each bucket. + */ + abstract ImmutableLongArray getCounts(); +} diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java new file mode 100644 index 00000000000..49c3e40bf88 --- /dev/null +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java @@ -0,0 +1,54 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.metrics.aggregator; + +import io.opentelemetry.sdk.common.InstrumentationLibraryInfo; +import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor; +import io.opentelemetry.sdk.resources.Resource; + +final class HistogramAggregatorFactory implements AggregatorFactory { + private final ImmutableDoubleArray boundaries; + private final boolean stateful; + + HistogramAggregatorFactory(double[] boundaries, boolean stateful) { + this.boundaries = ImmutableDoubleArray.copyOf(boundaries); + this.stateful = stateful; + + for (int i = 1; i < this.boundaries.length(); ++i) { + if (Double.compare(this.boundaries.get(i - 1), this.boundaries.get(i)) >= 0) { + throw new IllegalArgumentException( + "invalid bucket boundary: " + + this.boundaries.get(i - 1) + + " >= " + + this.boundaries.get(i)); + } + } + if (this.boundaries.length() > 0) { + if (this.boundaries.get(0) == Double.NEGATIVE_INFINITY) { + throw new IllegalArgumentException("invalid bucket boundary: -Inf"); + } + if (this.boundaries.get(this.boundaries.length() - 1) == Double.POSITIVE_INFINITY) { + throw new IllegalArgumentException("invalid bucket boundary: +Inf"); + } + } + } + + @Override + @SuppressWarnings("unchecked") + public Aggregator create( + Resource resource, + InstrumentationLibraryInfo instrumentationLibraryInfo, + InstrumentDescriptor descriptor) { + switch (descriptor.getValueType()) { + case LONG: + case DOUBLE: + return (Aggregator) + new DoubleHistogramAggregator( + resource, instrumentationLibraryInfo, descriptor, this.boundaries, this.stateful); + } + throw new IllegalArgumentException("Invalid instrument value type"); + } +} diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableDoubleArray.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableDoubleArray.java new file mode 100644 index 00000000000..22c9f146bbc --- /dev/null +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableDoubleArray.java @@ -0,0 +1,100 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.metrics.aggregator; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import javax.annotation.Nullable; +import javax.annotation.concurrent.Immutable; + +@Immutable +class ImmutableDoubleArray { + private static final ImmutableDoubleArray EMPTY = new ImmutableDoubleArray(new double[0]); + + /** Returns an immutable array containing the given values, in order. */ + public static ImmutableDoubleArray copyOf(double[] values) { + return values.length == 0 + ? EMPTY + : new ImmutableDoubleArray(Arrays.copyOf(values, values.length)); + } + + private final double[] array; + + private ImmutableDoubleArray(double[] array) { + this.array = array; + } + + /** Returns a copy of the underlying data as list. */ + public List toList() { + List result = new ArrayList<>(array.length); + for (double v : array) { + result.add(v); + } + return result; + } + + /** Returns the number of values in this array. */ + public int length() { + return array.length; + } + + /** + * Returns the {@code double} value present at the given index. + * + * @throws IndexOutOfBoundsException if {@code index} is negative, or greater than or equal to + * {@link #length} + */ + public double get(int index) { + return array[index]; + } + + /** + * Returns {@code true} if {@code object} is an {@code ImmutableDoubleArray} containing the same + * values as this one, in the same order. + */ + @Override + public boolean equals(@Nullable Object object) { + if (object == this) { + return true; + } + if (!(object instanceof ImmutableDoubleArray)) { + return false; + } + ImmutableDoubleArray that = (ImmutableDoubleArray) object; + return Arrays.equals(this.array, that.array); + } + + /** Returns an unspecified hash code for the contents of this immutable array. */ + @Override + public int hashCode() { + int hash = 1; + for (double value : array) { + hash *= 31; + hash += ((Double) value).hashCode(); + } + return hash; + } + + /** + * Returns a string representation of this array in the same form as {@link + * Arrays#toString(double[])}, for example {@code "[1, 2, 3]"}. + */ + @Override + public String toString() { + if (length() == 0) { + return "[]"; + } + StringBuilder builder = new StringBuilder(length() * 5); // rough estimate is fine + builder.append('[').append(array[0]); + + for (int i = 1; i < length(); i++) { + builder.append(", ").append(array[i]); + } + builder.append(']'); + return builder.toString(); + } +} diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableLongArray.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableLongArray.java new file mode 100644 index 00000000000..e1ec8a1db60 --- /dev/null +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableLongArray.java @@ -0,0 +1,113 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.metrics.aggregator; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import javax.annotation.Nullable; +import javax.annotation.concurrent.Immutable; + +@Immutable +class ImmutableLongArray { + private static final ImmutableLongArray EMPTY = new ImmutableLongArray(new long[0]); + + /** Returns an immutable array containing a single value. */ + public static ImmutableLongArray of(long e0) { + return new ImmutableLongArray(new long[] {e0}); + } + + /** Returns an immutable array containing the given values, in order. */ + public static ImmutableLongArray copyOf(long[] values) { + return values.length == 0 + ? EMPTY + : new ImmutableLongArray(Arrays.copyOf(values, values.length)); + } + + private final long[] array; + + private ImmutableLongArray(long[] array) { + this.array = array; + } + + /** Returns a copy of the underlying data as list. */ + public List toList() { + List result = new ArrayList<>(array.length); + for (long v : array) { + result.add(v); + } + return result; + } + + /** Returns the number of values in this array. */ + public int length() { + return array.length; + } + + /** + * Returns the {@code long} value present at the given index. + * + * @throws IndexOutOfBoundsException if {@code index} is negative, or greater than or equal to + * {@link #length} + */ + public long get(int index) { + return array[index]; + } + + /** + * Returns {@code true} if {@code object} is an {@code ImmutableLongArray} containing the same + * values as this one, in the same order. + */ + @Override + public boolean equals(@Nullable Object object) { + if (object == this) { + return true; + } + if (!(object instanceof ImmutableLongArray)) { + return false; + } + ImmutableLongArray that = (ImmutableLongArray) object; + if (this.length() != that.length()) { + return false; + } + for (int i = 0; i < length(); i++) { + if (this.get(i) != that.get(i)) { + return false; + } + } + return true; + } + + /** Returns an unspecified hash code for the contents of this immutable array. */ + @Override + public int hashCode() { + int hash = 1; + for (long value : array) { + hash *= 31; + hash += (int) (value ^ (value >>> 32)); + } + return hash; + } + + /** + * Returns a string representation of this array in the same form as {@link + * Arrays#toString(long[])}, for example {@code "[1, 2, 3]"}. + */ + @Override + public String toString() { + if (length() == 0) { + return "[]"; + } + StringBuilder builder = new StringBuilder(length() * 5); // rough estimate is fine + builder.append('[').append(array[0]); + + for (int i = 1; i < length(); i++) { + builder.append(", ").append(array[i]); + } + builder.append(']'); + return builder.toString(); + } +} diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactoryTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactoryTest.java index f908ab243d0..b0a3cc12c1c 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactoryTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactoryTest.java @@ -13,6 +13,7 @@ import io.opentelemetry.sdk.metrics.common.InstrumentValueType; import io.opentelemetry.sdk.metrics.data.AggregationTemporality; import io.opentelemetry.sdk.resources.Resource; +import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; class AggregatorFactoryTest { @@ -123,4 +124,73 @@ void getSumAggregatorFactory() { InstrumentValueType.DOUBLE))) .isInstanceOf(DoubleSumAggregator.class); } + + @Test + void getHistogramAggregatorFactory() { + AggregatorFactory histogram = + AggregatorFactory.histogram(new double[] {1.0}, /* stateful= */ false); + assertThat( + histogram.create( + Resource.getDefault(), + InstrumentationLibraryInfo.empty(), + InstrumentDescriptor.create( + "name", + "description", + "unit", + InstrumentType.VALUE_RECORDER, + InstrumentValueType.LONG))) + .isInstanceOf(DoubleHistogramAggregator.class); + assertThat( + histogram.create( + Resource.getDefault(), + InstrumentationLibraryInfo.empty(), + InstrumentDescriptor.create( + "name", + "description", + "unit", + InstrumentType.VALUE_RECORDER, + InstrumentValueType.DOUBLE))) + .isInstanceOf(DoubleHistogramAggregator.class); + + assertThat( + histogram + .create( + Resource.getDefault(), + InstrumentationLibraryInfo.empty(), + InstrumentDescriptor.create( + "name", + "description", + "unit", + InstrumentType.VALUE_RECORDER, + InstrumentValueType.LONG)) + .isStateful()) + .isFalse(); + assertThat( + AggregatorFactory.histogram(new double[] {1.0}, /* stateful= */ true) + .create( + Resource.getDefault(), + InstrumentationLibraryInfo.empty(), + InstrumentDescriptor.create( + "name", + "description", + "unit", + InstrumentType.VALUE_RECORDER, + InstrumentValueType.DOUBLE)) + .isStateful()) + .isTrue(); + + Assertions.assertThrows( + IllegalArgumentException.class, + () -> + AggregatorFactory.histogram( + new double[] {Double.NEGATIVE_INFINITY}, /* stateful= */ false)); + Assertions.assertThrows( + IllegalArgumentException.class, + () -> + AggregatorFactory.histogram( + new double[] {1, Double.POSITIVE_INFINITY}, /* stateful= */ false)); + Assertions.assertThrows( + IllegalArgumentException.class, + () -> AggregatorFactory.histogram(new double[] {2, 1, 3}, /* stateful= */ false)); + } } diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java new file mode 100644 index 00000000000..7443db7a0d7 --- /dev/null +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java @@ -0,0 +1,150 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.metrics.aggregator; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.google.errorprone.annotations.concurrent.GuardedBy; +import io.opentelemetry.sdk.common.InstrumentationLibraryInfo; +import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor; +import io.opentelemetry.sdk.metrics.common.InstrumentType; +import io.opentelemetry.sdk.metrics.common.InstrumentValueType; +import io.opentelemetry.sdk.resources.Resource; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import javax.annotation.Nullable; +import org.junit.jupiter.api.Test; + +public class DoubleHistogramAggregatorTest { + private static final DoubleHistogramAggregator aggregator = + new DoubleHistogramAggregator( + Resource.getDefault(), + InstrumentationLibraryInfo.empty(), + InstrumentDescriptor.create( + "name", + "description", + "unit", + InstrumentType.VALUE_RECORDER, + InstrumentValueType.LONG), + ImmutableDoubleArray.copyOf(new double[] {10.0, 100.0, 1000.0}), + /* stateful= */ false); + + @Test + void createHandle() { + assertThat(aggregator.createHandle()).isInstanceOf(DoubleHistogramAggregator.Handle.class); + } + + @Test + void testRecordings() { + AggregatorHandle aggregatorHandle = aggregator.createHandle(); + aggregatorHandle.recordLong(20); + aggregatorHandle.recordLong(5); + aggregatorHandle.recordLong(150); + aggregatorHandle.recordLong(2000); + assertThat(aggregatorHandle.accumulateThenReset()) + .isEqualTo( + HistogramAccumulation.create(2175, ImmutableLongArray.copyOf(new long[] {1, 1, 1, 1}))); + } + + @Test + void toAccumulationAndReset() { + AggregatorHandle aggregatorHandle = aggregator.createHandle(); + assertThat(aggregatorHandle.accumulateThenReset()).isNull(); + + aggregatorHandle.recordLong(100); + assertThat(aggregatorHandle.accumulateThenReset()) + .isEqualTo( + HistogramAccumulation.create(100, ImmutableLongArray.copyOf(new long[] {0, 1, 0, 0}))); + assertThat(aggregatorHandle.accumulateThenReset()).isNull(); + + aggregatorHandle.recordLong(0); + assertThat(aggregatorHandle.accumulateThenReset()) + .isEqualTo( + HistogramAccumulation.create(0, ImmutableLongArray.copyOf(new long[] {1, 0, 0, 0}))); + assertThat(aggregatorHandle.accumulateThenReset()).isNull(); + } + + @Test + void accumulateData() { + assertThat(aggregator.accumulateDouble(2.0)) + .isEqualTo(HistogramAccumulation.create(2.0, ImmutableLongArray.of(1))); + assertThat(aggregator.accumulateLong(10)) + .isEqualTo(HistogramAccumulation.create(10.0, ImmutableLongArray.of(1))); + } + + @Test + void testMultithreadedUpdates() throws Exception { + final AggregatorHandle aggregatorHandle = aggregator.createHandle(); + final Histogram summarizer = new Histogram(); + int numberOfThreads = 10; + final long[] updates = new long[] {1, 2, 3, 5, 7, 11, 13, 17, 19, 23}; + final int numberOfUpdates = 1000; + final CountDownLatch startingGun = new CountDownLatch(numberOfThreads); + List workers = new ArrayList<>(); + for (int i = 0; i < numberOfThreads; i++) { + final int index = i; + Thread t = + new Thread( + () -> { + long update = updates[index]; + try { + startingGun.await(); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + for (int j = 0; j < numberOfUpdates; j++) { + aggregatorHandle.recordLong(update); + if (ThreadLocalRandom.current().nextInt(10) == 0) { + summarizer.process(aggregatorHandle.accumulateThenReset()); + } + } + }); + workers.add(t); + t.start(); + } + for (int i = 0; i <= numberOfThreads; i++) { + startingGun.countDown(); + } + + for (Thread worker : workers) { + worker.join(); + } + // make sure everything gets merged when all the aggregation is done. + summarizer.process(aggregatorHandle.accumulateThenReset()); + + assertThat(summarizer.accumulation) + .isEqualTo( + HistogramAccumulation.create( + 101000, ImmutableLongArray.copyOf(new long[] {5000, 5000, 0, 0}))); + } + + private static final class Histogram { + private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); + + @GuardedBy("lock") + @Nullable + private HistogramAccumulation accumulation; + + void process(@Nullable HistogramAccumulation other) { + if (other == null) { + return; + } + lock.writeLock().lock(); + try { + if (accumulation == null) { + accumulation = other; + return; + } + accumulation = aggregator.merge(accumulation, other); + } finally { + lock.writeLock().unlock(); + } + } + } +} From 7ffbc2a206c2d5d75798499b0d7dc16d3a4c31cd Mon Sep 17 00:00:00 2001 From: beanliu Date: Thu, 25 Feb 2021 21:48:27 +1100 Subject: [PATCH 02/16] implement DoubleHistogramAggregator.toMetricData --- .../aggregator/DoubleHistogramAggregator.java | 15 ++++++++----- .../metrics/aggregator/MetricDataUtils.java | 20 +++++++++++++++++ .../DoubleHistogramAggregatorTest.java | 22 +++++++++++++++++++ 3 files changed, 52 insertions(+), 5 deletions(-) diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java index f516fcba250..d7bf3a71cd1 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java @@ -8,11 +8,11 @@ import io.opentelemetry.api.metrics.common.Labels; import io.opentelemetry.sdk.common.InstrumentationLibraryInfo; import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor; -import io.opentelemetry.sdk.metrics.data.DoubleGaugeData; +import io.opentelemetry.sdk.metrics.data.AggregationTemporality; +import io.opentelemetry.sdk.metrics.data.DoubleHistogramData; import io.opentelemetry.sdk.metrics.data.MetricData; import io.opentelemetry.sdk.resources.Resource; import java.util.Arrays; -import java.util.Collections; import java.util.Map; import java.util.concurrent.locks.ReentrantLock; import javax.annotation.concurrent.GuardedBy; @@ -56,14 +56,19 @@ public final MetricData toMetricData( long startEpochNanos, long lastCollectionEpoch, long epochNanos) { - // effectively no-op, will convert to histogram data in other PRs - return MetricData.createDoubleGauge( + return MetricData.createDoubleHistogram( getResource(), getInstrumentationLibraryInfo(), getInstrumentDescriptor().getName(), getInstrumentDescriptor().getDescription(), getInstrumentDescriptor().getUnit(), - DoubleGaugeData.create(Collections.emptyList())); + DoubleHistogramData.create( + isStateful() ? AggregationTemporality.CUMULATIVE : AggregationTemporality.DELTA, + MetricDataUtils.toDoubleHistogramPointList( + accumulationByLabels, + isStateful() ? startEpochNanos : lastCollectionEpoch, + epochNanos, + boundaries.toList()))); } @Override diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/MetricDataUtils.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/MetricDataUtils.java index 9574632271b..af156848831 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/MetricDataUtils.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/MetricDataUtils.java @@ -6,6 +6,7 @@ package io.opentelemetry.sdk.metrics.aggregator; import io.opentelemetry.api.metrics.common.Labels; +import io.opentelemetry.sdk.metrics.data.DoubleHistogramPointData; import io.opentelemetry.sdk.metrics.data.DoublePointData; import io.opentelemetry.sdk.metrics.data.DoubleSummaryPointData; import io.opentelemetry.sdk.metrics.data.LongPointData; @@ -44,4 +45,23 @@ static List toDoubleSummaryPointList( points.add(aggregator.toPoint(startEpochNanos, epochNanos, labels))); return points; } + + static List toDoubleHistogramPointList( + Map accumulationMap, + long startEpochNanos, + long epochNanos, + List boundaries) { + List points = new ArrayList<>(accumulationMap.size()); + accumulationMap.forEach( + (labels, aggregator) -> + points.add( + DoubleHistogramPointData.create( + startEpochNanos, + epochNanos, + labels, + aggregator.getSum(), + boundaries, + aggregator.getCounts().toList()))); + return points; + } } diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java index 7443db7a0d7..9a617890996 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java @@ -8,12 +8,17 @@ import static org.assertj.core.api.Assertions.assertThat; import com.google.errorprone.annotations.concurrent.GuardedBy; +import io.opentelemetry.api.metrics.common.Labels; import io.opentelemetry.sdk.common.InstrumentationLibraryInfo; import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor; import io.opentelemetry.sdk.metrics.common.InstrumentType; import io.opentelemetry.sdk.metrics.common.InstrumentValueType; +import io.opentelemetry.sdk.metrics.data.AggregationTemporality; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.metrics.data.MetricDataType; import io.opentelemetry.sdk.resources.Resource; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ThreadLocalRandom; @@ -78,6 +83,23 @@ void accumulateData() { .isEqualTo(HistogramAccumulation.create(10.0, ImmutableLongArray.of(1))); } + @Test + void toMetricData() { + AggregatorHandle aggregatorHandle = aggregator.createHandle(); + aggregatorHandle.recordLong(10); + + MetricData metricData = + aggregator.toMetricData( + Collections.singletonMap(Labels.empty(), aggregatorHandle.accumulateThenReset()), + 0, + 10, + 100); + assertThat(metricData).isNotNull(); + assertThat(metricData.getType()).isEqualTo(MetricDataType.HISTOGRAM); + assertThat(metricData.getDoubleHistogramData().getAggregationTemporality()) + .isEqualTo(AggregationTemporality.DELTA); + } + @Test void testMultithreadedUpdates() throws Exception { final AggregatorHandle aggregatorHandle = aggregator.createHandle(); From a0aff16c8da72c78d61c22e19e5154eb1101a098 Mon Sep 17 00:00:00 2001 From: beanliu Date: Thu, 25 Feb 2021 21:57:46 +1100 Subject: [PATCH 03/16] pass temporality instead of a boolean for the creation of histogram aggregator --- .../metrics/aggregator/DoubleHistogramBenchmark.java | 3 ++- .../sdk/metrics/aggregator/AggregatorFactory.java | 6 +++--- .../metrics/aggregator/HistogramAggregatorFactory.java | 10 ++++++---- .../sdk/metrics/aggregator/AggregatorFactoryTest.java | 10 +++++----- 4 files changed, 16 insertions(+), 13 deletions(-) diff --git a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java index f70ce64f656..1e77f05083e 100644 --- a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java +++ b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java @@ -9,6 +9,7 @@ import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor; import io.opentelemetry.sdk.metrics.common.InstrumentType; import io.opentelemetry.sdk.metrics.common.InstrumentValueType; +import io.opentelemetry.sdk.metrics.data.AggregationTemporality; import io.opentelemetry.sdk.resources.Resource; import java.util.concurrent.TimeUnit; import org.openjdk.jmh.annotations.Benchmark; @@ -25,7 +26,7 @@ @State(Scope.Benchmark) public class DoubleHistogramBenchmark { private static final Aggregator aggregator = - AggregatorFactory.histogram(new double[] {10, 100, 1_000}, /* stateful= */ false) + AggregatorFactory.histogram(new double[] {10, 100, 1_000}, AggregationTemporality.DELTA) .create( Resource.getDefault(), InstrumentationLibraryInfo.empty(), diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactory.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactory.java index 1c4a627d374..c8d5fa51194 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactory.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactory.java @@ -81,12 +81,12 @@ static AggregatorFactory minMaxSumCount() { * Returns an {@code AggregatorFactory} that calculates an approximation of the distribution of * the measurements taken. * - * @param stateful configures if the aggregator is stateful. + * @param temporality configures what temporality to be produced for the Histogram metrics. * @param boundaries configures the fixed bucket boundaries. * @return an {@code AggregationFactory} that calculates histogram of recorded measurements. */ - static AggregatorFactory histogram(double[] boundaries, boolean stateful) { - return new HistogramAggregatorFactory(boundaries, stateful); + static AggregatorFactory histogram(double[] boundaries, AggregationTemporality temporality) { + return new HistogramAggregatorFactory(boundaries, temporality); } /** diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java index 49c3e40bf88..f7477efc171 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java @@ -7,15 +7,16 @@ import io.opentelemetry.sdk.common.InstrumentationLibraryInfo; import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor; +import io.opentelemetry.sdk.metrics.data.AggregationTemporality; import io.opentelemetry.sdk.resources.Resource; final class HistogramAggregatorFactory implements AggregatorFactory { private final ImmutableDoubleArray boundaries; - private final boolean stateful; + private final AggregationTemporality temporality; - HistogramAggregatorFactory(double[] boundaries, boolean stateful) { + HistogramAggregatorFactory(double[] boundaries, AggregationTemporality temporality) { this.boundaries = ImmutableDoubleArray.copyOf(boundaries); - this.stateful = stateful; + this.temporality = temporality; for (int i = 1; i < this.boundaries.length(); ++i) { if (Double.compare(this.boundaries.get(i - 1), this.boundaries.get(i)) >= 0) { @@ -42,12 +43,13 @@ public Aggregator create( Resource resource, InstrumentationLibraryInfo instrumentationLibraryInfo, InstrumentDescriptor descriptor) { + final boolean stateful = this.temporality == AggregationTemporality.CUMULATIVE; switch (descriptor.getValueType()) { case LONG: case DOUBLE: return (Aggregator) new DoubleHistogramAggregator( - resource, instrumentationLibraryInfo, descriptor, this.boundaries, this.stateful); + resource, instrumentationLibraryInfo, descriptor, this.boundaries, stateful); } throw new IllegalArgumentException("Invalid instrument value type"); } diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactoryTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactoryTest.java index b0a3cc12c1c..f956cbdddee 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactoryTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactoryTest.java @@ -128,7 +128,7 @@ void getSumAggregatorFactory() { @Test void getHistogramAggregatorFactory() { AggregatorFactory histogram = - AggregatorFactory.histogram(new double[] {1.0}, /* stateful= */ false); + AggregatorFactory.histogram(new double[] {1.0}, AggregationTemporality.DELTA); assertThat( histogram.create( Resource.getDefault(), @@ -166,7 +166,7 @@ void getHistogramAggregatorFactory() { .isStateful()) .isFalse(); assertThat( - AggregatorFactory.histogram(new double[] {1.0}, /* stateful= */ true) + AggregatorFactory.histogram(new double[] {1.0}, AggregationTemporality.CUMULATIVE) .create( Resource.getDefault(), InstrumentationLibraryInfo.empty(), @@ -183,14 +183,14 @@ void getHistogramAggregatorFactory() { IllegalArgumentException.class, () -> AggregatorFactory.histogram( - new double[] {Double.NEGATIVE_INFINITY}, /* stateful= */ false)); + new double[] {Double.NEGATIVE_INFINITY}, AggregationTemporality.DELTA)); Assertions.assertThrows( IllegalArgumentException.class, () -> AggregatorFactory.histogram( - new double[] {1, Double.POSITIVE_INFINITY}, /* stateful= */ false)); + new double[] {1, Double.POSITIVE_INFINITY}, AggregationTemporality.DELTA)); Assertions.assertThrows( IllegalArgumentException.class, - () -> AggregatorFactory.histogram(new double[] {2, 1, 3}, /* stateful= */ false)); + () -> AggregatorFactory.histogram(new double[] {2, 1, 3}, AggregationTemporality.DELTA)); } } From 9456910b48661b8ea3c65bc4060eb105e58b3107 Mon Sep 17 00:00:00 2001 From: beanliu Date: Thu, 25 Feb 2021 22:30:00 +1100 Subject: [PATCH 04/16] From a53b6c909916e54b3ea3e42f7b4f16631df31bad Mon Sep 17 00:00:00 2001 From: beanliu Date: Fri, 26 Feb 2021 08:00:38 +1100 Subject: [PATCH 05/16] remove ImmutableDoubleArray --- .../aggregator/DoubleHistogramAggregator.java | 24 +++-- .../HistogramAggregatorFactory.java | 19 ++-- .../aggregator/ImmutableDoubleArray.java | 100 ------------------ .../DoubleHistogramAggregatorTest.java | 2 +- 4 files changed, 24 insertions(+), 121 deletions(-) delete mode 100644 sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableDoubleArray.java diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java index d7bf3a71cd1..ddac7f76648 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java @@ -12,19 +12,21 @@ import io.opentelemetry.sdk.metrics.data.DoubleHistogramData; import io.opentelemetry.sdk.metrics.data.MetricData; import io.opentelemetry.sdk.resources.Resource; +import java.util.ArrayList; import java.util.Arrays; +import java.util.List; import java.util.Map; import java.util.concurrent.locks.ReentrantLock; import javax.annotation.concurrent.GuardedBy; final class DoubleHistogramAggregator extends AbstractAggregator { - private final ImmutableDoubleArray boundaries; + private final double[] boundaries; DoubleHistogramAggregator( Resource resource, InstrumentationLibraryInfo instrumentationLibraryInfo, InstrumentDescriptor instrumentDescriptor, - ImmutableDoubleArray boundaries, + double[] boundaries, boolean stateful) { super(resource, instrumentationLibraryInfo, instrumentDescriptor, stateful); this.boundaries = boundaries; @@ -56,6 +58,10 @@ public final MetricData toMetricData( long startEpochNanos, long lastCollectionEpoch, long epochNanos) { + List boundaries = new ArrayList<>(this.boundaries.length); + for (double v : this.boundaries) { + boundaries.add(v); + } return MetricData.createDoubleHistogram( getResource(), getInstrumentationLibraryInfo(), @@ -68,7 +74,7 @@ public final MetricData toMetricData( accumulationByLabels, isStateful() ? startEpochNanos : lastCollectionEpoch, epochNanos, - boundaries.toList()))); + boundaries))); } @Override @@ -82,27 +88,27 @@ public HistogramAccumulation accumulateLong(long value) { } static final class Handle extends AggregatorHandle { - private final ImmutableDoubleArray boundaries; + private final double[] boundaries; private final ReentrantLock lock = new ReentrantLock(); @GuardedBy("lock") private final State current; - Handle(ImmutableDoubleArray boundaries) { + Handle(double[] boundaries) { this.boundaries = boundaries; - this.current = new State(this.boundaries.length() + 1); + this.current = new State(this.boundaries.length + 1); } // Benchmark shows that linear search performs better than binary search with ordinary // buckets. private int findBucketIndex(double value) { - for (int i = 0; i < boundaries.length(); ++i) { - if (Double.compare(value, boundaries.get(i)) <= 0) { + for (int i = 0; i < boundaries.length; ++i) { + if (Double.compare(value, boundaries[i]) <= 0) { return i; } } - return boundaries.length(); + return boundaries.length; } @Override diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java index f7477efc171..e92d27ec935 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java @@ -11,27 +11,24 @@ import io.opentelemetry.sdk.resources.Resource; final class HistogramAggregatorFactory implements AggregatorFactory { - private final ImmutableDoubleArray boundaries; + private final double[] boundaries; private final AggregationTemporality temporality; HistogramAggregatorFactory(double[] boundaries, AggregationTemporality temporality) { - this.boundaries = ImmutableDoubleArray.copyOf(boundaries); + this.boundaries = boundaries; this.temporality = temporality; - for (int i = 1; i < this.boundaries.length(); ++i) { - if (Double.compare(this.boundaries.get(i - 1), this.boundaries.get(i)) >= 0) { + for (int i = 1; i < this.boundaries.length; ++i) { + if (Double.compare(this.boundaries[i - 1], this.boundaries[i]) >= 0) { throw new IllegalArgumentException( - "invalid bucket boundary: " - + this.boundaries.get(i - 1) - + " >= " - + this.boundaries.get(i)); + "invalid bucket boundary: " + this.boundaries[i - 1] + " >= " + this.boundaries[i]); } } - if (this.boundaries.length() > 0) { - if (this.boundaries.get(0) == Double.NEGATIVE_INFINITY) { + if (this.boundaries.length > 0) { + if (this.boundaries[0] == Double.NEGATIVE_INFINITY) { throw new IllegalArgumentException("invalid bucket boundary: -Inf"); } - if (this.boundaries.get(this.boundaries.length() - 1) == Double.POSITIVE_INFINITY) { + if (this.boundaries[this.boundaries.length - 1] == Double.POSITIVE_INFINITY) { throw new IllegalArgumentException("invalid bucket boundary: +Inf"); } } diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableDoubleArray.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableDoubleArray.java deleted file mode 100644 index 22c9f146bbc..00000000000 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableDoubleArray.java +++ /dev/null @@ -1,100 +0,0 @@ -/* - * Copyright The OpenTelemetry Authors - * SPDX-License-Identifier: Apache-2.0 - */ - -package io.opentelemetry.sdk.metrics.aggregator; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; -import javax.annotation.Nullable; -import javax.annotation.concurrent.Immutable; - -@Immutable -class ImmutableDoubleArray { - private static final ImmutableDoubleArray EMPTY = new ImmutableDoubleArray(new double[0]); - - /** Returns an immutable array containing the given values, in order. */ - public static ImmutableDoubleArray copyOf(double[] values) { - return values.length == 0 - ? EMPTY - : new ImmutableDoubleArray(Arrays.copyOf(values, values.length)); - } - - private final double[] array; - - private ImmutableDoubleArray(double[] array) { - this.array = array; - } - - /** Returns a copy of the underlying data as list. */ - public List toList() { - List result = new ArrayList<>(array.length); - for (double v : array) { - result.add(v); - } - return result; - } - - /** Returns the number of values in this array. */ - public int length() { - return array.length; - } - - /** - * Returns the {@code double} value present at the given index. - * - * @throws IndexOutOfBoundsException if {@code index} is negative, or greater than or equal to - * {@link #length} - */ - public double get(int index) { - return array[index]; - } - - /** - * Returns {@code true} if {@code object} is an {@code ImmutableDoubleArray} containing the same - * values as this one, in the same order. - */ - @Override - public boolean equals(@Nullable Object object) { - if (object == this) { - return true; - } - if (!(object instanceof ImmutableDoubleArray)) { - return false; - } - ImmutableDoubleArray that = (ImmutableDoubleArray) object; - return Arrays.equals(this.array, that.array); - } - - /** Returns an unspecified hash code for the contents of this immutable array. */ - @Override - public int hashCode() { - int hash = 1; - for (double value : array) { - hash *= 31; - hash += ((Double) value).hashCode(); - } - return hash; - } - - /** - * Returns a string representation of this array in the same form as {@link - * Arrays#toString(double[])}, for example {@code "[1, 2, 3]"}. - */ - @Override - public String toString() { - if (length() == 0) { - return "[]"; - } - StringBuilder builder = new StringBuilder(length() * 5); // rough estimate is fine - builder.append('[').append(array[0]); - - for (int i = 1; i < length(); i++) { - builder.append(", ").append(array[i]); - } - builder.append(']'); - return builder.toString(); - } -} diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java index 9a617890996..dd82b88c79a 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java @@ -37,7 +37,7 @@ public class DoubleHistogramAggregatorTest { "unit", InstrumentType.VALUE_RECORDER, InstrumentValueType.LONG), - ImmutableDoubleArray.copyOf(new double[] {10.0, 100.0, 1000.0}), + new double[] {10.0, 100.0, 1000.0}, /* stateful= */ false); @Test From 9062d33ca175c289846c7f2dce54cda2e5006294 Mon Sep 17 00:00:00 2001 From: beanliu Date: Fri, 26 Feb 2021 08:13:44 +1100 Subject: [PATCH 06/16] remove ImmutableLongArray --- .../aggregator/DoubleHistogramAggregator.java | 18 +-- .../aggregator/HistogramAccumulation.java | 13 +- .../aggregator/ImmutableLongArray.java | 113 ------------------ .../metrics/aggregator/MetricDataUtils.java | 18 +-- .../DoubleHistogramAggregatorTest.java | 17 +-- 5 files changed, 33 insertions(+), 146 deletions(-) delete mode 100644 sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableLongArray.java diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java index ddac7f76648..8bc2b4d0617 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java @@ -20,6 +20,8 @@ import javax.annotation.concurrent.GuardedBy; final class DoubleHistogramAggregator extends AbstractAggregator { + private static final long[] countsOfOne = new long[] {1}; + private final double[] boundaries; DoubleHistogramAggregator( @@ -44,12 +46,11 @@ public AggregatorHandle createHandle() { */ @Override public final HistogramAccumulation merge(HistogramAccumulation x, HistogramAccumulation y) { - long[] mergedCounts = new long[x.getCounts().length()]; - for (int i = 0; i < x.getCounts().length(); ++i) { - mergedCounts[i] = x.getCounts().get(i) + y.getCounts().get(i); + long[] mergedCounts = new long[x.getCounts().length]; + for (int i = 0; i < x.getCounts().length; ++i) { + mergedCounts[i] = x.getCounts()[i] + y.getCounts()[i]; } - return HistogramAccumulation.create( - x.getSum() + y.getSum(), ImmutableLongArray.copyOf(mergedCounts)); + return HistogramAccumulation.create(x.getSum() + y.getSum(), mergedCounts); } @Override @@ -79,12 +80,12 @@ public final MetricData toMetricData( @Override public HistogramAccumulation accumulateDouble(double value) { - return HistogramAccumulation.create(value, ImmutableLongArray.of(1)); + return HistogramAccumulation.create(value, countsOfOne); } @Override public HistogramAccumulation accumulateLong(long value) { - return HistogramAccumulation.create(value, ImmutableLongArray.of(1)); + return HistogramAccumulation.create(value, countsOfOne); } static final class Handle extends AggregatorHandle { @@ -116,7 +117,8 @@ protected HistogramAccumulation doAccumulateThenReset() { lock.lock(); try { HistogramAccumulation acc = - HistogramAccumulation.create(current.sum, ImmutableLongArray.copyOf(current.counts)); + HistogramAccumulation.create( + current.sum, Arrays.copyOf(current.counts, current.counts.length)); current.reset(); return acc; } finally { diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAccumulation.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAccumulation.java index dd2aad43608..7a8557f9820 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAccumulation.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAccumulation.java @@ -12,11 +12,12 @@ @AutoValue abstract class HistogramAccumulation { /** - * Creates a new {@link HistogramAccumulation} with the given values. + * Creates a new {@link HistogramAccumulation} with the given values. Assume `counts` is read-only + * so we don't need a defensive-copy here. * * @return a new {@link HistogramAccumulation} with the given values. */ - static HistogramAccumulation create(double sum, ImmutableLongArray counts) { + static HistogramAccumulation create(double sum, long[] counts) { return new AutoValue_HistogramAccumulation(sum, counts); } @@ -30,9 +31,11 @@ static HistogramAccumulation create(double sum, ImmutableLongArray counts) { abstract double getSum(); /** - * The counts in each bucket. + * The counts in each bucket. The returned type is a mutable object, but it should be fine because + * the class is only used internally. * - * @return the counts in each bucket. + * @return the counts in each bucket. do not mutate the returned object. */ - abstract ImmutableLongArray getCounts(); + @SuppressWarnings("mutable") + abstract long[] getCounts(); } diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableLongArray.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableLongArray.java deleted file mode 100644 index e1ec8a1db60..00000000000 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/ImmutableLongArray.java +++ /dev/null @@ -1,113 +0,0 @@ -/* - * Copyright The OpenTelemetry Authors - * SPDX-License-Identifier: Apache-2.0 - */ - -package io.opentelemetry.sdk.metrics.aggregator; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; -import javax.annotation.Nullable; -import javax.annotation.concurrent.Immutable; - -@Immutable -class ImmutableLongArray { - private static final ImmutableLongArray EMPTY = new ImmutableLongArray(new long[0]); - - /** Returns an immutable array containing a single value. */ - public static ImmutableLongArray of(long e0) { - return new ImmutableLongArray(new long[] {e0}); - } - - /** Returns an immutable array containing the given values, in order. */ - public static ImmutableLongArray copyOf(long[] values) { - return values.length == 0 - ? EMPTY - : new ImmutableLongArray(Arrays.copyOf(values, values.length)); - } - - private final long[] array; - - private ImmutableLongArray(long[] array) { - this.array = array; - } - - /** Returns a copy of the underlying data as list. */ - public List toList() { - List result = new ArrayList<>(array.length); - for (long v : array) { - result.add(v); - } - return result; - } - - /** Returns the number of values in this array. */ - public int length() { - return array.length; - } - - /** - * Returns the {@code long} value present at the given index. - * - * @throws IndexOutOfBoundsException if {@code index} is negative, or greater than or equal to - * {@link #length} - */ - public long get(int index) { - return array[index]; - } - - /** - * Returns {@code true} if {@code object} is an {@code ImmutableLongArray} containing the same - * values as this one, in the same order. - */ - @Override - public boolean equals(@Nullable Object object) { - if (object == this) { - return true; - } - if (!(object instanceof ImmutableLongArray)) { - return false; - } - ImmutableLongArray that = (ImmutableLongArray) object; - if (this.length() != that.length()) { - return false; - } - for (int i = 0; i < length(); i++) { - if (this.get(i) != that.get(i)) { - return false; - } - } - return true; - } - - /** Returns an unspecified hash code for the contents of this immutable array. */ - @Override - public int hashCode() { - int hash = 1; - for (long value : array) { - hash *= 31; - hash += (int) (value ^ (value >>> 32)); - } - return hash; - } - - /** - * Returns a string representation of this array in the same form as {@link - * Arrays#toString(long[])}, for example {@code "[1, 2, 3]"}. - */ - @Override - public String toString() { - if (length() == 0) { - return "[]"; - } - StringBuilder builder = new StringBuilder(length() * 5); // rough estimate is fine - builder.append('[').append(array[0]); - - for (int i = 1; i < length(); i++) { - builder.append(", ").append(array[i]); - } - builder.append(']'); - return builder.toString(); - } -} diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/MetricDataUtils.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/MetricDataUtils.java index af156848831..dd7a233810f 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/MetricDataUtils.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/MetricDataUtils.java @@ -53,15 +53,15 @@ static List toDoubleHistogramPointList( List boundaries) { List points = new ArrayList<>(accumulationMap.size()); accumulationMap.forEach( - (labels, aggregator) -> - points.add( - DoubleHistogramPointData.create( - startEpochNanos, - epochNanos, - labels, - aggregator.getSum(), - boundaries, - aggregator.getCounts().toList()))); + (labels, aggregator) -> { + List counts = new ArrayList<>(aggregator.getCounts().length); + for (long v : aggregator.getCounts()) { + counts.add(v); + } + points.add( + DoubleHistogramPointData.create( + startEpochNanos, epochNanos, labels, aggregator.getSum(), boundaries, counts)); + }); return points; } } diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java index dd82b88c79a..d8762143d1e 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java @@ -53,8 +53,7 @@ void testRecordings() { aggregatorHandle.recordLong(150); aggregatorHandle.recordLong(2000); assertThat(aggregatorHandle.accumulateThenReset()) - .isEqualTo( - HistogramAccumulation.create(2175, ImmutableLongArray.copyOf(new long[] {1, 1, 1, 1}))); + .isEqualTo(HistogramAccumulation.create(2175, new long[] {1, 1, 1, 1})); } @Test @@ -64,23 +63,21 @@ void toAccumulationAndReset() { aggregatorHandle.recordLong(100); assertThat(aggregatorHandle.accumulateThenReset()) - .isEqualTo( - HistogramAccumulation.create(100, ImmutableLongArray.copyOf(new long[] {0, 1, 0, 0}))); + .isEqualTo(HistogramAccumulation.create(100, new long[] {0, 1, 0, 0})); assertThat(aggregatorHandle.accumulateThenReset()).isNull(); aggregatorHandle.recordLong(0); assertThat(aggregatorHandle.accumulateThenReset()) - .isEqualTo( - HistogramAccumulation.create(0, ImmutableLongArray.copyOf(new long[] {1, 0, 0, 0}))); + .isEqualTo(HistogramAccumulation.create(0, new long[] {1, 0, 0, 0})); assertThat(aggregatorHandle.accumulateThenReset()).isNull(); } @Test void accumulateData() { assertThat(aggregator.accumulateDouble(2.0)) - .isEqualTo(HistogramAccumulation.create(2.0, ImmutableLongArray.of(1))); + .isEqualTo(HistogramAccumulation.create(2.0, new long[] {1})); assertThat(aggregator.accumulateLong(10)) - .isEqualTo(HistogramAccumulation.create(10.0, ImmutableLongArray.of(1))); + .isEqualTo(HistogramAccumulation.create(10.0, new long[] {1})); } @Test @@ -141,9 +138,7 @@ void testMultithreadedUpdates() throws Exception { summarizer.process(aggregatorHandle.accumulateThenReset()); assertThat(summarizer.accumulation) - .isEqualTo( - HistogramAccumulation.create( - 101000, ImmutableLongArray.copyOf(new long[] {5000, 5000, 0, 0}))); + .isEqualTo(HistogramAccumulation.create(101000, new long[] {5000, 5000, 0, 0})); } private static final class Histogram { From 590f0dfe33e92012f9fbe5d5aca65834131e2800 Mon Sep 17 00:00:00 2001 From: beanliu Date: Fri, 26 Feb 2021 10:30:23 +1100 Subject: [PATCH 07/16] fixup! remove ImmutableDoubleArray --- .../aggregator/DoubleHistogramAggregator.java | 16 +++++++++++----- .../aggregator/HistogramAggregatorFactory.java | 3 ++- 2 files changed, 13 insertions(+), 6 deletions(-) diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java index 8bc2b4d0617..5a7c6cf34a3 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java @@ -14,6 +14,7 @@ import io.opentelemetry.sdk.resources.Resource; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.locks.ReentrantLock; @@ -24,6 +25,9 @@ final class DoubleHistogramAggregator extends AbstractAggregator boundaryList; + DoubleHistogramAggregator( Resource resource, InstrumentationLibraryInfo instrumentationLibraryInfo, @@ -32,6 +36,12 @@ final class DoubleHistogramAggregator extends AbstractAggregator boundaryList = new ArrayList<>(this.boundaries.length); + for (double v : this.boundaries) { + boundaryList.add(v); + } + this.boundaryList = Collections.unmodifiableList(boundaryList); } @Override @@ -59,10 +69,6 @@ public final MetricData toMetricData( long startEpochNanos, long lastCollectionEpoch, long epochNanos) { - List boundaries = new ArrayList<>(this.boundaries.length); - for (double v : this.boundaries) { - boundaries.add(v); - } return MetricData.createDoubleHistogram( getResource(), getInstrumentationLibraryInfo(), @@ -75,7 +81,7 @@ public final MetricData toMetricData( accumulationByLabels, isStateful() ? startEpochNanos : lastCollectionEpoch, epochNanos, - boundaries))); + boundaryList))); } @Override diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java index e92d27ec935..e6e7b14b5bc 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java @@ -9,13 +9,14 @@ import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor; import io.opentelemetry.sdk.metrics.data.AggregationTemporality; import io.opentelemetry.sdk.resources.Resource; +import java.util.Arrays; final class HistogramAggregatorFactory implements AggregatorFactory { private final double[] boundaries; private final AggregationTemporality temporality; HistogramAggregatorFactory(double[] boundaries, AggregationTemporality temporality) { - this.boundaries = boundaries; + this.boundaries = Arrays.copyOf(boundaries, boundaries.length); this.temporality = temporality; for (int i = 1; i < this.boundaries.length; ++i) { From bc80f0267760c8afac97248ad684ff7b88f6ba85 Mon Sep 17 00:00:00 2001 From: beanliu Date: Fri, 26 Feb 2021 10:39:48 +1100 Subject: [PATCH 08/16] simplify the implementation of DoubleHistogramAggregator --- .../aggregator/DoubleHistogramAggregator.java | 41 +++++++------------ 1 file changed, 14 insertions(+), 27 deletions(-) diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java index 5a7c6cf34a3..9e6c2623121 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java @@ -95,16 +95,22 @@ public HistogramAccumulation accumulateLong(long value) { } static final class Handle extends AggregatorHandle { + // read-only private final double[] boundaries; - private final ReentrantLock lock = new ReentrantLock(); + @GuardedBy("lock") + private double sum; @GuardedBy("lock") - private final State current; + private final long[] counts; + + private final ReentrantLock lock = new ReentrantLock(); Handle(double[] boundaries) { this.boundaries = boundaries; - this.current = new State(this.boundaries.length + 1); + this.counts = new long[this.boundaries.length + 1]; + this.sum = 0; + Arrays.fill(this.counts, 0); } // Benchmark shows that linear search performs better than binary search with ordinary @@ -123,9 +129,9 @@ protected HistogramAccumulation doAccumulateThenReset() { lock.lock(); try { HistogramAccumulation acc = - HistogramAccumulation.create( - current.sum, Arrays.copyOf(current.counts, current.counts.length)); - current.reset(); + HistogramAccumulation.create(sum, Arrays.copyOf(counts, counts.length)); + this.sum = 0; + Arrays.fill(this.counts, 0); return acc; } finally { lock.unlock(); @@ -138,7 +144,8 @@ protected void doRecordDouble(double value) { lock.lock(); try { - current.record(bucketIndex, value); + this.sum += value; + this.counts[bucketIndex]++; } finally { lock.unlock(); } @@ -148,25 +155,5 @@ protected void doRecordDouble(double value) { protected void doRecordLong(long value) { doRecordDouble((double) value); } - - private static final class State { - private double sum; - private final long[] counts; - - public State(int bucketSize) { - this.counts = new long[bucketSize]; - reset(); - } - - private void reset() { - this.sum = 0; - Arrays.fill(this.counts, 0); - } - - private void record(int bucketIndex, double value) { - this.sum += value; - this.counts[bucketIndex]++; - } - } } } From 826523b7dff3eabce7b2145c1c795cfc9fd464c2 Mon Sep 17 00:00:00 2001 From: beanliu Date: Fri, 26 Feb 2021 13:55:35 +1100 Subject: [PATCH 09/16] accumulate value with configured boundaries --- .../aggregator/DoubleHistogramAggregator.java | 33 +++++++++---------- .../DoubleHistogramAggregatorTest.java | 22 ++++++++++--- 2 files changed, 34 insertions(+), 21 deletions(-) diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java index 9e6c2623121..110c98131c8 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java @@ -21,8 +21,6 @@ import javax.annotation.concurrent.GuardedBy; final class DoubleHistogramAggregator extends AbstractAggregator { - private static final long[] countsOfOne = new long[] {1}; - private final double[] boundaries; // a cache for converting to MetricData @@ -86,12 +84,25 @@ public final MetricData toMetricData( @Override public HistogramAccumulation accumulateDouble(double value) { - return HistogramAccumulation.create(value, countsOfOne); + long[] counts = new long[this.boundaries.length + 1]; + counts[findBucketIndex(this.boundaries, value)] = 1; + return HistogramAccumulation.create(value, counts); } @Override public HistogramAccumulation accumulateLong(long value) { - return HistogramAccumulation.create(value, countsOfOne); + return accumulateDouble((double) value); + } + + // Benchmark shows that linear search performs better than binary search with ordinary + // buckets. + private static int findBucketIndex(double[] boundaries, double value) { + for (int i = 0; i < boundaries.length; ++i) { + if (Double.compare(value, boundaries[i]) <= 0) { + return i; + } + } + return boundaries.length; } static final class Handle extends AggregatorHandle { @@ -110,18 +121,6 @@ static final class Handle extends AggregatorHandle { this.boundaries = boundaries; this.counts = new long[this.boundaries.length + 1]; this.sum = 0; - Arrays.fill(this.counts, 0); - } - - // Benchmark shows that linear search performs better than binary search with ordinary - // buckets. - private int findBucketIndex(double value) { - for (int i = 0; i < boundaries.length; ++i) { - if (Double.compare(value, boundaries[i]) <= 0) { - return i; - } - } - return boundaries.length; } @Override @@ -140,7 +139,7 @@ protected HistogramAccumulation doAccumulateThenReset() { @Override protected void doRecordDouble(double value) { - int bucketIndex = findBucketIndex(value); + int bucketIndex = findBucketIndex(this.boundaries, value); lock.lock(); try { diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java index d8762143d1e..868e3b6c482 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java @@ -27,6 +27,7 @@ import org.junit.jupiter.api.Test; public class DoubleHistogramAggregatorTest { + private static final double[] boundaries = new double[] {10.0, 100.0, 1000.0}; private static final DoubleHistogramAggregator aggregator = new DoubleHistogramAggregator( Resource.getDefault(), @@ -37,7 +38,7 @@ public class DoubleHistogramAggregatorTest { "unit", InstrumentType.VALUE_RECORDER, InstrumentValueType.LONG), - new double[] {10.0, 100.0, 1000.0}, + boundaries, /* stateful= */ false); @Test @@ -74,10 +75,10 @@ void toAccumulationAndReset() { @Test void accumulateData() { - assertThat(aggregator.accumulateDouble(2.0)) - .isEqualTo(HistogramAccumulation.create(2.0, new long[] {1})); + assertThat(aggregator.accumulateDouble(11.1)) + .isEqualTo(HistogramAccumulation.create(11.1, new long[] {0, 1, 0, 0})); assertThat(aggregator.accumulateLong(10)) - .isEqualTo(HistogramAccumulation.create(10.0, new long[] {1})); + .isEqualTo(HistogramAccumulation.create(10.0, new long[] {1, 0, 0, 0})); } @Test @@ -97,6 +98,19 @@ void toMetricData() { .isEqualTo(AggregationTemporality.DELTA); } + @Test + void testHistogramCounts() { + assertThat(aggregator.accumulateDouble(1.1).getCounts().length) + .isEqualTo(boundaries.length + 1); + assertThat(aggregator.accumulateLong(1).getCounts().length).isEqualTo(boundaries.length + 1); + + AggregatorHandle aggregatorHandle = aggregator.createHandle(); + aggregatorHandle.recordDouble(1.1); + HistogramAccumulation histogramAccumulation = aggregatorHandle.accumulateThenReset(); + assertThat(histogramAccumulation).isNotNull(); + assertThat(histogramAccumulation.getCounts().length).isEqualTo(boundaries.length + 1); + } + @Test void testMultithreadedUpdates() throws Exception { final AggregatorHandle aggregatorHandle = aggregator.createHandle(); From 8af20d167702d065371513f364a82393e1631c00 Mon Sep 17 00:00:00 2001 From: beanliu Date: Fri, 26 Feb 2021 13:56:51 +1100 Subject: [PATCH 10/16] use nanoseconds as timeunit --- .../sdk/metrics/aggregator/DoubleHistogramBenchmark.java | 6 +++--- .../metrics/aggregator/DoubleMinMaxSumCountBenchmark.java | 6 +++--- .../sdk/metrics/aggregator/LongMinMaxSumCountBenchmark.java | 6 +++--- 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java index 1e77f05083e..9849dc2e171 100644 --- a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java +++ b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java @@ -47,7 +47,7 @@ public final void setup() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) - @OutputTimeUnit(TimeUnit.MILLISECONDS) + @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 10) public void aggregate_10Threads() { aggregatorHandle.recordDouble(100.0056); @@ -57,7 +57,7 @@ public void aggregate_10Threads() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) - @OutputTimeUnit(TimeUnit.MILLISECONDS) + @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 5) public void aggregate_5Threads() { aggregatorHandle.recordDouble(100.0056); @@ -67,7 +67,7 @@ public void aggregate_5Threads() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) - @OutputTimeUnit(TimeUnit.MILLISECONDS) + @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 1) public void aggregate_1Threads() { aggregatorHandle.recordDouble(100.0056); diff --git a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleMinMaxSumCountBenchmark.java b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleMinMaxSumCountBenchmark.java index a1fb8d575c8..f18d7cfb467 100644 --- a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleMinMaxSumCountBenchmark.java +++ b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleMinMaxSumCountBenchmark.java @@ -46,7 +46,7 @@ public final void setup() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) - @OutputTimeUnit(TimeUnit.MILLISECONDS) + @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 10) public void aggregate_10Threads() { aggregatorHandle.recordDouble(100.0056); @@ -56,7 +56,7 @@ public void aggregate_10Threads() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) - @OutputTimeUnit(TimeUnit.MILLISECONDS) + @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 5) public void aggregate_5Threads() { aggregatorHandle.recordDouble(100.0056); @@ -66,7 +66,7 @@ public void aggregate_5Threads() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) - @OutputTimeUnit(TimeUnit.MILLISECONDS) + @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 1) public void aggregate_1Threads() { aggregatorHandle.recordDouble(100.0056); diff --git a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/LongMinMaxSumCountBenchmark.java b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/LongMinMaxSumCountBenchmark.java index 895a06d4e25..20c140a7520 100644 --- a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/LongMinMaxSumCountBenchmark.java +++ b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/LongMinMaxSumCountBenchmark.java @@ -46,7 +46,7 @@ public final void setup() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) - @OutputTimeUnit(TimeUnit.MILLISECONDS) + @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 10) public void aggregate_10Threads() { aggregatorHandle.recordLong(100); @@ -56,7 +56,7 @@ public void aggregate_10Threads() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) - @OutputTimeUnit(TimeUnit.MILLISECONDS) + @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 5) public void aggregate_5Threads() { aggregatorHandle.recordLong(100); @@ -66,7 +66,7 @@ public void aggregate_5Threads() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) - @OutputTimeUnit(TimeUnit.MILLISECONDS) + @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 1) public void aggregate_1Threads() { aggregatorHandle.recordLong(100); From 9c41d8dd9bd93a6c6f0feb87f7b125e68383a7bb Mon Sep 17 00:00:00 2001 From: beanliu Date: Fri, 26 Feb 2021 14:29:39 +1100 Subject: [PATCH 11/16] update benchmark mode --- .../sdk/metrics/aggregator/DoubleHistogramBenchmark.java | 5 +++++ .../metrics/aggregator/DoubleMinMaxSumCountBenchmark.java | 5 +++++ .../sdk/metrics/aggregator/LongMinMaxSumCountBenchmark.java | 5 +++++ 3 files changed, 15 insertions(+) diff --git a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java index 9849dc2e171..42022cdf2cd 100644 --- a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java +++ b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java @@ -13,9 +13,11 @@ import io.opentelemetry.sdk.resources.Resource; import java.util.concurrent.TimeUnit; import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; import org.openjdk.jmh.annotations.Fork; import org.openjdk.jmh.annotations.Level; import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; import org.openjdk.jmh.annotations.OutputTimeUnit; import org.openjdk.jmh.annotations.Scope; import org.openjdk.jmh.annotations.Setup; @@ -47,6 +49,7 @@ public final void setup() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) + @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 10) public void aggregate_10Threads() { @@ -57,6 +60,7 @@ public void aggregate_10Threads() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) + @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 5) public void aggregate_5Threads() { @@ -67,6 +71,7 @@ public void aggregate_5Threads() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) + @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 1) public void aggregate_1Threads() { diff --git a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleMinMaxSumCountBenchmark.java b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleMinMaxSumCountBenchmark.java index f18d7cfb467..52365ce1e5e 100644 --- a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleMinMaxSumCountBenchmark.java +++ b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleMinMaxSumCountBenchmark.java @@ -12,9 +12,11 @@ import io.opentelemetry.sdk.resources.Resource; import java.util.concurrent.TimeUnit; import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; import org.openjdk.jmh.annotations.Fork; import org.openjdk.jmh.annotations.Level; import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; import org.openjdk.jmh.annotations.OutputTimeUnit; import org.openjdk.jmh.annotations.Scope; import org.openjdk.jmh.annotations.Setup; @@ -46,6 +48,7 @@ public final void setup() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) + @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 10) public void aggregate_10Threads() { @@ -56,6 +59,7 @@ public void aggregate_10Threads() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) + @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 5) public void aggregate_5Threads() { @@ -66,6 +70,7 @@ public void aggregate_5Threads() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) + @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 1) public void aggregate_1Threads() { diff --git a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/LongMinMaxSumCountBenchmark.java b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/LongMinMaxSumCountBenchmark.java index 20c140a7520..b15d4704298 100644 --- a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/LongMinMaxSumCountBenchmark.java +++ b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/LongMinMaxSumCountBenchmark.java @@ -12,9 +12,11 @@ import io.opentelemetry.sdk.resources.Resource; import java.util.concurrent.TimeUnit; import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; import org.openjdk.jmh.annotations.Fork; import org.openjdk.jmh.annotations.Level; import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; import org.openjdk.jmh.annotations.OutputTimeUnit; import org.openjdk.jmh.annotations.Scope; import org.openjdk.jmh.annotations.Setup; @@ -46,6 +48,7 @@ public final void setup() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) + @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 10) public void aggregate_10Threads() { @@ -56,6 +59,7 @@ public void aggregate_10Threads() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) + @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 5) public void aggregate_5Threads() { @@ -66,6 +70,7 @@ public void aggregate_5Threads() { @Fork(1) @Warmup(iterations = 5, time = 1) @Measurement(iterations = 10, time = 1) + @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.NANOSECONDS) @Threads(value = 1) public void aggregate_1Threads() { From 6e2dfdea0707dd96c7d82543507c5ef1739206ff Mon Sep 17 00:00:00 2001 From: beanliu Date: Mon, 1 Mar 2021 16:40:40 +1100 Subject: [PATCH 12/16] List instead of double[] for histogram factory --- .../aggregator/DoubleHistogramBenchmark.java | 3 +- .../metrics/aggregator/AggregatorFactory.java | 3 +- .../HistogramAggregatorFactory.java | 13 +++-- .../aggregator/AggregatorFactoryTest.java | 47 ++++++++++++------- 4 files changed, 44 insertions(+), 22 deletions(-) diff --git a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java index 42022cdf2cd..f1ae34ba6b4 100644 --- a/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java +++ b/sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramBenchmark.java @@ -11,6 +11,7 @@ import io.opentelemetry.sdk.metrics.common.InstrumentValueType; import io.opentelemetry.sdk.metrics.data.AggregationTemporality; import io.opentelemetry.sdk.resources.Resource; +import java.util.Arrays; import java.util.concurrent.TimeUnit; import org.openjdk.jmh.annotations.Benchmark; import org.openjdk.jmh.annotations.BenchmarkMode; @@ -28,7 +29,7 @@ @State(Scope.Benchmark) public class DoubleHistogramBenchmark { private static final Aggregator aggregator = - AggregatorFactory.histogram(new double[] {10, 100, 1_000}, AggregationTemporality.DELTA) + AggregatorFactory.histogram(Arrays.asList(10.0, 100.0, 1_000.0), AggregationTemporality.DELTA) .create( Resource.getDefault(), InstrumentationLibraryInfo.empty(), diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactory.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactory.java index c8d5fa51194..777b13e4029 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactory.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactory.java @@ -9,6 +9,7 @@ import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor; import io.opentelemetry.sdk.metrics.data.AggregationTemporality; import io.opentelemetry.sdk.resources.Resource; +import java.util.List; import javax.annotation.concurrent.Immutable; /** Factory class for {@link Aggregator}. */ @@ -85,7 +86,7 @@ static AggregatorFactory minMaxSumCount() { * @param boundaries configures the fixed bucket boundaries. * @return an {@code AggregationFactory} that calculates histogram of recorded measurements. */ - static AggregatorFactory histogram(double[] boundaries, AggregationTemporality temporality) { + static AggregatorFactory histogram(List boundaries, AggregationTemporality temporality) { return new HistogramAggregatorFactory(boundaries, temporality); } diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java index e6e7b14b5bc..1ce974d081d 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/HistogramAggregatorFactory.java @@ -9,18 +9,23 @@ import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor; import io.opentelemetry.sdk.metrics.data.AggregationTemporality; import io.opentelemetry.sdk.resources.Resource; -import java.util.Arrays; +import java.util.List; final class HistogramAggregatorFactory implements AggregatorFactory { private final double[] boundaries; private final AggregationTemporality temporality; - HistogramAggregatorFactory(double[] boundaries, AggregationTemporality temporality) { - this.boundaries = Arrays.copyOf(boundaries, boundaries.length); + HistogramAggregatorFactory(List boundaries, AggregationTemporality temporality) { + this.boundaries = boundaries.stream().mapToDouble(i -> i).toArray(); this.temporality = temporality; + for (double v : this.boundaries) { + if (Double.isNaN(v)) { + throw new IllegalArgumentException("invalid bucket boundary: NaN"); + } + } for (int i = 1; i < this.boundaries.length; ++i) { - if (Double.compare(this.boundaries[i - 1], this.boundaries[i]) >= 0) { + if (this.boundaries[i - 1] >= this.boundaries[i]) { throw new IllegalArgumentException( "invalid bucket boundary: " + this.boundaries[i - 1] + " >= " + this.boundaries[i]); } diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactoryTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactoryTest.java index f956cbdddee..d528a553e63 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactoryTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/AggregatorFactoryTest.java @@ -6,6 +6,7 @@ package io.opentelemetry.sdk.metrics.aggregator; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import io.opentelemetry.sdk.common.InstrumentationLibraryInfo; import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor; @@ -13,7 +14,8 @@ import io.opentelemetry.sdk.metrics.common.InstrumentValueType; import io.opentelemetry.sdk.metrics.data.AggregationTemporality; import io.opentelemetry.sdk.resources.Resource; -import org.junit.jupiter.api.Assertions; +import java.util.Arrays; +import java.util.Collections; import org.junit.jupiter.api.Test; class AggregatorFactoryTest { @@ -128,7 +130,7 @@ void getSumAggregatorFactory() { @Test void getHistogramAggregatorFactory() { AggregatorFactory histogram = - AggregatorFactory.histogram(new double[] {1.0}, AggregationTemporality.DELTA); + AggregatorFactory.histogram(Collections.singletonList(1.0), AggregationTemporality.DELTA); assertThat( histogram.create( Resource.getDefault(), @@ -166,7 +168,8 @@ void getHistogramAggregatorFactory() { .isStateful()) .isFalse(); assertThat( - AggregatorFactory.histogram(new double[] {1.0}, AggregationTemporality.CUMULATIVE) + AggregatorFactory.histogram( + Collections.singletonList(1.0), AggregationTemporality.CUMULATIVE) .create( Resource.getDefault(), InstrumentationLibraryInfo.empty(), @@ -179,18 +182,30 @@ void getHistogramAggregatorFactory() { .isStateful()) .isTrue(); - Assertions.assertThrows( - IllegalArgumentException.class, - () -> - AggregatorFactory.histogram( - new double[] {Double.NEGATIVE_INFINITY}, AggregationTemporality.DELTA)); - Assertions.assertThrows( - IllegalArgumentException.class, - () -> - AggregatorFactory.histogram( - new double[] {1, Double.POSITIVE_INFINITY}, AggregationTemporality.DELTA)); - Assertions.assertThrows( - IllegalArgumentException.class, - () -> AggregatorFactory.histogram(new double[] {2, 1, 3}, AggregationTemporality.DELTA)); + assertThatThrownBy( + () -> + AggregatorFactory.histogram( + Collections.singletonList(Double.NEGATIVE_INFINITY), + AggregationTemporality.DELTA)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("invalid bucket boundary: -Inf"); + assertThatThrownBy( + () -> + AggregatorFactory.histogram( + Arrays.asList(1.0, Double.POSITIVE_INFINITY), AggregationTemporality.DELTA)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("invalid bucket boundary: +Inf"); + assertThatThrownBy( + () -> + AggregatorFactory.histogram( + Arrays.asList(1.0, Double.NaN), AggregationTemporality.DELTA)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("invalid bucket boundary: NaN"); + assertThatThrownBy( + () -> + AggregatorFactory.histogram( + Arrays.asList(2.0, 1.0, 3.0), AggregationTemporality.DELTA)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("invalid bucket boundary: 2.0 >= 1.0"); } } From cb8bf6bd38aec21e0ad4b63345bad0e6116960e2 Mon Sep 17 00:00:00 2001 From: beanliu Date: Mon, 1 Mar 2021 16:43:07 +1100 Subject: [PATCH 13/16] update var names --- .../metrics/aggregator/DoubleHistogramAggregatorTest.java | 6 +++--- .../aggregator/DoubleMinMaxSumCountAggregatorTest.java | 6 +++--- .../aggregator/LongMinMaxSumCountAggregatorTest.java | 6 +++--- 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java index 868e3b6c482..86280f12e32 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java @@ -118,7 +118,7 @@ void testMultithreadedUpdates() throws Exception { int numberOfThreads = 10; final long[] updates = new long[] {1, 2, 3, 5, 7, 11, 13, 17, 19, 23}; final int numberOfUpdates = 1000; - final CountDownLatch startingGun = new CountDownLatch(numberOfThreads); + final CountDownLatch starter = new CountDownLatch(numberOfThreads); List workers = new ArrayList<>(); for (int i = 0; i < numberOfThreads; i++) { final int index = i; @@ -127,7 +127,7 @@ void testMultithreadedUpdates() throws Exception { () -> { long update = updates[index]; try { - startingGun.await(); + starter.await(); } catch (InterruptedException e) { throw new RuntimeException(e); } @@ -142,7 +142,7 @@ void testMultithreadedUpdates() throws Exception { t.start(); } for (int i = 0; i <= numberOfThreads; i++) { - startingGun.countDown(); + starter.countDown(); } for (Thread worker : workers) { diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleMinMaxSumCountAggregatorTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleMinMaxSumCountAggregatorTest.java index 267fc4e21bf..74d29f97882 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleMinMaxSumCountAggregatorTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleMinMaxSumCountAggregatorTest.java @@ -97,7 +97,7 @@ void testMultithreadedUpdates() throws Exception { int numberOfThreads = 10; final double[] updates = new double[] {1, 2, 3, 5, 7, 11, 13, 17, 19, 23}; final int numberOfUpdates = 1000; - final CountDownLatch startingGun = new CountDownLatch(numberOfThreads); + final CountDownLatch starter = new CountDownLatch(numberOfThreads); List workers = new ArrayList<>(); for (int i = 0; i < numberOfThreads; i++) { final int index = i; @@ -106,7 +106,7 @@ void testMultithreadedUpdates() throws Exception { () -> { double update = updates[index]; try { - startingGun.await(); + starter.await(); } catch (InterruptedException e) { throw new RuntimeException(e); } @@ -121,7 +121,7 @@ void testMultithreadedUpdates() throws Exception { t.start(); } for (int i = 0; i <= numberOfThreads; i++) { - startingGun.countDown(); + starter.countDown(); } for (Thread worker : workers) { diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/LongMinMaxSumCountAggregatorTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/LongMinMaxSumCountAggregatorTest.java index 6485cec4693..81a76999e67 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/LongMinMaxSumCountAggregatorTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/LongMinMaxSumCountAggregatorTest.java @@ -94,7 +94,7 @@ void testMultithreadedUpdates() throws Exception { int numberOfThreads = 10; final long[] updates = new long[] {1, 2, 3, 5, 7, 11, 13, 17, 19, 23}; final int numberOfUpdates = 1000; - final CountDownLatch startingGun = new CountDownLatch(numberOfThreads); + final CountDownLatch starter = new CountDownLatch(numberOfThreads); List workers = new ArrayList<>(); for (int i = 0; i < numberOfThreads; i++) { final int index = i; @@ -103,7 +103,7 @@ void testMultithreadedUpdates() throws Exception { () -> { long update = updates[index]; try { - startingGun.await(); + starter.await(); } catch (InterruptedException e) { throw new RuntimeException(e); } @@ -118,7 +118,7 @@ void testMultithreadedUpdates() throws Exception { t.start(); } for (int i = 0; i <= numberOfThreads; i++) { - startingGun.countDown(); + starter.countDown(); } for (Thread worker : workers) { From dc92511d6c4d721ce13071786ae30db47b23f74c Mon Sep 17 00:00:00 2001 From: beanliu Date: Mon, 1 Mar 2021 16:43:39 +1100 Subject: [PATCH 14/16] switch to using assertj --- .../sdk/metrics/data/MetricDataTest.java | 52 +++++++++---------- 1 file changed, 26 insertions(+), 26 deletions(-) diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/data/MetricDataTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/data/MetricDataTest.java index c72fa92d25b..eb44b57053e 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/data/MetricDataTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/data/MetricDataTest.java @@ -6,6 +6,7 @@ package io.opentelemetry.sdk.metrics.data; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import com.google.common.collect.ImmutableList; import io.opentelemetry.api.metrics.common.Labels; @@ -14,7 +15,6 @@ import java.util.Arrays; import java.util.Collections; import java.util.concurrent.TimeUnit; -import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; /** Unit tests for {@link io.opentelemetry.sdk.metrics.data.MetricData}. */ @@ -178,31 +178,31 @@ void metricData_HistogramPoints() { AggregationTemporality.DELTA, Collections.singleton(HISTOGRAM_POINT))); assertThat(metricData.getDoubleHistogramData().getPoints()).containsExactly(HISTOGRAM_POINT); - Assertions.assertThrows( - IllegalArgumentException.class, - () -> - DoubleHistogramPointData.create( - 0, 0, Labels.empty(), 0.0, ImmutableList.of(), ImmutableList.of())); - Assertions.assertThrows( - IllegalArgumentException.class, - () -> - DoubleHistogramPointData.create( - 0, - 0, - Labels.empty(), - 0.0, - ImmutableList.of(1.0, 1.0), - ImmutableList.of(0L, 0L, 0L))); - Assertions.assertThrows( - IllegalArgumentException.class, - () -> - DoubleHistogramPointData.create( - 0, - 0, - Labels.empty(), - 0.0, - ImmutableList.of(Double.NEGATIVE_INFINITY), - ImmutableList.of(0L, 0L))); + assertThatThrownBy( + () -> + DoubleHistogramPointData.create( + 0, 0, Labels.empty(), 0.0, ImmutableList.of(), ImmutableList.of())) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy( + () -> + DoubleHistogramPointData.create( + 0, + 0, + Labels.empty(), + 0.0, + ImmutableList.of(1.0, 1.0), + ImmutableList.of(0L, 0L, 0L))) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy( + () -> + DoubleHistogramPointData.create( + 0, + 0, + Labels.empty(), + 0.0, + ImmutableList.of(Double.NEGATIVE_INFINITY), + ImmutableList.of(0L, 0L))) + .isInstanceOf(IllegalArgumentException.class); } @Test From 4d75f5a6062a8014bd15f24c9b59a6275f0c6270 Mon Sep 17 00:00:00 2001 From: beanliu Date: Mon, 1 Mar 2021 17:02:16 +1100 Subject: [PATCH 15/16] simpler boundary check --- .../sdk/metrics/aggregator/DoubleHistogramAggregator.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java index 110c98131c8..35a8f1cd5ad 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregator.java @@ -98,7 +98,7 @@ public HistogramAccumulation accumulateLong(long value) { // buckets. private static int findBucketIndex(double[] boundaries, double value) { for (int i = 0; i < boundaries.length; ++i) { - if (Double.compare(value, boundaries[i]) <= 0) { + if (value <= boundaries[i]) { return i; } } From cd43a0e862b2667a2d7a9d95270f1101feef7775 Mon Sep 17 00:00:00 2001 From: beanliu Date: Mon, 1 Mar 2021 17:36:21 +1100 Subject: [PATCH 16/16] simplify multi-threaded test --- .../DoubleHistogramAggregatorTest.java | 78 ++++++++----------- 1 file changed, 31 insertions(+), 47 deletions(-) diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java index 86280f12e32..efbc387adfc 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/aggregator/DoubleHistogramAggregatorTest.java @@ -7,7 +7,7 @@ import static org.assertj.core.api.Assertions.assertThat; -import com.google.errorprone.annotations.concurrent.GuardedBy; +import com.google.common.collect.ImmutableList; import io.opentelemetry.api.metrics.common.Labels; import io.opentelemetry.sdk.common.InstrumentationLibraryInfo; import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor; @@ -17,12 +17,11 @@ import io.opentelemetry.sdk.metrics.data.MetricData; import io.opentelemetry.sdk.metrics.data.MetricDataType; import io.opentelemetry.sdk.resources.Resource; -import java.util.ArrayList; import java.util.Collections; -import java.util.List; -import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; import java.util.concurrent.ThreadLocalRandom; -import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.stream.Collectors; import javax.annotation.Nullable; import org.junit.jupiter.api.Test; @@ -112,69 +111,54 @@ void testHistogramCounts() { } @Test - void testMultithreadedUpdates() throws Exception { + void testMultithreadedUpdates() throws InterruptedException { final AggregatorHandle aggregatorHandle = aggregator.createHandle(); final Histogram summarizer = new Histogram(); - int numberOfThreads = 10; - final long[] updates = new long[] {1, 2, 3, 5, 7, 11, 13, 17, 19, 23}; - final int numberOfUpdates = 1000; - final CountDownLatch starter = new CountDownLatch(numberOfThreads); - List workers = new ArrayList<>(); - for (int i = 0; i < numberOfThreads; i++) { - final int index = i; - Thread t = - new Thread( - () -> { - long update = updates[index]; - try { - starter.await(); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } - for (int j = 0; j < numberOfUpdates; j++) { - aggregatorHandle.recordLong(update); - if (ThreadLocalRandom.current().nextInt(10) == 0) { - summarizer.process(aggregatorHandle.accumulateThenReset()); - } - } - }); - workers.add(t); - t.start(); - } - for (int i = 0; i <= numberOfThreads; i++) { - starter.countDown(); - } + final ImmutableList updates = + ImmutableList.of(1L, 2L, 3L, 5L, 7L, 11L, 13L, 17L, 19L, 23L); + final int numberOfThreads = updates.size(); + final int numberOfUpdates = 10000; + final ThreadPoolExecutor executor = + (ThreadPoolExecutor) Executors.newFixedThreadPool(numberOfThreads); + + executor.invokeAll( + updates.stream() + .map( + v -> + Executors.callable( + () -> { + for (int j = 0; j < numberOfUpdates; j++) { + aggregatorHandle.recordLong(v); + if (ThreadLocalRandom.current().nextInt(10) == 0) { + summarizer.process(aggregatorHandle.accumulateThenReset()); + } + } + })) + .collect(Collectors.toList())); - for (Thread worker : workers) { - worker.join(); - } // make sure everything gets merged when all the aggregation is done. summarizer.process(aggregatorHandle.accumulateThenReset()); assertThat(summarizer.accumulation) - .isEqualTo(HistogramAccumulation.create(101000, new long[] {5000, 5000, 0, 0})); + .isEqualTo(HistogramAccumulation.create(1010000, new long[] {50000, 50000, 0, 0})); } private static final class Histogram { - private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); + private final Object mutex = new Object(); - @GuardedBy("lock") - @Nullable - private HistogramAccumulation accumulation; + @Nullable private HistogramAccumulation accumulation; void process(@Nullable HistogramAccumulation other) { if (other == null) { return; } - lock.writeLock().lock(); - try { + + synchronized (mutex) { if (accumulation == null) { accumulation = other; return; } accumulation = aggregator.merge(accumulation, other); - } finally { - lock.writeLock().unlock(); } } }