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/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/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,59 @@ abstract class AbstractServerSmokeTest extends AbstractSmokeTest {
     }
     return remaining
   }
+
+  /** Set to false in a test suite to skip telemetry tests. */
+  boolean testTelemetry() {
+    true
+  }
+
+  @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() {
+    []
+  }
+
+  @SuppressWarnings('UnnecessaryBooleanExpression')
+  @RunLast
+  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
@@ -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"
   }

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

@@ -0,0 +1,19 @@
+package datadog.smoketest
+
+import org.spockframework.runtime.extension.ExtensionAnnotation
+
+import java.lang.annotation.ElementType
+import java.lang.annotation.Retention
+import java.lang.annotation.RetentionPolicy
+import java.lang.annotation.Target
+
+/**
+ * Spock test methods annotated with this will be executed last.
+ * This is useful for tests that need to wait for some test to settle while other tests run (e.g. telemetry), so it is
+ * more efficient to run them at the end.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target([ElementType.METHOD])
+@ExtensionAnnotation(RunLastExtension)
+@interface RunLast {
+}

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

@@ -0,0 +1,11 @@
+package datadog.smoketest
+
+import org.spockframework.runtime.extension.IAnnotationDrivenExtension
+import org.spockframework.runtime.model.FeatureInfo
+
+class RunLastExtension implements IAnnotationDrivenExtension<RunLast> {
+  @Override
+  void visitFeatureAnnotations(List<RunLast> annotations, FeatureInfo feature) {
+    feature.setExecutionOrder(Integer.MAX_VALUE)
+  }
+}