Batch endpoint collection & remove Wordpress Endpoint collection (#3764)

* Batch endpoint collection

Signed-off-by: Bob Weinand <[email protected]>

* Avoid duplicate Symfony endpoint collections & don't collect every page on wordpress
Batch ack sending too

Signed-off-by: Bob Weinand <[email protected]>

* Match SHM cache Integration type with sidecar writer

The sidecar's write_shm_file() serializes buffered_integrations as
HashSet<Integration>, but the PHP-side reader in components-rs
deserialized it as HashSet<String>. This mismatch has existed since
the SHM cache was introduced in 91222ad ("feat: reduce telemetry
sent", #3316), but was latent: config_sent was set true before any
integrations appeared in SHM, and integration dedup (the only other
consumer) was silently broken with the sidecar deduplicating
server-side as a fallback.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

* Use a proper struct for Telemetry SHM data

Signed-off-by: Bob Weinand <[email protected]>

* Update Makefile

Signed-off-by: Bob Weinand <[email protected]>

---------

Signed-off-by: Bob Weinand <[email protected]>
Co-authored-by: Gustavo André dos Santos Lopes <[email protected]>
Co-authored-by: Claude Opus 4.6 (1M context) <[email protected]>

Author: 3 people

Cargo.lock

@@ -48,7 +48,7 @@ RUN_TESTS_CMD := DD_SERVICE= DD_ENV= REPORT_EXIT_STATUS=1 TEST_PHP_SRCDIR=$(PROJ
 
 C_FILES = $(shell find components components-rs ext src/dogstatsd 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,datadog-remote-config,datadog-sidecar,datadog-sidecar-ffi,datadog-sidecar-macros,libdd-alloc,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-libunwind-sys,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 "*/libunwind*" -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,datadog-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-libunwind-sys,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 "*/libunwind*" -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")
 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 }' )

Makefile

@@ -3,6 +3,7 @@
 echo "are_endpoints_collected before: " . (\DDTrace\are_endpoints_collected() ? 'true' : 'false') . "\n";
 
 \DDTrace\add_endpoint('/test_random_endpoint', 'http.request', 'GET /test_random_endpoint', 'GET');
+\DDTrace\flush_endpoints();
 
 sleep(1);
 

appsec/tests/integration/src/test/www/base/public/endpoint_collection.php

@@ -313,6 +313,19 @@ typedef enum ddog_Log {
   DDOG_LOG_HOOK_TRACE = (5 | (4 << 4)),
 } ddog_Log;
 
+typedef enum ddog_Method {
+  DDOG_METHOD_GET = 0,
+  DDOG_METHOD_POST = 1,
+  DDOG_METHOD_PUT = 2,
+  DDOG_METHOD_DELETE = 3,
+  DDOG_METHOD_PATCH = 4,
+  DDOG_METHOD_HEAD = 5,
+  DDOG_METHOD_OPTIONS = 6,
+  DDOG_METHOD_TRACE = 7,
+  DDOG_METHOD_CONNECT = 8,
+  DDOG_METHOD_OTHER = 9,
+} ddog_Method;
+
 typedef enum ddog_MetricKind {
   DDOG_METRIC_KIND_COUNT,
   DDOG_METRIC_KIND_GAUGE,
@@ -1007,19 +1020,6 @@ typedef enum ddog_DynamicInstrumentationConfigState {
   DDOG_DYNAMIC_INSTRUMENTATION_CONFIG_STATE_NOT_SET,
 } ddog_DynamicInstrumentationConfigState;
 
-typedef enum ddog_Method {
-  DDOG_METHOD_GET = 0,
-  DDOG_METHOD_POST = 1,
-  DDOG_METHOD_PUT = 2,
-  DDOG_METHOD_DELETE = 3,
-  DDOG_METHOD_PATCH = 4,
-  DDOG_METHOD_HEAD = 5,
-  DDOG_METHOD_OPTIONS = 6,
-  DDOG_METHOD_TRACE = 7,
-  DDOG_METHOD_CONNECT = 8,
-  DDOG_METHOD_OTHER = 9,
-} ddog_Method;
-
 typedef struct ddog_AgentInfoReader ddog_AgentInfoReader;
 
 typedef struct ddog_AgentRemoteConfigReader ddog_AgentRemoteConfigReader;

components-rs/common.h

@@ -153,6 +153,15 @@ void ddog_sidecar_telemetry_addDependency_buffer(struct ddog_SidecarActionsBuffe
                                                  ddog_CharSlice dependency_name,
                                                  ddog_CharSlice dependency_version);
 
+/**
+ * Enqueues an endpoint into a telemetry actions buffer (to be sent via ddog_sidecar_telemetry_buffer_flush).
+ */
+void ddog_sidecar_telemetry_addEndpoint_buffer(struct ddog_SidecarActionsBuffer *buffer,
+                                               enum ddog_Method method,
+                                               ddog_CharSlice path,
+                                               ddog_CharSlice operation_name,
+                                               ddog_CharSlice resource_name);
+
 void ddog_sidecar_telemetry_enqueueConfig_buffer(struct ddog_SidecarActionsBuffer *buffer,
                                                  ddog_CharSlice config_key,
                                                  ddog_CharSlice config_value,

components-rs/ddtrace.h

@@ -1,11 +1,10 @@
 use crate::log::Log;
-use datadog_sidecar::service::telemetry::path_for_telemetry;
+use datadog_sidecar::service::telemetry::{path_for_telemetry, TelemetryCachedClientShmData};
 
 use hashbrown::{Equivalent, HashMap};
-use std::collections::HashSet;
 use std::ffi::CString;
 use std::path::PathBuf;
-use std::time::{Duration, SystemTime};
+use std::time::Duration;
 
 use datadog_ipc::platform::NamedShmHandle;
 use datadog_sidecar::one_way_shared_memory::{open_named_shm, OneWayShmReader};
@@ -126,6 +125,24 @@ pub unsafe extern "C" fn ddog_sidecar_telemetry_addDependency_buffer(
     buffer.buffer.push(SidecarAction::Telemetry(action));
 }
 
+/// Enqueues an endpoint into a telemetry actions buffer (to be sent via ddog_sidecar_telemetry_buffer_flush).
+#[no_mangle]
+pub unsafe extern "C" fn ddog_sidecar_telemetry_addEndpoint_buffer(
+    buffer: &mut SidecarActionsBuffer,
+    method: data::Method,
+    path: CharSlice,
+    operation_name: CharSlice,
+    resource_name: CharSlice,
+) {
+    let action = TelemetryActions::AddEndpoint(data::Endpoint {
+        method: Some(method),
+        path: Some(path.to_utf8_lossy().into_owned()),
+        operation_name: operation_name.to_utf8_lossy().into_owned(),
+        resource_name: resource_name.to_utf8_lossy().into_owned(),
+    });
+    buffer.buffer.push(SidecarAction::Telemetry(action));
+}
+
 #[no_mangle]
 pub unsafe extern "C" fn ddog_sidecar_telemetry_enqueueConfig_buffer(
     buffer: &mut SidecarActionsBuffer,
@@ -237,10 +254,7 @@ pub unsafe extern "C" fn ddog_sidecar_telemetry_add_integration_log_buffer(
 }
 
 pub struct ShmCache {
-    pub config_sent: bool,
-    pub integrations: HashSet<String>,
-    pub composer_paths: HashSet<PathBuf>,
-    pub last_endpoints_push: SystemTime,
+    pub shared: TelemetryCachedClientShmData,
     pub reader: OneWayShmReader<NamedShmHandle, CString>,
 }
 
@@ -269,7 +283,7 @@ pub unsafe extern "C" fn ddog_sidecar_telemetry_config_sent(
     service: CharSlice,
     env: CharSlice,
 ) -> bool {
-    ddog_sidecar_telemetry_cache_get_or_update(cache, service, env).config_sent
+    ddog_sidecar_telemetry_cache_get_or_update(cache, service, env).shared.config_sent
 }
 
 unsafe fn ddog_sidecar_telemetry_cache_get_or_update<'a>(
@@ -287,21 +301,13 @@ unsafe fn ddog_sidecar_telemetry_cache_get_or_update<'a>(
                 if changed {
                     buf = newbuf;
                 } else {
-                    cache.config_sent = false;
-                    cache.integrations.clear();
-                    cache.composer_paths.clear();
-                    cache.last_endpoints_push = SystemTime::UNIX_EPOCH;
+                    cache.shared = TelemetryCachedClientShmData::default();
                     return;
                 }
             }
 
-            if let Ok((config_sent, integrations, composer_paths, last_endpoints_push)) =
-                bincode::deserialize::<(bool, HashSet<String>, HashSet<PathBuf>, SystemTime)>(buf)
-            {
-                cache.config_sent = config_sent;
-                cache.integrations = integrations;
-                cache.composer_paths = composer_paths;
-                cache.last_endpoints_push = last_endpoints_push;
+            if let Ok(shared) = bincode::deserialize::<TelemetryCachedClientShmData>(buf) {
+                cache.shared = shared;
             }
         }
     }
@@ -319,10 +325,7 @@ unsafe fn ddog_sidecar_telemetry_cache_get_or_update<'a>(
     let reader = OneWayShmReader::<NamedShmHandle, _>::new(open_named_shm(&shm_path).ok(), shm_path);
     let cached_entry = cache.entry(ShmCacheKey(service_str.into(), env_str.into())).insert(ShmCache {
         reader,
-        config_sent: false,
-        integrations: HashSet::new(),
-        composer_paths: HashSet::new(),
-        last_endpoints_push: SystemTime::UNIX_EPOCH,
+        shared: TelemetryCachedClientShmData::default(),
     }).into_mut();
 
     refresh_cache(cached_entry);
@@ -347,10 +350,10 @@ pub unsafe extern "C" fn ddog_sidecar_telemetry_filter_flush(
         .into_iter()
         .filter(|action| match action {
             SidecarAction::Telemetry(TelemetryActions::AddIntegration(integration)) => {
-                !cache_entry.integrations.contains(&integration.name)
+                !cache_entry.shared.integrations.contains(integration)
             }
             SidecarAction::PhpComposerTelemetryFile(path) => {
-                !cache_entry.composer_paths.contains(path)
+                !cache_entry.shared.composer_paths.contains(path)
             }
             _ => true,
         })
@@ -374,6 +377,5 @@ pub unsafe extern "C" fn ddog_sidecar_telemetry_are_endpoints_collected(
     env: CharSlice,
 ) -> bool {
     let cache_entry = ddog_sidecar_telemetry_cache_get_or_update(cache, service, env);
-    let result = cache_entry.last_endpoints_push.elapsed().map_or(false, |d| d < Duration::from_secs(1800)); // 30 minutes
-    result
+    cache_entry.shared.last_endpoints_push.elapsed().map_or(false, |d| d < Duration::from_secs(1800)) // 30 minutes
 }

components-rs/telemetry.rs

@@ -2811,6 +2811,10 @@ PHP_FUNCTION(DDTrace_dogstatsd_set) {
 PHP_FUNCTION(DDTrace_are_endpoints_collected) {
     UNUSED(execute_data);
 
+    if (!ddtrace_sidecar || !ddtrace_sidecar_instance_id || !DDTRACE_G(sidecar_queue_id)) {
+        RETURN_TRUE; // Skip overhead if unnecessary
+    }
+
     if (!DDTRACE_G(last_service_name) || !DDTRACE_G(last_env_name)) {
         RETURN_FALSE;
     }
@@ -2872,25 +2876,35 @@ PHP_FUNCTION(DDTrace_add_endpoint) {
     ddog_CharSlice operation_name_slice = dd_zend_string_to_CharSlice(operation_name);
     ddog_CharSlice resource_name_slice = dd_zend_string_to_CharSlice(resource_name);
 
-    ddog_MaybeError result = ddog_sidecar_telemetry_addEndpoint(
-        &ddtrace_sidecar, ddtrace_sidecar_instance_id, &DDTRACE_G(sidecar_queue_id), method_enum, path_slice, operation_name_slice,
-        resource_name_slice);
-
     LOG_LINE(DEBUG,
              "Adding endpoint: %.*s (%zu) - operation_name: %.*s (%zu) - resource_name: %.*s (%zu) - method: %.*s (%zu)",
              (int)path_slice.len, (char *)path_slice.ptr, path_slice.len,
              (int)operation_name_slice.len, (char *)operation_name_slice.ptr, operation_name_slice.len,
              (int)resource_name_slice.len, (char *)resource_name_slice.ptr, resource_name_slice.len,
              (int)method->len, ZSTR_VAL(method), (size_t)method->len);
 
-    if (result.tag == DDOG_OPTION_ERROR_SOME_ERROR) {
-        ddog_CharSlice message = ddog_Error_message(&result.some);
-        LOG_LINE(ERROR, "Error submitting  endpoint to sidecar: %.*s", (int)message.len, (char *)message.ptr);
-        ZVAL_FALSE(return_value);
-    } else {
-        ZVAL_TRUE(return_value);
+    ddog_sidecar_telemetry_addEndpoint_buffer(ddtrace_telemetry_buffer(), method_enum, path_slice, operation_name_slice, resource_name_slice);
+
+    RETURN_TRUE;
+}
+
+PHP_FUNCTION(DDTrace_flush_endpoints) {
+    UNUSED(execute_data);
+    UNUSED(return_value);
+
+    if (!ddtrace_sidecar || !ddtrace_sidecar_instance_id || !DDTRACE_G(sidecar_queue_id) || !DDTRACE_G(telemetry_buffer)) {
+        return;
+    }
+
+    if (!DDTRACE_G(last_service_name) || !DDTRACE_G(last_env_name)) {
+        return;
     }
-    ddog_MaybeError_drop(result);
+
+    ddog_CharSlice service_name = dd_zend_string_to_CharSlice(DDTRACE_G(last_service_name));
+    ddog_CharSlice env_name = dd_zend_string_to_CharSlice(DDTRACE_G(last_env_name));
+
+    ddtrace_ffi_try("Failed flushing endpoint telemetry buffer",
+        ddog_sidecar_telemetry_filter_flush(&ddtrace_sidecar, ddtrace_sidecar_instance_id, &DDTRACE_G(sidecar_queue_id), ddtrace_telemetry_buffer(), ddtrace_telemetry_cache(), service_name, env_name));
 }
 
 PHP_FUNCTION(dd_trace_send_traces_via_thread) {

ext/ddtrace.c

@@ -839,6 +839,12 @@ function are_endpoints_collected(): bool {}
      * @param string $method The method of the endpoint
      */
     function add_endpoint(string $path, string $operation_name, string $resource_name, string $method): bool {}
+
+    /**
+     * Flush all buffered endpoints to the sidecar immediately.
+     * Call this once after batching all add_endpoint() calls.
+     */
+    function flush_endpoints(): void {}
 }
 
 namespace DDTrace\System {