Memory Mode support: Adding memory mode, and implementing it for Asynchronous Instruments

dependencyManagement/build.gradle.kts

@@ -49,6 +49,7 @@ val DEPENDENCIES = listOf(
   "io.opencensus:opencensus-contrib-exemplar-util:${opencensusVersion}",
   "org.openjdk.jmh:jmh-core:${jmhVersion}",
   "org.openjdk.jmh:jmh-generator-bytecode:${jmhVersion}",
+  "org.openjdk.jmh:jmh-generator-annprocess:${jmhVersion}",
   "org.mockito:mockito-core:${mockitoVersion}",
   "org.mockito:mockito-junit-jupiter:${mockitoVersion}",
   "org.slf4j:slf4j-simple:${slf4jVersion}",

sdk/metrics/build.gradle.kts

@@ -28,6 +28,7 @@ dependencies {
 
   jmh(project(":sdk:trace"))
   jmh(project(":sdk:testing"))
+  jmh(project(":sdk:metrics:testing-internal"))
 }
 
 testing {
@@ -41,6 +42,14 @@ testing {
         }
       }
     }
+    register<JvmTestSuite>("jmhBasedTest") {
+      dependencies {
+        implementation("org.openjdk.jmh:jmh-core")
+        implementation("org.openjdk.jmh:jmh-generator-bytecode")
+        annotationProcessor("org.openjdk.jmh:jmh-generator-annprocess")
+        implementation(project(":sdk:metrics:testing-internal"))
+      }
+    }
   }
 }
 

sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/internal/aggregator/HistogramCollectBenchmark.java

@@ -7,23 +7,19 @@
 
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.metrics.DoubleHistogram;
-import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.metrics.Aggregation;
-import io.opentelemetry.sdk.metrics.InstrumentType;
 import io.opentelemetry.sdk.metrics.SdkMeterProvider;
 import io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder;
 import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
-import io.opentelemetry.sdk.metrics.data.MetricData;
-import io.opentelemetry.sdk.metrics.export.MetricExporter;
 import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
 import io.opentelemetry.sdk.metrics.internal.SdkMeterProviderUtil;
 import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarFilter;
 import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Collection;
 import java.util.List;
 import java.util.Random;
 import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
 import org.openjdk.jmh.annotations.Fork;
@@ -37,6 +33,7 @@
 import org.openjdk.jmh.annotations.TearDown;
 import org.openjdk.jmh.annotations.Threads;
 import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jol.info.GraphLayout;
 
 /**
  * Measures the collection path for various histogram aggregations. Should be used primarily to
@@ -48,6 +45,7 @@
 @Warmup(iterations = 2, batchSize = 10)
 @Fork(1)
 public class HistogramCollectBenchmark {
+  private static final Logger logger = Logger.getLogger(HistogramCollectBenchmark.class.getName());
 
   private static final int cardinality = 100;
   private static final int measurementsPerSeries = 10_000;
@@ -64,31 +62,24 @@ public static class ThreadState {
 
     @Setup
     public void setup() {
-      SdkMeterProviderBuilder builder =
-          SdkMeterProvider.builder()
-              .registerMetricReader(
-                  PeriodicMetricReader.builder(
-                          // Configure an exporter that configures the temporality and aggregation
-                          // for the test case, but otherwise drops the data on export
-                          new NoopMetricExporter(
-                              aggregationTemporality, aggregationGenerator.aggregation))
-                      // Effectively disable periodic reading so reading is only done on #flush()
-                      .setInterval(Duration.ofSeconds(Integer.MAX_VALUE))
-                      .build());
+      PeriodicMetricReader metricReader =
+          PeriodicMetricReader.builder(
+                  // Configure an exporter that configures the temporality and aggregation
+                  // for the test case, but otherwise drops the data on export
+                  new NoopMetricExporter(aggregationTemporality, aggregationGenerator.aggregation))
+              // Effectively disable periodic reading so reading is only done on #flush()
+              .setInterval(Duration.ofSeconds(Integer.MAX_VALUE))
+              .build();
+      SdkMeterProviderBuilder builder = SdkMeterProvider.builder();
+      SdkMeterProviderUtil.registerMetricReaderWithCardinalitySelector(
+          builder, metricReader, unused -> cardinality);
       // Disable examplars
       SdkMeterProviderUtil.setExemplarFilter(builder, ExemplarFilter.alwaysOff());
       sdkMeterProvider = builder.build();
       histogram = sdkMeterProvider.get("meter").histogramBuilder("histogram").build();
 
       random = new Random();
-      attributesList = new ArrayList<>(cardinality);
-      String last = "aaaaaaaaaaaaaaaaaaaaaaaaaa";
-      for (int i = 0; i < cardinality; i++) {
-        char[] chars = last.toCharArray();
-        chars[random.nextInt(last.length())] = (char) (random.nextInt(26) + 'a');
-        last = new String(chars);
-        attributesList.add(Attributes.builder().put("key", last).build());
-      }
+      attributesList = AttributesGenerator.generate(cardinality);
     }
 
     @TearDown
@@ -122,40 +113,4 @@ public enum AggregationGenerator {
       this.aggregation = aggregation;
     }
   }
-
-  private static class NoopMetricExporter implements MetricExporter {
-    private final AggregationTemporality aggregationTemporality;
-    private final Aggregation aggregation;
-
-    private NoopMetricExporter(
-        AggregationTemporality aggregationTemporality, Aggregation aggregation) {
-      this.aggregationTemporality = aggregationTemporality;
-      this.aggregation = aggregation;
-    }
-
-    @Override
-    public CompletableResultCode export(Collection<MetricData> metrics) {
-      return CompletableResultCode.ofSuccess();
-    }
-
-    @Override
-    public CompletableResultCode flush() {
-      return CompletableResultCode.ofSuccess();
-    }
-
-    @Override
-    public CompletableResultCode shutdown() {
-      return CompletableResultCode.ofSuccess();
-    }
-
-    @Override
-    public Aggregation getDefaultAggregation(InstrumentType instrumentType) {
-      return aggregation;
-    }
-
-    @Override
-    public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) {
-      return aggregationTemporality;
-    }
-  }
 }

sdk/metrics/src/jmhBasedTest/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousCounterMemoryUsageBenchmark.java

@@ -0,0 +1,186 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.metrics.internal.state;
+
+import static org.assertj.core.api.Fail.fail;
+
+import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
+import io.opentelemetry.sdk.metrics.export.MemoryMode;
+import java.lang.management.ManagementFactory;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Benchmarks the memory usage of different memory modes and aggregation temporalities
+ *
+ * <p>This benchmark helps to see the difference in heap usage between the memory modes. You can
+ * change the aggregation temporality, number of counters and cardinality (attribute sets count).
+ * You have to run it manually for each parameter, since it's not JMH-based, and we need a clean
+ * heap before we start.
+ *
+ * <p>This benchmark class has additional usage: You can use it to see memory allocation frame graphs for a
+ * single run.
+ *
+ * <p>Steps:
+ * <ol>
+ * <li> Follow download instructions for async-profiler, located at
+ * https://github.com/async-profiler/async-profiler
+ * <li>Assuming you have extracted it at
+ * /tmp/async-profiler-2.9-macos, add the following to your JVM arguments of your run configuration:
+ *
+ * <p>-agentpath:/tmp/async-profiler-2.9-macos/build/libasyncProfiler.so=start,event=alloc,flamegraph,file=/tmp/profiled_data.html
+ *
+ * <li>Tune the parameters as you see fit
+ * <li>Run the class
+ * <li>Open /tmp/profiled_data.html with your browser
+ * </ol>
+ */
+@SuppressWarnings("SystemOut")
+public class AsynchronousCounterMemoryUsageBenchmark {
+
+  ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
+  AtomicLong maxUsedMemory = new AtomicLong(0);
+
+  private AsynchronousCounterMemoryUsageBenchmark() {}
+
+  /**
+   * Runs the memory usage benchmark.
+   *
+   * <p>Set the parameters in {@link AsynchronousCounterMemoryUsageBenchmark#measure()}
+   *
+   * @param args Unused
+   */
+  public static void main(String[] args)
+      throws ExecutionException, InterruptedException, TimeoutException {
+    AsynchronousCounterMemoryUsageBenchmark asynchronousCounterMemoryUsageBenchmark =
+        new AsynchronousCounterMemoryUsageBenchmark();
+    asynchronousCounterMemoryUsageBenchmark.measure();
+  }
+
+  @SuppressWarnings("DefaultCharset")
+  private void measure()
+      throws ExecutionException, InterruptedException, TimeoutException {
+    // Parameters
+    AggregationTemporality aggregationTemporality = AggregationTemporality.CUMULATIVE;
+    MemoryMode memoryMode = MemoryMode.IMMUTABLE_DATA;
+    int countersCount = 1;
+    int cardinality = 100_000;
+
+    AsynchronousMetricStorageGarbageCollectionBenchmark.ThreadState benchmarkSetup =
+        new AsynchronousMetricStorageGarbageCollectionBenchmark.ThreadState(
+            countersCount, cardinality);
+
+    benchmarkSetup.aggregationTemporality = aggregationTemporality;
+    benchmarkSetup.memoryMode = memoryMode;
+
+    AsynchronousMetricStorageGarbageCollectionBenchmark benchmark =
+        new AsynchronousMetricStorageGarbageCollectionBenchmark();
+
+    benchmarkSetup.setup();
+
+    waitForGarbageCollection();
+    long usedMemoryBefore = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed();
+
+    ScheduledFuture<?> scheduledFuture = startMeasuringUsedMemoryInBackground();
+    try {
+      runBenchmark(benchmark, benchmarkSetup);
+    } finally {
+      executorService.shutdown();
+    }
+
+    try {
+      scheduledFuture.get(10, TimeUnit.SECONDS);
+    } catch (CancellationException e) {
+      // Due to the shutdown, ignore
+    }
+
+    waitForGarbageCollection();
+    long usedMemoryAfter = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed();
+    long memoryUsedButFreed = maxUsedMemory.get() - usedMemoryAfter;
+
+    System.out.printf(
+        ""
+            + "\nCounters = %d, Cardinality = %,d"
+            + "\n%s, %s: "
+            + "\nmemoryUsedButFreed                 = %,15d [bytes]"
+            + "\nmaxMemoryUsedDuringCollection      = %,15d [bytes]"
+            + "\nmemoryUsedBeforeCollectionStart    = %,15d [bytes]"
+            + "\nmemoryUsedAfterCollectionFinished  = %,15d [bytes]%n",
+        countersCount,
+        cardinality,
+        benchmarkSetup.aggregationTemporality,
+        benchmarkSetup.memoryMode,
+        memoryUsedButFreed,
+        maxUsedMemory.get(),
+        usedMemoryBefore,
+        usedMemoryAfter);
+  }
+
+  private static void waitForGarbageCollection() throws InterruptedException {
+    List<Long> collectionCountBefore = getGarbageCollectorsCollectionCount();
+    boolean oneGcCountIncreased;
+    int attempts = 0;
+    do {
+      System.gc();
+      Thread.sleep(TimeUnit.SECONDS.toMillis(2));
+    } while ((oneGcCountIncreased = oneOfGarbageCollectorsCountIncreased(collectionCountBefore))
+        && ++attempts < 3);
+    if (!oneGcCountIncreased) {
+      fail("Failed to get GC in " + attempts + " attempts");
+    }
+  }
+
+  private static boolean oneOfGarbageCollectorsCountIncreased(List<Long> collectionCountBefore) {
+    List<Long> gcCollectionCountNow = getGarbageCollectorsCollectionCount();
+    for (int i = 0; i < gcCollectionCountNow.size(); i++) {
+      Long now = gcCollectionCountNow.get(i);
+      if (now > collectionCountBefore.get(i)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  private static List<Long> getGarbageCollectorsCollectionCount() {
+    List<Long> collectionCountBefore = new ArrayList<>();
+    ManagementFactory.getGarbageCollectorMXBeans()
+        .forEach(
+            garbageCollectorMXBean ->
+                collectionCountBefore.add(garbageCollectorMXBean.getCollectionCount()));
+    return collectionCountBefore;
+  }
+
+  private ScheduledFuture<?> startMeasuringUsedMemoryInBackground() {
+    Runnable measureMaxMemory =
+        () -> {
+          try {
+            long used = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed();
+            if (used > maxUsedMemory.get()) {
+              maxUsedMemory.set(used);
+            }
+          } catch (Throwable t) {
+            t.printStackTrace();
+            throw t;
+          }
+        };
+
+    return executorService.scheduleWithFixedDelay(measureMaxMemory, 0, 2, TimeUnit.MILLISECONDS);
+  }
+
+  private static void runBenchmark(
+      AsynchronousMetricStorageGarbageCollectionBenchmark benchmark,
+      AsynchronousMetricStorageGarbageCollectionBenchmark.ThreadState threadState) {
+    benchmark.recordAndCollect(threadState);
+  }
+}