Add a new config for push queue capacity PYROSCOPE_PUSH_QUEUE_CAPACITY (

#37)

agent/src/main/java/io/pyroscope/javaagent/PyroscopeAgent.java

@@ -1,7 +1,6 @@
 package io.pyroscope.javaagent;
 
 import io.pyroscope.javaagent.config.Config;
-import one.profiler.Events;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedNoReferenceMessageFactory;
 import org.apache.logging.log4j.simple.SimpleLogger;
@@ -15,10 +14,6 @@
 import java.util.concurrent.TimeUnit;
 
 public class PyroscopeAgent {
-    // The number of snapshots simultaneously stored in memory is limited by this.
-    // The number is fairly arbitrary. If an average snapshot is 5KB, it's about 1 MB.
-    private static final int UPLOAD_QUEUE_CAPACITY = 200;
-    private static final OverfillQueue<Snapshot> uploadQueue = new OverfillQueue<>(UPLOAD_QUEUE_CAPACITY);
 
     public static void premain(final String agentArgs,
                                final Instrumentation inst) {
@@ -39,6 +34,10 @@ public static void premain(final String agentArgs,
             PreConfigLogger.LOGGER.error("Error starting profiler", e);
             return;
         }
+        logger.debug("Config {}", config);
+
+        final OverfillQueue<Snapshot> pushQueue = new OverfillQueue<>(config.pushQueueCapacity);
+
 
         try {
             final Profiler profiler = new Profiler(
@@ -60,7 +59,7 @@ public Thread newThread(Runnable r) {
 
             final Runnable dumpProfile = () -> {
                 try {
-                    uploadQueue.put(profiler.dump());
+                    pushQueue.put(profiler.dump());
                 } catch (final InterruptedException ignored) {
                     // It's fine to swallow InterruptedException here and exit.
                     // It's a cue to end the work and exit and we have nothing to clean up.
@@ -69,7 +68,7 @@ public Thread newThread(Runnable r) {
             executor.scheduleAtFixedRate(dumpProfile,
                     config.uploadInterval.toMillis(), config.uploadInterval.toMillis(), TimeUnit.MILLISECONDS);
 
-            final Thread uploaderThread = new Thread(new Uploader(logger, uploadQueue, config));
+            final Thread uploaderThread = new Thread(new Uploader(logger, pushQueue, config));
             uploaderThread.setDaemon(true);
             uploaderThread.start();
         } catch (final Throwable e) {

agent/src/main/java/io/pyroscope/javaagent/config/Config.java

@@ -23,6 +23,7 @@ public final class Config {
     private static final String PYROSCOPE_ADHOC_SERVER_ADDRESS_CONFIG = "PYROSCOPE_ADHOC_SERVER_ADDRESS";
     private static final String PYROSCOPE_AUTH_TOKEN_CONFIG = "PYROSCOPE_AUTH_TOKEN";
     private static final String PYROSCOPE_FORMAT_CONFIG = "PYROSCOPE_FORMAT";
+    private static final String PYROSCOPE_PUSH_QUEUE_CAPACITY_CONFIG = "PYROSCOPE_PUSH_QUEUE_CAPACITY";
 
     private static final String DEFAULT_SPY_NAME = "javaspy";
     private static final Duration DEFAULT_PROFILING_INTERVAL = Duration.ofMillis(10);
@@ -32,6 +33,9 @@ public final class Config {
     private static final Duration DEFAULT_UPLOAD_INTERVAL = Duration.ofSeconds(10);
     private static final String DEFAULT_SERVER_ADDRESS = "http://localhost:4040";
     private static final Format DEFAULT_FORMAT = Format.COLLAPSED;
+    // The number of snapshots simultaneously stored in memory is limited by this.
+    // The number is fairly arbitrary. If an average snapshot is 5KB, it's about 160 KB.
+    private static final int DEFAULT_PUSH_QUEUE_CAPACITY = 32;
 
     public final String spyName = DEFAULT_SPY_NAME;
     public final String applicationName;
@@ -45,6 +49,7 @@ public final class Config {
     public final String authToken;
     public final String timeseriesName;
     public final Format format;
+    public final int pushQueueCapacity;
 
     Config(final String applicationName,
            final Duration profilingInterval,
@@ -55,7 +60,8 @@ public final class Config {
            final Level logLevel,
            final String serverAddress,
            final String authToken,
-           final Format format
+           final Format format,
+           final int pushQueueCapacity
         ) {
         this.applicationName = applicationName;
         this.profilingInterval = profilingInterval;
@@ -68,12 +74,32 @@ public final class Config {
         this.authToken = authToken;
         this.timeseriesName = timeseriesName(applicationName, profilingEvent, format);
         this.format = format;
+        this.pushQueueCapacity = pushQueueCapacity;
     }
 
     public long profilingIntervalInHertz() {
         return durationToHertz(this.profilingInterval);
     }
 
+    @Override
+    public String toString() {
+        return "Config{" +
+            "spyName='" + spyName + '\'' +
+            ", applicationName='" + applicationName + '\'' +
+            ", profilingInterval=" + profilingInterval +
+            ", profilingEvent=" + profilingEvent +
+            ", profilingAlloc='" + profilingAlloc + '\'' +
+            ", profilingLock='" + profilingLock + '\'' +
+            ", uploadInterval=" + uploadInterval +
+            ", logLevel=" + logLevel +
+            ", serverAddress='" + serverAddress + '\'' +
+            ", authToken='" + authToken + '\'' +
+            ", timeseriesName='" + timeseriesName + '\'' +
+            ", format=" + format +
+            ", pushQueueCapacity=" + pushQueueCapacity +
+            '}';
+    }
+
     private static long durationToHertz(Duration duration) {
         Duration oneSecond = Duration.ofSeconds(1);
         return oneSecond.toNanos() / duration.toNanos();
@@ -90,7 +116,8 @@ public static Config build() {
             logLevel(),
             serverAddress(),
             authToken(),
-            format()
+            format(),
+            pushQueueCapacity()
         );
     }
 
@@ -233,4 +260,21 @@ private static Format format() {
                 return DEFAULT_FORMAT;
         }
     }
+
+    private static int pushQueueCapacity() {
+        final String strPushQueueCapacity = System.getenv(PYROSCOPE_PUSH_QUEUE_CAPACITY_CONFIG);
+        if (strPushQueueCapacity == null || strPushQueueCapacity.isEmpty()) {
+            return DEFAULT_PUSH_QUEUE_CAPACITY;
+        }
+        try {
+            int pushQueueCapacity = Integer.parseInt(strPushQueueCapacity);
+            if (pushQueueCapacity <= 0) {
+                return DEFAULT_PUSH_QUEUE_CAPACITY;
+            } else {
+                return pushQueueCapacity;
+            }
+        } catch (NumberFormatException e) {
+            return DEFAULT_PUSH_QUEUE_CAPACITY;
+        }
+    }
 }