[SVLS-8351] Add CPU Enhanced Metrics in Linux Azure Functions

.github/workflows/build-datadog-serverless-compat.yml

@@ -56,7 +56,7 @@ jobs:
           retention-days: 3
       - if: ${{ inputs.runner == 'windows-2022' }}
         shell: bash
-        run: cargo build --release -p datadog-serverless-compat --features windows-pipes
+        run: cargo build --release -p datadog-serverless-compat --features windows-pipes,windows-enhanced-metrics
       - if: ${{ inputs.runner == 'windows-2022' }}
         uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # 4.6.2
         with:
@@ -69,7 +69,7 @@ jobs:
           rustup target add i686-pc-windows-msvc
           cargo build --release -p datadog-serverless-compat \
             --target i686-pc-windows-msvc \
-            --features windows-pipes
+            --features windows-pipes,windows-enhanced-metrics
       - if: ${{ inputs.runner == 'windows-2022' }}
         uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # 4.6.2
         with:

.github/workflows/cargo.yml

@@ -32,7 +32,12 @@ jobs:
         shell: bash
         run: chmod +x ./scripts/install-protoc.sh && ./scripts/install-protoc.sh $HOME
       - shell: bash
-        run: cargo check --workspace
+        run: |
+          if [[ "${{ inputs.runner }}" == "windows-2022" ]]; then
+            cargo check --workspace --features datadog-serverless-compat/windows-pipes,datadog-serverless-compat/windows-enhanced-metrics
+          else
+            cargo check --workspace
+          fi
 
   format:
     name: Format
@@ -72,7 +77,12 @@ jobs:
         shell: bash
         run: chmod +x ./scripts/install-protoc.sh && ./scripts/install-protoc.sh $HOME
       - shell: bash
-        run: cargo build --all
+        run: |
+          if [[ "${{ inputs.runner }}" == "windows-2022" ]]; then
+            cargo build --all --features datadog-serverless-compat/windows-pipes,datadog-serverless-compat/windows-enhanced-metrics
+          else
+            cargo build --all
+          fi
 
   build-datadog-serverless-compat:
     name: Build Datadog Serverless Compat
@@ -95,7 +105,7 @@ jobs:
       - shell: bash
         run: |
           if [[ "${{ inputs.runner }}" == "windows-2022" ]]; then
-            cargo nextest run --workspace --features datadog-serverless-compat/windows-pipes
+            cargo nextest run --workspace --features datadog-serverless-compat/windows-pipes,datadog-serverless-compat/windows-enhanced-metrics
           else
             cargo nextest run --workspace
           fi

crates/datadog-metrics-collector/Cargo.toml

@@ -9,3 +9,9 @@ description = "Collector to read, compute, and submit enhanced metrics in Server
 dogstatsd = { path = "../dogstatsd", default-features = true }
 tracing = { version = "0.1", default-features = false }
 libdd-common = { git = "https://github.com/DataDog/libdatadog", rev = "4ae8ebe252451374c292efd159ce254c3f5a72e0", default-features = false }
+
+[dev-dependencies]
+tokio = { version = "1", default-features = false, features = ["macros", "rt-multi-thread"] }
+
+[features]
+windows-enhanced-metrics = []

crates/datadog-metrics-collector/src/azure_cpu.rs

@@ -0,0 +1,277 @@
+// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/
+// SPDX-License-Identifier: Apache-2.0
+
+//! CPU metrics collector for Azure Functions
+//!
+//! This module provides OS-agnostic CPU stats collection, CPU usage
+//! computation, and metrics submission to Datadog.
+//!
+//! All CPU metrics are reported in nanocores (1 core = 1,000,000,000 nanocores).
+
+use dogstatsd::aggregator::AggregatorHandle;
+use dogstatsd::metric::{Metric, MetricValue, SortedTags};
+use tracing::{debug, error};
+
+const CPU_USAGE_METRIC: &str = "azure.functions.enhanced.cpu.usage";
+
+/// Computed CPU total usage metric
+pub struct CpuStats {
+    pub total: u64, // Cumulative CPU usage in nanoseconds
+}
+
+pub trait CpuStatsReader {
+    fn read(&self) -> Option<CpuStats>;
+}
+
+pub struct CpuMetricsCollector {
+    reader: Box<dyn CpuStatsReader>,
+    aggregator: AggregatorHandle,
+    tags: Option<SortedTags>,
+    last_usage_ns: Option<u64>,
+    last_collection_time: std::time::Instant,
+}
+
+impl CpuMetricsCollector {
+    /// Creates a new CpuMetricsCollector
+    ///
+    /// # Arguments
+    ///
+    /// * `aggregator` - The aggregator handle to submit metrics to
+    /// * `tags` - Optional tags to attach to all metrics
+    pub fn new(aggregator: AggregatorHandle, tags: Option<SortedTags>) -> Self {
+        #[cfg(all(windows, feature = "windows-enhanced-metrics"))]
+        let reader: Box<dyn CpuStatsReader> = Box::new(crate::azure_windows::WindowsCpuStatsReader);
+        #[cfg(not(windows))]
+        let reader: Box<dyn CpuStatsReader> = Box::new(crate::azure_linux::LinuxCpuStatsReader);
+        Self {
+            reader,
+            aggregator,
+            tags,
+            last_usage_ns: None,
+            last_collection_time: std::time::Instant::now(),
+        }
+    }
+
+    pub fn collect_and_submit(&mut self) {
+        let Some(cpu_stats) = self.reader.read() else {
+            debug!(
+                "Skipping CPU enhanced metrics collection - could not find data to generate CPU usage metrics"
+            );
+            return;
+        };
+
+        let current_usage_ns = cpu_stats.total;
+        let now_instant = std::time::Instant::now();
+        let now = std::time::UNIX_EPOCH
+            .elapsed()
+            .map(|d| d.as_secs())
+            .unwrap_or(0)
+            .try_into()
+            .unwrap_or(0);
+
+        // Skip first collection
+        let Some(last_usage_ns) = self.last_usage_ns else {
+            debug!("First CPU collection, skipping interval");
+            self.last_usage_ns = Some(current_usage_ns);
+            self.last_collection_time = now_instant;
+            return;
+        };
+
+        let elapsed_secs = now_instant
+            .duration_since(self.last_collection_time)
+            .as_secs_f64();
+
+        // Update state so the next collection always compares against the most recent reading, even if the interval is skipped
+        self.last_usage_ns = Some(current_usage_ns);
+        self.last_collection_time = now_instant;
+
+        let Some(usage_rate_nc) =
+            compute_usage_rate_nc(current_usage_ns, last_usage_ns, elapsed_secs)
+        else {
+            return;
+        };
+
+        let usage_metric = Metric::new(
+            CPU_USAGE_METRIC.into(),
+            MetricValue::distribution(usage_rate_nc),
+            self.tags.clone(),
+            Some(now),
+        );
+
+        if let Err(e) = self.aggregator.insert_batch(vec![usage_metric]) {
+            error!("Failed to insert CPU usage metric: {}", e);
+        }
+    }
+}
+
+fn compute_usage_rate_nc(
+    current_usage_ns: u64,
+    last_usage_ns: u64,
+    elapsed_secs: f64,
+) -> Option<f64> {
+    if current_usage_ns < last_usage_ns {
+        debug!("Current CPU usage is less than last usage, skipping interval");
+        return None;
+    }
+    if elapsed_secs <= 0.0 {
+        debug!("Elapsed time is less than or equal to 0, skipping interval");
+        return None;
+    }
+    let delta_ns = (current_usage_ns - last_usage_ns) as f64;
+    Some(delta_ns / elapsed_secs)
+}
+
+// For testing only since CpuMetricsCollector::new() hardcodes the reader based on OS
+#[cfg(test)]
+impl CpuMetricsCollector {
+    pub(crate) fn with_reader(
+        reader: Box<dyn CpuStatsReader>,
+        aggregator: AggregatorHandle,
+        tags: Option<SortedTags>,
+    ) -> Self {
+        Self {
+            reader,
+            aggregator,
+            tags,
+            last_usage_ns: None,
+            last_collection_time: std::time::Instant::now(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use dogstatsd::aggregator::AggregatorService;
+    use dogstatsd::metric::EMPTY_TAGS;
+    use std::cell::Cell;
+
+    #[test]
+    fn test_normal_delta() {
+        assert_eq!(
+            compute_usage_rate_nc(2_000_000_000, 1_000_000_000, 1.0),
+            Some(1_000_000_000.0)
+        );
+    }
+
+    #[test]
+    fn test_current_usage_less_than_last_usage_returns_none() {
+        assert!(compute_usage_rate_nc(1_000_000_000, 2_000_000_000, 1.0).is_none());
+    }
+
+    #[test]
+    fn test_zero_elapsed_time_returns_none() {
+        assert!(compute_usage_rate_nc(2_000_000_000, 1_000_000_000, 0.0).is_none());
+    }
+
+    struct MockCpuStatsReader {
+        idx: Cell<usize>,
+        values: Vec<Option<u64>>,
+    }
+
+    impl MockCpuStatsReader {
+        fn new(values: Vec<Option<u64>>) -> Self {
+            Self {
+                idx: Cell::new(0),
+                values,
+            }
+        }
+    }
+
+    impl CpuStatsReader for MockCpuStatsReader {
+        fn read(&self) -> Option<CpuStats> {
+            let i = self.idx.get();
+            self.idx.set(i + 1);
+            self.values
+                .get(i)
+                .and_then(|v| v.map(|total| CpuStats { total }))
+        }
+    }
+
+    #[tokio::test]
+    async fn test_first_collection_skipped() {
+        let (service, handle) =
+            AggregatorService::new(EMPTY_TAGS, 1000).expect("Aggregator creation failed");
+        let task = tokio::spawn(service.run());
+
+        let mut collector = CpuMetricsCollector::with_reader(
+            Box::new(MockCpuStatsReader::new(vec![Some(1_000_000_000)])),
+            handle.clone(),
+            None,
+        );
+        collector.collect_and_submit();
+
+        let response = handle.flush().await.expect("flush failed");
+        assert!(
+            response.distributions.is_empty(),
+            "Expected no batches flushed on first collection, got {:?}",
+            response.distributions.len()
+        );
+
+        handle.shutdown().expect("Shutdown failed");
+        task.await.expect("Service task panicked");
+    }
+
+    #[tokio::test]
+    async fn test_second_collection_submits_metric() {
+        let (service, handle) =
+            AggregatorService::new(EMPTY_TAGS, 1000).expect("Aggregator creation failed");
+        let task = tokio::spawn(service.run());
+
+        let mut collector = CpuMetricsCollector::with_reader(
+            Box::new(MockCpuStatsReader::new(vec![
+                Some(1_000_000_000),
+                Some(2_000_000_000),
+            ])),
+            handle.clone(),
+            None,
+        );
+        collector.collect_and_submit();
+        collector.collect_and_submit();
+
+        let response = handle.flush().await.expect("Flush failed");
+        assert_eq!(
+            response.distributions.len(),
+            1,
+            "Expected 1 batch to be flushed, got {:?}",
+            response.distributions.len()
+        );
+        assert_eq!(
+            response.distributions[0].sketches.len(),
+            1,
+            "Expected 1 metric, got {:?}",
+            response.distributions[0].sketches.len()
+        );
+
+        handle.shutdown().expect("Shutdown failed");
+        task.await.expect("Service task panicked");
+    }
+
+    #[tokio::test]
+    async fn test_current_usage_less_than_last_usage_skips_interval() {
+        let (service, handle) =
+            AggregatorService::new(EMPTY_TAGS, 1000).expect("Aggregator creation failed");
+        let task = tokio::spawn(service.run());
+
+        let mut collector = CpuMetricsCollector::with_reader(
+            Box::new(MockCpuStatsReader::new(vec![
+                Some(2_000_000_000),
+                Some(1_000_000_000),
+            ])),
+            handle.clone(),
+            None,
+        );
+        collector.collect_and_submit();
+        collector.collect_and_submit();
+
+        let response = handle.flush().await.expect("Flush failed");
+        assert!(
+            response.distributions.is_empty(),
+            "Expected no batches flushed when current usage is less than last usage, got {:?}",
+            response.distributions.len()
+        );
+
+        handle.shutdown().expect("Shutdown failed");
+        task.await.expect("Service task panicked");
+    }
+}