Fix in-process submission of telemetry (#1172)

Author: cataphract

datadog-sidecar-ffi/src/lib.rs

@@ -660,7 +660,8 @@ impl<'a> TryInto<SerializedTracerHeaderTags> for &'a TracerHeaderTags<'a> {
 pub unsafe extern "C" fn ddog_sidecar_enqueue_telemetry_log(
     session_id_ffi: CharSlice,
     runtime_id_ffi: CharSlice,
-    queue_id: u64,
+    service_name_ffi: CharSlice,
+    env_name_ffi: CharSlice,
     identifier_ffi: CharSlice,
     level: ddtelemetry::data::LogLevel,
     message_ffi: CharSlice,
@@ -671,7 +672,8 @@ pub unsafe extern "C" fn ddog_sidecar_enqueue_telemetry_log(
     try_c!(ddog_sidecar_enqueue_telemetry_log_impl(
         session_id_ffi,
         runtime_id_ffi,
-        queue_id,
+        service_name_ffi,
+        env_name_ffi,
         identifier_ffi,
         level,
         message_ffi,
@@ -695,7 +697,8 @@ fn char_slice_to_string(slice: CharSlice) -> Result<String, String> {
 fn ddog_sidecar_enqueue_telemetry_log_impl(
     session_id_ffi: CharSlice,
     runtime_id_ffi: CharSlice,
-    queue_id: u64,
+    service_name_ffi: CharSlice,
+    env_name_ffi: CharSlice,
     identifier_ffi: CharSlice,
     level: ddtelemetry::data::LogLevel,
     message_ffi: CharSlice,
@@ -705,7 +708,8 @@ fn ddog_sidecar_enqueue_telemetry_log_impl(
 ) -> Result<(), String> {
     if session_id_ffi.is_empty()
         || runtime_id_ffi.is_empty()
-        || queue_id == 0
+        || service_name_ffi.is_empty()
+        || env_name_ffi.is_empty()
         || identifier_ffi.is_empty()
         || message_ffi.is_empty()
     {
@@ -723,7 +727,8 @@ fn ddog_sidecar_enqueue_telemetry_log_impl(
         char_slice_to_string(session_id_ffi)?,
         char_slice_to_string(runtime_id_ffi)?,
     );
-    let queue_id: QueueId = queue_id.into();
+    let service_name: String = char_slice_to_string(service_name_ffi)?;
+    let env_name: String = char_slice_to_string(env_name_ffi)?;
     let identifier: String = char_slice_to_string(identifier_ffi)?;
     let message: String = char_slice_to_string(message_ffi)?;
 
@@ -753,7 +758,8 @@ fn ddog_sidecar_enqueue_telemetry_log_impl(
 
     let msg = InternalTelemetryActions {
         instance_id,
-        queue_id,
+        service_name,
+        env_name,
         actions: vec![log_action],
     };
 

datadog-sidecar/src/service/sidecar_server.rs

@@ -99,7 +99,7 @@ pub struct SidecarServer {
     /// A `Mutex` guarded `HashMap` that keeps a count of each session.
     session_counter: Arc<Mutex<HashMap<String, u32>>>,
     /// A `Mutex` guarded `HashMap` that stores the active telemetry clients.
-    telemetry_clients: TelemetryCachedClientSet,
+    pub(crate) telemetry_clients: TelemetryCachedClientSet,
     /// A `Mutex` guarded optional `ManualFutureCompleter` for telemetry configuration.
     pub self_telemetry_config:
         Arc<Mutex<Option<ManualFutureCompleter<ddtelemetry::config::Config>>>>,
@@ -359,98 +359,6 @@ impl SidecarServer {
         }
     }
 
-    pub async fn process_telemetry_action(
-        &self,
-        instance_id: &InstanceId,
-        queue_id: &QueueId,
-        actions: Vec<TelemetryActions>,
-    ) -> anyhow::Result<()> {
-        tracing::debug!(
-            "Processing telemetry action for target {:?}/{:?}: {:?}",
-            instance_id,
-            queue_id,
-            actions
-        );
-        let session = self.get_session(&instance_id.session_id);
-        let trace_config = session.get_trace_config();
-        let runtime_metadata = RuntimeMetadata::new(
-            trace_config.language.clone(),
-            trace_config.language_version.clone(),
-            trace_config.tracer_version.clone(),
-        );
-
-        let rt_info = self.get_runtime(instance_id);
-        let mut applications = rt_info.lock_applications();
-
-        match applications.entry(*queue_id) {
-            Entry::Occupied(mut entry) => {
-                let value = entry.get_mut();
-
-                let env = value.env.as_deref().unwrap_or("none");
-                let service = value.service_name.as_deref().unwrap_or("unknown-service");
-
-                let mut telemetry = match self.telemetry_clients.get_or_create(
-                    service,
-                    env,
-                    instance_id,
-                    &runtime_metadata,
-                    || {
-                        self.get_session(&instance_id.session_id)
-                            .session_config
-                            .lock_or_panic()
-                            .clone()
-                    },
-                ) {
-                    Some(client) => client,
-                    None => return Ok(()),
-                };
-
-                let mut actions_to_send = vec![];
-                let mut buffered_info_changed = false;
-
-                for action in actions {
-                    match action {
-                        TelemetryActions::AddIntegration(ref integration) => {
-                            if telemetry.buffered_integrations.insert(integration.clone()) {
-                                actions_to_send.push(action);
-                                buffered_info_changed = true;
-                            }
-                        }
-                        _ => {
-                            actions_to_send.push(action);
-                        }
-                    }
-                }
-
-                let client_clone = telemetry.clone();
-                let mut handle = telemetry.handle.lock_or_panic();
-                let last_handle = handle.take();
-                *handle = Some(tokio::spawn(async move {
-                    if let Some(last_handle) = last_handle {
-                        last_handle.await.ok();
-                    };
-                    debug!("Sending Telemetry Actions: {actions_to_send:?}");
-                    client_clone.client.send_msgs(actions_to_send).await.ok();
-                }));
-
-                if buffered_info_changed {
-                    info!(
-                    "Buffered telemetry info changed for instance {instance_id:?} and queue_id {queue_id:?}"
-                );
-                    telemetry.write_shm_file();
-                }
-            }
-
-            Entry::Vacant(_) => {
-                info!(
-                    "No application found for instance {instance_id:?} and queue_id {queue_id:?}"
-                );
-            }
-        }
-
-        Ok(())
-    }
-
     pub fn shutdown(&self) {
         self.remote_configs.shutdown();
     }

datadog-sidecar/src/service/telemetry.rs

@@ -1,12 +1,12 @@
 // Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::service::{InstanceId, QueueId, RuntimeMetadata, SidecarAction, SidecarServer};
+use crate::service::{InstanceId, RuntimeMetadata, SidecarAction, SidecarServer};
 use anyhow::{anyhow, Result};
 use ddcommon::MutexExt;
 use std::sync::OnceLock;
 use tokio::sync::mpsc;
-use tracing::{info, warn};
+use tracing::{debug, info, warn};
 
 use crate::one_way_shared_memory::OneWayShmWriter;
 use crate::primary_sidecar_identifier;
@@ -356,7 +356,8 @@ pub fn path_for_telemetry(service: &str, env: &str) -> CString {
 #[derive(Debug)]
 pub struct InternalTelemetryActions {
     pub instance_id: InstanceId,
-    pub queue_id: QueueId,
+    pub service_name: String,
+    pub env_name: String,
     pub actions: Vec<TelemetryActions>,
 }
 
@@ -377,16 +378,66 @@ pub(crate) async fn telemetry_action_receiver_task(sidecar: SidecarServer) {
         return;
     }
 
-    while let Some(msg) = rx.recv().await {
-        if let Err(e) = sidecar
-            .process_telemetry_action(&msg.instance_id, &msg.queue_id, msg.actions)
-            .await
-        {
+    while let Some(actions) = rx.recv().await {
+        let telemetry_client = get_telemetry_client(
+            &sidecar,
+            &actions.instance_id,
+            &actions.service_name,
+            &actions.env_name,
+        );
+
+        if let Some(telemetry_client) = telemetry_client {
+            for action in actions.actions {
+                let action_str = format!("{action:?}");
+                match telemetry_client.client.send_msg(action).await {
+                    Ok(_) => {
+                        debug!("Sent telemetry action to TelemetryWorker: {action_str}");
+                    }
+                    Err(e) => {
+                        warn!(
+                            "Failed to send telemetry action {action_str} to TelemetryWorker: {e}"
+                        );
+                    }
+                }
+            }
+        } else {
             warn!(
-                "Could not process telemetry action for target {:?}/{:?}: {}. Action dropped.",
-                msg.instance_id, msg.queue_id, e
+                "Could not get or client telemetry client for target {:?}, service {:?}, env {:?}. Action dropped.",
+                actions.instance_id, actions.service_name, actions.env_name
             );
         }
     }
     info!("Telemetry action receiver task shutting down.");
 }
+
+fn get_telemetry_client(
+    sidecar: &SidecarServer,
+    instance_id: &InstanceId,
+    service_name: &str,
+    env_name: &str,
+) -> Option<TelemetryCachedClient> {
+    let session = sidecar.get_session(&instance_id.session_id);
+    let trace_config = session.get_trace_config();
+    let runtime_meta = RuntimeMetadata::new(
+        trace_config.language.as_str(),
+        trace_config.language_version.as_str(),
+        trace_config.tracer_version.as_str(),
+    );
+
+    let get_config = || {
+        sidecar
+            .get_session(&instance_id.session_id)
+            .session_config
+            .lock_or_panic()
+            .clone()
+    };
+
+    TelemetryCachedClientSet::get_or_create(
+        &sidecar.telemetry_clients,
+        service_name,
+        env_name,
+        instance_id,
+        &runtime_meta,
+        get_config,
+    )
+}