Merge pull request #3865 from DataDog/glopes/fix-extended-hearbeat

appsec: assert app-extended-heartbeat re-emits full telemetry state

Author: cataphract

Cargo.lock

@@ -2,6 +2,8 @@ package com.datadog.appsec.php
 
 import groovy.transform.Canonical
 import groovy.transform.ToString
+import groovy.transform.stc.ClosureParams
+import groovy.transform.stc.FromString
 import java.net.http.HttpRequest
 import java.net.http.HttpResponse
 import java.util.Base64
@@ -13,12 +15,31 @@ import static java.net.http.HttpResponse.BodyHandlers.ofString
  * @link https://github.com/DataDog/instrumentation-telemetry-api-docs/blob/main/GeneratedDocumentation/ApiDocs/v2/producing-telemetry.md
  */
 class TelemetryHelpers {
-   static <T> List<T> filterMessages(List<Map> telemetryData, Class<T> type) {
-       (telemetryData.findAll { it['request_type'] in type.names } +
-                telemetryData.findAll { it['request_type'] == 'message-batch'}
-                        *.payload*.findAll { it['request_type'] in type.names }.flatten())
-        .collect { type.newInstance([it['payload']] as Object[]) }
-   }
+    /**
+     * Filters drained telemetry to messages of the given wrapper type, unwrapping any
+     * {@code message-batch} envelopes.
+     *
+     * <p>The sidecar runs two telemetry workers per session: one for the host app and
+     * one for {@code service_name == "datadog-ipc-helper"} (its own self-telemetry).
+     * Tests almost always want only the former; pass {@code userAppOnly = false} to
+     * include the sidecar's own telemetry too.
+     */
+    static <T> List<T> filterMessages(List<Map> telemetryData, Class<T> type, boolean userAppOnly = true) {
+        List<Map> payloads = []
+        for (msg in telemetryData) {
+            if (userAppOnly && msg.application?.service_name == 'datadog-ipc-helper') continue
+            if (msg.request_type in type.names) {
+                payloads << (msg.payload as Map)
+            } else if (msg.request_type == 'message-batch') {
+                for (inner in (msg.payload as List)) {
+                    if (inner.request_type in type.names) {
+                        payloads << (inner.payload as Map)
+                    }
+                }
+            }
+        }
+        payloads.collect { type.newInstance([it] as Object[]) }
+    }
 
     static class GenerateMetrics {
         static names = ['generate-metrics']
@@ -361,6 +382,19 @@ class TelemetryHelpers {
         }
     }
 
+    static class WithExtendedHeartbeat {
+        static names = ['app-extended-heartbeat']
+        List<ConfigurationEntry> configuration
+        List<DependencyEntry> dependencies
+        List<IntegrationEntry> integrations
+
+        WithExtendedHeartbeat(Map m) {
+            configuration = m.configuration?.collect { new ConfigurationEntry(it) }
+            dependencies  = m.dependencies?.collect { new DependencyEntry(it) }
+            integrations  = m.integrations?.collect { new IntegrationEntry(it) }
+        }
+    }
+
     static class ConfigurationEntry {
         String name
         String value
@@ -399,7 +433,10 @@ class TelemetryHelpers {
         }
     }
 
-    public static <T> List<T> waitForTelemetryData(AppSecContainer container, int timeoutSec, Closure<Boolean> cl, Class<T> cls, String path = '/hello.php') {
+    static <T> List<T> waitForTelemetryData(AppSecContainer container, int timeoutSec,
+                                            @ClosureParams(value = FromString, options = 'java.util.List<T>')
+                                                    Closure<Boolean> cl,
+                                            Class<T> cls, String path = '/hello.php') {
         List<T> messages = []
         def deadline = System.currentTimeSeconds() + timeoutSec
         def lastHttpReq = System.currentTimeSeconds() - 6
@@ -422,23 +459,46 @@ class TelemetryHelpers {
         messages
     }
 
-    public static List<AppEndpoints> waitForAppEndpoints(AppSecContainer container, int timeoutSec, Closure<Boolean> cl, String path = '/') {
+    static List<AppEndpoints> waitForAppEndpoints(AppSecContainer container, int timeoutSec,
+                                                  @ClosureParams(value = FromString,
+                                                          options = 'java.util.List<com.datadog.appsec.php.TelemetryHelpers.AppEndpoints>')
+                                                          Closure<Boolean> cl,
+                                                  String path = '/') {
         waitForTelemetryData(container, timeoutSec, cl, AppEndpoints, path)
     }
 
-    public static List<GenerateDistributions> waitForDistributions(AppSecContainer container, int timeoutSec, Closure<Boolean> cl) {
+    static List<GenerateDistributions> waitForDistributions(AppSecContainer container, int timeoutSec,
+                                                            @ClosureParams(value = FromString,
+                                                                    options = 'java.util.List<com.datadog.appsec.php.TelemetryHelpers.GenerateDistributions>')
+                                                                    Closure<Boolean> cl) {
         waitForTelemetryData(container, timeoutSec, cl, GenerateDistributions)
     }
 
-    public static List<GenerateMetrics> waitForMetrics(AppSecContainer container, int timeoutSec, Closure<Boolean> cl) {
+    static List<GenerateMetrics> waitForMetrics(AppSecContainer container, int timeoutSec,
+                                                @ClosureParams(value = FromString,
+                                                        options = 'java.util.List<com.datadog.appsec.php.TelemetryHelpers.GenerateMetrics>')
+                                                        Closure<Boolean> cl) {
         waitForTelemetryData(container, timeoutSec, cl, GenerateMetrics)
     }
 
-    public static List<WithIntegrations> waitForIntegrations(AppSecContainer container, int timeoutSec, Closure<Boolean> cl) {
+    static List<WithIntegrations> waitForIntegrations(AppSecContainer container, int timeoutSec,
+                                                      @ClosureParams(value = FromString,
+                                                              options = 'java.util.List<com.datadog.appsec.php.TelemetryHelpers.WithIntegrations>')
+                                                              Closure<Boolean> cl) {
         waitForTelemetryData(container, timeoutSec, cl, WithIntegrations)
     }
 
-    public static List<Logs> waitForLogs(AppSecContainer container, int timeoutSec, Closure<Boolean> cl) {
+    static List<WithExtendedHeartbeat> waitForExtendedHeartbeat(AppSecContainer container, int timeoutSec,
+                                                                @ClosureParams(value = FromString,
+                                                                        options = 'java.util.List<com.datadog.appsec.php.TelemetryHelpers.WithExtendedHeartbeat>')
+                                                                        Closure<Boolean> cl) {
+        waitForTelemetryData(container, timeoutSec, cl, WithExtendedHeartbeat)
+    }
+
+    static List<Logs> waitForLogs(AppSecContainer container, int timeoutSec,
+                                  @ClosureParams(value = FromString,
+                                          options = 'java.util.List<com.datadog.appsec.php.TelemetryHelpers.Logs>')
+                                          Closure<Boolean> cl) {
         waitForTelemetryData(container, timeoutSec, cl, Logs)
     }
 }

appsec/tests/integration/src/main/groovy/com/datadog/appsec/php/TelemetryHelpers.groovy

@@ -96,7 +96,6 @@ class AppSecContainer<SELF extends AppSecContainer<SELF>> extends GenericContain
         withEnv '_DD_DEBUG_SIDECAR_LOG_METHOD', 'file:///tmp/logs/sidecar.log'
         withEnv 'DD_SPAWN_WORKER_USE_EXEC', '1' // gdb fails following child with fdexec
         withEnv 'DD_TELEMETRY_HEARTBEAT_INTERVAL', '10'
-        withEnv 'DD_TELEMETRY_EXTENDED_HEARTBEAT_INTERVAL', '10'
         // withEnv '_DD_SHARED_LIB_DEBUG', '1'
         if (System.getProperty('XDEBUG') == '1') {
             Testcontainers.exposeHostPorts(9003)

appsec/tests/integration/src/main/groovy/com/datadog/appsec/php/docker/AppSecContainer.groovy

@@ -0,0 +1,102 @@
+package com.datadog.appsec.php.integration
+
+import com.datadog.appsec.php.TelemetryHelpers
+import com.datadog.appsec.php.docker.AppSecContainer
+import com.datadog.appsec.php.docker.FailOnUnmatchedTraces
+import groovy.util.logging.Slf4j
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.condition.DisabledIf
+import org.testcontainers.junit.jupiter.Container
+import org.testcontainers.junit.jupiter.Testcontainers
+
+import java.net.http.HttpResponse
+
+import static com.datadog.appsec.php.integration.TestParams.getPhpVersion
+import static com.datadog.appsec.php.integration.TestParams.getVariant
+
+/**
+ * Standalone test class for the extended-heartbeat re-emission behaviour.
+ *
+ * The default extended-heartbeat interval is 24h. To exercise it within the test
+ * timeout, this class starts its container with DD_TELEMETRY_EXTENDED_HEARTBEAT_INTERVAL=15
+ * (kept separate from TelemetryTests so the persistent telemetry worker for that
+ * class isn't already running with the long interval before this test runs).
+ *
+ * Per the spec (instrumentation-telemetry-api-docs/Source/ApiDocs/v2/openapi.yaml
+ * `AppExtendedHeartbeat`), the payload must carry configuration + dependencies +
+ * integrations so the agent can reconstruct application records on data loss.
+ */
+@Testcontainers
+@Slf4j
+@DisabledIf('isDisabled')
+class TelemetryExtendedHeartbeatTests {
+    static boolean disabled = variant.contains('zts') || phpVersion != '8.4'
+
+    @Container
+    @FailOnUnmatchedTraces
+    public static final AppSecContainer CONTAINER =
+            new AppSecContainer(
+                    workVolume: this.name,
+                    baseTag: 'apache2-fpm-php',
+                    phpVersion: phpVersion,
+                    phpVariant: variant,
+                    www: 'base',
+            ) {
+                @Override
+                void configure() {
+                    super.configure()
+                    // Short interval so the extended heartbeat fires within the test.
+                    withEnv 'DD_TELEMETRY_EXTENDED_HEARTBEAT_INTERVAL', '15'
+                }
+            }
+
+    @Test
+    void 'extended heartbeat re-emits configuration, dependencies and integrations'() {
+        // Trigger the integration loaders for phpredis (Redis class instantiation) and
+        // exec (system() call). These cause the tracer to push AddIntegration to the
+        // sidecar.
+        CONTAINER.traceFromRequest('/custom_integrations.php?integrations[]=redis&integrations[]=exec') {
+            HttpResponse<InputStream> resp -> assert resp.statusCode() == 200
+        }
+
+        // Stage 1: verify the regular flush path actually emits phpredis (and exec)
+        // in an app-started or app-integrations-change message. This proves we're
+        // about to test the heartbeat re-emitting *previously-flushed* data.
+        Set<String> flushed = [] as Set
+        TelemetryHelpers.waitForIntegrations(CONTAINER, 30) { msgs ->
+            msgs.each { (it.integrations ?: []).each { flushed << it.name } }
+            'phpredis' in flushed && 'exec' in flushed
+        }
+        assert 'phpredis' in flushed : "phpredis not emitted via app-started/app-integrations-change; saw: ${flushed}"
+        assert 'exec' in flushed : "exec not emitted via app-started/app-integrations-change; saw: ${flushed}"
+
+        // Stage 2: the extended heartbeat must re-emit the full triple
+        // (configuration + dependencies + integrations).
+        CONTAINER.drainTelemetry(0)
+        def hbs = TelemetryHelpers.waitForExtendedHeartbeat(CONTAINER, 35) { !it.empty }
+        assert !hbs.empty : 'No app-extended-heartbeat received within 35s'
+        def hb = hbs.first()
+        assert hb.configuration != null : 'heartbeat must include configuration field'
+        assert hb.dependencies  != null : 'heartbeat must include dependencies field'
+        assert hb.integrations  != null : 'heartbeat must include integrations field'
+        assert 'phpredis' in hb.integrations*.name :
+                "heartbeat should re-emit phpredis; got: ${hb.integrations*.name}"
+
+        // Stage 3: regression check for the leak that the libdd_telemetry fix addresses.
+        // After the heartbeat, the worker should pop the re-queued items out of unflushed.
+        // So when a NEW integration is loaded next, the following regular app-integrations-change
+        // must contain only the new integration, NOT the previously-flushed-and-re-queued phpredis.
+        CONTAINER.drainTelemetry(0)
+        CONTAINER.traceFromRequest('/custom_integrations.php?integrations[]=fs') {
+            HttpResponse<InputStream> resp -> assert resp.statusCode() == 200
+        }
+        Set<String> followup = [] as Set
+        TelemetryHelpers.waitForIntegrations(CONTAINER, 30) { msgs ->
+            msgs.each { (it.integrations ?: []).each { followup << it.name } }
+            'filesystem' in followup
+        }
+        assert 'filesystem' in followup : 'No app-integrations-change with filesystem within 30s after heartbeat'
+        assert !('phpredis' in followup) :
+                "phpredis must not be re-emitted after the heartbeat; got: ${followup}"
+    }
+}

appsec/tests/integration/src/test/groovy/com/datadog/appsec/php/integration/TelemetryExtendedHeartbeatTests.groovy

@@ -18,6 +18,10 @@ public function __construct()
         system('true');
     } elseif ($int === 'redis') {
         new Redis();
+    } elseif ($int === 'fs') {
+        // Triggers the filesystem integration via deferred-loaded fopen() hook.
+        $fp = fopen('/dev/null', 'r');
+        if ($fp) { fclose($fp); }
     } else {
         http_response_code(400);
     }