Periodically Export Stacktraces From StagingArea

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

@@ -81,6 +81,15 @@ 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_STAGING_EMPTY_INTERVAL =
+      "splunk.snapshot.profiler.staging.empty.interval";
+  private static final Duration DEFAULT_SNAPSHOT_PROFILER_STAGING_EMPTY_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 +237,15 @@ public static Duration getSnapshotProfilerSamplingInterval(ConfigProperties prop
         CONFIG_KEY_SNAPSHOT_PROFILER_SAMPLING_INTERVAL,
         DEFAULT_SNAPSHOT_PROFILER_SAMPLING_INTERVAL);
   }
+
+  public static Duration getSnapshotProfilerStagingEmptyInterval(ConfigProperties properties) {
+    return properties.getDuration(
+        CONFIG_KEY_SNAPSHOT_PROFILER_STAGING_EMPTY_INTERVAL,
+        DEFAULT_SNAPSHOT_PROFILER_STAGING_EMPTY_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,88 @@
+/*
+ * 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 com.splunk.opentelemetry.profiler.util.HelpfulExecutors;
+import java.io.Closeable;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+class PeriodicallyExportingStagingArea implements StagingArea, Closeable {
+  private final ScheduledExecutorService scheduler =
+      HelpfulExecutors.newSingleThreadedScheduledExecutor("periodically-exporting-staging-area");
+  private final Set<StackTrace> stackTraces = Collections.newSetFromMap(new ConcurrentHashMap<>());
+  private final Supplier<StackTraceExporter> exporter;
+  private final int capacity;
+  private volatile boolean closed = false;
+
+  PeriodicallyExportingStagingArea(
+      Supplier<StackTraceExporter> exporter, Duration emptyDuration, int capacity) {
+    this.exporter = exporter;
+    this.capacity = capacity;
+    scheduler.scheduleAtFixedRate(
+        this::empty, emptyDuration.toMillis(), emptyDuration.toMillis(), TimeUnit.MILLISECONDS);
+  }
+
+  @Override
+  public void stage(StackTrace stackTrace) {
+    if (closed) {
+      return;
+    }
+    stackTraces.add(stackTrace);
+    if (stackTraces.size() >= capacity) {
+      empty();
+    }
+  }
+
+  @Override
+  public void empty() {
+    if (stackTraces.isEmpty()) {
+      return;
+    }
+
+    List<StackTrace> stackTracesToExport = new ArrayList<>(stackTraces);
+    exporter.get().export(stackTracesToExport);
+    stackTracesToExport.forEach(stackTraces::remove);
+  }
+
+  @Override
+  public void close() {
+    this.closed = true;
+
+    stopScheduledExecutor();
+    empty();
+  }
+
+  private void stopScheduledExecutor() {
+    try {
+      scheduler.shutdown();
+      if (!scheduler.awaitTermination(1, TimeUnit.SECONDS)) {
+        scheduler.shutdownNow();
+      }
+    } catch (InterruptedException e) {
+      scheduler.shutdownNow();
+      Thread.currentThread().interrupt();
+    }
+  }
+}

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.getSnapshotProfilerStagingEmptyInterval(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/StackTrace.java

@@ -20,6 +20,8 @@
 import java.lang.management.ThreadInfo;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Arrays;
+import java.util.Objects;
 
 class StackTrace {
   static StackTrace from(
@@ -95,4 +97,31 @@ String getTraceId() {
   String getSpanId() {
     return spanId;
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (o == null || getClass() != o.getClass()) return false;
+    StackTrace that = (StackTrace) o;
+    return threadId == that.threadId
+        && Objects.equals(timestamp, that.timestamp)
+        && Objects.equals(duration, that.duration)
+        && Objects.equals(threadName, that.threadName)
+        && threadState == that.threadState
+        && Objects.deepEquals(stackFrames, that.stackFrames)
+        && Objects.equals(traceId, that.traceId)
+        && Objects.equals(spanId, that.spanId);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(
+        timestamp,
+        duration,
+        threadId,
+        threadName,
+        threadState,
+        Arrays.hashCode(stackFrames),
+        traceId,
+        spanId);
+  }
 }

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,38 @@ 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.staging.empty.interval", String.valueOf(milliseconds)));
+    assertEquals(
+        Duration.ofMillis(milliseconds),
+        Configuration.getSnapshotProfilerStagingEmptyInterval(properties));
+  }
+
+  @Test
+  void getDefaultSnapshotProfilerEmptyStagingInterval() {
+    var properties = DefaultConfigProperties.create(Collections.emptyMap());
+    assertEquals(
+        Duration.ofSeconds(5), Configuration.getSnapshotProfilerStagingEmptyInterval(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));
+  }
 }