Add smoke tests for telemetry

Author: smola

dd-smoke-tests/spring-boot-2.3-webmvc-jetty/src/test/groovy/SpringBootWebmvcIntegrationTest.groovy

@@ -86,4 +86,9 @@ class SpringBootWebmvcIntegrationTest extends AbstractServerSmokeTest {
     response.code() == 404
     waitForTraceCount(1)
   }
+
+  @Override
+  List<String> expectedTelemetryDependencies() {
+    ['org.eclipse.jetty:jetty-client']
+  }
 }

dd-smoke-tests/src/main/groovy/datadog/smoketest/AbstractServerSmokeTest.groovy

@@ -116,4 +116,49 @@ abstract class AbstractServerSmokeTest extends AbstractSmokeTest {
     }
     return remaining
   }
+
+  void 'receive telemetry app-started'() {
+    when:
+    waitForTelemetryCount(1)
+
+    then:
+    telemetryMessages.size() >= 1
+    Object msg = telemetryMessages.get(0)
+    msg['request_type'] == 'app-started'
+  }
+
+  List<String> expectedTelemetryDependencies() {
+    []
+  }
+
+  void 'receive telemetry app-dependencies-loaded'() {
+    when:
+    // app-started + 3 message-batch
+    waitForTelemetryCount(4)
+    waitForTelemetryFlat { it.get('request_type') == 'app-dependencies-loaded' }
+
+    then: 'received some dependencies'
+    def dependenciesLoaded = telemetryFlatMessages.findAll { it.get('request_type') == 'app-dependencies-loaded' }
+    def dependencies = []
+    dependenciesLoaded.each {
+      def payload = it.get('payload') as Map<String, Object>
+      dependencies.addAll(payload.get('dependencies')) }
+    dependencies.size() > 0
+
+    Set<String> dependencyNames = dependencies.collect {
+      def dependency = it as Map<String, Object>
+      dependency.get('name') as String
+    }.toSet()
+
+    and: 'received tracer dependencies'
+    // Not exhaustive list of tracer dependencies.
+    Set<String> missingDependencyNames = ['com.github.jnr:jnr-ffi', 'net.bytebuddy:byte-buddy-agent',].toSet()
+    missingDependencyNames.removeAll(dependencyNames) || true
+    missingDependencyNames.isEmpty()
+
+    and: 'received application dependencies'
+    Set<String> missingExtraDependencyNames = expectedTelemetryDependencies().toSet()
+    missingExtraDependencyNames.removeAll(dependencyNames) || true
+    missingExtraDependencyNames.isEmpty()
+  }
 }

dd-smoke-tests/src/main/groovy/datadog/smoketest/AbstractSmokeTest.groovy

@@ -7,6 +7,7 @@ import datadog.trace.test.agent.decoder.Decoder
 import datadog.trace.test.agent.decoder.DecodedMessage
 import datadog.trace.test.agent.decoder.DecodedTrace
 import datadog.trace.util.Strings
+import groovy.json.JsonSlurper
 
 import java.nio.charset.StandardCharsets
 import java.util.concurrent.CopyOnWriteArrayList
@@ -28,6 +29,12 @@ abstract class AbstractSmokeTest extends ProcessManager {
   @Shared
   protected CopyOnWriteArrayList<DecodedTrace> decodeTraces = new CopyOnWriteArrayList()
 
+  @Shared
+  protected CopyOnWriteArrayList<Map<String, Object>> telemetryMessages = new CopyOnWriteArrayList()
+
+  @Shared
+  protected CopyOnWriteArrayList<Map<String, Object>> telemetryFlatMessages = new CopyOnWriteArrayList()
+
   @Shared
   private Closure decode = decodedTracesCallback()
 
@@ -37,6 +44,9 @@ abstract class AbstractSmokeTest extends ProcessManager {
   @Shared
   private Throwable traceDecodingFailure = null
 
+  @Shared
+  private Throwable telemetryDecodingFailure = null
+
   @Shared
   @AutoCleanup
   protected TestHttpServer server = httpServer {
@@ -111,6 +121,24 @@ abstract class AbstractSmokeTest extends ProcessManager {
         response.status(200).send(remoteConfigResponse)
       }
       prefix("/telemetry/proxy/api/v2/apmtelemetry") {
+        def body = request.getBody()
+        if (body != null) {
+          Map<String, Object> msg = null
+          try {
+            msg = new JsonSlurper().parseText(new String(body, StandardCharsets.UTF_8)) as Map<String, Object>
+          } catch (Throwable t) {
+            println("=== Failure during telemetry decoding ===")
+            t.printStackTrace(System.out)
+            telemetryDecodingFailure = t
+            throw t
+          }
+          telemetryMessages.add(msg)
+          if (msg.get("request_type") == "message-batch") {
+            msg.get("payload")?.each { telemetryFlatMessages.add(it as Map<String, Object>) }
+          } else {
+            telemetryFlatMessages.add(msg)
+          }
+        }
         response.status(202).send()
       }
     }
@@ -147,7 +175,8 @@ abstract class AbstractSmokeTest extends ProcessManager {
       "-Ddd.profiling.ddprof.alloc.enabled=${isDdprofSafe()}",
       "-Ddatadog.slf4j.simpleLogger.defaultLogLevel=${logLevel()}",
       "-Dorg.slf4j.simpleLogger.defaultLogLevel=${logLevel()}",
-      "-Ddd.site="
+      "-Ddd.site=",
+      "-Ddd.telemetry.heartbeat.interval=2",
     ]
     if (inferServiceName())  {
       ret += "-Ddd.service.name=${SERVICE_NAME}"
@@ -268,6 +297,31 @@ abstract class AbstractSmokeTest extends ProcessManager {
     decodeTraces
   }
 
+  void waitForTelemetryCount(final int count) {
+    def conditions = new PollingConditions(timeout: 30, initialDelay: 0, delay: 1, factor: 1)
+    waitForTelemetryCount(conditions, count)
+  }
+
+  void waitForTelemetryCount(final PollingConditions poll, final int count) {
+    poll.eventually {
+      telemetryMessages.size() >= count
+    }
+  }
+
+  void waitForTelemetryFlat(final Function<Map<String, Object>, Boolean> predicate) {
+    def conditions = new PollingConditions(timeout: 30, initialDelay: 0, delay: 1, factor: 1)
+    waitForTelemetryFlat(conditions, predicate)
+  }
+
+  void waitForTelemetryFlat(final PollingConditions poll, final Function<Map<String, Object>, Boolean> predicate) {
+    poll.eventually {
+      if (telemetryDecodingFailure != null) {
+        throw telemetryDecodingFailure
+      }
+      assert telemetryFlatMessages.find { predicate.apply(it) } != null
+    }
+  }
+
   def logLevel() {
     return "info"
   }