Add smoke tests for telemetry

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/springboot-openliberty-20/src/test/groovy/datadog/smoketest/SpringBootOpenLibertySmokeTest.groovy

@@ -28,22 +28,23 @@ class SpringBootOpenLibertySmokeTest extends AbstractServerSmokeTest {
 
     List<String> envParams = new ArrayList<>()
     envParams.addAll(defaultJavaProperties)
-    envParams.addAll(
+    envParams.addAll([
       "-Ddd.writer.type=MultiWriter:TraceStructureWriter:${output.getAbsolutePath()},DDAgentWriter",
       "-Ddd.jmxfetch.enabled=false",
       "-Ddd.appsec.enabled=true",
       "-Ddatadog.slf4j.simpleLogger.defaultLogLevel=debug",
       "-Dorg.slf4j.simpleLogger.defaultLogLevel=debug",
-      "-Ddd.iast.enabled=true", "-Ddd.iast.request-sampling=100",
+      "-Ddd.iast.enabled=true",
+      "-Ddd.iast.request-sampling=100",
       "-Ddd.integration.spring-boot.enabled=true"
-      )
+      ])
 
 
     String javaToolOptions = envParams.stream().collect(Collectors.joining(" "))
 
 
     ProcessBuilder processBuilder = new ProcessBuilder(command)
-    processBuilder.environment().put("JAVA_TOOL_OPTIONS", javaToolOptions)
+    processBuilder.environment().put("JVM_ARGS", javaToolOptions)
     processBuilder.directory(new File(buildDirectory))
     return processBuilder
   }
@@ -61,6 +62,12 @@ class SpringBootOpenLibertySmokeTest extends AbstractServerSmokeTest {
     ].toSet()
   }
 
+  @Override
+  boolean testTelemetry() {
+    // FIXME: Breaks this test suite, not sure why.
+    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

@@ -30,35 +30,42 @@ class SpringBootOpenLibertySmokeVulnerabilityTest extends AbstractServerSmokeTes
 
     List<String> envParams = new ArrayList<>()
     envParams.addAll(defaultJavaProperties)
-    envParams.addAll(
+    envParams.addAll([
       "-Ddd.writer.type=MultiWriter:TraceStructureWriter:${output.getAbsolutePath()},DDAgentWriter",
       "-Ddd.jmxfetch.enabled=false",
       "-Ddd.appsec.enabled=true",
       "-Ddatadog.slf4j.simpleLogger.defaultLogLevel=debug",
       "-Dorg.slf4j.simpleLogger.defaultLogLevel=debug",
-      "-Ddd.iast.enabled=true", "-Ddd.iast.request-sampling=100"
-      )
+      "-Ddd.iast.enabled=true",
+      "-Ddd.iast.request-sampling=100"
+      ])
 
 
     String javaToolOptions = envParams.stream().collect(Collectors.joining(" "))
 
 
     ProcessBuilder processBuilder = new ProcessBuilder(command)
-    processBuilder.environment().put("JAVA_TOOL_OPTIONS", javaToolOptions)
+    processBuilder.environment().put("JVM_ARGS", javaToolOptions)
     processBuilder.directory(new File(buildDirectory))
     return processBuilder
   }
 
   @Override
   File createTemporaryFile() {
-    return new File("${buildDirectory}/tmp/springboot-openliberty.out")
+    return new File("${buildDirectory}/tmp/springboot-openliberty-iast.out")
   }
 
   @Override
   Closure decodedTracesCallback() {
     return {} // force traces decoding
   }
 
+  @Override
+  boolean testTelemetry() {
+    // FIXME: Breaks this test suite, not sure why.
+    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

@@ -55,6 +55,12 @@ class SpringBootOpenLibertySmokeTest extends AbstractServerSmokeTest {
     false // will use spring properties
   }
 
+  @Override
+  boolean testTelemetry() {
+    // FIXME: Breaks this test suite, not sure why.
+    false
+  }
+
   @Override
   File createTemporaryFile() {
     return new File("${buildDirectory}/tmp/springboot-openliberty.out")

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

@@ -59,6 +59,12 @@ class SpringBootOpenLibertySmokeVulnerabilityTest extends AbstractServerSmokeTes
     return {} // force traces decoding
   }
 
+  @Override
+  boolean testTelemetry() {
+    // FIXME: Breaks this test suite, not sure why.
+    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
@@ -55,6 +56,11 @@ abstract class AbstractServerSmokeTest extends AbstractSmokeTest {
     return Collections.emptySet()
   }
 
+  @Override
+  boolean testTelemetry() {
+    return false
+  }
+
   protected Set<String> expectedTraces(int processIndex) {
     if (processIndex > 0) {
       throw new IllegalArgumentException("Override expectedTraces(int processIndex) for multi process tests")
@@ -116,4 +122,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() {
+    []
+  }
+
+  @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()
 
@@ -40,6 +47,9 @@ abstract class AbstractSmokeTest extends ProcessManager {
   @Shared
   protected TestHttpServer.Headers lastTraceRequestHeaders = null
 
+  @Shared
+  private Throwable telemetryDecodingFailure = null
+
   @Shared
   @AutoCleanup
   protected TestHttpServer server = httpServer {
@@ -119,6 +129,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()
       }
     }
@@ -155,8 +183,11 @@ 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=",
     ]
+    if (testTelemetry()) {
+      ret += "-Ddd.telemetry.heartbeat.interval=2"
+    }
     if (inferServiceName())  {
       ret += "-Ddd.service.name=${SERVICE_NAME}"
     }
@@ -167,6 +198,11 @@ abstract class AbstractSmokeTest extends ProcessManager {
     true
   }
 
+  /** Set to false in a test suite to skip telemetry tests. */
+  boolean testTelemetry() {
+    true
+  }
+
   private static boolean isDdprofSafe() {
     // currently the J9 handling of jmethodIDs will cause frequent crashes
     return !Platform.isJ9()
@@ -276,7 +312,32 @@ 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"
+    return "debug"
   }
 }

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,13 @@
+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) {
+    if (!annotations.isEmpty()) {
+      feature.setExecutionOrder(Integer.MAX_VALUE)
+    }
+  }
+}