Epic

This is the first PR out of several which implements #5105.

The problem

Issue #5105 describes it quite well, so I'll add a very short motivation.
Currently the SDK uses immutable objects during metric collection, where the bulk of the work is located at the LeasedMetricProducer.
The side effect of immutable objects is that they generate a lot of garbage collected objects. Given enough instruments and/or attribute sets, this can amount to such a substantial amount, the garbage collector will start to kick in more often, and for longer periods of time.
There are systems which are ultra sensitive to latency, like databases and messaging systems (e.g. Apache Pulsar). Those systems will have a problem using OpenTelemetry as it is today.

The solution

The SDK will have two modes of operation:

1. Immutability - in which it behaves as it is now during metric collection and uses immutable objects.
2. Reusability - in which most objects used during collection would be reusable - mutable objects that are pooled into an object pool between collection runs.

What does this PR include?

This PR is focused on adding support for reusability, and implementing it for Asynchronous Instruments.

• The introduction of a MemoryMode enum dictating the two different modes.
• Adding a memoryMode property to MetricReader. This is used by the SDK collection process (in this PR only the Asynchronous Instrument related code) to switch between using immutable objects of reusable objects (from a pool).
• For PeriodicMetricReader we've added memoryMode property to MetricExporter as well, so both the metric exporter can act upon it, and also the PeriodicMetricReader delegates the value of memoryMode to the MetricExporter it has.
• A set of mutable implementations of DoublePointData and LongPointData. This are used to carry the current value for each (instrument, attributes) pair.
• Adding a mutable version of Measurement, by changing it from an abstract class into an interface and creating two classes implementing it: ImmutableMeasurement and LeasedMeasurement.
• SdkObservableMeasurement has been changed to object reuse if reusable memory mode is selected, for the active metric reader. Since it only creates Measurement object only as DTO for calling AsynchronousMetricStorage.record(measurement), I've added a single LeasedMeasurement object and keep re-using it across calls to record() from the callbacks. It was named "leased" since the SdkObserableMeasurement leases the measurement to AsynchronousMetricStorage for the duration of the record() call - it should not store it hence "leased".

Data structures for reusable objects

• I've added an ObjectPool class which as it sounds, allows you to borrow an object from the pool. If the pool is empty it uses an objectCreator function to generate one and return it. When the object is done with, instead of being garbage collected, it is returned back to the pool using returnObject(o).
  The pool it self must avoid any O(n) memory allocations upon borrow or return operations, hence using a simple array-based stack I wrote ArrayBasedStack. I decided to write one instead of using ArrayDequeue since I want in next PR improve the extra space it occupies (instead of x2 switching to changing double factor, or trim after x collections)

• Also I've added PooledHashMap, which is a bucket-based HashMap implementation, which uses an object pool when it needs a MapEntry object. This is needed primarily in AsynchronousMetricStorage.

Storage

• AsynchronousMetricStorage - here the bulk of the work was done in this PR. This is where all the callback record() calls gets funneled into. The primary changes were:

  • Using PooledHashMap instead of HashMap to reduce MapEntry memory allocations which were O(n). The maps are used to store the recordings (attributes -> point) and in case of DELTA temporality, the previous data points.
  • in record(measurement), the measurement is now converted into a reusable data point, coming from a pool of
    reusable data points, in the case of REUSABLE memory mode.
  • The result returned by collect() is now a reusable list (array-based), in the case of REUSABLE memory mode. Array-based removes any internal wrapper objects allocations.
  • In collect() it return a list of reusable points, which needs to return back to the pool. We do so when the next collect() is called. We "save" the list returned, return each reusable data point back to the pool, and reset the list for another usage.
  • The DELTA mode was a bit refactored to allow for easier reasoning when reading it.

• The *Aggregator classes were modified to allow for reusable data points to avoid allocations.

Benchmark tests

Garbage Collection (memory allocation) benchmark

AsynchronousMetricStorageGarbageCollectionBenchmarkTest runs a JMH benchmark named AsynchronousMetricStorageGarbageCollectionBenchmark with gc profiler, which measures the number
of objects (and size) the garbage collector collected (unused objects).

The AsynchronousMetricStorageGarbageCollectionBenchmark benchmark creates 10 asynchronous counters (any asynchronous instrument will do as the code path that was changes is almost the same for all async instrument types), and 1000 attribute sets. Each time the test runs, it calls flush which effectively calls the callback for each counter. Each such callback records a random number for each of the 1000 attribute sets. The result list ends up in NoOpMetricExporter which does nothing with it.

This is repeated 100 times, collectively called operation in the statistics and each such Operation is repeated 20 times - known as Iteration.

Each such test is repeated, with a brand new JVM, for all combinations of memory mode (REUSBLE_DATA, IMMUTABLE_DATA) and aggregation temporality (CUMULATIVE, DELTA). This is done since each combination has a different code path.

The statistics of this run can be seen in the output of the test and also pasted here below:

Benchmark                                                                                (aggregationTemporality)    (memoryMode)  Mode  Cnt          Score          Error   Units
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect                                        DELTA   REUSABLE_DATA    ss   20  253203950.050 ± 14647630.858   ns/op
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect:gc.alloc.rate                          DELTA   REUSABLE_DATA    ss   20          0.968 ±        0.038  MB/sec
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect:gc.alloc.rate.norm                     DELTA   REUSABLE_DATA    ss   20     257078.800 ±     8031.841    B/op
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect:gc.count                               DELTA   REUSABLE_DATA    ss   20            ≈ 0                 counts
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect                                        DELTA  IMMUTABLE_DATA    ss   20  181222052.050 ± 20645573.792   ns/op
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect:gc.alloc.rate                          DELTA  IMMUTABLE_DATA    ss   20       1437.476 ±      125.219  MB/sec
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect:gc.alloc.rate.norm                     DELTA  IMMUTABLE_DATA    ss   20  271877200.400 ±   527883.090    B/op
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect:gc.count                               DELTA  IMMUTABLE_DATA    ss   20         18.000                 counts
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect:gc.time                                DELTA  IMMUTABLE_DATA    ss   20         75.000                     ms
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect                                   CUMULATIVE   REUSABLE_DATA    ss   20  191358368.900 ± 13650017.689   ns/op
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect:gc.alloc.rate                     CUMULATIVE   REUSABLE_DATA    ss   20          1.166 ±        0.074  MB/sec
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect:gc.alloc.rate.norm                CUMULATIVE   REUSABLE_DATA    ss   20     233689.200 ±     8675.469    B/op
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect:gc.count                          CUMULATIVE   REUSABLE_DATA    ss   20            ≈ 0                 counts
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect                                   CUMULATIVE  IMMUTABLE_DATA    ss   20  142355889.500 ± 15351913.671   ns/op
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect:gc.alloc.rate                     CUMULATIVE  IMMUTABLE_DATA    ss   20       1403.485 ±      125.334  MB/sec
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect:gc.alloc.rate.norm                CUMULATIVE  IMMUTABLE_DATA    ss   20  208828913.600 ±   407209.901    B/op
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect:gc.count                          CUMULATIVE  IMMUTABLE_DATA    ss   20         13.000                 counts
AsynchronousMetricStorageGarbageCollectionBenchmark.recordAndCollect:gc.time                           CUMULATIVE  IMMUTABLE_DATA    ss   20         66.000                     ms

If we focus only on the normalized rate:

Benchmark                                            (aggregationTemporality)    (memoryMode)  Mode  Cnt          Score          Error   Units
recordAndCollect:gc.alloc.rate.norm                     DELTA   REUSABLE_DATA    ss   20     257078.800 ±     8031.841    B/op
recordAndCollect:gc.alloc.rate.norm                     DELTA  IMMUTABLE_DATA    ss   20  271877200.400 ±   527883.090    B/op
recordAndCollect:gc.alloc.rate.norm                CUMULATIVE   REUSABLE_DATA    ss   20     233689.200 ±     8675.469    B/op
recordAndCollect:gc.alloc.rate.norm                CUMULATIVE  IMMUTABLE_DATA    ss   20  208828913.600 ±   407209.901    B/op

You can see that for DELTA aggregation temporality, REUSABLE_DATA scores a rate of 257,078 [bytes/operation] compared with 271,877,200 [bytes/operation], which means REUSABLE_DATA memory mode reduced the size of
garbage collected objects by 99.9% compared with IMMUTABLE_DATA memory mode. Similar results (99.88%) exists for CUMULATIVE.

The test verifies that. Aside from the convenience of running this inside your code editor with one click, it protects from any future regression that can easily arise from a simple refactor or change.

Memory usage benchmark

IMMUTABLE_DATA "pays" with lots of one-off objects which almost immediately gets garbage collected. Memory is allocated to certain peak and then released. On the other hand, REUSABLE_DATA uses reusable objects, located in object pools, which stays in memory even after the collection finished. So it "pays" with constant memory usage. In a perfect implementation the peak memory consumption would almost be equal in both cases.

The object pools are not yet tuned, so they aggressively allocates more than what they need. In subsequent PRs I would like to minimize that effect, so the max heap for both is not far off each other.

You can run AsynchronousCounterMemoryUsageBenchmark main() method to view the memory usage for each combination of aggregation temporality, memory mode, number of asynchronous instruments and number of attribute sets.

The results are:

Cumulative

Delta

So you can see for example that for 50 counters and 100k attribute sets each, the maximum used memory is 958mb for reusable data and 581mb for immutable data, meaning 64% more memory needed. For 1 instrument, it's 20%. The reason it's bigger percentage as it has more attribute sets is because currently the object pool grows by 2 each time it runs out of capacity. This is something to be be tuned in next PRs.

opentelemetry-sdk-metrics.txt

Comparing source compatibility of  against 
+++  NEW ENUM: PUBLIC(+) FINAL(+) io.opentelemetry.sdk.metrics.export.MemoryMode  (compatible)
	+++  CLASS FILE FORMAT VERSION: 52.0 <- n.a.
	+++  NEW INTERFACE: java.lang.constant.Constable
	+++  NEW INTERFACE: java.lang.Comparable
	+++  NEW INTERFACE: java.io.Serializable
	+++  NEW SUPERCLASS: java.lang.Enum
	+++  NEW FIELD: PUBLIC(+) STATIC(+) FINAL(+) io.opentelemetry.sdk.metrics.export.MemoryMode REUSABLE_DATA
	+++  NEW FIELD: PUBLIC(+) STATIC(+) FINAL(+) io.opentelemetry.sdk.metrics.export.MemoryMode IMMUTABLE_DATA
	+++  NEW METHOD: PUBLIC(+) STATIC(+) io.opentelemetry.sdk.metrics.export.MemoryMode valueOf(java.lang.String)
	+++  NEW METHOD: PUBLIC(+) STATIC(+) io.opentelemetry.sdk.metrics.export.MemoryMode[] values()
***  MODIFIED INTERFACE: PUBLIC ABSTRACT io.opentelemetry.sdk.metrics.export.MetricExporter  (not serializable)
	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.metrics.export.MemoryMode getMemoryMode()
***  MODIFIED INTERFACE: PUBLIC ABSTRACT io.opentelemetry.sdk.metrics.export.MetricReader  (not serializable)
	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.metrics.export.MemoryMode getMemoryMode()
***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.sdk.metrics.export.PeriodicMetricReader  (not serializable)
	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.metrics.export.MemoryMode getMemoryMode()
***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.sdk.metrics.ViewBuilder  (not serializable)
	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.metrics.ViewBuilder setAttributeFilter(java.util.Set<java.lang.String>)

opentelemetry-sdk-testing.txt

Comparing source compatibility of  against 
***  MODIFIED CLASS: PUBLIC io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader  (not serializable)
	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
	+++  NEW METHOD: PUBLIC(+) STATIC(+) io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader$InMemoryMetricReaderBuilder builder()
	+++  NEW METHOD: PUBLIC(+) STATIC(+) io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader create(io.opentelemetry.sdk.metrics.export.AggregationTemporalitySelector)
	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.metrics.export.MemoryMode getMemoryMode()
+++  NEW CLASS: PUBLIC(+) STATIC(+) io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader$InMemoryMetricReaderBuilder  (not serializable)
	+++  CLASS FILE FORMAT VERSION: 52.0 <- n.a.
	+++  NEW SUPERCLASS: java.lang.Object
	+++  NEW CONSTRUCTOR: PUBLIC(+) InMemoryMetricReader$InMemoryMetricReaderBuilder()
	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader build()
	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader$InMemoryMetricReaderBuilder setAggregationTemporalitySelector(io.opentelemetry.sdk.metrics.export.AggregationTemporalitySelector)
	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader$InMemoryMetricReaderBuilder setDefaultAggregationSelector(io.opentelemetry.sdk.metrics.export.DefaultAggregationSelector)
	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader$InMemoryMetricReaderBuilder setMemoryMode(io.opentelemetry.sdk.metrics.export.MemoryMode)

Future PRs

• Tune ObjectPool so it won't occupy so much space - change growth factor as size increases, or introduce trimming.
• Activate the memory mode. I haven't changed any exporter not reader yet.