Use a sidecar connection per PHP thread

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

@@ -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);
 

components-rs/common.h

@@ -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/ddtrace.h

@@ -19,8 +19,6 @@ extern ddog_VecRemoteConfigCapabilities DDTRACE_REMOTE_CONFIG_CAPABILITIES;
 
 extern const uint8_t *DDOG_PHP_FUNCTION;
 
-extern struct ddog_SidecarTransport *ddtrace_sidecar;
-
 /**
  * # Safety
  * Must be called from a single-threaded context, such as MINIT.
@@ -153,6 +151,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/sidecar.rs

@@ -170,10 +170,6 @@ pub extern "C" fn ddog_sidecar_connect_php(
     MaybeError::None
 }
 
-#[no_mangle]
-#[allow(non_upper_case_globals)]
-pub static mut ddtrace_sidecar: *mut SidecarTransport = std::ptr::null_mut();
-
 #[no_mangle]
 pub extern "C" fn ddtrace_sidecar_reconnect(
     transport: &mut Box<SidecarTransport>,

components-rs/telemetry.rs

@@ -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
 }

ext/auto_flush.c

@@ -47,7 +47,7 @@ ZEND_RESULT_CODE ddtrace_flush_tracer(bool force_on_startup, bool collect_cycles
     char *url = ddtrace_agent_url();
 
     if (get_global_DD_TRACE_SIDECAR_TRACE_SENDER()) {
-        if (ddtrace_sidecar) {
+        if (DDTRACE_G(sidecar)) {
             ddog_SenderParameters parameters = {
                 .tracer_headers_tags = {
                     .container_id = ddtrace_get_container_id(),
@@ -59,7 +59,7 @@ ZEND_RESULT_CODE ddtrace_flush_tracer(bool force_on_startup, bool collect_cycles
                     .client_computed_top_level = false,
                     .client_computed_stats = !get_global_DD_APM_TRACING_ENABLED(),
                 },
-                .transport = ddtrace_sidecar,
+                .transport = DDTRACE_G(sidecar),
                 .instance_id = ddtrace_sidecar_instance_id,
                 .limit = limit,
                 .n_requests = get_global_DD_TRACE_AGENT_FLUSH_AFTER_N_REQUESTS(),