refactor enhanced metrics monitoring (#901)

Use a long-running task that resumes at the start of an invocation,
instead of spawning a new task on each invocation. Uses a channel-based
service to pause and resume monitoring.

Tested with a function that opens a large number of file descriptors to
make sure it still captures data accurately. Tested with self-monitoring
as well (new implementation, previous implementation):
<img width="1337" height="316" alt="Screenshot 2025-10-20 at 4 16 47 PM"
src="https://github.com/user-attachments/assets/3e3a0470-8566-42f9-926b-2ae3df34a4f7"
/>

Author: shreyamalpani

bottlecap/src/lifecycle/invocation/context.rs

@@ -432,7 +432,6 @@ mod tests {
     use crate::proc::{CPUData, NetworkData};
     use serde_json::json;
     use std::collections::HashMap;
-    use tokio::sync::watch;
 
     use super::*;
 
@@ -567,15 +566,10 @@ mod tests {
         });
 
         let uptime_offset = Some(50f64);
-        let (tmp_chan_tx, _) = watch::channel(());
-        let (process_chan_tx, _) = watch::channel(());
-
         let enhanced_metric_data = Some(EnhancedMetricData {
             network_offset,
             cpu_offset,
             uptime_offset,
-            tmp_chan_tx,
-            process_chan_tx,
         });
 
         buffer.add_enhanced_metric_data(&request_id, enhanced_metric_data.clone());

bottlecap/src/lifecycle/invocation/processor.rs

@@ -8,7 +8,6 @@ use chrono::{DateTime, Utc};
 use datadog_trace_protobuf::pb::Span;
 use datadog_trace_utils::tracer_header_tags;
 use serde_json::Value;
-use tokio::sync::watch;
 use tokio::time::Instant;
 use tracing::{debug, warn};
 
@@ -103,11 +102,14 @@ impl Processor {
             config.trace_aws_service_representation_enabled,
         );
 
+        let enhanced_metrics = EnhancedMetrics::new(metrics_aggregator, Arc::clone(&config));
+        enhanced_metrics.start_usage_metrics_task(); // starts the long-running task that monitors usage metrics (fd_use, threads_use, tmp_used)
+
         Processor {
             context_buffer: ContextBuffer::default(),
             inferrer: SpanInferrer::new(Arc::clone(&config)),
             propagator,
-            enhanced_metrics: EnhancedMetrics::new(metrics_aggregator, Arc::clone(&config)),
+            enhanced_metrics,
             aws_config,
             tracer_detected: false,
             runtime: None,
@@ -136,26 +138,18 @@ impl Processor {
             .unwrap_or_default();
 
         if self.config.lambda_proc_enhanced_metrics {
+            // Resume tmp, fd, and threads enhanced metrics monitoring
+            self.enhanced_metrics.resume_usage_metrics_monitoring();
+
             // Collect offsets for network and cpu metrics
             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();
 
-            // Start a channel for monitoring tmp enhanced data
-            let (tmp_chan_tx, tmp_chan_rx) = watch::channel(());
-            self.enhanced_metrics.set_tmp_enhanced_metrics(tmp_chan_rx);
-
-            // Start a channel for monitoring file descriptor and thread count
-            let (process_chan_tx, process_chan_rx) = watch::channel(());
-            self.enhanced_metrics
-                .set_process_enhanced_metrics(process_chan_rx);
-
             let enhanced_metric_offsets = Some(EnhancedMetricData {
                 network_offset,
                 cpu_offset,
                 uptime_offset,
-                tmp_chan_tx,
-                process_chan_tx,
             });
             self.context_buffer
                 .add_enhanced_metric_data(&request_id, enhanced_metric_offsets);
@@ -313,6 +307,10 @@ impl Processor {
             }
         }
 
+        // Set tmp, fd, and threads enhanced metrics
+        self.enhanced_metrics.set_max_enhanced_metrics();
+        self.enhanced_metrics.set_usage_enhanced_metrics(); // sets use metric values and pauses monitoring task
+
         self.context_buffer
             .add_runtime_duration(request_id, metrics.duration_ms);
 
@@ -388,10 +386,6 @@ impl Processor {
                 offsets.cpu_offset.clone(),
                 offsets.uptime_offset,
             );
-            // Send the signal to stop monitoring tmp
-            _ = offsets.tmp_chan_tx.send(());
-            // Send the signal to stop monitoring file descriptors and threads
-            _ = offsets.process_chan_tx.send(());
         }
 
         // todo(duncanista): Add missing metric tags for ASM

bottlecap/src/lifecycle/invocation/processor_service.rs

@@ -100,7 +100,7 @@ pub enum ProcessorCommand {
     SendCtxSpans {
         tags_provider: Arc<provider::Provider>,
         trace_sender: Arc<SendingTraceProcessor>,
-        context: Context,
+        context: Box<Context>,
     },
     Shutdown,
 }
@@ -338,7 +338,7 @@ impl InvocationProcessorHandle {
             .send(ProcessorCommand::SendCtxSpans {
                 tags_provider: Arc::clone(tags_provider),
                 trace_sender: Arc::clone(trace_sender),
-                context,
+                context: Box::new(context),
             })
             .await
     }
@@ -491,7 +491,7 @@ impl InvocationProcessorService {
                     context,
                 } => {
                     self.processor
-                        .send_ctx_spans(&tags_provider, &trace_sender, context)
+                        .send_ctx_spans(&tags_provider, &trace_sender, *context)
                         .await;
                 }
                 ProcessorCommand::Shutdown => {