Merge branch 'main' of github.com:DataDog/libdatadog into r1viollet/run-ffi-examples-on-windows

Author: r1viollet

Cargo.lock

@@ -65,7 +65,7 @@ resolver = "2"
 [workspace.package]
 rust-version = "1.84.1"
 edition = "2021"
-version = "25.0.0"
+version = "26.0.0"
 license = "Apache-2.0"
 authors = ["Datadog Inc. <info@datadoghq.com>"]
 

Cargo.toml

@@ -3,8 +3,7 @@
 
 use crate::handles::TransferHandles;
 use crate::platform::{Message, PlatformHandle};
-use nix::sys::select::FdSet;
-use nix::sys::time::{TimeVal, TimeValLike};
+use nix::poll::{poll, PollFd, PollFlags, PollTimeout};
 use std::{
     io::{self, ErrorKind, Read, Write},
     os::{
@@ -61,11 +60,13 @@ impl Channel {
     pub fn probe_readable(&self) -> bool {
         #[allow(clippy::unwrap_used)]
         let raw_fd = self.inner.as_owned_fd().unwrap().as_fd();
-        let mut fds = FdSet::new();
-        fds.insert(raw_fd);
-        nix::sys::select::select(None, Some(&mut fds), None, None, Some(&mut TimeVal::zero()))
-            .is_err()
-            || fds.contains(raw_fd)
+
+        let mut fds = [PollFd::new(raw_fd, PollFlags::POLLIN)];
+        poll(&mut fds, PollTimeout::ZERO).is_err()
+            || fds[0]
+                .revents()
+                .unwrap_or(PollFlags::empty())
+                .intersects(PollFlags::POLLIN | PollFlags::POLLHUP | PollFlags::POLLERR)
     }
 
     pub fn create_message<T>(&mut self, item: T) -> Result<Message<T>, io::Error>

LICENSE-3rdparty.yml

@@ -48,9 +48,12 @@ fn test_file_locking_works_as_expected() {
         .unwrap();
 
     let mut buf = [0; 10];
-    local
-        .set_read_timeout(Some(Duration::from_millis(500)))
-        .unwrap();
+    // give macOS runners on CI more time to read
+    #[cfg(target_os = "macos")]
+    let read_timeout = Duration::from_secs(10);
+    #[cfg(not(target_os = "macos"))]
+    let read_timeout = Duration::from_millis(500);
+    local.set_read_timeout(Some(read_timeout)).unwrap();
     // wait for child to signal its ready
     assert!(local.read(&mut buf).unwrap() > 0);
 

datadog-ipc/src/platform/unix/channel.rs

@@ -27,6 +27,7 @@ paste = "1"
 libc = "0.2"
 tracing = { version = "0.1", default-features = false }
 rmp-serde = "1.1.1"
+serde_json = "1.0"
 
 
 [target.'cfg(windows)'.dependencies]

datadog-ipc/tests/flock.rs

@@ -24,6 +24,7 @@ use datadog_sidecar::config::LogMethod;
 use datadog_sidecar::crashtracker::crashtracker_unix_socket_path;
 use datadog_sidecar::one_way_shared_memory::{OneWayShmReader, ReaderOpener};
 use datadog_sidecar::service::agent_info::AgentInfoReader;
+use datadog_sidecar::service::telemetry::InternalTelemetryAction;
 use datadog_sidecar::service::{
     blocking::{self, SidecarTransport},
     DynamicInstrumentationConfigState, InstanceId, QueueId, RuntimeMetadata,
@@ -39,6 +40,8 @@ use libdd_common_ffi::{self as ffi, MaybeError};
 #[cfg(windows)]
 use libdd_crashtracker_ffi::Metadata;
 use libdd_dogstatsd_client::DogStatsDActionOwned;
+use libdd_telemetry::data::metrics::{MetricNamespace, MetricType};
+use libdd_telemetry::metrics::MetricContext;
 use libdd_telemetry::{
     data::{self, Dependency, Integration},
     worker::{LifecycleAction, LogIdentifier, TelemetryActions},
@@ -402,6 +405,35 @@ pub unsafe extern "C" fn ddog_sidecar_telemetry_enqueueConfig(
     MaybeError::None
 }
 
+/// Reports an endpoint to the telemetry.
+#[no_mangle]
+#[allow(clippy::missing_safety_doc)]
+pub unsafe extern "C" fn ddog_sidecar_telemetry_addEndpoint(
+    transport: &mut Box<SidecarTransport>,
+    instance_id: &InstanceId,
+    queue_id: &QueueId,
+    method: libdd_telemetry::data::Method,
+    path: CharSlice,
+    operation_name: CharSlice,
+    resource_name: CharSlice,
+) -> MaybeError {
+    let endpoint = TelemetryActions::AddEndpoint(libdd_telemetry::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(),
+    });
+
+    try_c!(blocking::enqueue_actions(
+        transport,
+        instance_id,
+        queue_id,
+        vec![SidecarAction::Telemetry(endpoint)],
+    ));
+
+    MaybeError::None
+}
+
 /// Reports a dependency to the telemetry.
 #[no_mangle]
 #[allow(clippy::missing_safety_doc)]
@@ -683,6 +715,60 @@ fn char_slice_to_string(slice: CharSlice) -> Result<String, String> {
         .map_err(|e| format!("Failed to convert CharSlice to String: {e}"))
 }
 
+struct TelemetryContext {
+    instance_id: InstanceId,
+    service_name: String,
+    env_name: String,
+}
+
+impl TelemetryContext {
+    fn from_ffi(
+        session_id_ffi: CharSlice,
+        runtime_id_ffi: CharSlice,
+        service_name_ffi: CharSlice,
+        env_name_ffi: CharSlice,
+    ) -> Result<Self, String> {
+        if session_id_ffi.is_empty() {
+            return Err("Null or empty session_id".into());
+        }
+        if runtime_id_ffi.is_empty() {
+            return Err("Null or empty runtime_id".into());
+        }
+        if service_name_ffi.is_empty() {
+            return Err("Null or empty service_name".into());
+        }
+        if env_name_ffi.is_empty() {
+            return Err("Null or empty env_name".into());
+        }
+
+        Ok(Self {
+            instance_id: InstanceId::new(
+                char_slice_to_string(session_id_ffi)?,
+                char_slice_to_string(runtime_id_ffi)?,
+            ),
+            service_name: char_slice_to_string(service_name_ffi)?,
+            env_name: char_slice_to_string(env_name_ffi)?,
+        })
+    }
+
+    /// Sends a telemetry action through the internal telemetry channel
+    fn send_action(self, action: InternalTelemetryAction) -> Result<(), String> {
+        let sender = get_telemetry_action_sender()
+            .map_err(|e| format!("Failed to get telemetry action sender: {e}"))?;
+
+        let msg = InternalTelemetryActions {
+            instance_id: self.instance_id,
+            service_name: self.service_name,
+            env_name: self.env_name,
+            actions: vec![action],
+        };
+
+        sender
+            .try_send(msg)
+            .map_err(|e| format!("Failed to send telemetry action: {e}"))
+    }
+}
+
 #[allow(clippy::too_many_arguments)]
 fn ddog_sidecar_enqueue_telemetry_log_impl(
     session_id_ffi: CharSlice,
@@ -696,31 +782,19 @@ fn ddog_sidecar_enqueue_telemetry_log_impl(
     tags_ffi: Option<NonNull<CharSlice>>,
     is_sensitive: bool,
 ) -> Result<(), String> {
-    if session_id_ffi.is_empty()
-        || runtime_id_ffi.is_empty()
-        || service_name_ffi.is_empty()
-        || env_name_ffi.is_empty()
-        || identifier_ffi.is_empty()
-        || message_ffi.is_empty()
-    {
+    if identifier_ffi.is_empty() || message_ffi.is_empty() {
         return Err("Null or empty required arguments".into());
     }
 
-    let sender = match get_telemetry_action_sender() {
-        Ok(s) => s,
-        Err(e) => {
-            return Err(format!("Failed to get telemetry action sender: {e}"));
-        }
-    };
+    let ctx = TelemetryContext::from_ffi(
+        session_id_ffi,
+        runtime_id_ffi,
+        service_name_ffi,
+        env_name_ffi,
+    )?;
 
-    let instance_id = InstanceId::new(
-        char_slice_to_string(session_id_ffi)?,
-        char_slice_to_string(runtime_id_ffi)?,
-    );
-    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)?;
+    let identifier = char_slice_to_string(identifier_ffi)?;
+    let message = char_slice_to_string(message_ffi)?;
 
     let stack_trace = stack_trace_ffi
         .map(|s| char_slice_to_string(*unsafe { s.as_ref() }))
@@ -746,17 +820,136 @@ fn ddog_sidecar_enqueue_telemetry_log_impl(
     };
     let log_action = TelemetryActions::AddLog((log_id, log_data));
 
-    let msg = InternalTelemetryActions {
-        instance_id,
-        service_name,
-        env_name,
-        actions: vec![log_action],
+    ctx.send_action(InternalTelemetryAction::TelemetryAction(log_action))
+}
+
+/// Enqueues a telemetry point to be processed internally.
+///
+/// # Safety
+/// Pointers must be valid, strings must be null-terminated if not null.
+#[no_mangle]
+pub unsafe extern "C" fn ddog_sidecar_enqueue_telemetry_point(
+    session_id_ffi: CharSlice,
+    runtime_id_ffi: CharSlice,
+    service_name_ffi: CharSlice,
+    env_name_ffi: CharSlice,
+    metric_name_ffi: CharSlice,
+    value: f64,
+    tags_ffi: Option<NonNull<CharSlice>>,
+) -> MaybeError {
+    try_c!(ddog_sidecar_enqueue_telemetry_point_impl(
+        session_id_ffi,
+        runtime_id_ffi,
+        service_name_ffi,
+        env_name_ffi,
+        metric_name_ffi,
+        value,
+        tags_ffi,
+    ));
+    MaybeError::None
+}
+
+fn ddog_sidecar_enqueue_telemetry_point_impl(
+    session_id_ffi: CharSlice,
+    runtime_id_ffi: CharSlice,
+    service_name_ffi: CharSlice,
+    env_name_ffi: CharSlice,
+    metric_name_ffi: CharSlice,
+    value: f64,
+    tags_ffi: Option<NonNull<CharSlice>>,
+) -> Result<(), String> {
+    if metric_name_ffi.is_empty() {
+        return Err("Null or empty metric_name".into());
+    }
+
+    let ctx = TelemetryContext::from_ffi(
+        session_id_ffi,
+        runtime_id_ffi,
+        service_name_ffi,
+        env_name_ffi,
+    )?;
+
+    let metric_name = char_slice_to_string(metric_name_ffi)?;
+
+    fn get_tags(tags_slice: CharSlice) -> Result<Vec<Tag>, String> {
+        let tags = char_slice_to_string(tags_slice)?;
+        let (tags, error) = libdd_common::tag::parse_tags(tags.as_str());
+        if let Some(error) = error {
+            return Err(error.to_string());
+        }
+        Ok(tags)
+    }
+
+    let tags = match tags_ffi {
+        Some(tags_slice) => get_tags(*unsafe { tags_slice.as_ref() })?,
+        None => Vec::default(),
     };
 
-    match sender.try_send(msg) {
-        Ok(_) => Ok(()),
-        Err(err) => Err(format!("Failed to send telemetry action: {err}")),
+    ctx.send_action(InternalTelemetryAction::AddMetricPoint((
+        value,
+        metric_name,
+        tags,
+    )))
+}
+
+/// Registers a telemetry metric to be processed internally.
+///
+/// # Safety
+/// Pointers must be valid, strings must be null-terminated if not null.
+#[no_mangle]
+pub unsafe extern "C" fn ddog_sidecar_enqueue_telemetry_metric(
+    session_id_ffi: CharSlice,
+    runtime_id_ffi: CharSlice,
+    service_name_ffi: CharSlice,
+    env_name_ffi: CharSlice,
+    metric_name_ffi: CharSlice,
+    metric_type: MetricType,
+    metric_namespace: MetricNamespace,
+) -> MaybeError {
+    try_c!(ddog_sidecar_enqueue_telemetry_metric_impl(
+        session_id_ffi,
+        runtime_id_ffi,
+        service_name_ffi,
+        env_name_ffi,
+        metric_name_ffi,
+        metric_type,
+        metric_namespace,
+    ));
+    MaybeError::None
+}
+
+#[allow(clippy::too_many_arguments)]
+fn ddog_sidecar_enqueue_telemetry_metric_impl(
+    session_id_ffi: CharSlice,
+    runtime_id_ffi: CharSlice,
+    service_name_ffi: CharSlice,
+    env_name_ffi: CharSlice,
+    metric_name_ffi: CharSlice,
+    metric_type: MetricType,
+    metric_namespace: MetricNamespace,
+) -> Result<(), String> {
+    if metric_name_ffi.is_empty() {
+        return Err("Null or empty metric_name".into());
     }
+
+    let ctx = TelemetryContext::from_ffi(
+        session_id_ffi,
+        runtime_id_ffi,
+        service_name_ffi,
+        env_name_ffi,
+    )?;
+
+    let metric_name = char_slice_to_string(metric_name_ffi)?;
+
+    ctx.send_action(InternalTelemetryAction::RegisterTelemetryMetric(
+        MetricContext {
+            name: metric_name,
+            tags: Vec::default(),
+            metric_type,
+            common: true,
+            namespace: metric_namespace,
+        },
+    ))
 }
 
 /// Sends a trace to the sidecar via shared memory.

datadog-sidecar-ffi/Cargo.toml

@@ -29,7 +29,7 @@ use std::collections::{HashMap, HashSet};
 use std::pin::Pin;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
-use std::time::Duration;
+use std::time::{Duration, SystemTime};
 use tracing::{debug, error, info, trace, warn};
 
 use futures::FutureExt;
@@ -441,7 +441,7 @@ impl SidecarInterface for SidecarServer {
             );
             let mut telemetry = telemetry_mutex.lock_or_panic();
 
-            let mut actions_to_process = vec![];
+            let mut actions_to_process: Vec<SidecarAction> = vec![];
             let mut composer_paths_to_process = vec![];
             let mut buffered_info_changed = false;
             let mut remove_entry = false;
@@ -469,6 +469,11 @@ impl SidecarInterface for SidecarServer {
                     SidecarAction::ClearQueueId => {
                         remove_entry = true;
                     }
+                    SidecarAction::Telemetry(TelemetryActions::AddEndpoint(_)) => {
+                        telemetry.last_endpoints_push = SystemTime::now();
+                        buffered_info_changed = true;
+                        actions_to_process.push(action);
+                    }
                     SidecarAction::Telemetry(TelemetryActions::Lifecycle(
                         LifecycleAction::Stop,
                     )) => {