core_lib/exec/
metrics.rs

1#![allow(dead_code)]
2use core::f32;
3use std::{collections::HashMap, path::PathBuf, time::Duration};
4
5use chrono::{DateTime, Utc};
6use dirs::home_dir;
7use serde::{Deserialize, Serialize};
8use tokio::{
9    fs::{self},
10    io::AsyncWriteExt,
11    time::sleep,
12};
13
14#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
15pub enum JobStatus {
16    Pending,
17    Running,
18    Succeeded,
19    Failed,
20    Skipped,
21}
22
23/// Cap on the in-memory per-job sampling buffer (~36s at the 100ms sampling
24/// rate). Keeps memory bounded on long-running steps.
25const MAX_BUF_SAMPLES: usize = 360;
26
27#[derive(Debug, Clone, Serialize, Deserialize)]
28pub struct JobMetrics {
29    pub name: String,
30    pub status: JobStatus,
31    pub started_at: DateTime<Utc>,
32    pub finished_at: Option<DateTime<Utc>>,
33    pub duration_ms: Option<u128>,
34    pub cpu_usage: f32,
35    pub mem_usage: f32,
36    pub mem_usage_kb: u64,
37    pub max_cpu: f32,
38    pub max_mem: f32,
39    #[serde(skip)]
40    pub buf: Vec<(f32, u64)>,
41}
42
43#[derive(Debug, Clone, Serialize, Deserialize)]
44pub struct ExecMetrics {
45    pub project_id: String,
46    pub project_name: String,
47
48    pub started_at: DateTime<Utc>,
49    pub finished_at: Option<DateTime<Utc>>,
50    pub duration_ms: Option<u128>,
51
52    pub cpu_usage: f32,
53    pub mem_usage: f32,
54    pub mem_usage_kb: u64,
55    pub max_cpu: f32,
56    pub max_mem: f32,
57
58    pub jobs: HashMap<String, JobMetrics>,
59}
60
61impl ExecMetrics {
62    pub fn new(project_id: &str, project_name: &str) -> Self {
63        Self {
64            project_id: project_id.to_string(),
65            project_name: project_name.to_string(),
66            started_at: Utc::now(),
67            finished_at: None,
68            duration_ms: None,
69            cpu_usage: 0.0,
70            mem_usage: 0.0,
71            mem_usage_kb: 0,
72            max_cpu: 0.0,
73            max_mem: 0.0,
74            jobs: std::collections::HashMap::new(),
75        }
76    }
77
78    pub fn sys_push(&mut self, name: &str, cpu: f32, mem: u64) {
79        if let Some(j) = self.jobs.get_mut(name) {
80            j.buf.push((cpu, mem));
81            if j.buf.len() > MAX_BUF_SAMPLES {
82                j.buf.remove(0);
83            }
84        }
85    }
86
87    /// call at the end of the pipeline
88    pub fn finalize(&mut self) {
89        let end = Utc::now();
90        self.finished_at = Some(end);
91        self.duration_ms = Some(
92            end.signed_duration_since(self.started_at)
93                .num_milliseconds() as u128,
94        );
95        let v: Vec<(f32, f32, u64)> = self
96            .jobs
97            .values()
98            .map(|v| (v.cpu_usage, v.mem_usage, v.mem_usage_kb))
99            .collect();
100
101        if !v.is_empty() {
102            let mut sys = sysinfo::System::new_all();
103            sys.refresh_memory();
104            let cpu_sum: f32 = v.iter().map(|(cpu, _, _)| *cpu).sum();
105            let cpu_count = v.len() as f32;
106            let average_cpu = cpu_sum / cpu_count;
107
108            let max_cpu = v
109                .iter()
110                .map(|(cpu, _, _)| *cpu)
111                .fold(0.0_f32, |a, b| a.max(b));
112
113            let mem_usage_kb: Vec<u64> = v.iter().map(|(_, _, mem)| *mem).collect();
114            let avg_mem_usage_kb = mem_usage_kb.iter().sum::<u64>();
115
116            let mem_percentages: Vec<f32> = v.iter().map(|(_, mem, _)| *mem).collect(); // déjà en %
117            let average_mem_percent: f32 =
118                mem_percentages.iter().sum::<f32>() / mem_percentages.len() as f32;
119            let max_mem_percent: f32 = mem_percentages.iter().fold(0.0, |a, b| a.max(*b));
120
121            self.cpu_usage = average_cpu;
122            self.mem_usage_kb = avg_mem_usage_kb;
123            self.mem_usage = average_mem_percent;
124            self.max_cpu = max_cpu;
125            self.max_mem = max_mem_percent;
126        } else {
127            self.cpu_usage = 0.0;
128            self.mem_usage_kb = 0;
129            self.max_cpu = 0.0;
130            self.max_mem = 0.0;
131        }
132    }
133
134    /// start job (just insert it with Running status)
135    pub fn job_started(&mut self, name: &str) {
136        self.jobs.insert(
137            name.to_string(),
138            JobMetrics {
139                name: name.to_string(),
140                status: JobStatus::Running,
141                started_at: Utc::now(),
142                finished_at: None,
143                duration_ms: None,
144                cpu_usage: 0.0,
145                mem_usage: 0.0,
146                mem_usage_kb: 0,
147                max_cpu: 0.0,
148                max_mem: 0.0,
149                buf: Vec::new(),
150            },
151        );
152    }
153
154    pub fn job_finished(&mut self, name: &str, ok: bool) {
155        if let Some(j) = self.jobs.get_mut(name) {
156            if !j.buf.is_empty() {
157                let mut sys = sysinfo::System::new_all();
158                sys.refresh_memory();
159                let total_memory_kb = sys.total_memory() as f32;
160                let cpu_sum: f32 = j.buf.iter().map(|(value, _)| *value).sum();
161                let count = j.buf.len() as f32;
162
163                let average_cpu = cpu_sum / count;
164                let max_cpu = j.buf.iter().map(|(v, _)| *v).fold(0.0_f32, |a, b| a.max(b));
165
166                let mem_usage_kb: Vec<u64> = j.buf.iter().map(|(_, mem)| *mem).collect();
167                let avg_mem_usage_kb = mem_usage_kb.iter().sum::<u64>();
168
169                let mem_percentages: Vec<f32> = j
170                    .buf
171                    .iter()
172                    .map(|(_, mem)| (*mem as f32 / total_memory_kb) * 100.0)
173                    .collect();
174                let average_mem_percent: f32 =
175                    mem_percentages.iter().sum::<f32>() / mem_percentages.len() as f32;
176                let max_mem_percent: f32 = mem_percentages.iter().fold(0.0, |a, b| a.max(*b));
177
178                j.cpu_usage = average_cpu;
179                j.mem_usage_kb = avg_mem_usage_kb;
180                j.mem_usage = average_mem_percent;
181                j.max_mem = max_mem_percent;
182                j.max_cpu = max_cpu;
183            }
184            let end = Utc::now();
185            j.finished_at = Some(end);
186            j.duration_ms =
187                Some(end.signed_duration_since(j.started_at).num_milliseconds() as u128);
188            j.status = if ok {
189                JobStatus::Succeeded
190            } else {
191                JobStatus::Failed
192            }
193        }
194    }
195
196    // -----------------------
197    // Metrics persistance
198    // -----------------------
199
200    pub async fn ensure_metrics_dir() -> anyhow::Result<PathBuf> {
201        let home = home_dir().ok_or_else(|| anyhow::anyhow!("Failed to find HOME directory"))?;
202        let dir = home.join(".fleet").join("metrics");
203        if !fs::try_exists(&dir).await? {
204            fs::create_dir_all(&dir).await?;
205        }
206        Ok(dir)
207    }
208
209    pub fn get_metrics_path_by_id(id: &str) -> anyhow::Result<PathBuf> {
210        let home = home_dir().ok_or_else(|| anyhow::anyhow!("Failed to find HOME directory"))?;
211        let dir = home
212            .join(".fleet")
213            .join("metrics")
214            .join(id.to_owned() + ".ndjson");
215        Ok(dir)
216    }
217
218    pub fn rm_metrics_by_id(id: &str) -> anyhow::Result<()> {
219        let path = ExecMetrics::get_metrics_path_by_id(id)?;
220        if path.exists() {
221            std::fs::remove_file(path)?;
222        }
223        Ok(())
224    }
225
226    pub async fn open_metrics_file(project_id: &str) -> anyhow::Result<tokio::fs::File> {
227        let dir = Self::ensure_metrics_dir().await?;
228        let path = dir.join(format!("{project_id}.ndjson"));
229        let file = tokio::fs::OpenOptions::new()
230            .write(true)
231            .truncate(true)
232            .create(true)
233            .open(path)
234            .await?;
235        Ok(file)
236    }
237
238    pub async fn save(&self) -> anyhow::Result<()> {
239        let mut file = Self::open_metrics_file(&self.project_id).await?;
240        let line = serde_json::to_string(self)?;
241        file.write_all(line.as_bytes()).await?;
242        file.write_all(b"\n").await?;
243        file.flush().await?;
244        Ok(())
245    }
246}
247
248pub async fn monitor_process(pid: u32) -> (f32, u64) {
249    let mut sys = sysinfo::System::new_all();
250    let mut samples_cpu = vec![];
251    let mut max_mem = 0;
252
253    loop {
254        sys.refresh_processes(sysinfo::ProcessesToUpdate::All, true);
255        if let Some(proc) = sys.process(sysinfo::Pid::from(pid as usize)) {
256            samples_cpu.push(proc.cpu_usage());
257            max_mem = max_mem.max(proc.memory());
258        } else {
259            // Process finished
260            break;
261        }
262        sleep(Duration::from_millis(100)).await;
263    }
264
265    let avg_cpu = if samples_cpu.is_empty() {
266        0.0
267    } else {
268        samples_cpu.iter().sum::<f32>() / samples_cpu.len() as f32
269    };
270
271    (avg_cpu, max_mem)
272}
core_lib/exec/metrics.rs

core_lib/exec/
metrics.rs
