Add smoke tests for telemetry (#7955)

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/spring-boot-3.0-native/src/test/groovy/SpringBootNativeInstrumentationTest.groovy

@@ -55,6 +55,11 @@ class SpringBootNativeInstrumentationTest extends AbstractServerSmokeTest {
     return ["[servlet.request[spring.handler[WebController.doHello[WebController.sayHello]]]]"]
   }
 
+  @Override
+  boolean testTelemetry() {
+    false
+  }
+
   def "check native instrumentation"() {
     setup:
     String url = "http://localhost:${httpPort}/hello"

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

@@ -78,4 +78,9 @@ class SpringBootWebmvcIntegrationTest extends AbstractServerSmokeTest {
     responseBodyStr.contains("banana")
     waitForTraceCount(1)
   }
+
+  @Override
+  List<String> expectedTelemetryDependencies() {
+    ['spring-core']
+  }
 }

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

@@ -78,4 +78,9 @@ class SpringBootWebmvcIntegrationTest extends AbstractServerSmokeTest {
     responseBodyStr.contains("banana")
     waitForTraceCount(1)
   }
+
+  @Override
+  List<String> expectedTelemetryDependencies() {
+    ['spring-core']
+  }
 }

dd-smoke-tests/springboot-openliberty-20/src/test/groovy/datadog/smoketest/SpringBootOpenLibertySmokeTest.groovy

@@ -61,6 +61,11 @@ class SpringBootOpenLibertySmokeTest extends AbstractServerSmokeTest {
     ].toSet()
   }
 
+  @Override
+  boolean testTelemetry() {
+    false
+  }
+
   def "Test concurrent requests to Spring Boot running Open Liberty"() {
     setup:
     def url = "http://localhost:${httpPort}/connect"

dd-smoke-tests/springboot-openliberty-20/src/test/groovy/datadog/smoketest/SpringBootOpenLibertySmokeVulnerabilityTest.groovy

@@ -59,6 +59,11 @@ class SpringBootOpenLibertySmokeVulnerabilityTest extends AbstractServerSmokeTes
     return {} // force traces decoding
   }
 
+  @Override
+  boolean testTelemetry() {
+    false
+  }
+
   private static boolean contains(String s) {
     System.out.println("Checking span:" + s)
     return s.contains("MD5")

dd-smoke-tests/springboot-openliberty-23/src/test/groovy/datadog/smoketest/SpringBootOpenLibertySmokeTest.groovy

@@ -68,6 +68,11 @@ class SpringBootOpenLibertySmokeTest extends AbstractServerSmokeTest {
     ].toSet()
   }
 
+  @Override
+  boolean testTelemetry() {
+    false
+  }
+
   def "Test concurrent requests to Spring Boot running Open Liberty"() {
     setup:
     def url = "http://localhost:${httpPort}/connect"

dd-smoke-tests/springboot-openliberty-23/src/test/groovy/datadog/smoketest/SpringBootOpenLibertySmokeVulnerabilityTest.groovy

@@ -59,6 +59,11 @@ class SpringBootOpenLibertySmokeVulnerabilityTest extends AbstractServerSmokeTes
     return {} // force traces decoding
   }
 
+  @Override
+  boolean testTelemetry() {
+    false
+  }
+
   private static boolean contains(String s) {
     System.out.println("Checking span:" + s)
     return s.contains("MD5")

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

@@ -5,6 +5,7 @@ import datadog.trace.agent.test.utils.OkHttpUtils
 import datadog.trace.agent.test.utils.PortUtils
 import okhttp3.OkHttpClient
 import spock.lang.Shared
+import static org.junit.Assume.assumeTrue
 
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicInteger
@@ -116,4 +117,54 @@ abstract class AbstractServerSmokeTest extends AbstractSmokeTest {
     }
     return remaining
   }
+
+  @RunLast
+  void 'receive telemetry app-started'() {
+    when:
+    assumeTrue(testTelemetry())
+    waitForTelemetryCount(1)
+
+    then:
+    telemetryMessages.size() >= 1
+    Object msg = telemetryMessages.get(0)
+    msg['request_type'] == 'app-started'
+  }
+
+  List<String> expectedTelemetryDependencies() {
+    []
+  }
+
+  @RunLast
+  @SuppressWarnings('UnnecessaryBooleanExpression')
+  void 'receive telemetry app-dependencies-loaded'() {
+    when:
+    assumeTrue(testTelemetry())
+    // 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
@@ -37,6 +38,15 @@ abstract class AbstractSmokeTest extends ProcessManager {
   @Shared
   private Throwable traceDecodingFailure = null
 
+  @Shared
+  protected CopyOnWriteArrayList<Map<String, Object>> telemetryMessages = new CopyOnWriteArrayList()
+
+  @Shared
+  protected CopyOnWriteArrayList<Map<String, Object>> telemetryFlatMessages = new CopyOnWriteArrayList()
+
+  @Shared
+  private Throwable telemetryDecodingFailure = null
+
   @Shared
   protected TestHttpServer.Headers lastTraceRequestHeaders = null
 
@@ -119,6 +129,23 @@ abstract class AbstractSmokeTest extends ProcessManager {
         response.status(200).send(remoteConfigResponse)
       }
       prefix("/telemetry/proxy/api/v2/apmtelemetry") {
+        try {
+          byte[] body = request.getBody()
+          if (body != null) {
+            Map<String, Object> msg = new JsonSlurper().parseText(new String(body, StandardCharsets.UTF_8)) as Map<String, Object>
+            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)
+            }
+          }
+        } catch (Throwable t) {
+          println("=== Failure during telemetry decoding ===")
+          t.printStackTrace(System.out)
+          telemetryDecodingFailure = t
+          throw t
+        }
         response.status(202).send()
       }
     }
@@ -160,6 +187,9 @@ abstract class AbstractSmokeTest extends ProcessManager {
     if (inferServiceName())  {
       ret += "-Ddd.service.name=${SERVICE_NAME}"
     }
+    if (testTelemetry()) {
+      ret += "-Ddd.telemetry.heartbeat.interval=5"
+    }
     ret as String[]
   }
 
@@ -172,6 +202,11 @@ abstract class AbstractSmokeTest extends ProcessManager {
     return !Platform.isJ9()
   }
 
+
+  boolean testTelemetry() {
+    return true
+  }
+
   def setup() {
     traceCount.set(0)
     decodeTraces.clear()
@@ -272,6 +307,31 @@ abstract class AbstractSmokeTest extends ProcessManager {
     }
   }
 
+  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
+    }
+  }
+
   List<DecodedTrace> getTraces() {
     decodeTraces
   }