Add process-wide RSS metric to engine metrics (open-telemetry#2153)

Adds a memory_rss metric that reports the process-wide Resident Set Size
(physical RAM) using the memory-stats crate. Unlike the existing
jemalloc-based memory_usage (per-thread heap only), this captures the
full process memory footprint — matching what external tools like
kubectl top pod or htop report. Works on Linux, macOS, and Windows with
no feature flags required. This is reported under engine level. If the
direction is okay, will introduce more engine level metrics (total CPU
utilization etc.)

(OTel Collector has this metric already. Slightly different name, just
like every other metric)

## How are these changes tested?

Locally ran engine, and then query
`http://127.0.0.1:8080/metrics?reset=false&format=prometheus` and look
for the new metric - they match what I see using external tools for RSS
memory tracking.

```txt
# HELP memory_rss Process-wide Resident Set Size — physical RAM currently used by the process. Matches what external tools report (e.g. `kubectl top pod`, `htop`, `ps rss`).
# TYPE memory_rss gauge
memory_rss{set="engine.metrics",process_instance_id="AGOLC2UIVF4SFAEKCW6BLZ5XMM",host_id="",container_id=""} 354533376 1772503088570
```

## Are there any user-facing changes?

New metric, no change to existing.


Note: Decided to add a new dependency which brings libc which is already
a dependency. If concerns about external crate, we can hand roll this
ourselves.

Author: cijothomas

rust/otap-dataflow/Cargo.toml

@@ -106,6 +106,7 @@ libmimalloc-sys = { version = "0.1.44", features = ["extended", "v3"] }
 tikv-jemallocator = { version = "0.6.1" }
 tikv-jemalloc-ctl = { version = "0.6.1" }
 memmap2 = "0.9"
+memory-stats = "1"
 nix = { version = "0.31.0", features = ["process", "signal", "fs"] }
 notify = "8.0"  # Uses platform-native backend: inotify (Linux), kqueue (macOS), ReadDirectoryChanges (Windows)
 num_enum = "0.7"

rust/otap-dataflow/crates/controller/src/lib.rs

@@ -270,6 +270,48 @@ impl<PData: 'static + Clone + Send + Sync + std::fmt::Debug> Controller<PData> {
             move |cancellation_token| obs_state_store.run(cancellation_token),
         )?;
 
+        // Start the engine-wide metrics collection task.
+        // This samples engine-level metrics (e.g. RSS) on a fixed interval and
+        // reports them once per engine, rather than duplicating across pipelines.
+        let engine_entity_key = controller_ctx.register_engine_entity();
+        let engine_registry = controller_ctx.telemetry_registry();
+        let engine_reporter = metrics_reporter.clone();
+        let engine_metrics_handle = spawn_thread_local_task(
+            "engine-metrics",
+            admin_tracing_setup.clone(),
+            move |cancellation_token| async move {
+                use otap_df_engine::engine_metrics::EngineMetricsMonitor;
+                use std::time::Duration;
+                use tokio::time::{MissedTickBehavior, interval};
+
+                // TODO: Make this interval configurable via engine config.
+                const ENGINE_METRICS_INTERVAL: Duration = Duration::from_secs(5);
+
+                let mut monitor =
+                    EngineMetricsMonitor::new(engine_registry, engine_entity_key, engine_reporter);
+
+                let mut ticker = interval(ENGINE_METRICS_INTERVAL);
+                ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
+
+                loop {
+                    tokio::select! {
+                        _ = cancellation_token.cancelled() => {
+                            return Ok::<(), otap_df_telemetry::error::Error>(());
+                        }
+                        _ = ticker.tick() => {
+                            monitor.update();
+                            if let Err(err) = monitor.report() {
+                                otel_warn!(
+                                    "engine.metrics.reporting.fail",
+                                    error = err.to_string()
+                                );
+                            }
+                        }
+                    }
+                }
+            },
+        )?;
+
         let mut threads = Vec::new();
         let mut ctrl_msg_senders = Vec::new();
 
@@ -432,6 +474,7 @@ impl<PData: 'static + Clone + Send + Sync + std::fmt::Debug> Controller<PData> {
         }
 
         // All pipelines have finished; shut down the admin HTTP server and metric aggregator gracefully.
+        engine_metrics_handle.shutdown_and_join()?;
         admin_server_handle.shutdown_and_join()?;
         metrics_agg_handle.shutdown_and_join()?;
         if let Some(handle) = metrics_dispatcher_handle {

rust/otap-dataflow/crates/engine/Cargo.toml

@@ -45,6 +45,7 @@ data-encoding = { workspace = true }
 prost = { workspace = true }
 byte-unit = { workspace = true }
 cpu-time = { workspace = true }
+memory-stats = { workspace = true }
 nix = { workspace = true, features = ["resource"] }
 
 [target.'cfg(not(windows))'.dependencies]

rust/otap-dataflow/crates/engine/src/context.rs

@@ -160,6 +160,28 @@ impl ControllerContext {
             thread_id,
         )
     }
+
+    /// Registers the engine-level entity for engine-wide metrics.
+    ///
+    /// Returns the [`EntityKey`] to pass to
+    /// [`EngineMetricsMonitor::new`](crate::engine_metrics::EngineMetricsMonitor::new).
+    #[must_use]
+    pub fn register_engine_entity(&self) -> EntityKey {
+        use crate::attributes::ResourceAttributeSet;
+
+        self.telemetry_registry_handle
+            .register_entity(ResourceAttributeSet {
+                process_instance_id: self.process_instance_id.clone(),
+                host_id: self.host_id.clone(),
+                container_id: self.container_id.clone(),
+            })
+    }
+
+    /// Returns a handle to the telemetry registry.
+    #[must_use]
+    pub fn telemetry_registry(&self) -> TelemetryRegistryHandle {
+        self.telemetry_registry_handle.clone()
+    }
 }
 
 impl PipelineContext {

rust/otap-dataflow/crates/engine/src/engine_metrics.rs

@@ -0,0 +1,123 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Engine-level metrics for the OTAP dataflow engine.
+//!
+//! Unlike per-pipeline metrics (which are sampled on each pipeline thread),
+//! engine metrics are emitted **once per engine instance** by a dedicated
+//! background task spawned by the controller.
+//!
+//! **Metrics**
+//! - `memory_rss` (`ObserveUpDownCounter<u64>`, `{By}`):
+//!   Process-wide Resident Set Size — physical memory currently held in RAM.
+//!   Matches what external tools report (e.g. `kubectl top pod`, `htop`, `ps rss`).
+
+use otap_df_telemetry::instrument::ObserveUpDownCounter;
+use otap_df_telemetry::metrics::MetricSet;
+use otap_df_telemetry::registry::{EntityKey, TelemetryRegistryHandle};
+use otap_df_telemetry::reporter::MetricsReporter;
+use otap_df_telemetry_macros::metric_set;
+
+/// Engine-wide metrics emitted once per engine instance.
+#[metric_set(name = "engine.metrics")]
+#[derive(Debug, Default, Clone)]
+pub struct EngineMetrics {
+    /// Process-wide Resident Set Size — physical RAM currently used by the process.
+    /// Matches what external tools report (e.g. `kubectl top pod`, `htop`, `ps rss`).
+    #[metric(unit = "{By}")]
+    pub memory_rss: ObserveUpDownCounter<u64>,
+}
+
+/// Monitors and reports engine-wide metrics.
+///
+/// Created by the controller and driven by a periodic timer in a dedicated
+/// background task. Call [`update`](Self::update) to sample current values
+/// and [`report`](Self::report) to flush them to the metrics pipeline.
+pub struct EngineMetricsMonitor {
+    metrics: MetricSet<EngineMetrics>,
+    reporter: MetricsReporter,
+    registry: TelemetryRegistryHandle,
+}
+
+impl EngineMetricsMonitor {
+    /// Creates a new engine metrics monitor.
+    ///
+    /// The caller must have already registered the engine entity via
+    /// [`ControllerContext::register_engine_entity`](crate::context::ControllerContext::register_engine_entity).
+    #[must_use]
+    pub fn new(
+        registry: TelemetryRegistryHandle,
+        entity_key: EntityKey,
+        reporter: MetricsReporter,
+    ) -> Self {
+        let metrics = registry.register_metric_set_for_entity::<EngineMetrics>(entity_key);
+        Self {
+            metrics,
+            reporter,
+            registry,
+        }
+    }
+
+    /// Samples current engine-wide metrics (RSS, etc.).
+    pub fn update(&mut self) {
+        self.metrics.memory_rss.observe(get_rss_bytes());
+    }
+
+    /// Flushes sampled metrics to the reporting pipeline.
+    ///
+    /// Returns an error only if the metrics channel is permanently closed.
+    /// A full channel is silently tolerated (non-blocking, try-send semantics).
+    pub fn report(&mut self) -> Result<(), otap_df_telemetry::error::Error> {
+        self.reporter.report(&mut self.metrics)
+    }
+}
+
+/// Returns the current process-wide RSS (Resident Set Size) in bytes.
+fn get_rss_bytes() -> u64 {
+    memory_stats::memory_stats()
+        .map(|stats| stats.physical_mem as u64)
+        .unwrap_or(0)
+}
+
+impl Drop for EngineMetricsMonitor {
+    fn drop(&mut self) {
+        let _ = self
+            .registry
+            .unregister_metric_set(self.metrics.metric_set_key());
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::context::ControllerContext;
+    use otap_df_telemetry::registry::TelemetryRegistryHandle;
+
+    #[test]
+    fn engine_metrics_reports_nonzero_rss() {
+        let registry = TelemetryRegistryHandle::new();
+        let controller = ControllerContext::new(registry.clone());
+        let entity_key = controller.register_engine_entity();
+        let (_rx, reporter) = MetricsReporter::create_new_and_receiver(16);
+
+        let mut monitor = EngineMetricsMonitor::new(registry, entity_key, reporter);
+        monitor.update();
+
+        assert!(
+            monitor.metrics.memory_rss.get() > 0,
+            "memory_rss should report non-zero process RSS"
+        );
+    }
+
+    #[test]
+    fn engine_metrics_report_succeeds() {
+        let registry = TelemetryRegistryHandle::new();
+        let controller = ControllerContext::new(registry.clone());
+        let entity_key = controller.register_engine_entity();
+        let (_rx, reporter) = MetricsReporter::create_new_and_receiver(16);
+
+        let mut monitor = EngineMetricsMonitor::new(registry, entity_key, reporter);
+        monitor.update();
+        assert!(monitor.report().is_ok());
+    }
+}

rust/otap-dataflow/crates/engine/src/lib.rs

@@ -58,6 +58,7 @@ pub mod config;
 pub mod context;
 pub mod control;
 pub mod effect_handler;
+pub mod engine_metrics;
 pub mod entity_context;
 pub mod local;
 pub mod node;