Open History to see the last commit
1use serde::{Deserialize, Serialize};
2use std::collections::{HashMap, HashSet, VecDeque};
3use std::io::{self, BufRead, BufWriter, Write};
4use std::sync::mpsc::{self, Receiver, RecvTimeoutError};
5use std::thread;
6use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
7use sysinfo::{
8 MINIMUM_CPU_UPDATE_INTERVAL, Pid, ProcessRefreshKind, ProcessesToUpdate, System, UpdateKind,
9};
10
11const PROTOCOL_VERSION: u32 = 3;
12const MIN_SAMPLE_INTERVAL_MS: u64 = 250;
13const MAX_SAMPLE_INTERVAL_MS: u64 = 60_000;
14const PROCESS_START_TIME_PRECISION_MS: u64 = 1_000;
15const HISTORY_RETENTION_MS: u64 = 60 * 60_000;
16const MAX_HISTORY_SNAPSHOTS: usize = 3_600;
17const INPUT_QUEUE_CAPACITY: usize = 64;
18const MAX_HISTORY_RETAINED_ENTRIES: usize = 20_000;
19const MAX_HISTORY_RETAINED_BYTES: usize = 64 * 1024 * 1024;
20const MAX_PROCESS_NAME_BYTES: usize = 1_024;
21const MAX_PROCESS_COMMAND_BYTES: usize = 16 * 1_024;
22const MAX_PROCESS_STATUS_BYTES: usize = 256;
23const HISTORY_CHUNK_SNAPSHOTS: usize = 32;
24const TWO_STAGE_PROCESS_REFRESH: bool = cfg!(any(target_os = "linux", target_os = "macos"));
25
26#[derive(Debug, Clone, Deserialize, Serialize)]
27#[serde(rename_all = "camelCase")]
28struct ExternalProcess {
29 pid: u32,
30 #[serde(default)]
31 start_time_ms: Option<u64>,
32}
33
34impl ExternalProcess {
35 fn estimated_history_bytes(&self) -> usize {
36 std::mem::size_of::<Self>()
37 }
38}
39
40#[derive(Debug, Deserialize)]
41#[serde(
42 tag = "type",
43 rename_all = "camelCase",
44 rename_all_fields = "camelCase"
45)]
46enum Command {
47 Configure {
48 version: u32,
49 root_pid: u32,
50 sample_interval_ms: u64,
51 #[serde(default)]
52 external_processes: Vec<ExternalProcess>,
53 },
54 SetExternalProcesses {
55 version: u32,
56 processes: Vec<ExternalProcess>,
57 },
58 SetSampleInterval {
59 version: u32,
60 sample_interval_ms: u64,
61 },
62 SetStreaming {
63 version: u32,
64 enabled: bool,
65 },
66 SampleNow {
67 version: u32,
68 request_id: String,
69 },
70 ProcessTable {
71 version: u32,
72 request_id: String,
73 },
74 ReadHistory {
75 version: u32,
76 request_id: String,
77 window_ms: u64,
78 },
79 Shutdown {
80 version: u32,
81 },
82}
83
84impl Command {
85 fn version(&self) -> u32 {
86 match self {
87 Self::Configure { version, .. }
88 | Self::SetExternalProcesses { version, .. }
89 | Self::SetSampleInterval { version, .. }
90 | Self::SetStreaming { version, .. }
91 | Self::SampleNow { version, .. }
92 | Self::ProcessTable { version, .. }
93 | Self::ReadHistory { version, .. }
94 | Self::Shutdown { version } => *version,
95 }
96 }
97}
98
99enum Input {
100 Command(Command),
101 Invalid(String),
102}
103
104#[derive(Debug, Serialize)]
105#[serde(rename_all = "camelCase")]
106struct Capabilities {
107 cumulative_cpu_time: bool,
108 current_cpu_percent: bool,
109 resident_memory: bool,
110 virtual_memory: bool,
111 io_bytes: bool,
112 process_start_time: bool,
113 process_tree: bool,
114}
115
116#[derive(Debug, Serialize)]
117#[serde(rename_all = "camelCase")]
118struct HelloEvent {
119 version: u32,
120 #[serde(rename = "type")]
121 event_type: & str,
122 sidecar_version: & str,
123 sidecar_pid: u32,
124 platform: & str,
125 arch: & str,
126 capabilities: Capabilities,
127}
128
129#[derive(Debug, Clone, Copy, Serialize)]
130#[serde(rename_all = "kebab-case")]
131enum IoSemantics {
132 Storage,
133 AllIo,
134}
135
136#[derive(Debug, Clone, Serialize)]
137#[serde(rename_all = "camelCase")]
138struct ProcessSample {
139 pid: u32,
140 ppid: u32,
141 start_time_ms: u64,
142 run_time_ms: u64,
143 name: String,
144 command: String,
145 status: String,
146 cpu_percent: f32,
147 cpu_time_ms: u64,
148 resident_bytes: u64,
149 virtual_bytes: u64,
150 io_read_bytes: u64,
151 io_write_bytes: u64,
152 io_semantics: IoSemantics,
153}
154
155#[derive(Debug, Serialize)]
156#[serde(rename_all = "camelCase")]
157struct ProcessTableEntry {
158 pid: u32,
159 ppid: u32,
160 name: String,
161}
162
163#[derive(Debug, Serialize)]
164#[serde(rename_all = "camelCase")]
165struct ProcessTableEvent<> {
166 version: u32,
167 #[serde(rename = "type")]
168 event_type: & str,
169 request_id: & str,
170 processes: Vec<ProcessTableEntry>,
171}
172
173impl ProcessSample {
174 fn estimated_history_bytes(&self) -> usize {
175 std::mem::size_of::<Self>()
176 .saturating_add(self.name.len())
177 .saturating_add(self.command.len())
178 .saturating_add(self.status.len())
179 }
180}
181
182#[derive(Debug, Clone, Serialize)]
183#[serde(rename_all = "camelCase")]
184struct SnapshotEvent {
185 version: u32,
186 #[serde(rename = "type")]
187 event_type: & str,
188 sequence: u64,
189 sampled_at_unix_ms: u64,
190 collection_duration_micros: u64,
191 scanned_process_count: usize,
192 retained_process_count: usize,
193 inaccessible_process_count: usize,
194 #[serde(skip_serializing_if = "Option::is_none")]
195 request_id: Option<String>,
196 external_processes: Vec<ExternalProcess>,
197 processes: Vec<ProcessSample>,
198}
199
200impl SnapshotEvent {
201 fn retained_entry_count(&self) -> usize {
202 self.processes
203 .len()
204 .saturating_add(self.external_processes.len())
205 }
206
207 fn estimated_history_bytes(&self) -> usize {
208 std::mem::size_of::<Self>()
209 .saturating_add(
210 self.processes
211 .iter()
212 .map(ProcessSample::estimated_history_bytes)
213 .sum::<usize>(),
214 )
215 .saturating_add(
216 self.external_processes
217 .iter()
218 .map(ExternalProcess::estimated_history_bytes)
219 .sum::<usize>(),
220 )
221 }
222}
223
224#[derive(Debug, Serialize)]
225#[serde(rename_all = "camelCase")]
226struct HistoryChunkEvent<> {
227 version: u32,
228 #[serde(rename = "type")]
229 event_type: & str,
230 request_id: & str,
231 done: bool,
232 snapshots: & [SnapshotEvent],
233}
234
235#[derive(Debug, Serialize)]
236#[serde(rename_all = "camelCase")]
237struct ErrorEvent {
238 version: u32,
239 #[serde(rename = "type")]
240 event_type: & str,
241 code: & str,
242 message: String,
243 recoverable: bool,
244}
245
246#[derive(Debug, Clone)]
247struct CollectorConfig {
248 root_pid: u32,
249 sample_interval: Option<Duration>,
250 external_processes: HashMap<u32, Option<u64>>,
251}
252
253#[derive(Default)]
254struct HistoryRecorder {
255 snapshots: VecDeque<SnapshotEvent>,
256 retained_entry_count: usize,
257 retained_bytes: usize,
258}
259
260impl HistoryRecorder {
261 fn record(&mut self, snapshot: &SnapshotEvent) {
262 self.record_with_limits(
263 snapshot,
264 MAX_HISTORY_SNAPSHOTS,
265 MAX_HISTORY_RETAINED_ENTRIES,
266 MAX_HISTORY_RETAINED_BYTES,
267 );
268 }
269
270 fn record_with_limits(
271 &mut self,
272 snapshot: &SnapshotEvent,
273 max_snapshots: usize,
274 max_retained_entries: usize,
275 max_retained_bytes: usize,
276 ) {
277 let clock_moved_backward = self
278 .snapshots
279 .back()
280 .is_some_and(|previous| previous.sampled_at_unix_ms > snapshot.sampled_at_unix_ms);
281 let mut retained = snapshot.clone();
282 retained.request_id = None;
283 self.retained_entry_count = self
284 .retained_entry_count
285 .saturating_add(retained.retained_entry_count());
286 self.retained_bytes = self
287 .retained_bytes
288 .saturating_add(retained.estimated_history_bytes());
289 self.snapshots.push_back(retained);
290 self.trim_to_limits(
291 snapshot.sampled_at_unix_ms,
292 max_snapshots,
293 max_retained_entries,
294 max_retained_bytes,
295 clock_moved_backward,
296 );
297 }
298
299 fn trim_to_limits(
300 &mut self,
301 now_ms: u64,
302 max_snapshots: usize,
303 max_retained_entries: usize,
304 max_retained_bytes: usize,
305 clock_moved_backward: bool,
306 ) {
307 if clock_moved_backward {
308 let mut future_entry_count = 0usize;
309 let mut future_bytes = 0usize;
310 self.snapshots.retain(|snapshot| {
311 let keep = snapshot.sampled_at_unix_ms <= now_ms;
312 if !keep {
313 future_entry_count =
314 future_entry_count.saturating_add(snapshot.retained_entry_count());
315 future_bytes = future_bytes.saturating_add(snapshot.estimated_history_bytes());
316 }
317 keep
318 });
319 self.retained_entry_count =
320 self.retained_entry_count.saturating_sub(future_entry_count);
321 self.retained_bytes = self.retained_bytes.saturating_sub(future_bytes);
322 }
323
324 while self.snapshots.front().is_some_and(|snapshot| {
325 snapshot.sampled_at_unix_ms < now_ms.saturating_sub(HISTORY_RETENTION_MS)
326 || self.snapshots.len() > max_snapshots
327 || self.retained_entry_count > max_retained_entries
328 || self.retained_bytes > max_retained_bytes
329 }) {
330 if let Some(removed) = self.snapshots.pop_front() {
331 self.retained_entry_count = self
332 .retained_entry_count
333 .saturating_sub(removed.retained_entry_count());
334 self.retained_bytes = self
335 .retained_bytes
336 .saturating_sub(removed.estimated_history_bytes());
337 }
338 }
339 }
340
341 fn read(&self, window_ms: u64, now_ms: u64) -> Vec<SnapshotEvent> {
342 let started_at_ms = now_ms.saturating_sub(window_ms.min(HISTORY_RETENTION_MS));
343 self.snapshots
344 .iter()
345 .filter(|snapshot| {
346 snapshot.sampled_at_unix_ms >= started_at_ms
347 && snapshot.sampled_at_unix_ms <= now_ms
348 })
349 .cloned()
350 .collect()
351 }
352}
353
354struct Collector {
355 system: System,
356 sequence: u64,
357 cpu_baseline_refreshed_at: Option<Instant>,
358}
359
360impl Collector {
361 fn new() -> Self {
362 Self {
363 system: System::new(),
364 sequence: 0,
365 cpu_baseline_refreshed_at: None,
366 }
367 }
368
369 fn prime_cpu_usage(&mut self) {
370 self.system.refresh_processes_specifics(
371 ProcessesToUpdate::All,
372 true,
373 process_discovery_refresh_kind(),
374 );
375 self.cpu_baseline_refreshed_at = Some(Instant::now());
376 }
377
378 // Terminal polling asks for this every second while a shell is open. On
379 // Linux one read of /proc/<pid>/stat has everything; sysinfo also opens
380 // /proc/<pid>/status and builds a full record per process, ~3x the CPU.
381 #[cfg(target_os = "linux")]
382 fn process_table(&self) -> Vec<ProcessTableEntry> {
383 use std::io::Read;
384
385 let Ok(entries) = std::fs::read_dir("/proc") else {
386 return Vec::new();
387 };
388 let mut buffer = Vec::with_capacity(512);
389 let mut processes = entries
390 .filter_map(|entry| {
391 let entry = entry.ok()?;
392 let pid = entry.file_name().to_str()?.parse::<u32>().ok()?;
393 buffer.clear();
394 std::fs::File::open(entry.path().join("stat"))
395 .ok()?
396 .read_to_end(&mut buffer)
397 .ok()?;
398 let (name, ppid) = parse_proc_stat(&buffer)?;
399 Some(ProcessTableEntry { pid, ppid, name })
400 })
401 .collect::<Vec<_>>();
402 processes.sort_by_key(|process| process.pid);
403 processes
404 }
405
406 #[cfg(not(target_os = "linux"))]
407 fn process_table(&self) -> Vec<ProcessTableEntry> {
408 // A separate System preserves snapshot CPU baselines.
409 let mut process_table_system = System::new();
410 process_table_system.refresh_processes_specifics(
411 ProcessesToUpdate::All,
412 true,
413 ProcessRefreshKind::nothing().without_tasks(),
414 );
415 let mut processes = process_table_system
416 .processes()
417 .iter()
418 .filter_map(|(pid, process)| {
419 let pid = pid.as_u32();
420 // Kernel idle PID 0 would fail the server's positive-PID event schema.
421 if pid == 0 {
422 return None;
423 }
424 Some(ProcessTableEntry {
425 pid,
426 ppid: process.parent().map(Pid::as_u32).unwrap_or(0),
427 name: truncate_utf8(
428 process.name().to_string_lossy().into_owned(),
429 MAX_PROCESS_NAME_BYTES,
430 ),
431 })
432 })
433 .collect::<Vec<_>>();
434 processes.sort_by_key(|process| process.pid);
435 processes
436 }
437
438 fn sample(&mut self, config: &CollectorConfig, request_id: Option<String>) -> SnapshotEvent {
439 if let Some(delay) =
440 remaining_cpu_measurement_delay(self.cpu_baseline_refreshed_at.take(), Instant::now())
441 {
442 thread::sleep(delay);
443 }
444 let collection_started = Instant::now();
445 self.system.refresh_processes_specifics(
446 ProcessesToUpdate::All,
447 true,
448 process_discovery_refresh_kind(),
449 );
450 self.cpu_baseline_refreshed_at = Some(Instant::now());
451
452 let rows = self
453 .system
454 .processes()
455 .iter()
456 .map(|(pid, process)| {
457 let pid = pid.as_u32();
458 let ppid = process.parent().map(Pid::as_u32).unwrap_or(0);
459 (pid, ppid, process.start_time().saturating_mul(1_000))
460 })
461 .collect::<Vec<_>>();
462 let external_processes = config
463 .external_processes
464 .iter()
465 .filter_map(|(pid, expected_start_time_ms)| {
466 let (_, _, actual_start_time_ms) = rows
467 .iter()
468 .find(|(candidate_pid, _, _)| candidate_pid == pid)?;
469 matches_external_identity(*actual_start_time_ms, *expected_start_time_ms).then_some(
470 ExternalProcess {
471 pid: *pid,
472 start_time_ms: Some(*actual_start_time_ms),
473 },
474 )
475 })
476 .collect::<Vec<_>>();
477 let mut roots = external_processes
478 .iter()
479 .map(|process| process.pid)
480 .collect::<HashSet<_>>();
481 roots.insert(config.root_pid);
482 let tracked = select_tracked_pids(&rows, &roots);
483 let tracked_process_count = tracked.len();
484 let process_details = if TWO_STAGE_PROCESS_REFRESH && !tracked.is_empty() {
485 let monitor_pid = Pid::from_u32(std::process::id());
486 let mut detail_pids = tracked
487 .iter()
488 .copied()
489 .map(Pid::from_u32)
490 .collect::<Vec<_>>();
491 if !tracked.contains(&monitor_pid.as_u32()) {
492 detail_pids.push(monitor_pid);
493 }
494 // Detail fields need no baseline. Drop command data and OS handles after each sample.
495 let mut details = System::new();
496 details.refresh_processes_specifics(
497 ProcessesToUpdate::Some(&detail_pids),
498 true,
499 process_refresh_kind().without_cpu(),
500 );
501 // This process cannot be replaced during collection. Its start time
502 // exposes any boot-epoch shift between the two System instances.
503 let start_time_offset = self.system.process(monitor_pid).and_then(|process| {
504 details.process(monitor_pid).map(|detail| {
505 i128::from(detail.start_time()) - i128::from(process.start_time())
506 })
507 });
508 Some((details, start_time_offset))
509 } else {
510 None
511 };
512 let sample_details = process_details
513 .as_ref()
514 .map_or(&self.system, |(details, _)| details);
515 let start_time_offset = process_details
516 .as_ref()
517 .map_or(Some(0), |(_, offset)| *offset);
518 let mut processes = tracked
519 .into_iter()
520 .filter_map(|pid| {
521 let process = self.system.process(Pid::from_u32(pid))?;
522 let details = sample_details.process(Pid::from_u32(pid))?;
523 if !matches_process_start_time(
524 process.start_time(),
525 details.start_time(),
526 start_time_offset?,
527 ) {
528 return None;
529 }
530 let disk_usage = details.disk_usage();
531 let command = if details.cmd().is_empty() {
532 process.name().to_string_lossy().into_owned()
533 } else {
534 details
535 .cmd()
536 .iter()
537 .map(|part| part.to_string_lossy())
538 .collect::<Vec<_>>()
539 .join(" ")
540 };
541
542 Some(ProcessSample {
543 pid,
544 ppid: process.parent().map(Pid::as_u32).unwrap_or(0),
545 start_time_ms: process.start_time().saturating_mul(1_000),
546 run_time_ms: process.run_time().saturating_mul(1_000),
547 name: truncate_utf8(
548 process.name().to_string_lossy().into_owned(),
549 MAX_PROCESS_NAME_BYTES,
550 ),
551 command: truncate_utf8(command, MAX_PROCESS_COMMAND_BYTES),
552 status: truncate_utf8(
553 format!("{:?}", process.status()),
554 MAX_PROCESS_STATUS_BYTES,
555 ),
556 cpu_percent: process.cpu_usage(),
557 cpu_time_ms: process.accumulated_cpu_time(),
558 resident_bytes: details.memory(),
559 virtual_bytes: details.virtual_memory(),
560 io_read_bytes: disk_usage.total_read_bytes,
561 io_write_bytes: disk_usage.total_written_bytes,
562 io_semantics: io_semantics(),
563 })
564 })
565 .collect::<Vec<_>>();
566 drop(process_details);
567 processes.sort_by_key(|process| process.pid);
568 self.sequence = self.sequence.saturating_add(1);
569
570 SnapshotEvent {
571 version: PROTOCOL_VERSION,
572 event_type: "snapshot",
573 sequence: self.sequence,
574 sampled_at_unix_ms: unix_time_ms(),
575 collection_duration_micros: collection_started.elapsed().as_micros() as u64,
576 scanned_process_count: self.system.processes().len(),
577 retained_process_count: processes.len(),
578 inaccessible_process_count: inaccessible_process_count(
579 tracked_process_count,
580 processes.len(),
581 ),
582 request_id,
583 external_processes,
584 processes,
585 }
586 }
587}
588
589// Keep CPU baselines separate. Even a metadata refresh resets Linux process times.
590fn process_discovery_refresh_kind() -> ProcessRefreshKind {
591 if TWO_STAGE_PROCESS_REFRESH {
592 ProcessRefreshKind::nothing().with_cpu().without_tasks()
593 } else {
594 process_refresh_kind()
595 }
596}
597
598fn process_refresh_kind() -> ProcessRefreshKind {
599 ProcessRefreshKind::nothing()
600 .with_memory()
601 .with_cpu()
602 .with_disk_usage()
603 .with_cmd(UpdateKind::Always)
604 .without_tasks()
605}
606
607/// Name and parent pid from `/proc/<pid>/stat`: `pid (comm) state ppid ...`.
608/// The name may itself contain spaces and parentheses, so it ends at the last `)`.
609#[cfg(any(target_os = "linux", test))]
610fn parse_proc_stat(stat: &[u8]) -> Option<(String, u32)> {
611 let open = stat.iter().position(|&byte| byte == b'(')?;
612 let close = stat.iter().rposition(|&byte| byte == b')')?;
613 let name = stat.get(open + 1..close)?;
614 let mut fields = stat.get(close + 1..)?.split(|&byte| byte == b' ');
615 fields.next()?;
616 fields.next()?;
617 let ppid = std::str::from_utf8(fields.next()?).ok()?.parse().ok()?;
618 let name = truncate_utf8(
619 String::from_utf8_lossy(name).into_owned(),
620 MAX_PROCESS_NAME_BYTES,
621 );
622 Some((name, ppid))
623}
624
625fn inaccessible_process_count(selected: usize, materialized: usize) -> usize {
626 selected.saturating_sub(materialized)
627}
628
629fn matches_process_start_time(discovered: u64, detail: u64, epoch_offset: i128) -> bool {
630 i128::from(detail) - epoch_offset == i128::from(discovered)
631}
632
633fn remaining_cpu_measurement_delay(
634 baseline_refreshed_at: Option<Instant>,
635 now: Instant,
636) -> Option<Duration> {
637 baseline_refreshed_at
638 .and_then(|baseline| MINIMUM_CPU_UPDATE_INTERVAL.checked_sub(now.duration_since(baseline)))
639 .filter(|delay| !delay.is_zero())
640}
641
642fn matches_external_identity(
643 actual_start_time_ms: u64,
644 expected_start_time_ms: Option<u64>,
645) -> bool {
646 // Match sysinfo's whole-second precision without admitting reused PIDs from adjacent seconds.
647 expected_start_time_ms.is_none_or(|expected| {
648 actual_start_time_ms == expected - (expected % PROCESS_START_TIME_PRECISION_MS)
649 })
650}
651
652fn select_tracked_pids(rows: &[(u32, u32, u64)], roots: &HashSet<u32>) -> HashSet<u32> {
653 let mut children_by_parent = HashMap::<u32, Vec<(u32, u64)>>::new();
654 let mut start_time_by_pid = HashMap::<u32, u64>::new();
655 for (pid, ppid, start_time_ms) in rows {
656 children_by_parent
657 .entry(*ppid)
658 .or_default()
659 .push((*pid, *start_time_ms));
660 start_time_by_pid.insert(*pid, *start_time_ms);
661 }
662
663 let mut tracked = HashSet::new();
664 let mut visited_identities = HashSet::new();
665 let mut queue = roots
666 .iter()
667 .filter_map(|pid| {
668 start_time_by_pid
669 .get(pid)
670 .map(|start_time_ms| (*pid, *start_time_ms))
671 })
672 .collect::<VecDeque<_>>();
673
674 while let Some((pid, start_time_ms)) = queue.pop_front() {
675 if !visited_identities.insert((pid, start_time_ms)) {
676 continue;
677 }
678 tracked.insert(pid);
679 if let Some(children) = children_by_parent.get(&pid) {
680 queue.extend(
681 children
682 .iter()
683 .copied()
684 .filter(|(_, child_start_time_ms)| *child_start_time_ms >= start_time_ms),
685 );
686 }
687 }
688
689 tracked
690}
691
692fn truncate_utf8(mut value: String, max_bytes: usize) -> String {
693 if value.len() <= max_bytes {
694 return value;
695 }
696 let mut boundary = max_bytes;
697 while !value.is_char_boundary(boundary) {
698 boundary = boundary.saturating_sub(1);
699 }
700 value.truncate(boundary);
701 value
702}
703
704fn io_semantics() -> IoSemantics {
705 if cfg!(target_os = "windows") {
706 IoSemantics::AllIo
707 } else {
708 IoSemantics::Storage
709 }
710}
711
712fn unix_time_ms() -> u64 {
713 SystemTime::now()
714 .duration_since(UNIX_EPOCH)
715 .unwrap_or_default()
716 .as_millis() as u64
717}
718
719fn clamp_sample_interval(sample_interval_ms: u64) -> Option<Duration> {
720 (sample_interval_ms > 0).then(|| {
721 Duration::from_millis(
722 sample_interval_ms.clamp(MIN_SAMPLE_INTERVAL_MS, MAX_SAMPLE_INTERVAL_MS),
723 )
724 })
725}
726
727fn spawn_input_reader() -> Receiver<Input> {
728 let (sender, receiver) = mpsc::sync_channel(INPUT_QUEUE_CAPACITY);
729 thread::spawn(move || {
730 let stdin = io::stdin();
731 for line in stdin.lock().lines() {
732 let line = match line {
733 Ok(line) => line,
734 Err(error) => {
735 let _ = sender.send(Input::Invalid(format!(
736 "failed reading command stream: {error}"
737 )));
738 return;
739 }
740 };
741 if line.trim().is_empty() {
742 continue;
743 }
744 match serde_json::from_str::<Command>(&line) {
745 Ok(command) => {
746 if sender.send(Input::Command(command)).is_err() {
747 return;
748 }
749 }
750 Err(error) => {
751 if sender
752 .send(Input::Invalid(format!("invalid command: {error}")))
753 .is_err()
754 {
755 return;
756 }
757 }
758 }
759 }
760 });
761 receiver
762}
763
764fn sample_now_deadline(
765 current: Option<Instant>,
766 interval: Option<Duration>,
767 now: Instant,
768) -> Option<Instant> {
769 current.or_else(|| interval.map(|duration| now + duration))
770}
771
772fn write_event<T: Serialize>(writer: &mut impl Write, event: &T) -> io::Result<()> {
773 serde_json::to_writer(&mut *writer, event)?;
774 writer.write_all(b"\n")?;
775 writer.flush()
776}
777
778fn write_error(
779 writer: &mut impl Write,
780 code: & str,
781 message: impl Into<String>,
782 recoverable: bool,
783) -> io::Result<()> {
784 write_event(
785 writer,
786 &ErrorEvent {
787 version: PROTOCOL_VERSION,
788 event_type: "error",
789 code,
790 message: message.into(),
791 recoverable,
792 },
793 )
794}
795
796fn write_history(
797 writer: &mut impl Write,
798 request_id: &str,
799 snapshots: &[SnapshotEvent],
800) -> io::Result<()> {
801 if snapshots.is_empty() {
802 return write_event(
803 writer,
804 &HistoryChunkEvent {
805 version: PROTOCOL_VERSION,
806 event_type: "historyChunk",
807 request_id,
808 done: true,
809 snapshots,
810 },
811 );
812 }
813
814 let chunk_count = snapshots.len().div_ceil(HISTORY_CHUNK_SNAPSHOTS);
815 for (index, chunk) in snapshots.chunks(HISTORY_CHUNK_SNAPSHOTS).enumerate() {
816 write_event(
817 writer,
818 &HistoryChunkEvent {
819 version: PROTOCOL_VERSION,
820 event_type: "historyChunk",
821 request_id,
822 done: index + 1 == chunk_count,
823 snapshots: chunk,
824 },
825 )?;
826 }
827 Ok(())
828}
829
830fn main() -> io::Result<()> {
831 let mut writer = BufWriter::new(io::stdout().lock());
832 write_event(
833 &mut writer,
834 &HelloEvent {
835 version: PROTOCOL_VERSION,
836 event_type: "hello",
837 sidecar_version: env!("CARGO_PKG_VERSION"),
838 sidecar_pid: std::process::id(),
839 platform: std::env::consts::OS,
840 arch: std::env::consts::ARCH,
841 capabilities: Capabilities {
842 cumulative_cpu_time: true,
843 current_cpu_percent: true,
844 resident_memory: true,
845 virtual_memory: true,
846 io_bytes: true,
847 process_start_time: true,
848 process_tree: true,
849 },
850 },
851 )?;
852
853 let receiver = spawn_input_reader();
854 let mut collector = Collector::new();
855 let mut history = HistoryRecorder::default();
856 let mut config: Option<CollectorConfig> = None;
857 let mut next_sample_at: Option<Instant> = None;
858 let mut streaming_enabled = false;
859
860 loop {
861 if next_sample_at.is_some_and(|deadline| deadline <= Instant::now()) {
862 if let Some(current) = config.as_ref() {
863 if let Some(interval) = current.sample_interval {
864 let event = collector.sample(current, None);
865 history.record(&event);
866 if streaming_enabled {
867 write_event(&mut writer, &event)?;
868 }
869 next_sample_at = Some(Instant::now() + interval);
870 } else {
871 next_sample_at = None;
872 }
873 } else {
874 next_sample_at = None;
875 }
876 continue;
877 }
878
879 let timeout = next_sample_at
880 .map(|deadline| deadline.saturating_duration_since(Instant::now()))
881 .unwrap_or(Duration::from_secs(60));
882
883 match receiver.recv_timeout(timeout) {
884 Ok(Input::Invalid(message)) => {
885 write_error(&mut writer, "invalid-command", message, true)?;
886 }
887 Ok(Input::Command(command)) => {
888 if command.version() != PROTOCOL_VERSION {
889 write_error(
890 &mut writer,
891 "protocol-mismatch",
892 format!(
893 "unsupported protocol version {}; expected {PROTOCOL_VERSION}",
894 command.version()
895 ),
896 false,
897 )?;
898 continue;
899 }
900
901 match command {
902 Command::Configure {
903 root_pid,
904 sample_interval_ms,
905 external_processes,
906 ..
907 } => {
908 let sample_interval = clamp_sample_interval(sample_interval_ms);
909 config = Some(CollectorConfig {
910 root_pid,
911 sample_interval,
912 external_processes: external_processes
913 .into_iter()
914 .map(|process| (process.pid, process.start_time_ms))
915 .collect(),
916 });
917 collector.prime_cpu_usage();
918 next_sample_at = sample_interval.map(|_| Instant::now());
919 }
920 Command::SetExternalProcesses { processes, .. } => {
921 if let Some(current) = config.as_mut() {
922 current.external_processes = processes
923 .into_iter()
924 .map(|process| (process.pid, process.start_time_ms))
925 .collect();
926 } else {
927 write_error(
928 &mut writer,
929 "not-configured",
930 "configure must be sent before external processes",
931 true,
932 )?;
933 }
934 }
935 Command::SetSampleInterval {
936 sample_interval_ms, ..
937 } => {
938 if let Some(current) = config.as_mut() {
939 current.sample_interval = clamp_sample_interval(sample_interval_ms);
940 next_sample_at = current
941 .sample_interval
942 .map(|interval| Instant::now() + interval);
943 } else {
944 write_error(
945 &mut writer,
946 "not-configured",
947 "configure must be sent before changing the sample interval",
948 true,
949 )?;
950 }
951 }
952 Command::SetStreaming { enabled, .. } => {
953 streaming_enabled = enabled;
954 }
955 Command::SampleNow { request_id, .. } => {
956 if let Some(current) = config.as_ref() {
957 let event = collector.sample(current, Some(request_id));
958 history.record(&event);
959 write_event(&mut writer, &event)?;
960 next_sample_at = sample_now_deadline(
961 next_sample_at,
962 current.sample_interval,
963 Instant::now(),
964 );
965 } else {
966 write_error(
967 &mut writer,
968 "not-configured",
969 "configure must be sent before sampling",
970 true,
971 )?;
972 }
973 }
974 Command::ProcessTable { request_id, .. } => {
975 let event = ProcessTableEvent {
976 version: PROTOCOL_VERSION,
977 event_type: "processTable",
978 request_id: &request_id,
979 processes: collector.process_table(),
980 };
981 write_event(&mut writer, &event)?;
982 }
983 Command::ReadHistory {
984 request_id,
985 window_ms,
986 ..
987 } => {
988 if config.is_some() {
989 let snapshots = history.read(window_ms, unix_time_ms());
990 write_history(&mut writer, &request_id, &snapshots)?;
991 } else {
992 write_error(
993 &mut writer,
994 "not-configured",
995 "configure must be sent before reading history",
996 true,
997 )?;
998 }
999 }
1000 Command::Shutdown { .. } => return Ok(()),
1001 }
1002 }
1003 Err(RecvTimeoutError::Timeout) => {}
1004 Err(RecvTimeoutError::Disconnected) => return Ok(()),
1005 }
1006 }
1007}
1008
1009#[cfg(test)]
1010mod tests {
1011 use super::*;
1012
1013 #[test]
1014 fn selects_roots_and_all_descendants() {
1015 let rows = vec![
1016 (10, 1, 1_000),
1017 (11, 10, 1_100),
1018 (12, 11, 1_200),
1019 (20, 1, 2_000),
1020 (21, 20, 2_100),
1021 (30, 99, 3_000),
1022 ];
1023 let tracked = select_tracked_pids(&rows, &HashSet::from([10, 20]));
1024
1025 assert_eq!(tracked, HashSet::from([10, 11, 12, 20, 21]));
1026 }
1027
1028 #[test]
1029 fn rejects_descendants_older_than_a_reused_parent_pid() {
1030 let rows = vec![
1031 (20, 1, 5_000),
1032 (21, 20, 4_000),
1033 (22, 20, 5_100),
1034 (23, 21, 5_200),
1035 ];
1036 let tracked = select_tracked_pids(&rows, &HashSet::from([20]));
1037
1038 assert_eq!(tracked, HashSet::from([20, 22]));
1039 }
1040
1041 #[test]
1042 fn ignores_missing_roots() {
1043 let rows = vec![(10, 1, 1_000), (11, 10, 1_100)];
1044 let tracked = select_tracked_pids(&rows, &HashSet::from([99]));
1045
1046 assert!(tracked.is_empty());
1047 }
1048
1049 #[test]
1050 fn validates_external_process_start_identity() {
1051 assert!(matches_external_identity(10_000, None));
1052 assert!(matches_external_identity(10_000, Some(10_999)));
1053 assert!(!matches_external_identity(10_000, Some(11_000)));
1054 assert!(!matches_external_identity(10_000, Some(9_999)));
1055 }
1056
1057 #[test]
1058 fn loads_details_when_an_existing_process_becomes_selected() {
1059 let mut collector = Collector::new();
1060 let mut config = CollectorConfig {
1061 root_pid: u32::MAX,
1062 sample_interval: None,
1063 external_processes: HashMap::new(),
1064 };
1065 assert!(collector.sample(&config, None).processes.is_empty());
1066
1067 config.root_pid = std::process::id();
1068 let snapshot = collector.sample(&config, None);
1069 let process = snapshot
1070 .processes
1071 .iter()
1072 .find(|process| process.pid == config.root_pid)
1073 .expect("selected process");
1074
1075 assert!(!process.command.is_empty());
1076 assert!(process.resident_bytes > 0);
1077 assert!(process.cpu_percent.is_finite());
1078 }
1079
1080 #[cfg(any(target_os = "linux", target_os = "macos"))]
1081 #[test]
1082 fn keeps_command_lines_and_memory_out_of_the_discovery_table() {
1083 let mut collector = Collector::new();
1084 let mut config = CollectorConfig {
1085 root_pid: u32::MAX,
1086 sample_interval: None,
1087 external_processes: HashMap::new(),
1088 };
1089 assert!(collector.sample(&config, None).processes.is_empty());
1090
1091 let pid = Pid::from_u32(std::process::id());
1092 let unselected = collector.system.process(pid).expect("discovered process");
1093 assert!(unselected.cmd().is_empty());
1094 assert_eq!(unselected.memory(), 0);
1095 assert_eq!(unselected.virtual_memory(), 0);
1096
1097 config.external_processes.insert(pid.as_u32(), None);
1098 let snapshot = collector.sample(&config, None);
1099 let selected = snapshot
1100 .processes
1101 .iter()
1102 .find(|process| process.pid == pid.as_u32())
1103 .expect("external process details");
1104 assert!(!selected.command.is_empty());
1105 assert!(selected.resident_bytes > 0);
1106 assert!(selected.start_time_ms > 0);
1107
1108 let retained = collector
1109 .system
1110 .process(pid)
1111 .expect("retained CPU baseline");
1112 assert!(retained.cmd().is_empty());
1113 assert_eq!(retained.memory(), 0);
1114 assert_eq!(retained.virtual_memory(), 0);
1115 }
1116
1117 #[test]
1118 fn accepts_clock_shifts_without_accepting_reused_process_starts() {
1119 assert!(matches_process_start_time(10_000, 13_600, 3_600));
1120 assert!(!matches_process_start_time(10_000, 13_601, 3_600));
1121 assert!(matches_process_start_time(10_000, 6_400, -3_600));
1122 assert!(!matches_process_start_time(10_000, 6_401, -3_600));
1123 assert!(matches_process_start_time(10_000, 10_000, 0));
1124 assert!(!matches_process_start_time(10_000, 10_001, 0));
1125 }
1126
1127 #[test]
1128 fn decodes_protocol_commands() {
1129 let configure = serde_json::from_str::<Command>(
1130 r#"{"version":3,"type":"configure","rootPid":42,"sampleIntervalMs":1000,"externalProcesses":[{"pid":7}]}"#,
1131 )
1132 .expect("configure command");
1133
1134 match configure {
1135 Command::Configure {
1136 root_pid,
1137 sample_interval_ms,
1138 external_processes,
1139 ..
1140 } => {
1141 assert_eq!(root_pid, 42);
1142 assert_eq!(sample_interval_ms, 1_000);
1143 assert_eq!(external_processes[0].pid, 7);
1144 assert_eq!(external_processes[0].start_time_ms, None);
1145 }
1146 _ => panic!("unexpected command"),
1147 }
1148
1149 let read_history = serde_json::from_str::<Command>(
1150 r#"{"version":3,"type":"readHistory","requestId":"history-1","windowMs":60000}"#,
1151 )
1152 .expect("read history command");
1153 assert!(matches!(
1154 read_history,
1155 Command::ReadHistory {
1156 request_id,
1157 window_ms: 60_000,
1158 ..
1159 } if request_id == "history-1"
1160 ));
1161
1162 let process_table = serde_json::from_str::<Command>(
1163 r#"{"version":3,"type":"processTable","requestId":"processes-1"}"#,
1164 )
1165 .expect("process table command");
1166 assert!(matches!(
1167 process_table,
1168 Command::ProcessTable { request_id, .. } if request_id == "processes-1"
1169 ));
1170 }
1171
1172 #[test]
1173 fn clamps_sample_interval() {
1174 assert_eq!(clamp_sample_interval(0), None);
1175 assert_eq!(clamp_sample_interval(1), Some(Duration::from_millis(250)));
1176 assert_eq!(
1177 clamp_sample_interval(100_000),
1178 Some(Duration::from_millis(60_000))
1179 );
1180 }
1181
1182 #[test]
1183 fn counts_selected_processes_that_could_not_be_materialized() {
1184 assert_eq!(inaccessible_process_count(5, 3), 2);
1185 assert_eq!(inaccessible_process_count(3, 5), 0);
1186 }
1187
1188 #[test]
1189 fn waits_for_a_cpu_measurement_window_after_priming() {
1190 let baseline = Instant::now();
1191
1192 assert_eq!(
1193 remaining_cpu_measurement_delay(Some(baseline), baseline),
1194 Some(MINIMUM_CPU_UPDATE_INTERVAL)
1195 );
1196 assert_eq!(
1197 remaining_cpu_measurement_delay(Some(baseline), baseline + MINIMUM_CPU_UPDATE_INTERVAL),
1198 None
1199 );
1200 assert_eq!(remaining_cpu_measurement_delay(None, baseline), None);
1201 }
1202
1203 #[test]
1204 fn retains_bounded_history_without_request_ids() {
1205 let mut history = HistoryRecorder::default();
1206 for sequence in 0..=MAX_HISTORY_SNAPSHOTS {
1207 history.record(&SnapshotEvent {
1208 version: PROTOCOL_VERSION,
1209 event_type: "snapshot",
1210 sequence: sequence as u64,
1211 sampled_at_unix_ms: sequence as u64 * 1_000,
1212 collection_duration_micros: 1,
1213 scanned_process_count: 0,
1214 retained_process_count: 0,
1215 inaccessible_process_count: 0,
1216 request_id: Some("request".to_owned()),
1217 external_processes: vec![ExternalProcess {
1218 pid: 7,
1219 start_time_ms: Some(1_000),
1220 }],
1221 processes: Vec::new(),
1222 });
1223 }
1224
1225 assert_eq!(history.snapshots.len(), MAX_HISTORY_SNAPSHOTS);
1226 assert!(
1227 history
1228 .snapshots
1229 .iter()
1230 .all(|snapshot| snapshot.request_id.is_none())
1231 );
1232 assert!(history.snapshots.iter().all(|snapshot| {
1233 snapshot.external_processes.len() == 1
1234 && snapshot.external_processes[0].pid == 7
1235 && snapshot.external_processes[0].start_time_ms == Some(1_000)
1236 }));
1237 assert_eq!(
1238 history
1239 .read(10_000, MAX_HISTORY_SNAPSHOTS as u64 * 1_000)
1240 .len(),
1241 11
1242 );
1243 }
1244
1245 #[test]
1246 fn excludes_and_trims_future_history_after_the_clock_moves_backward() {
1247 let mut history = HistoryRecorder::default();
1248 let snapshot = SnapshotEvent {
1249 version: PROTOCOL_VERSION,
1250 event_type: "snapshot",
1251 sequence: 1,
1252 sampled_at_unix_ms: 2_000,
1253 collection_duration_micros: 1,
1254 scanned_process_count: 0,
1255 retained_process_count: 0,
1256 inaccessible_process_count: 0,
1257 request_id: None,
1258 external_processes: Vec::new(),
1259 processes: Vec::new(),
1260 };
1261 history.record(&snapshot);
1262
1263 assert!(history.read(0, 1_000).is_empty());
1264
1265 history.record(&SnapshotEvent {
1266 sequence: 2,
1267 sampled_at_unix_ms: 1_000,
1268 ..snapshot
1269 });
1270 assert_eq!(history.snapshots.len(), 1);
1271 assert_eq!(
1272 history.snapshots.front().map(|entry| entry.sequence),
1273 Some(2)
1274 );
1275 }
1276
1277 #[test]
1278 fn bounds_history_by_estimated_process_bytes() {
1279 let mut history = HistoryRecorder::default();
1280 let command = "x".repeat(128);
1281 let process = ProcessSample {
1282 pid: 1,
1283 ppid: 0,
1284 start_time_ms: 0,
1285 run_time_ms: 0,
1286 name: "process".to_owned(),
1287 command,
1288 status: "Run".to_owned(),
1289 cpu_percent: 0.0,
1290 cpu_time_ms: 0,
1291 resident_bytes: 0,
1292 virtual_bytes: 0,
1293 io_read_bytes: 0,
1294 io_write_bytes: 0,
1295 io_semantics: IoSemantics::Storage,
1296 };
1297 let snapshot_bytes =
1298 std::mem::size_of::<SnapshotEvent>() + process.estimated_history_bytes();
1299 for sequence in 0..3 {
1300 history.record_with_limits(
1301 &SnapshotEvent {
1302 version: PROTOCOL_VERSION,
1303 event_type: "snapshot",
1304 sequence,
1305 sampled_at_unix_ms: sequence * 1_000,
1306 collection_duration_micros: 1,
1307 scanned_process_count: 1,
1308 retained_process_count: 1,
1309 inaccessible_process_count: 0,
1310 request_id: None,
1311 external_processes: Vec::new(),
1312 processes: vec![ProcessSample {
1313 pid: sequence as u32 + 1,
1314 start_time_ms: sequence * 1_000,
1315 ..process.clone()
1316 }],
1317 },
1318 3,
1319 3,
1320 snapshot_bytes * 2,
1321 );
1322 }
1323
1324 assert!(history.retained_bytes <= snapshot_bytes * 2);
1325 assert_eq!(history.snapshots.len(), 2);
1326 assert_eq!(
1327 history.snapshots.front().map(|snapshot| snapshot.sequence),
1328 Some(1)
1329 );
1330 }
1331
1332 #[test]
1333 fn counts_external_processes_toward_history_limits() {
1334 let mut history = HistoryRecorder::default();
1335 let external_processes = (1..=128)
1336 .map(|pid| ExternalProcess {
1337 pid,
1338 start_time_ms: Some(u64::from(pid) * 1_000),
1339 })
1340 .collect::<Vec<_>>();
1341 let snapshot = SnapshotEvent {
1342 version: PROTOCOL_VERSION,
1343 event_type: "snapshot",
1344 sequence: 0,
1345 sampled_at_unix_ms: 0,
1346 collection_duration_micros: 1,
1347 scanned_process_count: 0,
1348 retained_process_count: 0,
1349 inaccessible_process_count: 0,
1350 request_id: None,
1351 external_processes,
1352 processes: Vec::new(),
1353 };
1354 let snapshot_bytes = snapshot.estimated_history_bytes();
1355 let snapshot_entries = snapshot.retained_entry_count();
1356
1357 for sequence in 0..3 {
1358 history.record_with_limits(
1359 &SnapshotEvent {
1360 sequence,
1361 sampled_at_unix_ms: sequence * 1_000,
1362 ..snapshot.clone()
1363 },
1364 3,
1365 snapshot_entries * 2,
1366 snapshot_bytes * 2,
1367 );
1368 }
1369
1370 assert_eq!(history.retained_entry_count, snapshot_entries * 2);
1371 assert!(history.retained_bytes <= snapshot_bytes * 2);
1372 assert_eq!(history.snapshots.len(), 2);
1373 assert_eq!(
1374 history.snapshots.front().map(|snapshot| snapshot.sequence),
1375 Some(1)
1376 );
1377 }
1378
1379 #[test]
1380 fn truncates_process_strings_at_utf8_boundaries() {
1381 let value = "é".repeat(MAX_PROCESS_NAME_BYTES);
1382 let truncated = truncate_utf8(value, MAX_PROCESS_NAME_BYTES - 1);
1383
1384 assert!(truncated.len() < MAX_PROCESS_NAME_BYTES);
1385 assert!(truncated.is_char_boundary(truncated.len()));
1386 }
1387
1388 #[test]
1389 fn parses_proc_stat_names_with_spaces_and_parentheses() {
1390 assert_eq!(
1391 parse_proc_stat(b"4242 (tmux: server) S 1 4242 4242 0 -1 4194560 3101"),
1392 Some(("tmux: server".to_owned(), 1))
1393 );
1394 assert_eq!(
1395 parse_proc_stat(b"77 (a) b (c)) R 4242 77 77 34816 77 4194304"),
1396 Some(("a) b (c)".to_owned(), 4242))
1397 );
1398 assert_eq!(parse_proc_stat(b"77 (truncated"), None);
1399 }
1400
1401 #[cfg(target_os = "linux")]
1402 #[test]
1403 fn linux_process_table_reports_own_process_like_sysinfo() {
1404 let collector = Collector::new();
1405 let mut system = System::new();
1406 system.refresh_processes_specifics(
1407 ProcessesToUpdate::All,
1408 true,
1409 ProcessRefreshKind::nothing().without_tasks(),
1410 );
1411 let own_pid = std::process::id();
1412 let table = collector.process_table();
1413 let own = table
1414 .iter()
1415 .find(|process| process.pid == own_pid)
1416 .expect("own process");
1417 let expected = system
1418 .process(Pid::from_u32(own_pid))
1419 .expect("sysinfo own process");
1420 assert_eq!(own.ppid, expected.parent().map(Pid::as_u32).unwrap_or(0));
1421 assert_eq!(own.name, expected.name().to_string_lossy());
1422 assert!(table.windows(2).all(|pair| pair[0].pid < pair[1].pid));
1423 }
1424
1425 #[test]
1426 fn refreshes_commands_without_enumerating_linux_tasks() {
1427 let refresh_kind = process_refresh_kind();
1428
1429 assert_eq!(refresh_kind.cmd(), UpdateKind::Always);
1430 assert!(!refresh_kind.tasks());
1431 assert!(refresh_kind.cpu());
1432 assert!(refresh_kind.memory());
1433 assert!(refresh_kind.disk_usage());
1434 }
1435
1436 #[test]
1437 fn sample_now_does_not_postpone_an_existing_periodic_deadline() {
1438 let now = Instant::now();
1439 let deadline = now + Duration::from_secs(1);
1440
1441 assert_eq!(
1442 sample_now_deadline(
1443 Some(deadline),
1444 Some(Duration::from_secs(5)),
1445 now + Duration::from_millis(100)
1446 ),
1447 Some(deadline)
1448 );
1449 }
1450}