Quick Pulse cleanup and add sample telemetry feature (#1852)

* initial commit - testing sample telemetry

* working code and some functional tests

* rename playback session records

* debug session-records in build

* working code with integration tests

* add comments gradle file

* fix integration postRequest test

* debug integrationtests

* debug with logger

* fix postRequest regex

* undo debug statements

* add sample telemetry to quickpulse working code

* address comments

* remove debug statements

* more cleanup

* Small refactorings

* address comments and more cleanup

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

Author: kryalama

agent/agent-tooling/build.gradle.kts

@@ -4,6 +4,13 @@ plugins {
   id("com.github.johnrengelman.shadow")
 }
 
+// Adding this step to copy playback test results from session-records to build/classes/java/test. Azure core testing framework follows this directory structure.
+sourceSets {
+  test {
+    output.setResourcesDir("build/classes/java/test")
+  }
+}
+
 dependencies {
   implementation(project(":agent:agent-profiler:agent-service-profiler"))
   implementation(project(":agent:agent-profiler:agent-alerting-api"))
@@ -61,6 +68,7 @@ dependencies {
   testImplementation("io.opentelemetry.instrumentation:opentelemetry-instrumentation-api-annotation-support")
 
   testImplementation("org.junit.jupiter:junit-jupiter")
+  testImplementation("com.azure:azure-core-test:1.7.0")
   testImplementation("org.assertj:assertj-core")
   testImplementation("org.awaitility:awaitility")
   testImplementation("org.mockito:mockito-core")
@@ -72,3 +80,5 @@ dependencies {
   testImplementation("com.microsoft.jfr:jfr-streaming")
   testImplementation("com.azure:azure-storage-blob")
 }
+
+

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

@@ -52,6 +52,7 @@ public class MainEntryPoint {
 
   private static RpConfiguration rpConfiguration;
   private static Configuration configuration;
+  private static String agentVersion = "(unknown)";
 
   private MainEntryPoint() {}
 
@@ -63,12 +64,15 @@ public static Configuration getConfiguration() {
     return configuration;
   }
 
+  public static String getAgentVersion() {
+    return agentVersion;
+  }
+
   // TODO turn this into an interceptor
   @SuppressWarnings("SystemOut")
   public static void start(Instrumentation instrumentation, File javaagentFile) {
     boolean success = false;
     Logger startupLogger = null;
-    String version = "(unknown)";
     try {
       if (DEBUG_SIGNED_JAR_ACCESS) {
         JarVerifierClassFileTransformer transformer = new JarVerifierClassFileTransformer();
@@ -78,7 +82,7 @@ public static void start(Instrumentation instrumentation, File javaagentFile) {
       }
       Path agentPath = javaagentFile.toPath();
       // need to initialize version before initializing DiagnosticsHelper
-      version = SdkVersionFinder.initVersion(agentPath);
+      agentVersion = SdkVersionFinder.initVersion(agentPath);
       DiagnosticsHelper.setAgentJarFile(agentPath);
       // configuration is only read this early in order to extract logging configuration
       rpConfiguration = RpConfigurationBuilder.create(agentPath);
@@ -92,17 +96,17 @@ public static void start(Instrumentation instrumentation, File javaagentFile) {
           instrumentation, ConfigOverride.getConfig(configuration), false);
       startupLogger.info(
           "ApplicationInsights Java Agent {} started successfully (PID {})",
-          version,
+          agentVersion,
           new PidFinder().getValue());
       success = true;
       LoggerFactory.getLogger(DiagnosticsHelper.DIAGNOSTICS_LOGGER_NAME)
-          .info("Application Insights Codeless Agent {} Attach Successful", version);
+          .info("Application Insights Codeless Agent {} Attach Successful", agentVersion);
     } catch (ThreadDeath td) {
       throw td;
     } catch (Throwable t) {
 
       FriendlyException friendlyException = getFriendlyException(t);
-      String banner = "ApplicationInsights Java Agent " + version + " failed to start";
+      String banner = "ApplicationInsights Java Agent " + agentVersion + " failed to start";
       if (friendlyException != null) {
         logErrorMessage(
             startupLogger, friendlyException.getMessageWithBanner(banner), true, t, javaagentFile);

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

@@ -24,12 +24,22 @@
 import com.microsoft.applicationinsights.agent.internal.exporter.models.MonitorDomain;
 import com.microsoft.applicationinsights.agent.internal.exporter.models.RemoteDependencyData;
 import com.microsoft.applicationinsights.agent.internal.exporter.models.RequestData;
+import com.microsoft.applicationinsights.agent.internal.exporter.models.StackFrame;
 import com.microsoft.applicationinsights.agent.internal.exporter.models.TelemetryExceptionData;
+import com.microsoft.applicationinsights.agent.internal.exporter.models.TelemetryExceptionDetails;
 import com.microsoft.applicationinsights.agent.internal.exporter.models.TelemetryItem;
 import com.microsoft.applicationinsights.agent.internal.perfcounter.CpuPerformanceCounterCalculator;
+import com.microsoft.applicationinsights.agent.internal.quickpulse.model.QuickPulseDependencyDocument;
+import com.microsoft.applicationinsights.agent.internal.quickpulse.model.QuickPulseDocument;
+import com.microsoft.applicationinsights.agent.internal.quickpulse.model.QuickPulseExceptionDocument;
+import com.microsoft.applicationinsights.agent.internal.quickpulse.model.QuickPulseRequestDocument;
 import com.microsoft.applicationinsights.agent.internal.telemetry.TelemetryClient;
 import java.lang.management.ManagementFactory;
 import java.lang.management.MemoryMXBean;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
@@ -50,6 +60,7 @@ static class FinalCounters {
     public final int unsuccessfulRdds;
     public final long memoryCommitted;
     public final double cpuUsage;
+    public final List<QuickPulseDocument> documentList = new ArrayList<>();
 
     public FinalCounters(
         Counters currentCounters,
@@ -82,6 +93,9 @@ public FinalCounters(
       this.rdds = countAndDuration.count;
       this.rddsDuration = countAndDuration.duration;
       this.unsuccessfulRdds = currentCounters.unsuccessfulRdds.get();
+      synchronized (currentCounters.documentList) {
+        this.documentList.addAll(currentCounters.documentList);
+      }
     }
   }
 
@@ -106,6 +120,7 @@ static class Counters {
 
     final AtomicLong rddsAndDuations = new AtomicLong(0);
     final AtomicInteger unsuccessfulRdds = new AtomicInteger(0);
+    final List<QuickPulseDocument> documentList = new ArrayList<>();
 
     static long encodeCountAndDuration(long count, long duration) {
       if (count > MAX_COUNT || duration > MAX_DURATION) {
@@ -166,7 +181,7 @@ public synchronized FinalCounters getAndRestart() {
     return null;
   }
 
-  /*@VisibleForTesting*/
+  // only used by tests
   synchronized FinalCounters peek() {
     Counters currentCounters = this.counters.get(); // this should be the only differece
     if (currentCounters != null) {
@@ -199,7 +214,7 @@ public void add(TelemetryItem telemetryItem) {
     } else if (data instanceof RemoteDependencyData) {
       addDependency((RemoteDependencyData) data, itemCount);
     } else if (data instanceof TelemetryExceptionData) {
-      addException(itemCount);
+      addException((TelemetryExceptionData) data, itemCount);
     }
   }
 
@@ -218,15 +233,59 @@ private void addDependency(RemoteDependencyData telemetry, int itemCount) {
     if (success != null && !success) { // success should not be null
       counters.unsuccessfulRdds.incrementAndGet();
     }
+    QuickPulseDependencyDocument quickPulseDependencyDocument = new QuickPulseDependencyDocument();
+    quickPulseDependencyDocument.setDocumentType("RemoteDependency");
+    quickPulseDependencyDocument.setType("DependencyTelemetryDocument");
+    quickPulseDependencyDocument.setOperationId(telemetry.getId());
+    quickPulseDependencyDocument.setVersion("1.0");
+    quickPulseDependencyDocument.setName(telemetry.getName());
+    quickPulseDependencyDocument.setCommandName(telemetry.getData());
+    quickPulseDependencyDocument.setTarget(telemetry.getTarget());
+    quickPulseDependencyDocument.setSuccess(telemetry.isSuccess());
+    quickPulseDependencyDocument.setDuration(telemetry.getDuration());
+    quickPulseDependencyDocument.setResultCode(telemetry.getResultCode());
+    quickPulseDependencyDocument.setOperationName(telemetry.getId());
+    quickPulseDependencyDocument.setDependencyTypeName(telemetry.getType());
+    quickPulseDependencyDocument.setProperties(
+        aggregateProperties(telemetry.getProperties(), telemetry.getMeasurements()));
+    synchronized (counters.documentList) {
+      counters.documentList.add(quickPulseDependencyDocument);
+    }
   }
 
-  private void addException(int itemCount) {
+  private void addException(TelemetryExceptionData exceptionData, int itemCount) {
     Counters counters = this.counters.get();
     if (counters == null) {
       return;
     }
 
     counters.exceptions.addAndGet(itemCount);
+    QuickPulseExceptionDocument quickPulseExceptionDocument = new QuickPulseExceptionDocument();
+    quickPulseExceptionDocument.setDocumentType("Exception");
+    quickPulseExceptionDocument.setType("ExceptionTelemetryDocument");
+    quickPulseExceptionDocument.setOperationId(exceptionData.getProblemId());
+    quickPulseExceptionDocument.setVersion("1.0");
+    List<TelemetryExceptionDetails> exceptionList = exceptionData.getExceptions();
+    StringBuilder exceptions = new StringBuilder();
+    if (exceptionList != null && exceptionList.size() > 0) {
+      List<StackFrame> parsedStack = exceptionList.get(0).getParsedStack();
+      String stack = exceptionList.get(0).getStack();
+      if (parsedStack != null && parsedStack.size() > 0) {
+        for (StackFrame stackFrame : parsedStack) {
+          if (stackFrame != null && stackFrame.getAssembly() != null) {
+            exceptions.append(stackFrame.getAssembly()).append("\n");
+          }
+        }
+      } else if (stack != null && stack.length() > 0) {
+        exceptions.append(stack);
+      }
+      quickPulseExceptionDocument.setException(exceptions.toString());
+      quickPulseExceptionDocument.setExceptionMessage(exceptionList.get(0).getMessage());
+      quickPulseExceptionDocument.setExceptionType(exceptionList.get(0).getTypeName());
+    }
+    synchronized (counters.documentList) {
+      counters.documentList.add(quickPulseExceptionDocument);
+    }
   }
 
   private void addRequest(RequestData requestTelemetry, int itemCount) {
@@ -241,6 +300,32 @@ private void addRequest(RequestData requestTelemetry, int itemCount) {
     if (!requestTelemetry.isSuccess()) {
       counters.unsuccessfulRequests.incrementAndGet();
     }
+    QuickPulseRequestDocument quickPulseRequestDocument = new QuickPulseRequestDocument();
+    quickPulseRequestDocument.setDocumentType("Request");
+    quickPulseRequestDocument.setType("RequestTelemetryDocument");
+    quickPulseRequestDocument.setOperationId(requestTelemetry.getId());
+    quickPulseRequestDocument.setVersion("1.0");
+    quickPulseRequestDocument.setSuccess(requestTelemetry.isSuccess());
+    quickPulseRequestDocument.setDuration(requestTelemetry.getDuration());
+    quickPulseRequestDocument.setResponseCode(requestTelemetry.getResponseCode());
+    quickPulseRequestDocument.setOperationName(requestTelemetry.getName());
+    quickPulseRequestDocument.setProperties(
+        aggregateProperties(requestTelemetry.getProperties(), requestTelemetry.getMeasurements()));
+    synchronized (counters.documentList) {
+      counters.documentList.add(quickPulseRequestDocument);
+    }
+  }
+
+  private static Map<String, String> aggregateProperties(
+      Map<String, String> properties, Map<String, Double> measurements) {
+    Map<String, String> aggregatedProperties = new HashMap<>();
+    if (measurements != null) {
+      measurements.forEach((k, v) -> aggregatedProperties.put(k, String.valueOf(v)));
+    }
+    if (properties != null) {
+      aggregatedProperties.putAll(properties);
+    }
+    return aggregatedProperties;
   }
 
   // TODO (trask) optimization: move live metrics request capture to OpenTelemetry layer so don't