Track disk IO via Non-essential statsbeat (#1925)

* Just track read and write failures

* Increment read/write failure counters

* Simplify a bit

* Fix tests

* Fix spotlessApply

* Remove a comment

* Update smoke test

* Fix smoke test

* Add unit tests

* Fix smoke test

* Add comments

* Don't increment read failure count where there is no file to be loaded

* Fix smoke test

* Increment read failure count at the exception level, not at the higher level

* Increment write failure count at the exception level for easier understanding

* Fix unit tests

* Be consistent and getter used by tests only

* Add a todo

* Add comment

Co-authored-by: Trask Stalnaker <[email protected]>

* Delete

Co-authored-by: Trask Stalnaker <[email protected]>

* Add a comment

Co-authored-by: Trask Stalnaker <[email protected]>

* Remove this

* Fix spotlessApply

Co-authored-by: Trask Stalnaker <[email protected]>

Author: heyams

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/localstorage/LocalFileLoader.java

@@ -22,6 +22,7 @@
 package com.microsoft.applicationinsights.agent.internal.localstorage;
 
 import com.microsoft.applicationinsights.agent.internal.common.OperationLogger;
+import com.microsoft.applicationinsights.agent.internal.statsbeat.NonessentialStatsbeat;
 import java.io.File;
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -37,13 +38,19 @@ public class LocalFileLoader {
 
   private final LocalFileCache localFileCache;
   private final File telemetryFolder;
+  // this is null for Statsbeat telemetry
+  @Nullable private final NonessentialStatsbeat nonessentialStatsbeat;
 
   private static final OperationLogger operationLogger =
       new OperationLogger(LocalFileLoader.class, "Loading telemetry from disk");
 
-  public LocalFileLoader(LocalFileCache localFileCache, File telemetryFolder) {
+  public LocalFileLoader(
+      LocalFileCache localFileCache,
+      File telemetryFolder,
+      @Nullable NonessentialStatsbeat nonessentialStatsbeat) {
     this.localFileCache = localFileCache;
     this.telemetryFolder = telemetryFolder;
+    this.nonessentialStatsbeat = nonessentialStatsbeat;
   }
 
   // Load ByteBuffer from persisted files on disk in FIFO order.
@@ -81,7 +88,7 @@ PersistedFile loadTelemetriesFromDisk() {
               + TEMPORARY_FILE_EXTENSION
               + " extension: ",
           e);
-      // TODO (heya) track number of failures to create a temp file via Statsbeat
+      incrementReadFailureCount();
       return null;
     }
 
@@ -90,8 +97,8 @@ PersistedFile loadTelemetriesFromDisk() {
       // TODO (trask) optimization: read this directly into ByteBuffer(s)
       result = Files.readAllBytes(tempFile.toPath());
     } catch (IOException ex) {
-      // TODO (heya) track deserialization failure via Statsbeat
       operationLogger.recordFailure("Fail to read telemetry from " + tempFile.getName(), ex);
+      incrementReadFailureCount();
       return null;
     }
 
@@ -139,6 +146,12 @@ private static void deleteFile(File file) {
     }
   }
 
+  private void incrementReadFailureCount() {
+    if (nonessentialStatsbeat != null) {
+      nonessentialStatsbeat.incrementReadFailureCount();
+    }
+  }
+
   static class PersistedFile {
     final File file;
     final ByteBuffer rawBytes;

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/localstorage/LocalFileSender.java

@@ -67,7 +67,6 @@ public void run() {
     } catch (RuntimeException ex) {
       logger.error(
           "Unexpected error occurred while sending telemetries from the local storage.", ex);
-      // TODO (heya) track sending persisted telemetries failure via Statsbeat.
     }
   }
 }

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/localstorage/LocalFileWriter.java

@@ -22,13 +22,15 @@
 package com.microsoft.applicationinsights.agent.internal.localstorage;
 
 import com.microsoft.applicationinsights.agent.internal.common.OperationLogger;
+import com.microsoft.applicationinsights.agent.internal.statsbeat.NonessentialStatsbeat;
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.util.Collection;
 import java.util.List;
+import javax.annotation.Nullable;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.FilenameUtils;
 
@@ -41,41 +43,48 @@ public final class LocalFileWriter {
 
   private final LocalFileCache localFileCache;
   private final File telemetryFolder;
+  // this is null for Statsbeat telemetry
+  @Nullable private final NonessentialStatsbeat nonessentialStatsbeat;
 
   private static final OperationLogger operationLogger =
       new OperationLogger(
           LocalFileWriter.class, "Writing telemetry to disk (telemetry is discarded on failure)");
 
-  public LocalFileWriter(LocalFileCache localFileCache, File telemetryFolder) {
+  public LocalFileWriter(
+      LocalFileCache localFileCache,
+      File telemetryFolder,
+      @Nullable NonessentialStatsbeat nonessentialStatsbeat) {
     this.telemetryFolder = telemetryFolder;
     this.localFileCache = localFileCache;
+    this.nonessentialStatsbeat = nonessentialStatsbeat;
   }
 
-  public boolean writeToDisk(List<ByteBuffer> buffers) {
+  public void writeToDisk(List<ByteBuffer> buffers) {
     long size = getTotalSizeOfPersistedFiles(telemetryFolder);
     if (size >= MAX_FILE_SIZE_IN_BYTES) {
       operationLogger.recordFailure(
           "Local persistent storage capacity has been reached. It's currently at ("
               + (size / 1024)
               + "KB). Telemetry will be lost");
-      return false;
+      incrementWriteFailureCount();
+      return;
     }
 
     File tempFile;
     try {
       tempFile = createTempFile(telemetryFolder);
     } catch (IOException e) {
       operationLogger.recordFailure("unable to create temporary file: " + e, e);
-      // TODO (heya) track number of failures to create a temp file via Statsbeat
-      return false;
+      incrementWriteFailureCount();
+      return;
     }
 
     try {
       write(tempFile, buffers);
     } catch (IOException e) {
       operationLogger.recordFailure(String.format("unable to write to file: %s", e), e);
-      // TODO (heya) track IO write failure via Statsbeat
-      return false;
+      incrementWriteFailureCount();
+      return;
     }
 
     File permanentFile;
@@ -93,15 +102,19 @@ public boolean writeToDisk(List<ByteBuffer> buffers) {
               + PERMANENT_FILE_EXTENSION
               + " extension: ",
           e);
-      // TODO (heya) track number of failures to rename a file via Statsbeat
-      return false;
+      incrementWriteFailureCount();
+      return;
     }
 
     localFileCache.addPersistedFilenameToMap(permanentFile.getName());
 
-    // TODO (heya) track data persistence success via Statsbeat
     operationLogger.recordSuccess();
-    return true;
+  }
+
+  private void incrementWriteFailureCount() {
+    if (nonessentialStatsbeat != null) {
+      nonessentialStatsbeat.incrementWriteFailureCount();
+    }
   }
 
   private static void write(File file, List<ByteBuffer> buffers) throws IOException {

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/statsbeat/NonessentialStatsbeat.java

@@ -0,0 +1,83 @@
+/*
+ * ApplicationInsights-Java
+ * Copyright (c) Microsoft Corporation
+ * All rights reserved.
+ *
+ * MIT License
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this
+ * software and associated documentation files (the ""Software""), to deal in the Software
+ * without restriction, including without limitation the rights to use, copy, modify, merge,
+ * publish, distribute, sublicense, and/or sell copies of the Software, and to permit
+ * persons to whom the Software is furnished to do so, subject to the following conditions:
+ * The above copyright notice and this permission notice shall be included in all copies or
+ * substantial portions of the Software.
+ * THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
+ * INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
+ * PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE
+ * FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+ * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+package com.microsoft.applicationinsights.agent.internal.statsbeat;
+
+import com.microsoft.applicationinsights.agent.internal.exporter.models.TelemetryItem;
+import com.microsoft.applicationinsights.agent.internal.telemetry.TelemetryClient;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class NonessentialStatsbeat extends BaseStatsbeat {
+
+  // TODO when there are more non-essential statsbeat(s) in the future, try to separate different
+  // kinds of non-essential statsbeat into different classes.
+
+  // Track local storage IO success and failure
+  private static final String READ_FAILURE_COUNT = "Read Failure Count";
+  private static final String WRITE_FAILURE_COUNT = "Write Failure Count";
+  private final AtomicLong readFailureCount = new AtomicLong();
+  private final AtomicLong writeFailureCount = new AtomicLong();
+
+  // only used by tests
+  public NonessentialStatsbeat() {
+    super(new CustomDimensions());
+  }
+
+  protected NonessentialStatsbeat(CustomDimensions customDimensions) {
+    super(customDimensions);
+  }
+
+  @Override
+  protected void send(TelemetryClient telemetryClient) {
+    if (readFailureCount.get() != 0) {
+      TelemetryItem telemetryItem =
+          createStatsbeatTelemetry(telemetryClient, READ_FAILURE_COUNT, readFailureCount.get());
+      telemetryClient.trackStatsbeatAsync(telemetryItem);
+    }
+
+    if (writeFailureCount.get() != 0) {
+      TelemetryItem telemetryItem =
+          createStatsbeatTelemetry(telemetryClient, WRITE_FAILURE_COUNT, writeFailureCount.get());
+      telemetryClient.trackStatsbeatAsync(telemetryItem);
+    }
+
+    readFailureCount.set(0L);
+    writeFailureCount.set(0L);
+  }
+
+  public void incrementReadFailureCount() {
+    readFailureCount.incrementAndGet();
+  }
+
+  // used by tests only
+  long getReadFailureCount() {
+    return readFailureCount.get();
+  }
+
+  public void incrementWriteFailureCount() {
+    writeFailureCount.incrementAndGet();
+  }
+
+  // used by tests only
+  public long getWriteFailureCount() {
+    return writeFailureCount.get();
+  }
+}

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/statsbeat/StatsbeatModule.java

@@ -46,6 +46,7 @@ public class StatsbeatModule {
   private final AttachStatsbeat attachStatsbeat;
   private final FeatureStatsbeat featureStatsbeat;
   private final FeatureStatsbeat instrumentationStatsbeat;
+  private final NonessentialStatsbeat nonessentialStatsbeat;
 
   private final AtomicBoolean started = new AtomicBoolean();
 
@@ -55,6 +56,7 @@ public StatsbeatModule(Cache<String, String> ikeyEndpointMap) {
     attachStatsbeat = new AttachStatsbeat(customDimensions);
     featureStatsbeat = new FeatureStatsbeat(customDimensions, FeatureType.FEATURE);
     instrumentationStatsbeat = new FeatureStatsbeat(customDimensions, FeatureType.INSTRUMENTATION);
+    nonessentialStatsbeat = new NonessentialStatsbeat(customDimensions);
   }
 
   public void start(TelemetryClient telemetryClient, Configuration config) {
@@ -104,11 +106,14 @@ public void start(TelemetryClient telemetryClient, Configuration config) {
 
     featureStatsbeat.trackConfigurationOptions(config);
 
-    if (config.preview.statsbeat.disabled) {
-      // disabled will disable non-essentials Statsbeat, such as tracking failure or success of disk
-      // persistence operations, optional network statsbeat, live metric,
-      // azure metadata service failure, profile endpoint, etc.
-      // TODO stop sending non-essential Statsbeat when applicable
+    if (!config.preview.statsbeat.disabled) {
+      scheduledExecutor.scheduleWithFixedDelay(
+          new StatsbeatSender(nonessentialStatsbeat, telemetryClient),
+          longIntervalSeconds,
+          longIntervalSeconds,
+          TimeUnit.SECONDS);
+    } else {
+      logger.debug("Non-essential Statsbeat is disabled.");
     }
   }
 
@@ -120,6 +125,10 @@ public FeatureStatsbeat getInstrumentationStatsbeat() {
     return instrumentationStatsbeat;
   }
 
+  public NonessentialStatsbeat getNonessentialStatsbeat() {
+    return nonessentialStatsbeat;
+  }
+
   /** Runnable which is responsible for calling the send method to transmit Statsbeat telemetry. */
   private static class StatsbeatSender implements Runnable {
 

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

@@ -246,10 +246,7 @@ private CompletableResultCode internalSend(
   }
 
   private void writeToDiskOnFailure(List<ByteBuffer> byteBuffers) {
-    if (!localFileWriter.writeToDisk(byteBuffers)) {
-      // TODO (heya) track # of write failure via Statsbeat
-    }
-
+    localFileWriter.writeToDisk(byteBuffers);
     byteBufferPool.offer(byteBuffers);
   }
 

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

@@ -220,8 +220,12 @@ public BatchSpanProcessor getChannelBatcher() {
         if (channelBatcher == null) {
           LocalFileCache localFileCache = new LocalFileCache();
           File telemetryFolder = LocalStorageUtils.getOfflineTelemetryFolder();
-          LocalFileLoader localFileLoader = new LocalFileLoader(localFileCache, telemetryFolder);
-          LocalFileWriter localFileWriter = new LocalFileWriter(localFileCache, telemetryFolder);
+          LocalFileLoader localFileLoader =
+              new LocalFileLoader(
+                  localFileCache, telemetryFolder, statsbeatModule.getNonessentialStatsbeat());
+          LocalFileWriter localFileWriter =
+              new LocalFileWriter(
+                  localFileCache, telemetryFolder, statsbeatModule.getNonessentialStatsbeat());
           TelemetryChannel channel =
               TelemetryChannel.create(
                   endpointProvider.getIngestionEndpointUrl(),
@@ -243,8 +247,10 @@ public BatchSpanProcessor getStatsbeatChannelBatcher() {
         if (statsbeatChannelBatcher == null) {
           LocalFileCache localFileCache = new LocalFileCache();
           File statsbeatFolder = LocalStorageUtils.getOfflineStatsbeatFolder();
-          LocalFileLoader localFileLoader = new LocalFileLoader(localFileCache, statsbeatFolder);
-          LocalFileWriter localFileWriter = new LocalFileWriter(localFileCache, statsbeatFolder);
+          LocalFileLoader localFileLoader =
+              new LocalFileLoader(localFileCache, statsbeatFolder, null);
+          LocalFileWriter localFileWriter =
+              new LocalFileWriter(localFileCache, statsbeatFolder, null);
           TelemetryChannel channel =
               TelemetryChannel.create(
                   endpointProvider.getStatsbeatEndpointUrl(),

agent/agent-tooling/src/test/java/com/microsoft/applicationinsights/agent/internal/localstorage/IntegrationTests.java

@@ -64,13 +64,13 @@ public void setup() throws MalformedURLException {
     when(mockedClient.send(mockedRequest)).thenReturn(Mono.just(mockedResponse));
     HttpPipelineBuilder pipelineBuilder = new HttpPipelineBuilder().httpClient(mockedClient);
     localFileCache = new LocalFileCache();
-    localFileLoader = new LocalFileLoader(localFileCache, tempFolder);
+    localFileLoader = new LocalFileLoader(localFileCache, tempFolder, null);
 
     telemetryChannel =
         new TelemetryChannel(
             pipelineBuilder.build(),
             new URL("http://foo.bar"),
-            new LocalFileWriter(localFileCache, tempFolder),
+            new LocalFileWriter(localFileCache, tempFolder, null),
             null);
   }