In-process log submission in sidecar (for appsec helper)

Author: cataphract

datadog-sidecar/Cargo.toml

@@ -104,6 +104,9 @@ datadog-remote-config = { path = "../datadog-remote-config", features = ["test"]
 datadog-trace-utils = { path = "../datadog-trace-utils", features = ["test-utils"] }
 datadog-dynamic-configuration = { path = "../datadog-dynamic-configuration", features = ["test"] }
 
+[build-dependencies]
+cbindgen = "0.26"
+
 [lints.rust]
 unexpected_cfgs = { level = "warn", check-cfg = [
     'cfg(tokio_taskdump,windows_seh_wrapper)',

datadog-sidecar/cbindgen.toml

@@ -0,0 +1,10 @@
+# Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/
+# SPDX-License-Identifier: Apache-2.0
+
+language = "C"
+
+include_guard = "DDOG_TELEMETRY_API_H"
+autogen_warning = "/* Warning, this file is autogenerated by cbindgen. Don't modify this manually. */"
+
+[export]
+item_types = ["functions", "structs", "enums"]

datadog-sidecar/include/sidecar_ffi.h

@@ -0,0 +1,47 @@
+#ifndef DDOG_TELEMETRY_API_H
+#define DDOG_TELEMETRY_API_H
+
+/* Warning, this file is autogenerated by cbindgen. Don't modify this manually. */
+
+#include <stdarg.h>
+#include <stdbool.h>
+#include <stdint.h>
+#include <stdlib.h>
+
+typedef enum CLogLevel {
+  Error = 1,
+  Warn = 2,
+  Debug = 3,
+} CLogLevel;
+
+typedef enum FfiError {
+  Ok = 0,
+  PointerNull = 1,
+  OperationFailed = 2,
+  InvalidUtf8 = 3,
+  QueueFull = 4,
+} FfiError;
+
+typedef struct FfiString {
+  const char *ptr;
+  uintptr_t len;
+} FfiString;
+
+/**
+ * Enqueues a telemetry log action to be processed internally.
+ * Non-blocking. Logs might be dropped if the internal queue is full.
+ *
+ * # Safety
+ * Pointers must be valid, strings must be null-terminated if not null.
+ */
+enum FfiError ddog_sidecar_enqueue_telemetry_log(struct FfiString session_id_ffi,
+                                                 struct FfiString runtime_id_ffi,
+                                                 uint64_t queue_id,
+                                                 struct FfiString identifier_ffi,
+                                                 enum CLogLevel level,
+                                                 struct FfiString message_ffi,
+                                                 struct FfiString *stack_trace_ffi,
+                                                 struct FfiString *tags_ffi,
+                                                 bool is_sensitive);
+
+#endif  /* DDOG_TELEMETRY_API_H */

datadog-sidecar/src/entry.rs

@@ -27,6 +27,7 @@ use crate::setup::{self, IpcClient, IpcServer, Liaison};
 
 use crate::config::{self, Config};
 use crate::self_telemetry::self_telemetry;
+use crate::service::telemetry_action_receiver_task;
 use crate::tracer::get_shm_limiter;
 use crate::watchdog::Watchdog;
 use crate::{ddog_daemon_entry_point, setup_daemon_process};
@@ -91,6 +92,9 @@ where
     drop(get_shm_limiter().lock());
 
     let server = SidecarServer::default();
+
+    tokio::spawn(telemetry_action_receiver_task(server.clone()));
+
     let (shutdown_complete_tx, shutdown_complete_rx) = mpsc::channel::<()>(1);
 
     let watchdog_handle =

datadog-sidecar/src/service/mod.rs

@@ -44,6 +44,8 @@ pub(crate) mod sidecar_server;
 mod telemetry;
 pub(crate) mod tracing;
 
+pub(crate) use telemetry::telemetry_action_receiver_task;
+
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct SessionConfig {
     pub endpoint: Endpoint,

datadog-sidecar/src/service/queue_id.rs

@@ -9,7 +9,7 @@ use serde::{Deserialize, Serialize};
 #[derive(Default, Copy, Clone, Hash, PartialEq, Eq, Debug, Serialize, Deserialize)]
 #[repr(transparent)]
 pub struct QueueId {
-    inner: u64,
+    pub(crate) inner: u64,
 }
 
 impl QueueId {

datadog-sidecar/src/service/sidecar_server.rs

@@ -109,7 +109,7 @@ pub struct SidecarServer {
     pub self_telemetry_config:
         Arc<Mutex<Option<ManualFutureCompleter<ddtelemetry::config::Config>>>>,
     /// Keeps track of the number of submitted payloads.
-    pub submitted_payloads: Arc<AtomicU64>,
+    pub(crate) submitted_payloads: Arc<AtomicU64>,
     /// All tracked agent infos per endpoint
     pub agent_infos: AgentInfos,
     /// All remote config handling
@@ -224,7 +224,7 @@ impl SidecarServer {
         }
     }
 
-    fn get_session(&self, session_id: &String) -> SessionInfo {
+    pub(crate) fn get_session(&self, session_id: &String) -> SessionInfo {
         let mut sessions = self.sessions.lock_or_panic();
         match sessions.get(session_id) {
             Some(session) => session.clone(),

datadog-sidecar/src/service/telemetry/ffi.rs

@@ -0,0 +1,189 @@
+// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/
+// SPDX-License-Identifier: Apache-2.0
+#![allow(clippy::missing_safety_doc)]
+
+use crate::service::{InstanceId, QueueId};
+use ddtelemetry::data::{Log as DDLog, LogLevel as DDLogLevel};
+use ddtelemetry::worker::{LogIdentifier, TelemetryActions};
+use std::collections::hash_map::DefaultHasher;
+use std::ffi::CStr;
+use std::hash::{Hash, Hasher};
+use std::ptr::NonNull;
+use tokio::sync::mpsc;
+use tracing::{error, warn};
+
+use super::{get_telemetry_action_sender, InternalTelemetryActions};
+
+#[repr(C)]
+pub enum FfiError {
+    Ok = 0,
+    PointerNull = 1,
+    OperationFailed = 2,
+    InvalidUtf8 = 3,
+    QueueFull = 4,
+}
+
+#[repr(C)]
+pub enum CLogLevel {
+    Error = 1,
+    Warn = 2,
+    Debug = 3,
+}
+
+impl From<CLogLevel> for DDLogLevel {
+    fn from(level: CLogLevel) -> Self {
+        match level {
+            CLogLevel::Error => DDLogLevel::Error,
+            CLogLevel::Warn => DDLogLevel::Warn,
+            CLogLevel::Debug => DDLogLevel::Debug,
+        }
+    }
+}
+
+#[repr(C)]
+pub struct FfiString {
+    ptr: *const libc::c_char,
+    len: usize,
+}
+
+impl FfiString {
+    fn is_empty(&self) -> bool {
+        self.len == 0 || self.ptr.is_null()
+    }
+}
+
+impl TryFrom<&FfiString> for String {
+    type Error = FfiError;
+
+    fn try_from(vref: &FfiString) -> Result<Self, Self::Error> {
+        if vref.ptr.is_null() {
+            return Err(FfiError::PointerNull);
+        }
+        let cstr = unsafe {
+            CStr::from_ptr(vref.ptr)
+                .to_str()
+                .map_err(|_| FfiError::InvalidUtf8)
+        }?;
+        Ok(cstr.to_string())
+    }
+}
+
+impl TryFrom<FfiString> for String {
+    type Error = FfiError;
+
+    fn try_from(value: FfiString) -> Result<Self, Self::Error> {
+        String::try_from(&value)
+    }
+}
+
+/// Enqueues a telemetry log action to be processed internally.
+/// Non-blocking. Logs might be dropped if the internal queue is full.
+///
+/// # Safety
+/// Pointers must be valid, strings must be null-terminated if not null.
+#[no_mangle]
+pub unsafe extern "C" fn ddog_sidecar_enqueue_telemetry_log(
+    session_id_ffi: FfiString,
+    runtime_id_ffi: FfiString,
+    queue_id: u64,
+    identifier_ffi: FfiString,
+    level: CLogLevel,
+    message_ffi: FfiString,
+    stack_trace_ffi: Option<NonNull<FfiString>>,
+    tags_ffi: Option<NonNull<FfiString>>,
+    is_sensitive: bool,
+) -> FfiError {
+    match ddog_sidecar_enqueue_telemetry_log_result(
+        session_id_ffi,
+        runtime_id_ffi,
+        queue_id,
+        identifier_ffi,
+        level,
+        message_ffi,
+        stack_trace_ffi,
+        tags_ffi,
+        is_sensitive,
+    ) {
+        Ok(result) => result,
+        Err(e) => e,
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+fn ddog_sidecar_enqueue_telemetry_log_result(
+    session_id_ffi: FfiString,
+    runtime_id_ffi: FfiString,
+    queue_id: u64,
+    identifier_ffi: FfiString,
+    level: CLogLevel,
+    message_ffi: FfiString,
+    stack_trace_ffi: Option<NonNull<FfiString>>,
+    tags_ffi: Option<NonNull<FfiString>>,
+    is_sensitive: bool,
+) -> Result<FfiError, FfiError> {
+    if session_id_ffi.is_empty()
+        || runtime_id_ffi.is_empty()
+        || queue_id == 0
+        || identifier_ffi.is_empty()
+        || message_ffi.is_empty()
+    {
+        return Err(FfiError::PointerNull);
+    }
+
+    let sender = match get_telemetry_action_sender() {
+        Ok(s) => s,
+        Err(e) => {
+            error!("Failed to get telemetry action sender: {}", e);
+            return Err(FfiError::OperationFailed);
+        }
+    };
+
+    let instance_id = InstanceId::new(
+        String::try_from(session_id_ffi)?,
+        String::try_from(runtime_id_ffi)?,
+    );
+    let queue_id = QueueId { inner: queue_id };
+    let identifier: String = identifier_ffi.try_into()?;
+    let message: String = message_ffi.try_into()?;
+
+    let stack_trace = stack_trace_ffi
+        .map(|s| String::try_from(unsafe { s.as_ref() }))
+        .transpose()?;
+    let tags: Option<String> = tags_ffi
+        .map(|s| String::try_from(unsafe { s.as_ref() }))
+        .transpose()?;
+
+    let mut hasher = DefaultHasher::new();
+    identifier.hash(&mut hasher);
+    let log_id = LogIdentifier {
+        indentifier: hasher.finish(),
+    };
+
+    let log_data = DDLog {
+        message,
+        level: level.into(),
+        stack_trace,
+        count: 1,
+        tags: tags.unwrap_or("".into()),
+        is_sensitive,
+    };
+    let log_action = TelemetryActions::AddLog((log_id, log_data));
+
+    let msg = InternalTelemetryActions {
+        instance_id,
+        queue_id,
+        actions: vec![log_action],
+    };
+
+    match sender.try_send(msg) {
+        Ok(_) => Ok(FfiError::Ok),
+        Err(mpsc::error::TrySendError::Full(_)) => {
+            warn!("Telemetry action queue full. Action dropped.");
+            Err(FfiError::QueueFull)
+        }
+        Err(mpsc::error::TrySendError::Closed(_)) => {
+            error!("Telemetry action receiver closed.");
+            Err(FfiError::OperationFailed)
+        }
+    }
+}