Separate queue for metrics to avoid dropping telemetry (#2062)

* Separate queue for metrics to avoid dropping telemetry

* And bump version

Author: trask

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/configuration/Configuration.java

@@ -244,7 +244,10 @@ public static class PreviewConfiguration {
 
     public List<InstrumentationKeyOverride> instrumentationKeyOverrides = new ArrayList<>();
 
-    public int exportQueueCapacity = 2048;
+    public int generalExportQueueCapacity = 2048;
+    // metrics get flooded every 60 seconds by default, so need larger queue size to avoid dropping
+    // telemetry (they are much smaller so a larger queue size is ok)
+    public int metricsExportQueueCapacity = 65536;
   }
 
   public static class InheritedAttribute {

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/init/AiComponentInstaller.java

@@ -173,7 +173,8 @@ private static AppIdSupplier start(Instrumentation instrumentation) {
             .setIkeyEndpointMap(ikeyEndpointMap)
             .setStatsbeatModule(statsbeatModule)
             .setReadOnlyFileSystem(readOnlyFileSystem)
-            .setMaxExportQueueSize(config.preview.exportQueueCapacity)
+            .setGeneralExportQueueSize(config.preview.generalExportQueueCapacity)
+            .setMetricsExportQueueSize(config.preview.metricsExportQueueCapacity)
             .setAadAuthentication(config.preview.authentication)
             .build();
 

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/telemetry/BatchSpanProcessor.java

@@ -61,7 +61,8 @@ public static BatchSpanProcessorBuilder builder(TelemetryChannel spanExporter) {
       long scheduleDelayNanos,
       int maxQueueSize,
       int maxExportBatchSize,
-      long exporterTimeoutNanos) {
+      long exporterTimeoutNanos,
+      String queueName) {
     MpscArrayQueue<TelemetryItem> queue = new MpscArrayQueue<>(maxQueueSize);
     this.worker =
         new Worker(
@@ -70,7 +71,8 @@ public static BatchSpanProcessorBuilder builder(TelemetryChannel spanExporter) {
             maxExportBatchSize,
             exporterTimeoutNanos,
             queue,
-            queue.capacity());
+            queue.capacity(),
+            queueName);
     Thread workerThread = new DaemonThreadFactory(WORKER_THREAD_NAME).newThread(worker);
     workerThread.start();
   }
@@ -103,6 +105,7 @@ private static final class Worker implements Runnable {
 
     private final Queue<TelemetryItem> queue;
     private final int queueCapacity;
+    private final String queueName;
     // When waiting on the spans queue, exporter thread sets this atomic to the number of more
     // spans it needs before doing an export. Writer threads would then wait for the queue to reach
     // spansNeeded size before notifying the exporter thread about new entries.
@@ -124,25 +127,32 @@ private Worker(
         int maxExportBatchSize,
         long exporterTimeoutNanos,
         Queue<TelemetryItem> queue,
-        int queueCapacity) {
+        int queueCapacity,
+        String queueName) {
       this.spanExporter = spanExporter;
       this.scheduleDelayNanos = scheduleDelayNanos;
       this.maxExportBatchSize = maxExportBatchSize;
       this.exporterTimeoutNanos = exporterTimeoutNanos;
       this.queue = queue;
       this.queueCapacity = queueCapacity;
+      this.queueName = queueName;
       this.signal = new ArrayBlockingQueue<>(1);
       this.batch = new ArrayList<>(this.maxExportBatchSize);
     }
 
     private void addSpan(TelemetryItem span) {
       if (!queue.offer(span)) {
         queuingSpanLogger.recordFailure(
-            "Max export queue size of "
+            "Max "
+                + queueName
+                + " export queue capacity of "
                 + queueCapacity
-                + " has been hit, dropping a span (max export queue size can be increased"
-                + " in the applicationinsights.json configuration file, e.g."
-                + " { \"preview\": { \"exportQueueCapacity\": "
+                + " has been hit, dropping a telemetry record (max "
+                + queueName
+                + " export queue capacity can be increased in the applicationinsights.json"
+                + " configuration file, e.g. { \"preview\": { \""
+                + queueName
+                + "ExportQueueCapacity\": "
                 + (queueCapacity * 2)
                 + " } }");
       } else {

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/telemetry/BatchSpanProcessorBuilder.java

@@ -129,8 +129,13 @@ public BatchSpanProcessorBuilder setMaxExportBatchSize(int maxExportBatchSize) {
    * @return a new {@link io.opentelemetry.sdk.trace.export.BatchSpanProcessor}.
    * @throws NullPointerException if the {@code spanExporter} is {@code null}.
    */
-  public BatchSpanProcessor build() {
+  public BatchSpanProcessor build(String queueName) {
     return new BatchSpanProcessor(
-        spanExporter, scheduleDelayNanos, maxQueueSize, maxExportBatchSize, exporterTimeoutNanos);
+        spanExporter,
+        scheduleDelayNanos,
+        maxQueueSize,
+        maxExportBatchSize,
+        exporterTimeoutNanos,
+        queueName);
   }
 }

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/telemetry/TelemetryClient.java

@@ -95,12 +95,14 @@ public class TelemetryClient {
   private final Cache<String, String> ikeyEndpointMap;
   private final StatsbeatModule statsbeatModule;
   private final boolean readOnlyFileSystem;
-  private final int exportQueueCapacity;
+  private final int generalExportQueueCapacity;
+  private final int metricsExportQueueCapacity;
 
   @Nullable private final Configuration.AadAuthentication aadAuthentication;
 
   private final Object channelInitLock = new Object();
-  private volatile @MonotonicNonNull BatchSpanProcessor channelBatcher;
+  private volatile @MonotonicNonNull BatchSpanProcessor generalChannelBatcher;
+  private volatile @MonotonicNonNull BatchSpanProcessor metricsChannelBatcher;
   private volatile @MonotonicNonNull BatchSpanProcessor statsbeatChannelBatcher;
 
   public static TelemetryClient.Builder builder() {
@@ -124,7 +126,8 @@ public TelemetryClient(Builder builder) {
     this.ikeyEndpointMap = builder.ikeyEndpointMap;
     this.statsbeatModule = builder.statsbeatModule;
     this.readOnlyFileSystem = builder.readOnlyFileSystem;
-    this.exportQueueCapacity = builder.exportQueueCapacity;
+    this.generalExportQueueCapacity = builder.generalExportQueueCapacity;
+    this.metricsExportQueueCapacity = builder.metricsExportQueueCapacity;
     this.aadAuthentication = builder.aadAuthentication;
   }
 
@@ -182,7 +185,11 @@ public void trackAsync(TelemetryItem telemetry) {
     // batching, retry, throttling, and writing to disk on failure occur downstream
     // for simplicity not reporting back success/failure from this layer
     // only that it was successfully delivered to the next layer
-    getChannelBatcher().trackAsync(telemetry);
+    if (data instanceof MetricsData) {
+      getMetricsChannelBatcher().trackAsync(telemetry);
+    } else {
+      getGeneralChannelBatcher().trackAsync(telemetry);
+    }
   }
 
   public void trackStatsbeatAsync(TelemetryItem telemetry) {
@@ -193,50 +200,69 @@ public void trackStatsbeatAsync(TelemetryItem telemetry) {
   }
 
   public CompletableResultCode flushChannelBatcher() {
-    if (channelBatcher != null) {
-      return channelBatcher.forceFlush();
+    if (generalChannelBatcher != null) {
+      return generalChannelBatcher.forceFlush();
     } else {
       return CompletableResultCode.ofSuccess();
     }
   }
 
-  public BatchSpanProcessor getChannelBatcher() {
-    if (channelBatcher == null) {
+  private BatchSpanProcessor getGeneralChannelBatcher() {
+    if (generalChannelBatcher == null) {
       synchronized (channelInitLock) {
-        if (channelBatcher == null) {
-          File telemetryFolder;
-          LocalFileLoader localFileLoader = null;
-          LocalFileWriter localFileWriter = null;
-          if (!readOnlyFileSystem) {
-            telemetryFolder = LocalStorageUtils.getOfflineTelemetryFolder();
-            LocalFileCache localFileCache = new LocalFileCache(telemetryFolder);
-            localFileLoader =
-                new LocalFileLoader(
-                    localFileCache, telemetryFolder, statsbeatModule.getNonessentialStatsbeat());
-            localFileWriter =
-                new LocalFileWriter(
-                    localFileCache, telemetryFolder, statsbeatModule.getNonessentialStatsbeat());
-          }
-
-          TelemetryChannel channel =
-              TelemetryChannel.create(
-                  endpointProvider.getIngestionEndpointUrl(),
-                  localFileWriter,
-                  ikeyEndpointMap,
-                  statsbeatModule,
-                  false,
-                  aadAuthentication);
-
-          if (!readOnlyFileSystem) {
-            LocalFileSender.start(localFileLoader, channel);
-          }
+        if (generalChannelBatcher == null) {
+          generalChannelBatcher = initChannelBatcher(generalExportQueueCapacity, 512, "general");
+        }
+      }
+    }
+    return generalChannelBatcher;
+  }
 
-          channelBatcher =
-              BatchSpanProcessor.builder(channel).setMaxQueueSize(exportQueueCapacity).build();
+  // metrics get flooded every 60 seconds by default, so need much larger queue size to avoid
+  // dropping telemetry (they are much smaller so a larger queue size and larger batch size are ok)
+  private BatchSpanProcessor getMetricsChannelBatcher() {
+    if (metricsChannelBatcher == null) {
+      synchronized (channelInitLock) {
+        if (metricsChannelBatcher == null) {
+          metricsChannelBatcher = initChannelBatcher(metricsExportQueueCapacity, 2048, "metrics");
         }
       }
     }
-    return channelBatcher;
+    return metricsChannelBatcher;
+  }
+
+  private BatchSpanProcessor initChannelBatcher(
+      int exportQueueCapacity, int maxExportBatchSize, String queueName) {
+    LocalFileLoader localFileLoader = null;
+    LocalFileWriter localFileWriter = null;
+    if (!readOnlyFileSystem) {
+      File telemetryFolder = LocalStorageUtils.getOfflineTelemetryFolder();
+      LocalFileCache localFileCache = new LocalFileCache(telemetryFolder);
+      localFileLoader =
+          new LocalFileLoader(
+              localFileCache, telemetryFolder, statsbeatModule.getNonessentialStatsbeat());
+      localFileWriter =
+          new LocalFileWriter(
+              localFileCache, telemetryFolder, statsbeatModule.getNonessentialStatsbeat());
+    }
+
+    TelemetryChannel channel =
+        TelemetryChannel.create(
+            endpointProvider.getIngestionEndpointUrl(),
+            localFileWriter,
+            ikeyEndpointMap,
+            statsbeatModule,
+            false,
+            aadAuthentication);
+
+    if (!readOnlyFileSystem) {
+      LocalFileSender.start(localFileLoader, channel);
+    }
+
+    return BatchSpanProcessor.builder(channel)
+        .setMaxQueueSize(exportQueueCapacity)
+        .setMaxExportBatchSize(maxExportBatchSize)
+        .build(queueName);
   }
 
   public BatchSpanProcessor getStatsbeatChannelBatcher() {
@@ -266,7 +292,7 @@ public BatchSpanProcessor getStatsbeatChannelBatcher() {
             LocalFileSender.start(localFileLoader, channel);
           }
 
-          statsbeatChannelBatcher = BatchSpanProcessor.builder(channel).build();
+          statsbeatChannelBatcher = BatchSpanProcessor.builder(channel).build("statsbeat");
         }
       }
     }
@@ -486,7 +512,8 @@ public static class Builder {
     private Cache<String, String> ikeyEndpointMap;
     private StatsbeatModule statsbeatModule;
     private boolean readOnlyFileSystem;
-    private int exportQueueCapacity;
+    private int generalExportQueueCapacity;
+    private int metricsExportQueueCapacity;
     @Nullable private Configuration.AadAuthentication aadAuthentication;
 
     public Builder setCustomDimensions(Map<String, String> customDimensions) {
@@ -533,8 +560,13 @@ public Builder setReadOnlyFileSystem(boolean readOnlyFileSystem) {
       return this;
     }
 
-    public Builder setMaxExportQueueSize(int exportQueueCapacity) {
-      this.exportQueueCapacity = exportQueueCapacity;
+    public Builder setGeneralExportQueueSize(int generalExportQueueCapacity) {
+      this.generalExportQueueCapacity = generalExportQueueCapacity;
+      return this;
+    }
+
+    public Builder setMetricsExportQueueSize(int metricsExportQueueCapacity) {
+      this.metricsExportQueueCapacity = metricsExportQueueCapacity;
       return this;
     }
 

gradle.properties

@@ -1,5 +1,5 @@
 // Project properties
-version=3.2.5-BETA
+version=3.2.5-BETA.2
 group=com.microsoft.azure
 # gradle default is 256m which causes sporadic failures - https://docs.gradle.org/current/userguide/build_environment.html#sec:configuring_jvm_memory
 org.gradle.jvmargs=-XX:MaxMetaspaceSize=512m