feat: OTel context sharing

Makefile

@@ -48,7 +48,7 @@ RUN_TESTS_CMD := DD_SERVICE= DD_ENV= DD_TRACE_RETRY_INTERVAL=1 REPORT_EXIT_STATU
 
 C_FILES = $(shell find components components-rs ext src/dogstatsd tracer zend_abstract_interface -name '*.c' -o -name '*.h' | awk '{ printf "$(BUILD_DIR)/%s\n", $$1 }' )
 TEST_FILES = $(shell find tests/ext -name '*.php*' -o -name '*.inc' -o -name '*.json' -o -name '*.yaml' -o -name 'CONFLICTS' | awk '{ printf "$(BUILD_DIR)/%s\n", $$1 }' )
-RUST_FILES = $(BUILD_DIR)/Cargo.toml $(BUILD_DIR)/Cargo.lock $(shell find components-rs -name '*.c' -o -name '*.rs' -o -name 'Cargo.toml' | awk '{ printf "$(BUILD_DIR)/%s\n", $$1 }' ) $(shell find libdatadog/{build-common,datadog-ffe,datadog-ipc,datadog-ipc-macros,datadog-live-debugger,datadog-live-debugger-ffi,libdd-remote-config,datadog-sidecar,datadog-sidecar-ffi,datadog-sidecar-macros,libdd-alloc,libdd-capabilities,libdd-capabilities-impl,libdd-common,libdd-common-ffi,libdd-crashtracker,libdd-crashtracker-ffi,libdd-data-pipeline,libdd-ddsketch,libdd-dogstatsd-client,libdd-library-config,libdd-library-config-ffi,libdd-log,libdd-shared-runtime,libdd-telemetry,libdd-telemetry-ffi,libdd-tinybytes,libdd-trace-*,spawn_worker,tools/{cc_utils,sidecar_mockgen},libdd-trace-*,Cargo.toml} \( -type l -o -type f \) \( -path "*/src*" -o -path "*/examples*" -o -path "*Cargo.toml" -o -path "*/build.rs" -o -path "*/tests/dataservice.rs" -o -path "*/tests/service_functional.rs" \) -not -path "*/datadog-ipc/build.rs" -not -path "*/datadog-sidecar-ffi/build.rs")
+RUST_FILES = $(BUILD_DIR)/Cargo.toml $(BUILD_DIR)/Cargo.lock $(shell find components-rs -name '*.c' -o -name '*.rs' -o -name 'Cargo.toml' | awk '{ printf "$(BUILD_DIR)/%s\n", $$1 }' ) $(shell find libdatadog/{build-common,datadog-ffe,datadog-ipc,datadog-ipc-macros,datadog-live-debugger,datadog-live-debugger-ffi,libdd-remote-config,datadog-sidecar,datadog-sidecar-ffi,datadog-sidecar-macros,libdd-alloc,libdd-capabilities,libdd-capabilities-impl,libdd-common,libdd-common-ffi,libdd-crashtracker,libdd-crashtracker-ffi,libdd-data-pipeline,libdd-ddsketch,libdd-dogstatsd-client,libdd-library-config,libdd-library-config-ffi,libdd-log,libdd-otel-thread-ctx*,libdd-shared-runtime,libdd-telemetry,libdd-telemetry-ffi,libdd-tinybytes,libdd-trace-*,spawn_worker,tools/{cc_utils,sidecar_mockgen},libdd-trace-*,Cargo.toml} \( -type l -o -type f \) \( -path "*/src*" -o -path "*/examples*" -o -path "*Cargo.toml" -o -path "*/build.rs" -o -path "*/tests/dataservice.rs" -o -path "*/tests/service_functional.rs" -o -name "tls-dynamic-list.txt" \) -not -path "*/datadog-ipc/build.rs" -not -path "*/datadog-sidecar-ffi/build.rs")
 ALL_OBJECT_FILES = $(C_FILES) $(RUST_FILES) $(BUILD_DIR)/Makefile
 TEST_OPCACHE_FILES = $(shell find tests/opcache -name '*.php*' -o -name '.gitkeep' | awk '{ printf "$(BUILD_DIR)/%s\n", $$1 }' )
 TEST_STUB_FILES = $(shell find tests/ext -type d -name 'stubs' -exec find '{}' -type f \; | awk '{ printf "$(BUILD_DIR)/%s\n", $$1 }' )
@@ -423,9 +423,9 @@ clang_format_fix:
 cbindgen: remove_cbindgen generate_cbindgen
 
 remove_cbindgen:
-	rm -f components-rs/datadog.h components-rs/live-debugger.h components-rs/telemetry.h components-rs/sidecar.h components-rs/common.h components-rs/crashtracker.h components-rs/library-config.h
+	rm -f components-rs/datadog.h components-rs/live-debugger.h components-rs/telemetry.h components-rs/sidecar.h components-rs/common.h components-rs/crashtracker.h components-rs/library-config.h components-rs/otel-thread-ctx.h
 
-generate_cbindgen: cbindgen_binary # Regenerate components-rs/datadog.h components-rs/live-debugger.h components-rs/telemetry.h components-rs/sidecar.h components-rs/common.h components-rs/crashtracker.h components-rs/library-config.h
+generate_cbindgen: cbindgen_binary # Regenerate components-rs/datadog.h components-rs/live-debugger.h components-rs/telemetry.h components-rs/sidecar.h components-rs/common.h components-rs/crashtracker.h components-rs/library-config.h components-rs/otel-thread-ctx.h
 	( \
 		$(command rustup && echo run nightly --) cbindgen --crate datadog-php  \
 			--config cbindgen.toml \
@@ -449,11 +449,14 @@ generate_cbindgen: cbindgen_binary # Regenerate components-rs/datadog.h componen
 		$(command rustup && echo run nightly --) cbindgen --crate libdd-library-config-ffi  \
 			--config libdd-library-config-ffi/cbindgen.toml \
 			--output $(PROJECT_ROOT)/components-rs/library-config.h; \
+		$(command rustup && echo run nightly --) cbindgen --crate libdd-otel-thread-ctx-ffi  \
+			--config libdd-otel-thread-ctx-ffi/cbindgen.toml \
+			--output $(PROJECT_ROOT)/components-rs/otel-thread-ctx.h; \
 		if test -d $(PROJECT_ROOT)/tmp; then \
 			mkdir -pv "$(BUILD_DIR)"; \
 			export CARGO_TARGET_DIR="$(BUILD_DIR)/target"; \
 		fi; \
-		cargo run -p tools --bin dedup_headers -- $(PROJECT_ROOT)/components-rs/common.h $(PROJECT_ROOT)/components-rs/datadog.h $(PROJECT_ROOT)/components-rs/live-debugger.h $(PROJECT_ROOT)/components-rs/telemetry.h $(PROJECT_ROOT)/components-rs/sidecar.h $(PROJECT_ROOT)/components-rs/crashtracker.h $(PROJECT_ROOT)/components-rs/library-config.h \
+		cargo run -p tools --bin dedup_headers -- $(PROJECT_ROOT)/components-rs/common.h $(PROJECT_ROOT)/components-rs/datadog.h $(PROJECT_ROOT)/components-rs/live-debugger.h $(PROJECT_ROOT)/components-rs/telemetry.h $(PROJECT_ROOT)/components-rs/sidecar.h $(PROJECT_ROOT)/components-rs/crashtracker.h $(PROJECT_ROOT)/components-rs/library-config.h $(PROJECT_ROOT)/components-rs/otel-thread-ctx.h \
 	)
 
 cbindgen_binary:

appsec/src/extension/ddappsec.c

@@ -14,6 +14,7 @@
 #include <string.h>
 #include <sys/stat.h>
 #include <sys/types.h>
+#include <unistd.h>
 
 #include <pthread.h>
 #include <stdatomic.h>
@@ -55,6 +56,7 @@ static atomic_int _thread_count;
 static void _check_enabled(void);
 #ifdef TESTING
 static void _register_testing_objects(void);
+volatile int ddappsec_debugger_wait_continue;
 #endif
 
 static PHP_MINIT_FUNCTION(ddappsec);
@@ -481,6 +483,32 @@ static PHP_FUNCTION(datadog_appsec_testing_stop_for_debugger)
     RETURN_TRUE;
 }
 
+static PHP_FUNCTION(datadog_appsec_testing_wait_for_debugger)
+{
+    if (zend_parse_parameters_none() == FAILURE) {
+        RETURN_FALSE;
+    }
+    ddappsec_debugger_wait_continue = 0;
+
+    int fd = open(
+        "/tmp/pid", O_CREAT | O_WRONLY | O_TRUNC | O_CLOEXEC, 0600); // NOLINT
+    if (fd < 0) {
+        RETURN_FALSE;
+    }
+    char pid[sizeof("-2147483648")] = "";
+    sprintf(pid, "%" PRIi32, (int32_t)getpid()); // NOLINT
+    ATTR_UNUSED ssize_t unused_ = write(fd, pid, strlen(pid));
+    close(fd);
+
+    while (!ddappsec_debugger_wait_continue) {
+        usleep(10000); // NOLINT
+    }
+    ddappsec_debugger_wait_continue = 0;
+    unlink("/tmp/pid"); // NOLINT
+
+    RETURN_TRUE;
+}
+
 static PHP_FUNCTION(datadog_appsec_testing_request_exec)
 {
     zend_array *data = NULL;
@@ -632,6 +660,7 @@ static const zend_function_entry testing_request_control_functions[] = {
     ZEND_RAW_FENTRY(DD_TESTING_NS "rinit", PHP_FN(datadog_appsec_testing_rinit), void_ret_bool_arginfo, 0, NULL, NULL)
     ZEND_RAW_FENTRY(DD_TESTING_NS "rshutdown", PHP_FN(datadog_appsec_testing_rshutdown), void_ret_bool_arginfo, 0, NULL, NULL)
     ZEND_RAW_FENTRY(DD_TESTING_NS "request_exec", PHP_FN(datadog_appsec_testing_request_exec), request_exec_arginfo, 0, NULL, NULL)
+    ZEND_RAW_FENTRY(DD_TESTING_NS "wait_for_debugger", PHP_FN(datadog_appsec_testing_wait_for_debugger), void_ret_bool_arginfo, 0, NULL, NULL)
     PHP_FE_END
 };
 static const zend_function_entry testing_functions[] = {

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

@@ -0,0 +1,261 @@
+package com.datadog.appsec.php.integration
+
+import com.datadog.appsec.php.docker.AppSecContainer
+import com.datadog.appsec.php.docker.InspectContainerHelper
+import groovy.json.JsonSlurper
+import groovy.util.logging.Slf4j
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.condition.DisabledIf
+import org.testcontainers.containers.Container.ExecResult
+import org.testcontainers.junit.jupiter.Container
+import org.testcontainers.junit.jupiter.Testcontainers
+
+import java.net.http.HttpResponse
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.TimeUnit
+
+import static com.datadog.appsec.php.integration.TestParams.getPhpVersion
+import static com.datadog.appsec.php.integration.TestParams.getVariant
+import static java.net.http.HttpResponse.BodyHandlers.ofString
+
+@Testcontainers
+@Slf4j
+@DisabledIf('isDisabled')
+class OtelThreadContextTests {
+    private static final String PID_FILE = '/tmp/pid'
+    private static final String GDB_SCRIPT =
+            '/project/appsec/tests/integration/src/test/resources/otel_context_gdb.py'
+    private static final String GDB_TIMEOUT = '20s'
+    private static final String EXPECTED_PROCESS_CONTEXT_SIGNATURE = 'OTEL_CTX'
+    private static final String EXPECTED_PROCESS_CONTEXT_VERSION = '2'
+
+    static boolean disabled = phpVersion != '8.3'
+
+    @Container
+    public static final AppSecContainer CONTAINER =
+            new AppSecContainer(
+                    workVolume: this.name,
+                    baseTag: 'apache2-mod-php',
+                    phpVersion: phpVersion,
+                    phpVariant: variant,
+                    www: 'base',
+            )
+
+    static void main(String[] args) {
+        InspectContainerHelper.run(CONTAINER)
+    }
+
+    @AfterEach
+    void afterEach() {
+        CONTAINER.clearTraces()
+    }
+
+    @Test
+    void 'otel thread context matches trace ids during regular request lifecycle'() {
+        PausedRequest pausedRequest = startPausedRequest('/otel_context_regular.php')
+        boolean released = false
+
+        try {
+            Map<String, String> threadContext = inspectThreadLocalAndContinue(pausedRequest.pid)
+            released = true
+            HttpResponse<String> response = awaitResponse(pausedRequest)
+            Map responseBody = parseJsonResponse(response)
+
+            assertThreadContextMatchesResponse(threadContext, responseBody)
+        } finally {
+            if (!released) {
+                continuePausedRequestQuietly(pausedRequest.pid)
+            }
+        }
+    }
+
+    @Test
+    void 'otel thread context matches trace ids during user request lifecycle'() {
+        PausedRequest pausedRequest = startPausedRequest('/otel_context_user_request.php')
+        boolean released = false
+
+        try {
+            Map<String, String> threadContext = inspectThreadLocalAndContinue(pausedRequest.pid)
+            released = true
+            HttpResponse<String> response = awaitResponse(pausedRequest)
+            Map responseBody = parseJsonResponse(response)
+
+            assert responseBody.outer_span_id != responseBody.span_id
+            assertThreadContextMatchesResponse(threadContext, responseBody)
+        } finally {
+            if (!released) {
+                continuePausedRequestQuietly(pausedRequest.pid)
+            }
+        }
+    }
+
+    @Test
+    void 'otel process context shared memory has expected threadlocal schema and version'() {
+        PausedRequest pausedRequest = startPausedRequest('/otel_context_regular.php')
+        boolean released = false
+
+        try {
+            Map<String, String> processContext = inspectProcessContext(pausedRequest.pid)
+            continuePausedRequest(pausedRequest.pid)
+            released = true
+
+            HttpResponse<String> response = awaitResponse(pausedRequest)
+            parseJsonResponse(response)
+
+            assert processContext.present == 'true'
+            assert processContext.signature == EXPECTED_PROCESS_CONTEXT_SIGNATURE
+            assert processContext.version == EXPECTED_PROCESS_CONTEXT_VERSION
+            assert processContext.payload_size.toInteger() > 0
+            assert processContext.published_at.toBigInteger() > 0
+            assert processContext.has_threadlocal_schema_key == 'true'
+            assert processContext.has_threadlocal_schema_value == 'true'
+            assert processContext.has_threadlocal_attribute_key_map == 'true'
+            assert processContext.has_local_root_span_key == 'true'
+        } finally {
+            if (!released) {
+                continuePausedRequestQuietly(pausedRequest.pid)
+            }
+        }
+    }
+
+    private static void assertThreadContextMatchesResponse(
+            Map<String, String> threadContext, Map responseBody) {
+        assert responseBody.waited == true
+        assert threadContext.ctx != '0x0'
+        assert threadContext.valid == '1'
+        assert threadContext.attrs_data_size.toInteger() >= 18
+        assert threadContext.attr0_key == '0'
+        assert threadContext.attr0_len == '16'
+        assert threadContext.trace_id == responseBody.trace_id
+        assert threadContext.span_id == responseBody.span_id
+        assert threadContext.local_root_span_id == responseBody.local_root_span_id
+    }
+
+    private static Map parseJsonResponse(HttpResponse<String> response) {
+        assert response.statusCode() == 200
+        new JsonSlurper().parseText(response.body()) as Map
+    }
+
+    private static PausedRequest startPausedRequest(String path) {
+        CONTAINER.execInContainer('rm', '-f', PID_FILE)
+
+        def request = CONTAINER.buildReq(path).GET().build()
+        CompletableFuture<HttpResponse<String>> responseFuture =
+                CONTAINER.httpClient.sendAsync(request, ofString())
+
+        new PausedRequest(
+                pid: waitForPausedPid(responseFuture),
+                responseFuture: responseFuture)
+    }
+
+    private static String waitForPausedPid(CompletableFuture<HttpResponse<String>> responseFuture) {
+        long deadline = System.currentTimeMillis() + 15_000
+
+        while (System.currentTimeMillis() < deadline) {
+            if (responseFuture.isDone()) {
+                HttpResponse<String> response = responseFuture.getNow(null)
+                throw new AssertionError(
+                        "Request completed before the debugger pause: HTTP ${response.statusCode()}\n${response.body()}".toString())
+            }
+
+            ExecResult res = CONTAINER.execInContainer(
+                    'bash', '-lc',
+                    "test -s ${PID_FILE} && cat ${PID_FILE} || true".toString())
+            if (res.exitCode == 0) {
+                String pid = res.stdout.trim()
+                if (pid) {
+                    return pid
+                }
+            }
+            Thread.sleep(100)
+        }
+
+        throw new AssertionError('Timed out waiting for the paused PHP worker pid')
+    }
+
+    private static HttpResponse<String> awaitResponse(PausedRequest pausedRequest) {
+        pausedRequest.responseFuture.get(30, TimeUnit.SECONDS)
+    }
+
+    private static Map<String, String> inspectThreadLocalAndContinue(String pid) {
+        List<String> commands = [
+                'set pagination off',
+                'otel-thread-context',
+                'ddappsec-continue',
+                'detach',
+                'quit',
+        ]
+
+        ExecResult res = runGdb(pid, commands)
+        parseKeyValueOutput(res.stdout)
+    }
+
+    private static void continuePausedRequest(String pid) {
+        runGdb(pid, [
+                'set pagination off',
+                'ddappsec-continue',
+                'detach',
+                'quit',
+        ])
+    }
+
+    private static void continuePausedRequestQuietly(String pid) {
+        try {
+            continuePausedRequest(pid)
+        } catch (Throwable ignored) {
+            // The original failure is more useful than a best-effort cleanup error.
+        }
+    }
+
+    private static ExecResult runGdb(String pid, List<String> commands) {
+        List<String> args = [
+                'timeout',
+                GDB_TIMEOUT,
+                'gdb',
+                '--batch',
+                '--quiet',
+                '-p',
+                pid,
+                '-ex',
+                "python exec(open('${GDB_SCRIPT}').read())".toString(),
+        ]
+        commands.each {
+            args.add('-ex')
+            args.add(it)
+        }
+
+        ExecResult res = CONTAINER.execInContainer(args as String[])
+        if (res.exitCode != 0 || res.stderr =~ /(Traceback|Python Exception|No symbol|Undefined command)/) {
+            throw new AssertionError(
+                    "gdb failed with exit code ${res.exitCode}\nstdout:\n${res.stdout}\nstderr:\n${res.stderr}".toString())
+        }
+        res
+    }
+
+    private static Map<String, String> inspectProcessContext(String pid) {
+        ExecResult res = runGdb(pid, [
+                'set pagination off',
+                'otel-process-context',
+                'detach',
+                'quit',
+        ])
+        parseKeyValueOutput(res.stdout)
+    }
+
+    private static Map<String, String> parseKeyValueOutput(String output) {
+        Map<String, String> result = [:]
+        output.readLines().each { String line ->
+            if (line ==~ /^[A-Za-z_][A-Za-z0-9_]*=.*/) {
+                int idx = line.indexOf('=')
+                result[line.substring(0, idx).trim()] = line.substring(idx + 1).trim()
+            }
+        }
+        result
+    }
+
+    private static class PausedRequest {
+        String pid
+        CompletableFuture<HttpResponse<String>> responseFuture
+    }
+}