fix quick pulse memory leak (#1879)

* fix quick pulse memory leak

* add/update tests

* use list instead of arrayblocking queue

* use quickpulsestatus instead of headerinfo

Author: kryalama

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/quickpulse/QuickPulseCoordinator.java

@@ -80,7 +80,8 @@ private long sendData() {
     QuickPulseHeaderInfo currentQuickPulseHeaderInfo = dataSender.getQuickPulseHeaderInfo();
 
     this.handleReceivedHeaders(currentQuickPulseHeaderInfo);
-
+    QuickPulseDataCollector.INSTANCE.setQuickPulseStatus(
+        currentQuickPulseHeaderInfo.getQuickPulseStatus());
     switch (currentQuickPulseHeaderInfo.getQuickPulseStatus()) {
       case ERROR:
         pingMode = true;
@@ -105,6 +106,7 @@ private long sendData() {
   private long ping() {
     QuickPulseHeaderInfo pingResult = pingSender.ping(qpsServiceRedirectedEndpoint);
     this.handleReceivedHeaders(pingResult);
+    QuickPulseDataCollector.INSTANCE.setQuickPulseStatus(pingResult.getQuickPulseStatus());
     switch (pingResult.getQuickPulseStatus()) {
       case ERROR:
         return waitOnErrorInMillis;

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/quickpulse/QuickPulseDataCollector.java

@@ -113,6 +113,7 @@ private CountAndDuration(long count, long duration) {
   static class Counters {
     private static final long MAX_COUNT = 524287L;
     private static final long MAX_DURATION = 17592186044415L;
+    private static final int MAX_DOCUMENTS_SIZE = 1000;
 
     public final AtomicInteger exceptions = new AtomicInteger(0);
 
@@ -139,6 +140,7 @@ static CountAndDuration decodeCountAndDuration(long countAndDuration) {
   private final AtomicReference<Counters> counters = new AtomicReference<>(null);
   private final MemoryMXBean memory;
   private final CpuPerformanceCounterCalculator cpuPerformanceCounterCalculator;
+  private volatile QuickPulseStatus quickPulseStatus;
 
   QuickPulseDataCollector() {
     CpuPerformanceCounterCalculator temp;
@@ -162,17 +164,28 @@ static CountAndDuration decodeCountAndDuration(long countAndDuration) {
     }
     cpuPerformanceCounterCalculator = temp;
     memory = ManagementFactory.getMemoryMXBean();
+    quickPulseStatus = QuickPulseStatus.QP_IS_OFF;
   }
 
   public synchronized void disable() {
     counters.set(null);
+    quickPulseStatus = QuickPulseStatus.QP_IS_OFF;
   }
 
   public synchronized void enable(TelemetryClient telemetryClient) {
     this.telemetryClient = telemetryClient;
     counters.set(new Counters());
   }
 
+  public synchronized void setQuickPulseStatus(QuickPulseStatus quickPulseStatus) {
+    this.quickPulseStatus = quickPulseStatus;
+  }
+
+  // Used only in tests
+  public synchronized QuickPulseStatus getQuickPulseStatus() {
+    return this.quickPulseStatus;
+  }
+
   @Nullable
   public synchronized FinalCounters getAndRestart() {
     Counters currentCounters = counters.getAndSet(new Counters());
@@ -194,8 +207,8 @@ synchronized FinalCounters peek() {
   }
 
   public void add(TelemetryItem telemetryItem) {
-    if (telemetryClient == null) {
-      // quick pulse is not enabled
+    if (telemetryClient == null || quickPulseStatus != QuickPulseStatus.QP_IS_ON) {
+      // quick pulse is not enabled or quick pulse data sender is not enabled
       return;
     }
 
@@ -252,7 +265,9 @@ private void addDependency(RemoteDependencyData telemetry, int itemCount) {
     quickPulseDependencyDocument.setProperties(
         aggregateProperties(telemetry.getProperties(), telemetry.getMeasurements()));
     synchronized (counters.documentList) {
-      counters.documentList.add(quickPulseDependencyDocument);
+      if (counters.documentList.size() < Counters.MAX_DOCUMENTS_SIZE) {
+        counters.documentList.add(quickPulseDependencyDocument);
+      }
     }
   }
 
@@ -287,7 +302,9 @@ private void addException(TelemetryExceptionData exceptionData, int itemCount) {
       quickPulseExceptionDocument.setExceptionType(exceptionList.get(0).getTypeName());
     }
     synchronized (counters.documentList) {
-      counters.documentList.add(quickPulseExceptionDocument);
+      if (counters.documentList.size() < Counters.MAX_DOCUMENTS_SIZE) {
+        counters.documentList.add(quickPulseExceptionDocument);
+      }
     }
   }
 
@@ -315,7 +332,9 @@ private void addRequest(RequestData requestTelemetry, int itemCount) {
     quickPulseRequestDocument.setProperties(
         aggregateProperties(requestTelemetry.getProperties(), requestTelemetry.getMeasurements()));
     synchronized (counters.documentList) {
-      counters.documentList.add(quickPulseRequestDocument);
+      if (counters.documentList.size() < Counters.MAX_DOCUMENTS_SIZE) {
+        counters.documentList.add(quickPulseRequestDocument);
+      }
     }
   }
 

agent/agent-tooling/src/test/java/com/microsoft/applicationinsights/agent/internal/quickpulse/QuickPulseCoordinatorTest.java

@@ -21,15 +21,29 @@
 
 package com.microsoft.applicationinsights.agent.internal.quickpulse;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.notNull;
 import static org.mockito.Mockito.mock;
 
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 
 class QuickPulseCoordinatorTest {
+
+  @BeforeEach
+  void setup() {
+    QuickPulseDataCollector.INSTANCE.disable();
+  }
+
+  @AfterEach
+  void tearDown() {
+    QuickPulseDataCollector.INSTANCE.disable();
+  }
+
   @Test
   void testOnlyPings() throws InterruptedException {
     QuickPulseDataFetcher mockFetcher = mock(QuickPulseDataFetcher.class);
@@ -65,6 +79,9 @@ void testOnlyPings() throws InterruptedException {
     Mockito.verify(mockSender, Mockito.never()).getQuickPulseHeaderInfo();
 
     Mockito.verify(mockPingSender, Mockito.atLeast(1)).ping(null);
+    // make sure QP_IS_OFF after ping
+    assertThat(QuickPulseDataCollector.INSTANCE.getQuickPulseStatus())
+        .isEqualTo(QuickPulseStatus.QP_IS_OFF);
   }
 
   @Test
@@ -107,6 +124,9 @@ void testOnePingAndThenOnePost() throws InterruptedException {
     Mockito.verify(mockSender, Mockito.times(1)).getQuickPulseHeaderInfo();
 
     Mockito.verify(mockPingSender, Mockito.atLeast(1)).ping(null);
+    // Make sure QP_IS_OFF after one post and ping
+    assertThat(QuickPulseDataCollector.INSTANCE.getQuickPulseStatus())
+        .isEqualTo(QuickPulseStatus.QP_IS_OFF);
   }
 
   // FIXME (trask) sporadically failing on CI

agent/agent-tooling/src/test/java/com/microsoft/applicationinsights/agent/internal/quickpulse/QuickPulseDataCollectorTests.java

@@ -78,6 +78,7 @@ void nullCountersAfterDisable() {
   void requestTelemetryIsCounted_DurationIsSum() {
     TelemetryClient telemetryClient = new TelemetryClient();
     telemetryClient.setInstrumentationKey(FAKE_INSTRUMENTATION_KEY);
+    QuickPulseDataCollector.INSTANCE.setQuickPulseStatus(QuickPulseStatus.QP_IS_ON);
     QuickPulseDataCollector.INSTANCE.enable(telemetryClient);
 
     // add a success and peek
@@ -120,6 +121,7 @@ void requestTelemetryIsCounted_DurationIsSum() {
   void dependencyTelemetryIsCounted_DurationIsSum() {
     TelemetryClient telemetryClient = new TelemetryClient();
     telemetryClient.setInstrumentationKey(FAKE_INSTRUMENTATION_KEY);
+    QuickPulseDataCollector.INSTANCE.setQuickPulseStatus(QuickPulseStatus.QP_IS_ON);
     QuickPulseDataCollector.INSTANCE.enable(telemetryClient);
 
     // add a success and peek.
@@ -162,6 +164,7 @@ void dependencyTelemetryIsCounted_DurationIsSum() {
   void exceptionTelemetryIsCounted() {
     TelemetryClient telemetryClient = new TelemetryClient();
     telemetryClient.setInstrumentationKey(FAKE_INSTRUMENTATION_KEY);
+    QuickPulseDataCollector.INSTANCE.setQuickPulseStatus(QuickPulseStatus.QP_IS_ON);
     QuickPulseDataCollector.INSTANCE.enable(telemetryClient);
 
     TelemetryItem telemetry = createExceptionTelemetry(new Exception());
@@ -253,4 +256,30 @@ private static void assertCountersReset(FinalCounters counters) {
 
     assertThat(counters.exceptions).isEqualTo(0);
   }
+
+  @Test
+  void checkDocumentsListSize() {
+    TelemetryClient telemetryClient = new TelemetryClient();
+    telemetryClient.setInstrumentationKey(FAKE_INSTRUMENTATION_KEY);
+    QuickPulseDataCollector.INSTANCE.setQuickPulseStatus(QuickPulseStatus.QP_IS_ON);
+    QuickPulseDataCollector.INSTANCE.enable(telemetryClient);
+
+    final long duration = 112233L;
+    TelemetryItem telemetry =
+        createRequestTelemetry("request-test", new Date(), duration, "200", true);
+    telemetry.setInstrumentationKey(FAKE_INSTRUMENTATION_KEY);
+    for (int i = 0; i < 1005; i++) {
+      QuickPulseDataCollector.INSTANCE.add(telemetry);
+    }
+    // check max documentList size
+    assertThat(QuickPulseDataCollector.INSTANCE.getAndRestart().documentList.size())
+        .isEqualTo(1000);
+
+    QuickPulseDataCollector.INSTANCE.setQuickPulseStatus(QuickPulseStatus.QP_IS_OFF);
+    for (int i = 0; i < 5; i++) {
+      QuickPulseDataCollector.INSTANCE.add(telemetry);
+    }
+    // no telemetry items are added when QP_IS_OFF
+    assertThat(QuickPulseDataCollector.INSTANCE.getAndRestart().documentList.size()).isEqualTo(0);
+  }
 }

agent/agent-tooling/src/test/java/com/microsoft/applicationinsights/agent/internal/quickpulse/QuickPulseIntegrationTests.java

@@ -110,6 +110,7 @@ public void testPostRequest() throws InterruptedException {
     telemetryClient.setConnectionString(connectionString);
     QuickPulseDataFetcher dataFetcher =
         new QuickPulseDataFetcher(sendQueue, telemetryClient, "machine1", "instance1", null);
+    QuickPulseDataCollector.INSTANCE.setQuickPulseStatus(QuickPulseStatus.QP_IS_ON);
     QuickPulseDataCollector.INSTANCE.enable(telemetryClient);
     final long duration = 112233L;
     // Request Telemetry
@@ -148,6 +149,8 @@ public void testPostRequest() throws InterruptedException {
     Thread.sleep(50);
     assertTrue(pingCountDown.await(1, TimeUnit.SECONDS));
     assertThat(quickPulseHeaderInfo.getQuickPulseStatus()).isEqualTo(QuickPulseStatus.QP_IS_ON);
+    assertThat(QuickPulseDataCollector.INSTANCE.getQuickPulseStatus())
+        .isEqualTo(QuickPulseStatus.QP_IS_ON);
     assertTrue(postCountDown.await(1, TimeUnit.SECONDS));
     senderThread.interrupt();
     coordinatorThread.interrupt();