Generate CPU Enhanced Metrics

bottlecap/Cargo.toml

@@ -20,6 +20,7 @@ figment = { version = "0.10", default-features = false, features = ["yaml", "env
 hyper = { version = "0.14", default-features = false, features = ["server"] }
 lazy_static = { version = "1.5", default-features = false }
 log = { version = "0.4", default-features = false }
+nix = { version = "0.26", default-features = false, features = ["feature"] }
 protobuf = { version = "3.5", default-features = false }
 regex = { version = "1.10", default-features = false }
 reqwest = { version = "0.12", features = ["json", "http2", "rustls-tls"], default-features = false }

bottlecap/src/bin/bottlecap/main.rs

@@ -425,8 +425,9 @@ async fn extension_loop_active(
                                     ..
                                 } => {
                                     let mut p = invocation_processor.lock().await;
+                                    let mut enhanced_metric_data = None;
                                     if let Some(metrics) = metrics {
-                                        p.on_platform_runtime_done(
+                                        enhanced_metric_data = p.on_platform_runtime_done(
                                             &request_id,
                                             metrics.duration_ms,
                                             config.clone(),
@@ -450,6 +451,11 @@ async fn extension_loop_active(
                                         request_id, status
                                     );
 
+                                    // set cpu utilization metrics here to avoid accounting for extra idle time
+                                    if let Some(offsets) = enhanced_metric_data {
+                                        lambda_enhanced_metrics.set_cpu_utilization_enhanced_metrics(offsets.cpu_offset, offsets.uptime_offset);
+                                    }
+
                                     // TODO(astuyve) it'll be easy to
                                     // pass the invocation deadline to
                                     // flush tasks here, so they can
@@ -462,6 +468,7 @@ async fn extension_loop_active(
                                             stats_flusher.manual_flush()
                                         );
                                     }
+
                                     break;
                                 }
                                 TelemetryRecord::PlatformReport {
@@ -476,11 +483,13 @@ async fn extension_loop_active(
                                     );
                                     lambda_enhanced_metrics.set_report_log_metrics(&metrics);
                                     let mut p = invocation_processor.lock().await;
-                                    if let Some((post_runtime_duration_ms, network_offset)) = p.on_platform_report(&request_id, metrics.duration_ms) {
-                                        lambda_enhanced_metrics.set_post_runtime_duration_metric(
-                                            post_runtime_duration_ms,
-                                        );
-                                        lambda_enhanced_metrics.set_network_enhanced_metrics(network_offset);
+                                    let (post_runtime_duration_ms, enhanced_metric_data) = p.on_platform_report(&request_id, metrics.duration_ms);
+                                    if let Some(duration) = post_runtime_duration_ms {
+                                        lambda_enhanced_metrics.set_post_runtime_duration_metric(duration);
+                                    }
+                                    if let Some(offsets) = enhanced_metric_data {
+                                        lambda_enhanced_metrics.set_network_enhanced_metrics(offsets.network_offset);
+                                        lambda_enhanced_metrics.set_cpu_time_enhanced_metrics(offsets.cpu_offset);
                                     }
                                     drop(p);
 

bottlecap/src/lifecycle/invocation/context.rs

@@ -1,4 +1,4 @@
-use crate::proc::NetworkData;
+use crate::metrics::enhanced::lambda::EnhancedMetricData;
 use std::collections::VecDeque;
 
 use tracing::debug;
@@ -9,7 +9,7 @@ pub struct Context {
     pub runtime_duration_ms: f64,
     pub init_duration_ms: f64,
     pub start_time: i64,
-    pub network_offset: Option<NetworkData>,
+    pub enhanced_metric_data: Option<EnhancedMetricData>,
 }
 
 impl Context {
@@ -19,14 +19,14 @@ impl Context {
         runtime_duration_ms: f64,
         init_duration_ms: f64,
         start_time: i64,
-        network_offset: Option<NetworkData>,
+        enhanced_metric_data: Option<EnhancedMetricData>,
     ) -> Self {
         Context {
             request_id,
             runtime_duration_ms,
             init_duration_ms,
             start_time,
-            network_offset,
+            enhanced_metric_data,
         }
     }
 }
@@ -143,13 +143,17 @@ impl ContextBuffer {
 
     /// Adds the network offset to a `Context` in the buffer.
     ///
-    pub fn add_network_offset(&mut self, request_id: &String, network_data: Option<NetworkData>) {
+    pub fn add_enhanced_metric_data(
+        &mut self,
+        request_id: &String,
+        enhanced_metric_data: Option<EnhancedMetricData>,
+    ) {
         if let Some(context) = self
             .buffer
             .iter_mut()
             .find(|context| context.request_id == *request_id)
         {
-            context.network_offset = network_data;
+            context.enhanced_metric_data = enhanced_metric_data;
         } else {
             debug!("Could not add network offset - context not found");
         }
@@ -166,6 +170,9 @@ impl ContextBuffer {
 #[cfg(test)]
 #[allow(clippy::unwrap_used)]
 mod tests {
+    use crate::proc::{CPUData, NetworkData};
+    use std::collections::HashMap;
+
     use super::*;
 
     #[test]
@@ -286,7 +293,7 @@ mod tests {
     }
 
     #[test]
-    fn test_add_network_offset() {
+    fn test_add_enhanced_metric_data() {
         let mut buffer = ContextBuffer::with_capacity(2);
 
         let request_id = String::from("1");
@@ -300,10 +307,28 @@ mod tests {
             tx_bytes: 254.0,
         });
 
-        buffer.add_network_offset(&request_id, network_offset);
-        assert_eq!(
-            buffer.get(&request_id).unwrap().network_offset,
+        let mut individual_cpu_idle_times = HashMap::new();
+        individual_cpu_idle_times.insert("cpu0".to_string(), 10.0);
+        individual_cpu_idle_times.insert("cpu1".to_string(), 20.0);
+        let cpu_offset = Some(CPUData {
+            total_user_time_ms: 100.0,
+            total_system_time_ms: 53.0,
+            total_idle_time_ms: 20.0,
+            individual_cpu_idle_times: individual_cpu_idle_times,
+        });
+
+        let uptime_offset = Some(50.0);
+
+        let enhanced_metric_data = Some(EnhancedMetricData {
             network_offset,
+            cpu_offset,
+            uptime_offset,
+        });
+
+        buffer.add_enhanced_metric_data(&request_id, enhanced_metric_data.clone());
+        assert_eq!(
+            buffer.get(&request_id).unwrap().enhanced_metric_data,
+            enhanced_metric_data,
         );
     }
 }

bottlecap/src/lifecycle/invocation/processor.rs

@@ -14,7 +14,8 @@ use tracing::debug;
 use crate::{
     config::{self, AwsConfig},
     lifecycle::invocation::{context::ContextBuffer, span_inferrer::SpanInferrer},
-    proc::{self, NetworkData},
+    metrics::enhanced::lambda::EnhancedMetricData,
+    proc::{self, CPUData, NetworkData},
     tags::provider,
     traces::{
         context::SpanContext,
@@ -34,6 +35,7 @@ pub struct Processor {
     propagator: DatadogCompositePropagator,
     aws_config: AwsConfig,
     tracer_detected: bool,
+    collect_enhanced_data: bool,
 }
 
 impl Processor {
@@ -73,17 +75,26 @@ impl Processor {
             propagator,
             aws_config: aws_config.clone(),
             tracer_detected: false,
+            collect_enhanced_data: config.enhanced_metrics,
         }
     }
 
-    /// Given a `request_id`, add the enhanced metric offsets to the context buffer.
+    /// Given a `request_id`, creates the context and adds the enhanced metric offsets to the context buffer.
     ///
     pub fn on_invoke_event(&mut self, request_id: String) {
         self.context_buffer.create_context(request_id.clone());
-
-        let network_offset: Option<NetworkData> = proc::get_network_data().ok();
-        self.context_buffer
-            .add_network_offset(&request_id, network_offset);
+        if self.collect_enhanced_data {
+            let network_offset: Option<NetworkData> = proc::get_network_data().ok();
+            let cpu_offset: Option<CPUData> = proc::get_cpu_data().ok();
+            let uptime_offset: Option<f64> = proc::get_uptime().ok();
+            let enhanced_metric_offsets = Some(EnhancedMetricData {
+                network_offset,
+                cpu_offset,
+                uptime_offset,
+            });
+            self.context_buffer
+                .add_enhanced_metric_data(&request_id, enhanced_metric_offsets);
+        }
     }
 
     /// Given a `request_id` and the time of the platform start, add the start time to the context buffer.
@@ -110,10 +121,11 @@ impl Processor {
         tags_provider: Arc<provider::Provider>,
         trace_processor: Arc<dyn trace_processor::TraceProcessor + Send + Sync>,
         trace_agent_tx: Sender<SendData>,
-    ) {
+    ) -> Option<EnhancedMetricData> {
         self.context_buffer
             .add_runtime_duration(request_id, duration_ms);
 
+        let mut enhanced_metric_data: Option<EnhancedMetricData> = None;
         if let Some(context) = self.context_buffer.get(request_id) {
             let span = &mut self.span;
             // `round` is intentionally meant to be a whole integer
@@ -129,6 +141,8 @@ impl Processor {
             // - error.type
             // - error.stack
             // - metrics tags (for asm)
+
+            enhanced_metric_data.clone_from(&context.enhanced_metric_data);
         }
 
         if let Some(trigger_tags) = self.inferrer.get_trigger_tags() {
@@ -169,6 +183,8 @@ impl Processor {
                 debug!("Failed to send invocation span to agent: {e}");
             }
         }
+
+        enhanced_metric_data
     }
 
     /// Given a `request_id` and the duration in milliseconds of the platform report,
@@ -181,18 +197,18 @@ impl Processor {
         &mut self,
         request_id: &String,
         duration_ms: f64,
-    ) -> Option<(f64, Option<NetworkData>)> {
+    ) -> (Option<f64>, Option<EnhancedMetricData>) {
         if let Some(context) = self.context_buffer.remove(request_id) {
-            if context.runtime_duration_ms == 0.0 {
-                return None;
-            }
+            let mut post_runtime_duration_ms: Option<f64> = None;
 
-            let post_runtime_duration_ms = duration_ms - context.runtime_duration_ms;
+            if context.runtime_duration_ms != 0.0 {
+                post_runtime_duration_ms = Some(duration_ms - context.runtime_duration_ms);
+            }
 
-            return Some((post_runtime_duration_ms, context.network_offset));
+            return (post_runtime_duration_ms, context.enhanced_metric_data);
         }
 
-        None
+        (None, None)
     }
 
     /// If this method is called, it means that we are operating in a Universally Instrumented

bottlecap/src/metrics/enhanced/constants.rs

@@ -24,5 +24,13 @@ pub const INVOCATIONS_METRIC: &str = "aws.lambda.enhanced.invocations";
 pub const RX_BYTES_METRIC: &str = "aws.lambda.enhanced.rx_bytes";
 pub const TX_BYTES_METRIC: &str = "aws.lambda.enhanced.tx_bytes";
 pub const TOTAL_NETWORK_METRIC: &str = "aws.lambda.enhanced.total_network";
+pub const CPU_SYSTEM_TIME_METRIC: &str = "aws.lambda.enhanced.cpu_system_time";
+pub const CPU_USER_TIME_METRIC: &str = "aws.lambda.enhanced.cpu_user_time";
+pub const CPU_TOTAL_TIME_METRIC: &str = "aws.lambda.enhanced.cpu_total_time";
+pub const CPU_TOTAL_UTILIZATION_PCT_METRIC: &str = "aws.lambda.enhanced.cpu_total_utilization_pct";
+pub const CPU_TOTAL_UTILIZATION_METRIC: &str = "aws.lambda.enhanced.cpu_total_utilization";
+pub const NUM_CORES_METRIC: &str = "aws.lambda.enhanced.num_cores";
+pub const CPU_MAX_UTILIZATION_METRIC: &str = "aws.lambda.enhanced.cpu_max_utilization";
+pub const CPU_MIN_UTILIZATION_METRIC: &str = "aws.lambda.enhanced.cpu_min_utilization";
 //pub const ASM_INVOCATIONS_METRIC: &str = "aws.lambda.enhanced.asm.invocations";
 pub const ENHANCED_METRICS_ENV_VAR: &str = "DD_ENHANCED_METRICS";