feat(telemetry)!: add process_tags to Application in telemetry (#1459)

# What does this PR do?

This PR adds a `process_tags` field to the telemetry `Application` struct. This tag can be set during configuration or later (when the process_tags are acquired).

# Motivation

`dd-trace-php` needs to be send the process_tags for its telemetry according to this [RFC](https://docs.google.com/document/d/1AFdLUuVk70i0bJd5335-RxqsvwAV9ovAqcO2z5mEMbA/edit?tab=t.0#heading=h.s9l1lctqlg11). 
It is important to note that the tracer will set the configuration of its sidecar before we can retrieve the process_tags. That's why I created a function specifically to set the processs_tags.

# How to test the change?

The change is tested in dd-trace-php through this PR: DataDog/dd-trace-php#3627


Co-authored-by: louis.tricot <louis.tricot@datadoghq.com>

Author: dubloom

datadog-sidecar-ffi/src/lib.rs

@@ -588,6 +588,7 @@ pub unsafe extern "C" fn ddog_sidecar_session_set_config(
     remote_config_capabilities_count: usize,
     remote_config_enabled: bool,
     is_fork: bool,
+    process_tags: ffi::CharSlice,
 ) -> MaybeError {
     #[cfg(unix)]
     let remote_config_notify_target = libc::getpid();
@@ -631,13 +632,31 @@ pub unsafe extern "C" fn ddog_sidecar_session_set_config(
             .as_slice()
             .to_vec(),
             remote_config_enabled,
+            process_tags: process_tags.to_utf8_lossy().into(),
         },
         is_fork
     ));
 
     MaybeError::None
 }
 
+/// Updates the process_tags for an existing session.
+#[no_mangle]
+#[allow(clippy::missing_safety_doc)]
+pub unsafe extern "C" fn ddog_sidecar_session_set_process_tags(
+    transport: &mut Box<SidecarTransport>,
+    session_id: ffi::CharSlice,
+    process_tags: ffi::CharSlice,
+) -> MaybeError {
+    try_c!(blocking::set_session_process_tags(
+        transport,
+        session_id.to_utf8_lossy().into(),
+        process_tags.to_utf8_lossy().into(),
+    ));
+
+    MaybeError::None
+}
+
 #[repr(C)]
 pub struct TracerHeaderTags<'a> {
     pub lang: ffi::CharSlice<'a>,

datadog-sidecar-ffi/tests/sidecar.rs

@@ -112,6 +112,7 @@ fn test_ddog_sidecar_register_app() {
             0,
             false,
             false,
+            "".into(),
         )
         .unwrap_none();
 
@@ -162,6 +163,7 @@ fn test_ddog_sidecar_register_app() {
             0,
             false,
             false,
+            "".into(),
         )
         .unwrap_none();
 

datadog-sidecar/src/service/blocking.rs

@@ -243,6 +243,28 @@ pub fn set_session_config(
     })
 }
 
+/// Updates the process tags for an existing session.
+///
+/// # Arguments
+///
+/// * `transport` - The transport used for communication.
+/// * `session_id` - The ID of the session.
+/// * `process_tags` - The process tags string to set.
+///
+/// # Returns
+///
+/// An `io::Result<()>` indicating the result of the operation.
+pub fn set_session_process_tags(
+    transport: &mut SidecarTransport,
+    session_id: String,
+    process_tags: String,
+) -> io::Result<()> {
+    transport.send(SidecarInterfaceRequest::SetSessionProcessTags {
+        session_id,
+        process_tags,
+    })
+}
+
 /// Sends a trace as bytes.
 ///
 /// # Arguments

datadog-sidecar/src/service/mod.rs

@@ -64,6 +64,7 @@ pub struct SessionConfig {
     pub remote_config_products: Vec<RemoteConfigProduct>,
     pub remote_config_capabilities: Vec<RemoteConfigCapabilities>,
     pub remote_config_enabled: bool,
+    pub process_tags: String,
 }
 
 #[derive(Debug, Deserialize, Serialize)]

datadog-sidecar/src/service/session_info.rs

@@ -43,6 +43,7 @@ pub(crate) struct SessionInfo {
     pub(crate) session_id: String,
     pub(crate) pid: Arc<AtomicI32>,
     pub(crate) remote_config_enabled: Arc<Mutex<bool>>,
+    pub(crate) process_tags: Arc<Mutex<Option<String>>>,
 }
 
 impl Clone for SessionInfo {
@@ -62,6 +63,7 @@ impl Clone for SessionInfo {
             session_id: self.session_id.clone(),
             pid: self.pid.clone(),
             remote_config_enabled: self.remote_config_enabled.clone(),
+            process_tags: self.process_tags.clone(),
         }
     }
 }

datadog-sidecar/src/service/sidecar_interface.rs

@@ -67,6 +67,14 @@ pub trait SidecarInterface {
         is_fork: bool,
     );
 
+    /// Updates the process tags for an existing session.
+    ///
+    /// # Arguments
+    ///
+    /// * `session_id` - The ID of the session.
+    /// * `process_tags` - The process tags string.
+    async fn set_session_process_tags(session_id: String, process_tags: String);
+
     /// Shuts down a runtime.
     ///
     /// # Arguments

datadog-sidecar/src/service/sidecar_server.rs

@@ -421,6 +421,8 @@ impl SidecarInterface for SidecarServer {
                 .unwrap_or("unknown-service");
             let env = entry.get().env.as_deref().unwrap_or("none");
 
+            let process_tags = session.process_tags.lock_or_panic().clone();
+
             // Lock telemetry client
             let telemetry_mutex = self.telemetry_clients.get_or_create(
                 service,
@@ -438,6 +440,7 @@ impl SidecarInterface for SidecarServer {
                             Config::default()
                         })
                 },
+                process_tags,
             );
             let mut telemetry = telemetry_mutex.lock_or_panic();
 
@@ -565,6 +568,8 @@ impl SidecarInterface for SidecarServer {
             *session.remote_config_notify_function.lock().unwrap() = remote_config_notify_function;
         }
         *session.remote_config_enabled.lock_or_panic() = config.remote_config_enabled;
+        *session.process_tags.lock_or_panic() =
+            (!config.process_tags.is_empty()).then_some(config.process_tags.clone());
         session.modify_telemetry_config(|cfg| {
             cfg.telemetry_heartbeat_interval = config.telemetry_heartbeat_interval;
             let endpoint = get_product_endpoint(
@@ -658,6 +663,19 @@ impl SidecarInterface for SidecarServer {
         })
     }
 
+    type SetSessionProcessTagsFut = NoResponse;
+
+    fn set_session_process_tags(
+        self,
+        _: Context,
+        session_id: String,
+        process_tags: String,
+    ) -> Self::SetSessionProcessTagsFut {
+        let session = self.get_session(&session_id);
+        *session.process_tags.lock_or_panic() = (!process_tags.is_empty()).then_some(process_tags);
+        no_response()
+    }
+
     type ShutdownRuntimeFut = NoResponse;
 
     fn shutdown_runtime(self, _: Context, instance_id: InstanceId) -> Self::ShutdownRuntimeFut {

datadog-sidecar/src/service/telemetry.rs

@@ -77,6 +77,7 @@ impl TelemetryCachedClient {
         instance_id: &InstanceId,
         runtime_meta: &RuntimeMetadata,
         get_config: impl FnOnce() -> libdd_telemetry::config::Config,
+        process_tags: Option<String>,
     ) -> Self {
         let mut builder = TelemetryWorkerBuilder::new_fetch_host(
             service.to_string(),
@@ -87,6 +88,7 @@ impl TelemetryCachedClient {
 
         builder.runtime_id = Some(instance_id.runtime_id.clone());
         builder.application.env = Some(env.to_string());
+        builder.application.process_tags = process_tags;
         let config = get_config();
         builder.config = config.clone();
 
@@ -298,6 +300,7 @@ impl TelemetryCachedClientSet {
         instance_id: &InstanceId,
         runtime_meta: &RuntimeMetadata,
         get_config: F,
+        process_tags: Option<String>,
     ) -> Arc<Mutex<TelemetryCachedClient>>
     where
         F: FnOnce() -> libdd_telemetry::config::Config,
@@ -330,6 +333,7 @@ impl TelemetryCachedClientSet {
                 instance_id,
                 runtime_meta,
                 get_config,
+                process_tags,
             ))),
         };
 
@@ -487,12 +491,15 @@ fn get_telemetry_client(
             })
     };
 
+    let process_tags = session.process_tags.lock_or_panic().clone();
+
     TelemetryCachedClientSet::get_or_create(
         &sidecar.telemetry_clients,
         service_name,
         env_name,
         instance_id,
         &runtime_meta,
         get_config,
+        process_tags,
     )
 }

libdd-telemetry/examples/tm-ping.rs

@@ -75,6 +75,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         runtime_name: None,
         runtime_version: None,
         runtime_patches: None,
+        process_tags: None,
     };
     let host = build_host();
     let payload = data::payload::Payload::AppStarted(build_app_started_payload());

libdd-telemetry/examples/tm-send-sketch.rs

@@ -78,6 +78,7 @@ async fn async_main() {
         runtime_name: None,
         runtime_version: None,
         runtime_patches: None,
+        process_tags: None,
     };
     let host = build_host();