Merge pull request #33287 from teskje/swap-usage

cluster,orchestrator-k8s: report swap usage

Author: teskje

src/catalog/src/builtin.rs

@@ -8857,9 +8857,9 @@ pub static MZ_CLUSTER_REPLICA_UTILIZATION: LazyLock<BuiltinView> = LazyLock::new
 SELECT
     r.id AS replica_id,
     m.process_id,
-    m.cpu_nano_cores::float8 / s.cpu_nano_cores * 100 AS cpu_percent,
-    m.memory_bytes::float8 / s.memory_bytes * 100 AS memory_percent,
-    m.disk_bytes::float8 / s.disk_bytes * 100 AS disk_percent
+    m.cpu_nano_cores::float8 / NULLIF(s.cpu_nano_cores, 0) * 100 AS cpu_percent,
+    m.memory_bytes::float8 / NULLIF(s.memory_bytes, 0) * 100 AS memory_percent,
+    m.disk_bytes::float8 / NULLIF(s.disk_bytes, 0) * 100 AS disk_percent
 FROM
     mz_catalog.mz_cluster_replicas AS r
         JOIN mz_catalog.mz_cluster_replica_sizes AS s ON r.size = s.size
@@ -8907,9 +8907,9 @@ pub static MZ_CLUSTER_REPLICA_UTILIZATION_HISTORY: LazyLock<BuiltinView> =
 SELECT
     r.id AS replica_id,
     m.process_id,
-    m.cpu_nano_cores::float8 / s.cpu_nano_cores * 100 AS cpu_percent,
-    m.memory_bytes::float8 / s.memory_bytes * 100 AS memory_percent,
-    m.disk_bytes::float8 / s.disk_bytes * 100 AS disk_percent,
+    m.cpu_nano_cores::float8 / NULLIF(s.cpu_nano_cores, 0) * 100 AS cpu_percent,
+    m.memory_bytes::float8 / NULLIF(s.memory_bytes, 0) * 100 AS memory_percent,
+    m.disk_bytes::float8 / NULLIF(s.disk_bytes, 0) * 100 AS disk_percent,
     m.occurred_at
 FROM
     mz_catalog.mz_cluster_replicas AS r
@@ -12279,9 +12279,9 @@ replica_metrics_history AS (
     m.occurred_at,
     m.replica_id,
     r.size,
-    (SUM(m.cpu_nano_cores::float8) / s.cpu_nano_cores) / s.processes AS cpu_percent,
-    (SUM(m.memory_bytes::float8) / s.memory_bytes) / s.processes AS memory_percent,
-    (SUM(m.disk_bytes::float8) / s.disk_bytes) / s.processes AS disk_percent,
+    (SUM(m.cpu_nano_cores::float8) / NULLIF(s.cpu_nano_cores, 0)) / s.processes AS cpu_percent,
+    (SUM(m.memory_bytes::float8) / NULLIF(s.memory_bytes, 0)) / s.processes AS memory_percent,
+    (SUM(m.disk_bytes::float8) / NULLIF(s.disk_bytes, 0)) / s.processes AS disk_percent,
     SUM(m.disk_bytes::float8) AS disk_bytes,
     SUM(m.memory_bytes::float8) AS memory_bytes,
     s.disk_bytes::numeric * s.processes AS total_disk_bytes,

src/clusterd/src/usage_metrics.rs

@@ -9,7 +9,8 @@
 
 //! Support for collecting system usage metrics.
 //!
-//! Currently only disk usage is supported. We may want to add CPU and memory usage in the future.
+//! Currently only disk and swap usage is supported.
+//! We may want to add CPU and memory usage in the future.
 
 use std::path::PathBuf;
 
@@ -26,6 +27,7 @@ impl Collector {
     pub fn collect(&self) -> Usage {
         Usage {
             disk_bytes: self.collect_disk_usage(),
+            swap_bytes: self.collect_swap_usage(),
         }
     }
 
@@ -49,10 +51,33 @@ impl Collector {
 
         Some(used_bytes)
     }
+
+    #[cfg(target_os = "linux")]
+    fn collect_swap_usage(&self) -> Option<u64> {
+        use mz_compute::memory_limiter::ProcStatus;
+        use mz_ore::cast::CastInto;
+
+        match ProcStatus::from_proc() {
+            Ok(status) => {
+                let bytes = status.vm_swap.cast_into();
+                Some(bytes)
+            }
+            Err(err) => {
+                error!("error reading /proc/self/status: {err}");
+                None
+            }
+        }
+    }
+
+    #[cfg(not(target_os = "linux"))]
+    fn collect_swap_usage(&self) -> Option<u64> {
+        None
+    }
 }
 
 /// A system usage measurement.
 #[derive(Serialize)]
 pub(crate) struct Usage {
     disk_bytes: Option<u64>,
+    swap_bytes: Option<u64>,
 }

src/compute/src/memory_limiter.rs

@@ -296,15 +296,17 @@ impl LimiterMetrics {
     }
 }
 
-struct ProcStatus {
+/// Helper for reading and parsing `/proc/self/status` on Linux.
+pub struct ProcStatus {
     /// Resident Set Size (RSS) in bytes.
-    vm_rss: usize,
+    pub vm_rss: usize,
     /// Swap memory in bytes.
-    vm_swap: usize,
+    pub vm_swap: usize,
 }
 
 impl ProcStatus {
-    fn from_proc() -> anyhow::Result<Self> {
+    /// Populate a new `ProcStatus` with information in /proc/self/status.
+    pub fn from_proc() -> anyhow::Result<Self> {
         let contents = std::fs::read_to_string("/proc/self/status")?;
         let mut vm_rss = 0;
         let mut vm_swap = 0;

src/orchestrator-kubernetes/src/lib.rs

@@ -215,8 +215,6 @@ impl Orchestrator for KubernetesOrchestrator {
 #[derive(Clone, Copy)]
 struct ServiceInfo {
     scale: u16,
-    disk: bool,
-    disk_limit: Option<DiskLimit>,
 }
 
 struct NamespacedKubernetesOrchestrator {
@@ -1228,14 +1226,10 @@ impl NamespacedOrchestrator for NamespacedKubernetesOrchestrator {
             },
         });
 
-        self.service_infos.lock().expect("poisoned lock").insert(
-            id.to_string(),
-            ServiceInfo {
-                scale,
-                disk,
-                disk_limit,
-            },
-        );
+        self.service_infos
+            .lock()
+            .expect("poisoned lock")
+            .insert(id.to_string(), ServiceInfo { scale });
 
         Ok(Box::new(KubernetesService { hosts, ports }))
     }
@@ -1445,31 +1439,16 @@ impl OrchestratorWorker {
             self_: &OrchestratorWorker,
             service_name: &str,
             i: usize,
-            disk: bool,
-            disk_limit: Option<DiskLimit>,
         ) -> ServiceProcessMetrics {
             let name = format!("{service_name}-{i}");
 
-            let disk_usage_fut = async {
-                if disk {
-                    Some(get_disk_usage(self_, service_name, i).await)
-                } else {
-                    None
-                }
-            };
+            let disk_usage_fut = get_disk_usage(self_, service_name, i);
             let (metrics, disk_usage) =
                 match futures::future::join(self_.metrics_api.get(&name), disk_usage_fut).await {
-                    (Ok(metrics), disk_usage) => {
-                        let disk_usage = match disk_usage {
-                            Some(Ok(disk_usage)) => Some(disk_usage),
-                            Some(Err(e)) => {
-                                warn!("Failed to fetch disk usage for {name}: {e}");
-                                None
-                            }
-                            _ => None,
-                        };
-
-                        (metrics, disk_usage)
+                    (Ok(metrics), Ok(disk_usage)) => (metrics, disk_usage),
+                    (Ok(metrics), Err(e)) => {
+                        warn!("Failed to fetch disk usage for {name}: {e}");
+                        (metrics, None)
                     }
                     (Err(e), _) => {
                         warn!("Failed to get metrics for {name}: {e}");
@@ -1516,21 +1495,6 @@ impl OrchestratorWorker {
                 }
             };
 
-            // We only populate a `disk_usage` if we have both:
-            // - a disk limit (so it must be an actual managed cluster with a real limit)
-            // - a reported disk usage
-            //
-            // The disk limit can be more up-to-date (from `service_infos`) than the
-            // reported metric. In that case, we report the minimum of the usage
-            // and the limit, which means we can report 100% usage temporarily
-            // if a replica is sized down.
-            let disk_usage = match (disk_usage, disk_limit) {
-                (Some(disk_usage), Some(DiskLimit(disk_limit))) => {
-                    Some(std::cmp::min(disk_usage, disk_limit.0))
-                }
-                _ => None,
-            };
-
             ServiceProcessMetrics {
                 cpu_nano_cores: cpu,
                 memory_bytes: memory,
@@ -1547,10 +1511,11 @@ impl OrchestratorWorker {
             self_: &OrchestratorWorker,
             service_name: &str,
             i: usize,
-        ) -> anyhow::Result<u64> {
+        ) -> anyhow::Result<Option<u64>> {
             #[derive(Deserialize)]
             pub(crate) struct Usage {
                 disk_bytes: Option<u64>,
+                swap_bytes: Option<u64>,
             }
 
             let service = self_
@@ -1584,16 +1549,21 @@ impl OrchestratorWorker {
                 .build()
                 .context("error building HTTP client")?;
             let resp = http_client.get(metrics_url).send().await?;
-            let usage: Usage = resp.json().await?;
+            let Usage {
+                disk_bytes,
+                swap_bytes,
+            } = resp.json().await?;
 
-            usage
-                .disk_bytes
-                .ok_or_else(|| anyhow!("process did not provide disk usage"))
+            let bytes = if let (Some(disk), Some(swap)) = (disk_bytes, swap_bytes) {
+                Some(disk + swap)
+            } else {
+                disk_bytes.or(swap_bytes)
+            };
+            Ok(bytes)
         }
 
-        let ret = futures::future::join_all(
-            (0..info.scale).map(|i| get_metrics(self, name, i.into(), info.disk, info.disk_limit)),
-        );
+        let ret =
+            futures::future::join_all((0..info.scale).map(|i| get_metrics(self, name, i.into())));
 
         ret.await
     }

test/sqllogictest/system-cluster.slt

@@ -411,7 +411,7 @@ ORDER BY r.id;
 Explained Query:
   Finish order_by=[#0{id} asc nulls_last] output=[#0..=#5]
     Project (#0{id}..=#3{size}, #5{name}, #29)
-      Map (((uint8_to_double(#27{memory_bytes}) / uint8_to_double(#21{memory_bytes})) * 100))
+      Map (((uint8_to_double(#27{memory_bytes}) / uint8_to_double(case when (0 = uint8_to_numeric(#21{memory_bytes})) then null else #21{memory_bytes} end)) * 100))
         Join on=(#0{id} = #15{id} = #24{replica_id} AND #2{cluster_id} = #4{id} AND #16{size} = #17{size}) type=delta
           ArrangeBy keys=[[#0{id}], [#2{cluster_id}]]
             Project (#0{id}..=#3{size})