`3b4c806`

add data collectors (cpu, memory, process) and history ring buffer

Authored by mfwolffe <wolffemf@dukes.jmu.edu> 3 months ago

SHA: 3b4c80697017735f22ec47b3c98fbf8207d2dfad
Parents: b9657a9
Tree: f77bcac

5 changed files

Status	File	+	-
A	`gartop/src/collector/cpu.rs`	128	0
A	`gartop/src/collector/history.rs`	105	0
A	`gartop/src/collector/memory.rs`	63	0
M	`gartop/src/collector/mod.rs`	9	1
A	`gartop/src/collector/process.rs`	174	0

gartop/src/collector/cpu.rsadded

 +//! CPU statistics collection from /proc/stat
++
 +use crate::error::Result;
 +use gartop_ipc::CpuStats;
 +use procfs::CurrentSI;
 +use std::time::{SystemTime, UNIX_EPOCH};
++
 +/// CPU times for calculating deltas.
 +#[derive(Debug, Clone, Default)]
 +struct CpuTimes {
 +    user: u64,
 +    nice: u64,
 +    system: u64,
 +    idle: u64,
 +    iowait: u64,
 +    irq: u64,
 +    softirq: u64,
 +    steal: u64,
 +}
++
 +impl CpuTimes {
 +    fn total(&self) -> u64 {
 +        self.user
 +            + self.nice
 +            + self.system
 +            + self.idle
 +            + self.iowait
 +            + self.irq
 +            + self.softirq
 +            + self.steal
 +    }
++
 +    fn active(&self) -> u64 {
 +        self.user + self.nice + self.system + self.irq + self.softirq + self.steal
 +    }
 +}
++
 +/// CPU collector with state for delta calculations.
 +pub struct CpuCollector {
 +    prev_total: CpuTimes,
 +    prev_per_core: Vec<CpuTimes>,
 +    core_count: usize,
 +}
++
 +impl CpuCollector {
 +    /// Create a new CPU collector.
 +    pub fn new() -> Result<Self> {
 +        let kernel_stats = procfs::KernelStats::current()?;
 +        let core_count = kernel_stats.cpu_time.len().max(1);
++
 +        Ok(Self {
 +            prev_total: CpuTimes::default(),
 +            prev_per_core: vec![CpuTimes::default(); core_count],
 +            core_count,
 +        })
 +    }
++
 +    /// Collect current CPU statistics.
 +    pub fn collect(&mut self) -> Result<CpuStats> {
 +        let kernel_stats = procfs::KernelStats::current()?;
++
 +        // Parse total CPU times
 +        let total = &kernel_stats.total;
 +        let curr_total = CpuTimes {
 +            user: total.user,
 +            nice: total.nice,
 +            system: total.system,
 +            idle: total.idle,
 +            iowait: total.iowait.unwrap_or(0),
 +            irq: total.irq.unwrap_or(0),
 +            softirq: total.softirq.unwrap_or(0),
 +            steal: total.steal.unwrap_or(0),
 +        };
++
 +        // Calculate overall usage
 +        let total_delta = curr_total.total().saturating_sub(self.prev_total.total());
 +        let active_delta = curr_total.active().saturating_sub(self.prev_total.active());
 +        let usage_percent = if total_delta > 0 {
 +            (active_delta as f64 / total_delta as f64) * 100.0
 +        } else {
 +            0.0
 +        };
++
 +        // Calculate per-core usage
 +        let mut per_core = Vec::with_capacity(self.core_count);
 +        for (i, cpu) in kernel_stats.cpu_time.iter().enumerate() {
 +            if i >= self.core_count {
 +                break;
 +            }
 +            let curr = CpuTimes {
 +                user: cpu.user,
 +                nice: cpu.nice,
 +                system: cpu.system,
 +                idle: cpu.idle,
 +                iowait: cpu.iowait.unwrap_or(0),
 +                irq: cpu.irq.unwrap_or(0),
 +                softirq: cpu.softirq.unwrap_or(0),
 +                steal: cpu.steal.unwrap_or(0),
 +            };
++
 +            let prev: &CpuTimes = &self.prev_per_core[i];
 +            let total_d = curr.total().saturating_sub(prev.total());
 +            let active_d = curr.active().saturating_sub(prev.active());
 +            let core_usage = if total_d > 0 {
 +                (active_d as f64 / total_d as f64) * 100.0
 +            } else {
 +                0.0
 +            };
 +            per_core.push(core_usage);
++
 +            self.prev_per_core[i] = curr;
 +        }
++
 +        self.prev_total = curr_total;
++
 +        let timestamp = SystemTime::now()
 +            .duration_since(UNIX_EPOCH)
 +            .map(|d| d.as_millis() as u64)
 +            .unwrap_or(0);
++
 +        Ok(CpuStats {
 +            usage_percent,
 +            per_core,
 +            core_count: self.core_count,
 +            timestamp,
 +        })
 +    }
 +}

gartop/src/collector/history.rsadded

 +//! Generic ring buffer for time-series history
++
 +use std::collections::VecDeque;
++
 +/// Ring buffer for maintaining a fixed-size history of samples.
 +#[derive(Debug, Clone)]
 +pub struct History<T> {
 +    buffer: VecDeque<T>,
 +    capacity: usize,
 +}
++
 +impl<T: Clone> History<T> {
 +    /// Create a new history buffer with the given capacity.
 +    pub fn new(capacity: usize) -> Self {
 +        Self {
 +            buffer: VecDeque::with_capacity(capacity),
 +            capacity,
 +        }
 +    }
++
 +    /// Push a new sample, evicting the oldest if at capacity.
 +    pub fn push(&mut self, value: T) {
 +        if self.buffer.len() >= self.capacity {
 +            self.buffer.pop_front();
 +        }
 +        self.buffer.push_back(value);
 +    }
++
 +    /// Get the most recent sample.
 +    pub fn latest(&self) -> Option<&T> {
 +        self.buffer.back()
 +    }
++
 +    /// Get all samples as owned values (oldest to newest).
 +    pub fn to_vec(&self) -> Vec<T> {
 +        self.buffer.iter().cloned().collect()
 +    }
++
 +    /// Get the last N samples as owned values (oldest to newest).
 +    pub fn last_n(&self, n: usize) -> Vec<T> {
 +        let skip = self.buffer.len().saturating_sub(n);
 +        self.buffer.iter().skip(skip).cloned().collect()
 +    }
++
 +    /// Current number of samples.
 +    pub fn len(&self) -> usize {
 +        self.buffer.len()
 +    }
++
 +    /// Check if empty.
 +    pub fn is_empty(&self) -> bool {
 +        self.buffer.is_empty()
 +    }
++
 +    /// Clear all samples.
 +    pub fn clear(&mut self) {
 +        self.buffer.clear();
 +    }
 +}
++
 +impl<T: Clone> Default for History<T> {
 +    fn default() -> Self {
 +        Self::new(300) // Default: 5 minutes at 1 sample/second
 +    }
 +}
++
 +#[cfg(test)]
 +mod tests {
 +    use super::*;
++
 +    #[test]
 +    fn test_push_and_latest() {
 +        let mut h: History<i32> = History::new(3);
 +        assert!(h.latest().is_none());
++
 +        h.push(1);
 +        assert_eq!(h.latest(), Some(&1));
++
 +        h.push(2);
 +        assert_eq!(h.latest(), Some(&2));
 +    }
++
 +    #[test]
 +    fn test_capacity() {
 +        let mut h: History<i32> = History::new(3);
 +        h.push(1);
 +        h.push(2);
 +        h.push(3);
 +        assert_eq!(h.len(), 3);
++
 +        h.push(4);
 +        assert_eq!(h.len(), 3);
 +        assert_eq!(h.to_vec(), vec![2, 3, 4]);
 +    }
++
 +    #[test]
 +    fn test_last_n() {
 +        let mut h: History<i32> = History::new(5);
 +        for i in 1..=5 {
 +            h.push(i);
 +        }
 +        assert_eq!(h.last_n(3), vec![3, 4, 5]);
 +        assert_eq!(h.last_n(10), vec![1, 2, 3, 4, 5]);
 +    }
 +}

gartop/src/collector/memory.rsadded

 +//! Memory statistics collection from /proc/meminfo
++
 +use crate::error::Result;
 +use gartop_ipc::MemoryStats;
 +use procfs::Current;
 +use std::time::{SystemTime, UNIX_EPOCH};
++
 +/// Memory collector.
 +pub struct MemoryCollector;
++
 +impl MemoryCollector {
 +    /// Create a new memory collector.
 +    pub fn new() -> Self {
 +        Self
 +    }
++
 +    /// Collect current memory statistics.
 +    pub fn collect(&self) -> Result<MemoryStats> {
 +        let meminfo = procfs::Meminfo::current()?;
++
 +        let total = meminfo.mem_total;
 +        let free = meminfo.mem_free;
 +        let available = meminfo.mem_available.unwrap_or(free);
 +        let buffers = meminfo.buffers;
 +        let cached = meminfo.cached;
 +        let slab_reclaimable = meminfo.s_reclaimable.unwrap_or(0);
++
 +        // Calculate used memory (excluding buffers/cache)
 +        let used = total.saturating_sub(free + buffers + cached + slab_reclaimable);
++
 +        let swap_total = meminfo.swap_total;
 +        let swap_free = meminfo.swap_free;
 +        let swap_used = swap_total.saturating_sub(swap_free);
++
 +        let usage_percent = if total > 0 {
 +            (used as f64 / total as f64) * 100.0
 +        } else {
 +            0.0
 +        };
++
 +        let timestamp = SystemTime::now()
 +            .duration_since(UNIX_EPOCH)
 +            .map(|d| d.as_millis() as u64)
 +            .unwrap_or(0);
++
 +        Ok(MemoryStats {
 +            total,
 +            used,
 +            free,
 +            available,
 +            swap_total,
 +            swap_used,
 +            usage_percent,
 +            timestamp,
 +        })
 +    }
 +}
++
 +impl Default for MemoryCollector {
 +    fn default() -> Self {
 +        Self::new()
 +    }
 +}

gartop/src/collector/mod.rsmodified

  //!
  //! Collects CPU, memory, and process data from procfs.
 -// TODO: Sprint 2 - implement collectors
 +mod cpu;
 +mod history;
 +mod memory;
 +mod process;
++
 +pub use cpu::CpuCollector;
 +pub use history::History;
 +pub use memory::MemoryCollector;
 +pub use process::ProcessCollector;

gartop/src/collector/process.rsadded

 +//! Process information collection from /proc/[pid]/
++
 +use crate::error::{Error, Result};
 +use gartop_ipc::{ProcessInfo, SortField};
 +use procfs::process::{all_processes, Process};
 +use procfs::{Current, CurrentSI};
 +use std::collections::HashMap;
++
 +/// Process collector with previous CPU times for percentage calculation.
 +pub struct ProcessCollector {
 +    /// Previous CPU times per PID for delta calculation.
 +    prev_times: HashMap<i32, (u64, u64)>, // (utime + stime, total_system_time)
 +    /// Total memory for percentage calculation.
 +    total_memory: u64,
 +}
++
 +impl ProcessCollector {
 +    /// Create a new process collector.
 +    pub fn new() -> Result<Self> {
 +        let meminfo = procfs::Meminfo::current()?;
++
 +        Ok(Self {
 +            prev_times: HashMap::new(),
 +            total_memory: meminfo.mem_total,
 +        })
 +    }
++
 +    /// Collect all running processes.
 +    pub fn collect(
 +        &mut self,
 +        sort_by: SortField,
 +        limit: Option<usize>,
 +    ) -> Result<Vec<ProcessInfo>> {
 +        let kernel_stats = procfs::KernelStats::current()?;
 +        let total = &kernel_stats.total;
 +        let system_total = total.user
 +            + total.nice
 +            + total.system
 +            + total.idle
 +            + total.iowait.unwrap_or(0)
 +            + total.irq.unwrap_or(0)
 +            + total.softirq.unwrap_or(0)
 +            + total.steal.unwrap_or(0);
++
 +        let mut processes = Vec::new();
 +        let mut current_pids = Vec::new();
++
 +        for proc_result in all_processes()? {
 +            let proc = match proc_result {
 +                Ok(p) => p,
 +                Err(_) => continue,
 +            };
++
 +            let pid = proc.pid();
 +            current_pids.push(pid);
++
 +            match self.process_info(&proc, system_total) {
 +                Ok(info) => processes.push(info),
 +                Err(_) => continue, // Skip processes we can't read
 +            }
 +        }
++
 +        // Clean up old entries
 +        self.prev_times.retain(|pid, _| current_pids.contains(pid));
++
 +        // Sort
 +        match sort_by {
 +            SortField::Cpu => processes.sort_by(|a, b| {
 +                b.cpu_percent
 +                    .partial_cmp(&a.cpu_percent)
 +                    .unwrap_or(std::cmp::Ordering::Equal)
 +            }),
 +            SortField::Memory => processes.sort_by(|a, b| {
 +                b.memory_percent
 +                    .partial_cmp(&a.memory_percent)
 +                    .unwrap_or(std::cmp::Ordering::Equal)
 +            }),
 +            SortField::Pid => processes.sort_by_key(|p| p.pid),
 +            SortField::Name => processes.sort_by(|a, b| a.name.cmp(&b.name)),
 +        }
++
 +        // Limit
 +        if let Some(n) = limit {
 +            processes.truncate(n);
 +        }
++
 +        Ok(processes)
 +    }
++
 +    /// Get info for a single process.
 +    fn process_info(&mut self, proc: &Process, system_total: u64) -> Result<ProcessInfo> {
 +        let stat = proc.stat()?;
 +        let status = proc.status()?;
++
 +        let pid = proc.pid();
 +        let name = stat.comm.clone();
 +        let cmdline = proc
 +            .cmdline()
 +            .map(|v| v.join(" "))
 +            .unwrap_or_else(|_| name.clone());
++
 +        // Calculate CPU percentage
 +        let proc_time = stat.utime + stat.stime;
 +        let cpu_percent =
 +            if let Some((prev_proc_time, prev_sys_time)) = self.prev_times.get(&pid) {
 +                let proc_delta = proc_time.saturating_sub(*prev_proc_time);
 +                let sys_delta = system_total.saturating_sub(*prev_sys_time);
 +                if sys_delta > 0 {
 +                    (proc_delta as f64 / sys_delta as f64) * 100.0
 +                } else {
 +                    0.0
 +                }
 +            } else {
 +                0.0
 +            };
 +        self.prev_times.insert(pid, (proc_time, system_total));
++
 +        // Memory
 +        let rss = stat.rss as u64 * 4096; // Pages to bytes
 +        let vsize = stat.vsize;
 +        let memory_percent = if self.total_memory > 0 {
 +            (rss as f64 / self.total_memory as f64) * 100.0
 +        } else {
 +            0.0
 +        };
++
 +        // State
 +        let state = match stat.state {
 +            'R' => "running",
 +            'S' => "sleeping",
 +            'D' => "disk sleep",
 +            'Z' => "zombie",
 +            'T' => "stopped",
 +            't' => "tracing stop",
 +            'X' | 'x' => "dead",
 +            _ => "unknown",
 +        }
 +        .to_string();
++
 +        // User
 +        let user = users::get_user_by_uid(status.ruid)
 +            .map(|u| u.name().to_string_lossy().to_string())
 +            .unwrap_or_else(|| status.ruid.to_string());
++
 +        Ok(ProcessInfo {
 +            pid,
 +            name,
 +            cmdline,
 +            cpu_percent,
 +            memory_percent,
 +            rss,
 +            vsize,
 +            state,
 +            user,
 +        })
 +    }
++
 +    /// Kill a process by PID.
 +    pub fn kill(&self, pid: i32, signal: i32) -> Result<()> {
 +        use nix::sys::signal::{kill, Signal};
 +        use nix::unistd::Pid;
++
 +        let sig = Signal::try_from(signal)
 +            .map_err(|_| Error::Ipc(format!("Invalid signal: {}", signal)))?;
++
 +        kill(Pid::from_raw(pid), sig).map_err(|e| match e {
 +            nix::errno::Errno::ESRCH => Error::ProcessNotFound(pid),
 +            nix::errno::Errno::EPERM => {
 +                Error::PermissionDenied(format!("Cannot kill PID {}", pid))
 +            }
 +            _ => Error::Io(std::io::Error::from_raw_os_error(e as i32)),
 +        })
 +    }
 +}