Periodically Export Stacktraces From StagingArea

profiler/src/main/java/com/splunk/opentelemetry/profiler/Configuration.java

@@ -81,6 +81,14 @@ public class Configuration implements AutoConfigurationCustomizerProvider {
       "splunk.snapshot.profiler.sampling.interval";
   private static final Duration DEFAULT_SNAPSHOT_PROFILER_SAMPLING_INTERVAL = Duration.ofMillis(10);
 
+  private static final String CONFIG_KEY_SNAPSHOT_PROFILER_EXPORT_INTERVAL =
+      "splunk.snapshot.profiler.export.interval";
+  private static final Duration DEFAULT_SNAPSHOT_PROFILER_EXPORT_INTERVAL = Duration.ofSeconds(5);
+
+  private static final String CONFIG_KEY_SNAPSHOT_PROFILER_STAGING_CAPACITY =
+      "splunk.snapshot.profiler.staging.capacity";
+  private static final int DEFAULT_SNAPSHOT_PROFILER_STAGING_CAPACITY = 2000;
+
   @Override
   public void customize(AutoConfigurationCustomizer autoConfiguration) {
     autoConfiguration.addPropertiesSupplier(this::defaultProperties);
@@ -228,4 +236,14 @@ public static Duration getSnapshotProfilerSamplingInterval(ConfigProperties prop
         CONFIG_KEY_SNAPSHOT_PROFILER_SAMPLING_INTERVAL,
         DEFAULT_SNAPSHOT_PROFILER_SAMPLING_INTERVAL);
   }
+
+  public static Duration getSnapshotProfilerExportInterval(ConfigProperties properties) {
+    return properties.getDuration(
+        CONFIG_KEY_SNAPSHOT_PROFILER_EXPORT_INTERVAL, DEFAULT_SNAPSHOT_PROFILER_EXPORT_INTERVAL);
+  }
+
+  public static int getSnapshotProfilerStagingCapacity(ConfigProperties properties) {
+    return properties.getInt(
+        CONFIG_KEY_SNAPSHOT_PROFILER_STAGING_CAPACITY, DEFAULT_SNAPSHOT_PROFILER_STAGING_CAPACITY);
+  }
 }

profiler/src/main/java/com/splunk/opentelemetry/profiler/snapshot/PeriodicallyExportingStagingArea.java

@@ -0,0 +1,133 @@
+/*
+ * Copyright Splunk Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.splunk.opentelemetry.profiler.snapshot;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+class PeriodicallyExportingStagingArea implements StagingArea {
+  private static final String WORKER_THREAD_NAME =
+      PeriodicallyExportingStagingArea.class.getSimpleName() + "_WorkerThread";
+
+  private volatile boolean closed = false;
+
+  private final Worker worker;
+
+  PeriodicallyExportingStagingArea(
+      Supplier<StackTraceExporter> exporter, Duration delay, int capacity) {
+    worker = new Worker(exporter, delay, capacity);
+    worker.setName(WORKER_THREAD_NAME);
+    worker.setDaemon(true);
+    worker.start();
+  }
+
+  @Override
+  public void stage(StackTrace stackTrace) {
+    if (closed) {
+      return;
+    }
+    worker.add(stackTrace);
+  }
+
+  @Override
+  public void empty() {}
+
+  @Override
+  public void close() {
+    this.closed = true;
+
+    // Wait for the worker thread to exit. Note that this does not guarantee that the pending items
+    // are exported as we don't attempt to wait for the actual export to complete.
+    try {
+      worker.shutdown();
+      worker.join();
+    } catch (InterruptedException exception) {
+      Thread.currentThread().interrupt();
+    }
+  }
+
+  private static class Worker extends Thread {
+    // when shutting down we queue a fake stack trace to ensure that shutdown process starts
+    // immediately
+    private static final Object SHUTDOWN_MARKER = new Object();
+
+    private final BlockingQueue<Object> queue;
+    private final Supplier<StackTraceExporter> exporter;
+    private final Duration delay;
+    private final int maxExportBatchSize;
+
+    private volatile boolean shutdown = false;
+    private long nextExportTime;
+
+    private Worker(Supplier<StackTraceExporter> exporter, Duration delay, int maxExportBatchSize) {
+      this.exporter = exporter;
+      this.delay = delay;
+      this.maxExportBatchSize = maxExportBatchSize;
+      // set the queue size to 4x the batch size, in sdk batch processors both of these are
+      // configurable but by default queue size is also 4*batch size
+      this.queue = new ArrayBlockingQueue<>(maxExportBatchSize * 4);
+
+      updateNextExportTime();
+    }
+
+    void add(StackTrace stackTrace) {
+      // If queue is full drop the stack trace, not much we can do.
+      queue.offer(stackTrace);
+    }
+
+    @Override
+    public void run() {
+      List<StackTrace> stackTracesToExport = new ArrayList<>();
+      try {
+        // run until shutdown is called and all queued spans are passed to the exporter
+        while (!shutdown || !queue.isEmpty() || !stackTracesToExport.isEmpty()) {
+          Object stackTrace = queue.poll(nextExportTime - System.nanoTime(), TimeUnit.NANOSECONDS);
+          if (stackTrace != null && stackTrace != SHUTDOWN_MARKER) {
+            stackTracesToExport.add((StackTrace) stackTrace);
+          }
+          // trigger export when either next export time is reached, we have max batch size, or we
+          // are shutting down and have read all the queued stacks
+          if (System.nanoTime() >= nextExportTime
+              || stackTracesToExport.size() >= maxExportBatchSize
+              || (shutdown && queue.isEmpty())) {
+            exporter.get().export(stackTracesToExport);
+            stackTracesToExport = new ArrayList<>();
+            updateNextExportTime();
+          }
+        }
+      } catch (InterruptedException exception) {
+        Thread.currentThread().interrupt();
+      }
+    }
+
+    private void updateNextExportTime() {
+      nextExportTime = System.nanoTime() + delay.toNanos();
+    }
+
+    private void shutdown() throws InterruptedException {
+      shutdown = true;
+      // we don't care if the queue is full and offer fails, we only wish to ensure that there is
+      // something in the queue so that shutdown could start immediately
+      queue.offer(SHUTDOWN_MARKER);
+    }
+  }
+}

profiler/src/main/java/com/splunk/opentelemetry/profiler/snapshot/ScheduledExecutorStackTraceSampler.java

@@ -70,7 +70,6 @@ public void stop(SpanContext spanContext) {
           if (spanContext.equals(sampler.getSpanContext())) {
             sampler.shutdown();
             waitForShutdown(sampler);
-            stagingArea.get().empty(spanContext.getTraceId());
             return null;
           }
           return sampler;
@@ -153,7 +152,7 @@ public void run() {
         String spanId = retrieveActiveSpan(thread).getSpanId();
         StackTrace stackTrace =
             StackTrace.from(Instant.now(), samplingPeriod, threadInfo, traceId, spanId);
-        stagingArea.get().stage(traceId, stackTrace);
+        stagingArea.get().stage(stackTrace);
       } catch (Exception e) {
         logger.log(Level.SEVERE, e, samplerErrorMessage(traceId, thread.getId()));
       } finally {

profiler/src/main/java/com/splunk/opentelemetry/profiler/snapshot/SnapshotProfilingSdkCustomizer.java

@@ -47,12 +47,18 @@ public SnapshotProfilingSdkCustomizer() {
   private static Function<ConfigProperties, StackTraceSampler> stackTraceSamplerProvider() {
     return properties -> {
       Duration samplingPeriod = Configuration.getSnapshotProfilerSamplingInterval(properties);
-      ConfigurableSupplier<StagingArea> supplier = StagingArea.SUPPLIER;
-      supplier.configure(new AccumulatingStagingArea(StackTraceExporter.SUPPLIER));
-      return new ScheduledExecutorStackTraceSampler(supplier, SpanTracker.SUPPLIER, samplingPeriod);
+      StagingArea.SUPPLIER.configure(createStagingArea(properties));
+      return new ScheduledExecutorStackTraceSampler(
+          StagingArea.SUPPLIER, SpanTracker.SUPPLIER, samplingPeriod);
     };
   }
 
+  private static StagingArea createStagingArea(ConfigProperties properties) {
+    Duration interval = Configuration.getSnapshotProfilerExportInterval(properties);
+    int capacity = Configuration.getSnapshotProfilerStagingCapacity(properties);
+    return new PeriodicallyExportingStagingArea(StackTraceExporter.SUPPLIER, interval, capacity);
+  }
+
   @VisibleForTesting
   SnapshotProfilingSdkCustomizer(
       TraceRegistry registry, StackTraceSampler sampler, SpanTrackingActivator activator) {

profiler/src/main/java/com/splunk/opentelemetry/profiler/snapshot/StagingArea.java

@@ -26,16 +26,16 @@ interface StagingArea extends Closeable {
   StagingArea NOOP =
       new StagingArea() {
         @Override
-        public void stage(String traceId, StackTrace stackTrace) {}
+        public void stage(StackTrace stackTrace) {}
 
         @Override
-        public void empty(String traceId) {}
+        public void empty() {}
       };
   ConfigurableSupplier<StagingArea> SUPPLIER = new ConfigurableSupplier<>(NOOP);
 
-  void stage(String traceId, StackTrace stackTrace);
+  void stage(StackTrace stackTrace);
 
-  void empty(String traceId);
+  void empty();
 
   default void close() {}
 }

profiler/src/test/java/com/splunk/opentelemetry/profiler/ConfigurationTest.java

@@ -204,4 +204,37 @@ void getDefaultSnapshotProfilerSamplingInterval() {
     assertEquals(
         Duration.ofMillis(10), Configuration.getSnapshotProfilerSamplingInterval(properties));
   }
+
+  @ParameterizedTest
+  @ValueSource(ints = {128, 512, 2056})
+  void getConfiguredSnapshotProfilerEmptyStagingInterval(int milliseconds) {
+    var properties =
+        DefaultConfigProperties.create(
+            Map.of("splunk.snapshot.profiler.export.interval", String.valueOf(milliseconds)));
+    assertEquals(
+        Duration.ofMillis(milliseconds),
+        Configuration.getSnapshotProfilerExportInterval(properties));
+  }
+
+  @Test
+  void getDefaultSnapshotProfilerEmptyStagingInterval() {
+    var properties = DefaultConfigProperties.create(Collections.emptyMap());
+    assertEquals(
+        Duration.ofSeconds(5), Configuration.getSnapshotProfilerExportInterval(properties));
+  }
+
+  @ParameterizedTest
+  @ValueSource(ints = {100, 1000, 10_000})
+  void getConfiguredSnapshotProfilerStagingCapacity(int value) {
+    var properties =
+        DefaultConfigProperties.create(
+            Map.of("splunk.snapshot.profiler.staging.capacity", String.valueOf(value)));
+    assertEquals(value, Configuration.getSnapshotProfilerStagingCapacity(properties));
+  }
+
+  @Test
+  void getDefaultSnapshotProfilerStagingCapacity() {
+    var properties = DefaultConfigProperties.create(Collections.emptyMap());
+    assertEquals(2000, Configuration.getSnapshotProfilerStagingCapacity(properties));
+  }
 }