Smoke Test For Live Metrics

smoke-tests/apps/LiveMetrics/build.gradle.kts

@@ -4,4 +4,8 @@ plugins {
 
 dependencies {
   smokeTestImplementation("org.awaitility:awaitility:4.2.0")
+  implementation("log4j:log4j:1.2.17")
+  implementation("io.opentelemetry.instrumentation:opentelemetry-instrumentation-annotations:1.22.1")
+  implementation("com.azure:azure-json:1.0.0")
+  implementation("com.azure:azure-monitor-opentelemetry-autoconfigure:1.0.0-beta.1")
 }

smoke-tests/apps/LiveMetrics/src/main/java/com/microsoft/applicationinsights/smoketestapp/TestServlet.java

@@ -3,19 +3,31 @@
 
 package com.microsoft.applicationinsights.smoketestapp;
 
+import io.opentelemetry.instrumentation.annotations.WithSpan;
 import java.io.IOException;
 import javax.servlet.ServletException;
 import javax.servlet.annotation.WebServlet;
 import javax.servlet.http.HttpServlet;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
 
 @WebServlet("/*")
 public class TestServlet extends HttpServlet {
 
+  private static final Logger logger = LogManager.getLogger("smoketestapp-livemetrics");
+
   @Override
   protected void doGet(HttpServletRequest req, HttpServletResponse resp)
       throws ServletException, IOException {
+    doWork();
     resp.getWriter().println("ok");
   }
+
+  @WithSpan
+  private void doWork() {
+    logger.error("This message should generate an exception!", new Exception("Fake Exception"));
+    logger.info("This message should generate a trace");
+  }
 }

smoke-tests/apps/LiveMetrics/src/smokeTest/java/com/microsoft/applicationinsights/smoketest/LiveMetricsTest.java

@@ -16,8 +16,20 @@
 import static com.microsoft.applicationinsights.smoketest.EnvironmentValue.TOMCAT_8_JAVA_8_OPENJ9;
 import static com.microsoft.applicationinsights.smoketest.EnvironmentValue.WILDFLY_13_JAVA_8;
 import static com.microsoft.applicationinsights.smoketest.EnvironmentValue.WILDFLY_13_JAVA_8_OPENJ9;
-
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.azure.json.JsonProviders;
+import com.azure.json.JsonReader;
+import com.azure.monitor.opentelemetry.autoconfigure.implementation.quickpulse.swagger.models.DocumentIngress;
+import com.azure.monitor.opentelemetry.autoconfigure.implementation.quickpulse.swagger.models.DocumentType;
+import com.azure.monitor.opentelemetry.autoconfigure.implementation.quickpulse.swagger.models.Exception;
+import com.azure.monitor.opentelemetry.autoconfigure.implementation.quickpulse.swagger.models.MetricPoint;
+import com.azure.monitor.opentelemetry.autoconfigure.implementation.quickpulse.swagger.models.MonitoringDataPoint;
+import com.azure.monitor.opentelemetry.autoconfigure.implementation.quickpulse.swagger.models.Trace;
+import java.io.IOException;
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
 import org.awaitility.Awaitility;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
@@ -27,14 +39,117 @@ abstract class LiveMetricsTest {
 
   @RegisterExtension static final SmokeTestExtension testing = SmokeTestExtension.create();
 
+  private boolean foundExceptionDoc = false;
+  private boolean foundTraceDoc = false;
+  private boolean foundDependency = false;
+  private boolean foundRequest = false;
+
   @Test
   @TargetUri("/test")
-  void doMostBasicTest() throws Exception {
-    testing.getTelemetry(0);
-
+  void testTelemetryDataFlow() throws java.lang.Exception {
     Awaitility.await()
-        .atMost(Duration.ofSeconds(10))
-        .until(() -> testing.mockedIngestion.isLiveMetricsPingReceived());
+        .atMost(Duration.ofSeconds(60))
+        .until(() -> testing.mockedIngestion.getCountForType("RequestData") == 1);
+
+    assertThat(testing.mockedIngestion.isPingReceived()).isTrue();
+
+    List<String> postBodies = testing.mockedIngestion.getPostBodies();
+    assertThat(postBodies).hasSizeGreaterThan(0); // should post at least once
+
+    for (String postBody : postBodies) {
+      searchPostBody(postBody);
+    }
+
+    assertThat(foundExceptionDoc).isTrue();
+    assertThat(foundTraceDoc).isTrue();
+    assertThat(foundDependency).isTrue();
+    assertThat(foundRequest).isTrue();
+  }
+
+  private void searchPostBody(String postBody) {
+    // Each post body is a list with a singular MonitoringDataPoint
+    List<MonitoringDataPoint> dataPoints = new ArrayList<>();
+    try {
+      JsonReader reader = JsonProviders.createReader(postBody);
+      dataPoints = reader.readArray(MonitoringDataPoint::fromJson);
+    } catch (IOException e) {
+      throw new IllegalStateException("Failed to parse post request body", e);
+    }
+
+    // Because the mock ping/posts should succeed, we should only have one MonitoringDataPoint per
+    // post
+    assertThat(dataPoints).hasSize(1);
+    MonitoringDataPoint dataPoint = dataPoints.get(0);
+
+    List<DocumentIngress> docs = dataPoint.getDocuments();
+    List<MetricPoint> metrics = dataPoint.getMetrics();
+
+    confirmDocsAreFiltered(docs);
+    confirmPerfCountersNonZero(metrics);
+    foundExceptionDoc = foundExceptionDoc || hasException(docs);
+    foundTraceDoc = foundTraceDoc || hasTrace(docs);
+    foundDependency = foundDependency || hasDependency(metrics);
+    foundRequest = foundRequest || hasRequest(metrics);
+  }
+
+  private void confirmDocsAreFiltered(List<DocumentIngress> docs) {
+    for (DocumentIngress doc : docs) {
+      assertThat(doc.getDocumentType()).isNotEqualTo(DocumentType.REMOTE_DEPENDENCY);
+      assertThat(doc.getDocumentType()).isNotEqualTo(DocumentType.REQUEST);
+    }
+  }
+
+  private boolean hasException(List<DocumentIngress> docs) {
+    for (DocumentIngress doc : docs) {
+      if (doc.getDocumentType().equals(DocumentType.EXCEPTION)
+          && ((Exception) doc).getExceptionMessage().equals("Fake Exception")) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  private boolean hasTrace(List<DocumentIngress> docs) {
+    for (DocumentIngress doc : docs) {
+      if (doc.getDocumentType().equals(DocumentType.TRACE)
+          && ((Trace) doc).getMessage().equals("This message should generate a trace")) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  private boolean hasDependency(List<MetricPoint> metrics) {
+    for (MetricPoint metric : metrics) {
+      String name = metric.getName();
+      double value = metric.getValue();
+      if (name.equals("\\ApplicationInsights\\Dependency Calls/Sec")) {
+        return value == 1;
+      }
+    }
+    return false;
+  }
+
+  private boolean hasRequest(List<MetricPoint> metrics) {
+    for (MetricPoint metric : metrics) {
+      String name = metric.getName();
+      double value = metric.getValue();
+      if (name.equals("\\ApplicationInsights\\Requests/Sec")) {
+        return value == 1;
+      }
+    }
+    return false;
+  }
+
+  private void confirmPerfCountersNonZero(List<MetricPoint> metrics) {
+    for (MetricPoint metric : metrics) {
+      String name = metric.getName();
+      double value = metric.getValue();
+      if (name.equals("\\Process\\Physical Bytes")
+          || name.equals("\\% Process\\Processor Time Normalized")) {
+        assertThat(value).isNotEqualTo(0);
+      }
+    }
   }
 
   @Environment(TOMCAT_8_JAVA_8)

smoke-tests/framework/src/main/java/com/microsoft/applicationinsights/smoketest/SmokeTestExtension.java

@@ -205,6 +205,7 @@ private void prepareEnvironment(Environment environment) throws Exception {
     }
     mockedIngestion.startServer();
     mockedIngestion.setRequestLoggingEnabled(true);
+    mockedIngestion.setQuickPulseRequestLoggingEnabled(true);
     if (useOtlpEndpoint) {
       mockedOtlpIngestion.startServer();
     }
@@ -536,6 +537,7 @@ public void afterAll(ExtensionContext context) throws Exception {
 
     mockedIngestion.stopServer();
     mockedIngestion.setRequestLoggingEnabled(false);
+    mockedIngestion.setQuickPulseRequestLoggingEnabled(false);
     if (useOtlpEndpoint) {
       mockedOtlpIngestion.stopServer();
     }

smoke-tests/framework/src/main/java/com/microsoft/applicationinsights/smoketest/fakeingestion/MockedAppInsightsIngestionServer.java

@@ -24,6 +24,7 @@ public class MockedAppInsightsIngestionServer {
 
   private final MockedAppInsightsIngestionServlet servlet;
   private final MockedProfilerSettingsServlet profilerSettingsServlet;
+  private final MockedQuickPulseServlet quickPulseServlet;
   private final Server server;
 
   public MockedAppInsightsIngestionServer() {
@@ -33,8 +34,10 @@ public MockedAppInsightsIngestionServer() {
 
     servlet = new MockedAppInsightsIngestionServlet();
     profilerSettingsServlet = new MockedProfilerSettingsServlet();
+    quickPulseServlet = new MockedQuickPulseServlet();
 
     handler.addServletWithMapping(new ServletHolder(profilerSettingsServlet), "/profiler/*");
+    handler.addServletWithMapping(new ServletHolder(quickPulseServlet), "/QuickPulseService.svc/*");
     handler.addServletWithMapping(new ServletHolder(servlet), "/*");
   }
 
@@ -276,8 +279,12 @@ public boolean test(Envelope input) {
     return items;
   }
 
-  public boolean isLiveMetricsPingReceived() {
-    return servlet.isLiveMetricsPingReceived();
+  public boolean isPingReceived() {
+    return quickPulseServlet.isPingReceived();
+  }
+
+  public List<String> getPostBodies() {
+    return quickPulseServlet.getPostBodies();
   }
 
   @SuppressWarnings("SystemOut")
@@ -304,4 +311,8 @@ public void run() {
   public void setRequestLoggingEnabled(boolean enabled) {
     servlet.setRequestLoggingEnabled(enabled);
   }
+
+  public void setQuickPulseRequestLoggingEnabled(boolean enabled) {
+    quickPulseServlet.setRequestLoggingEnabled(enabled);
+  }
 }

smoke-tests/framework/src/main/java/com/microsoft/applicationinsights/smoketest/fakeingestion/MockedAppInsightsIngestionServlet.java

@@ -32,8 +32,6 @@ class MockedAppInsightsIngestionServlet extends HttpServlet {
 
   private final Object multimapLock = new Object();
 
-  private volatile boolean liveMetricsPingReceived;
-
   private volatile boolean loggingEnabled;
 
   MockedAppInsightsIngestionServlet() {
@@ -51,7 +49,6 @@ void resetData() {
     synchronized (multimapLock) {
       type2envelope.clear();
     }
-    liveMetricsPingReceived = false;
   }
 
   boolean hasData() {
@@ -100,18 +97,8 @@ List<Envelope> waitForItems(
     throw new TimeoutException("timed out waiting for items");
   }
 
-  boolean isLiveMetricsPingReceived() {
-    return liveMetricsPingReceived;
-  }
-
   @Override
   protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws IOException {
-    if (!"/v2.1/track".equals(req.getPathInfo())
-        && !"/QuickPulseService.svc/ping".equals(req.getPathInfo())) {
-      resp.sendError(404, "Unknown URI");
-      return;
-    }
-
     String contentEncoding = req.getHeader("content-encoding");
     Readable reader;
     if ("gzip".equals(contentEncoding)) {
@@ -126,12 +113,6 @@ protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws I
     resp.setContentType("application/json");
     logit("raw payload:\n\n" + body + "\n");
 
-    if ("/QuickPulseService.svc/ping".equals(req.getPathInfo())) {
-      liveMetricsPingReceived = true;
-      resp.setHeader("x-ms-qps-subscribed", "false");
-      return;
-    }
-
     String[] lines = body.split("\n");
     for (String line : lines) {
       Envelope envelope = JsonHelper.GSON.fromJson(line.trim(), Envelope.class);

smoke-tests/framework/src/main/java/com/microsoft/applicationinsights/smoketest/fakeingestion/MockedQuickPulseServlet.java

@@ -0,0 +1,82 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.applicationinsights.smoketest.fakeingestion;
+
+import com.google.common.io.CharStreams;
+import java.io.IOException;
+import java.io.StringWriter;
+import java.rmi.ServerError;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+public class MockedQuickPulseServlet extends HttpServlet {
+
+  private final AtomicBoolean pingReceived = new AtomicBoolean(false);
+  private final List<String> postBodies = new ArrayList<>();
+  private final Object lock = new Object();
+
+  private static final String BODY =
+      "{\"ETag\":\"fake::etag\",\"Metrics\":[],\"QuotaInfo\":null,\"DocumentStreams\":[{\"Id\":\"all-types-default\",\"DocumentFilterGroups\":[{\"TelemetryType\":\"Request\",\"Filters\":{\"Filters\":[{\"FieldName\":\"Success\",\"Predicate\":\"Equal\",\"Comparand\":\"false\"}]}},{\"TelemetryType\":\"Dependency\",\"Filters\":{\"Filters\":[{\"FieldName\":\"Success\",\"Predicate\":\"Equal\",\"Comparand\":\"false\"}]}},{\"TelemetryType\":\"Exception\",\"Filters\":{\"Filters\":[]}},{\"TelemetryType\":\"Event\",\"Filters\":{\"Filters\":[]}},{\"TelemetryType\":\"Trace\",\"Filters\":{\"Filters\":[]}}]}]}";
+
+  private volatile boolean loggingEnabled;
+
+  public MockedQuickPulseServlet() {}
+
+  @SuppressWarnings("SystemOut")
+  private void logit(String message) {
+    if (loggingEnabled) {
+      System.out.println("FAKE INGESTION: INFO - " + message);
+    }
+  }
+
+  @Override
+  protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws IOException {
+    Readable reader = req.getReader();
+    StringWriter sw = new StringWriter();
+    CharStreams.copy(reader, sw);
+    String body = sw.toString();
+
+    String path = req.getPathInfo();
+    logit("QuickPulse received: " + path);
+    if (path.equals("/ping")) {
+      pingReceived.set(true);
+      logit("ping body: " + body);
+      // want the following request to be a post
+      resp.setHeader("x-ms-qps-configuration-etag", "fake::etag");
+      resp.setHeader("x-ms-qps-subscribed", "true");
+      resp.setContentType("application/json");
+      resp.getWriter().write(BODY);
+
+    } else if (path.equals("/post")) {
+      synchronized (lock) {
+        postBodies.add(body);
+      }
+      logit("post body: " + body);
+      // continue to post
+      resp.setHeader("x-ms-qps-subscribed", "true");
+      resp.setHeader("x-ms-qps-configuration-etag", "fake::etag");
+    } else {
+      throw new ServerError(
+          "Unexpected path: " + path + " please fix the test/mock server setup", new Error());
+    }
+  }
+
+  public boolean isPingReceived() {
+    return pingReceived.get();
+  }
+
+  public List<String> getPostBodies() {
+    synchronized (lock) {
+      return new ArrayList<>(postBodies);
+    }
+  }
+
+  public void setRequestLoggingEnabled(boolean enabled) {
+    loggingEnabled = enabled;
+  }
+}