1use crate::allocation::{
11 allocation_capability, allocation_snapshot, AllocationSessionToken, AllocationSnapshotV2,
12};
13use crate::collector::{CollectorCapability, SnapshotCollector};
14use crate::hardware::{milli_ratio, HardwareFieldSourceV2, SourcedEvidenceV2};
15use crate::schema::ResourceSnapshot;
16use crate::schema_v2::{Evidence, EvidenceGap};
17use serde::{Deserialize, Serialize};
18
19#[cfg(all(feature = "process-metrics", target_os = "linux"))]
20mod linux;
21#[cfg(any(not(feature = "process-metrics"), not(target_os = "linux")))]
22mod stubs;
23#[cfg(all(feature = "process-metrics", target_os = "linux"))]
24use linux as platform;
25#[cfg(any(not(feature = "process-metrics"), not(target_os = "linux")))]
26use stubs as platform;
27
28pub const SYSTEM_EVIDENCE_V2_VERSION: u16 = 1;
29
30const fn legacy_component_version() -> u16 {
31 1
32}
33
34fn gap<T>(reason: EvidenceGap) -> Evidence<T> {
35 Evidence::unavailable(reason)
36}
37
38#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
44#[serde(rename_all = "kebab-case")]
45#[repr(u8)]
46pub enum IoCacheStateV2 {
47 Cold = 1,
48 Warm = 2,
49 Direct = 3,
50}
51
52impl IoCacheStateV2 {
53 pub const fn as_value(self) -> u64 {
55 self as u64
56 }
57
58 pub fn from_value(value: u64) -> Option<Self> {
60 match value {
61 1 => Some(Self::Cold),
62 2 => Some(Self::Warm),
63 3 => Some(Self::Direct),
64 _ => None,
65 }
66 }
67}
68
69#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
71pub struct AllocationTotalsV2 {
72 #[serde(default = "legacy_component_version")]
73 pub version: u16,
74 pub allocations: u64,
75 pub deallocations: u64,
76 pub allocated_bytes: u64,
77 pub deallocated_bytes: u64,
78 pub live_bytes: u64,
80 pub peak_live_bytes: u64,
82}
83
84#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
87pub struct StageAllocationV2 {
88 #[serde(default = "legacy_component_version")]
89 pub version: u16,
90 #[serde(default)]
91 pub metric_id: Option<crate::MetricId>,
92 pub allocations: u64,
94 pub allocated_bytes: u64,
96 pub live_bytes: u64,
98 pub peak_live_bytes: u64,
100}
101
102#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
105pub struct AllocationEvidenceV2 {
106 #[serde(default = "legacy_component_version")]
107 pub version: u16,
108 pub totals: Evidence<AllocationTotalsV2>,
109 #[serde(default)]
110 pub stages: Vec<StageAllocationV2>,
111}
112
113#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
115pub struct FaultEvidenceV2 {
116 #[serde(default = "legacy_component_version")]
117 pub version: u16,
118 pub minor_faults: SourcedEvidenceV2<u64>,
119 pub major_faults: SourcedEvidenceV2<u64>,
120}
121
122#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
124pub struct IoEvidenceV2 {
125 #[serde(default = "legacy_component_version")]
126 pub version: u16,
127 pub read_bytes: SourcedEvidenceV2<u64>,
129 pub write_bytes: SourcedEvidenceV2<u64>,
131 pub read_syscalls: SourcedEvidenceV2<u64>,
132 pub write_syscalls: SourcedEvidenceV2<u64>,
133 pub cancelled_write_bytes: SourcedEvidenceV2<u64>,
135}
136
137#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
139pub struct MemoryEvidenceV2 {
140 #[serde(default = "legacy_component_version")]
141 pub version: u16,
142 pub resident_bytes: SourcedEvidenceV2<u64>,
143 pub virtual_bytes: SourcedEvidenceV2<u64>,
144 pub resident_high_water_bytes: SourcedEvidenceV2<u64>,
146 pub swap_bytes: SourcedEvidenceV2<u64>,
147}
148
149#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
151pub struct PressureEvidenceV2 {
152 #[serde(default = "legacy_component_version")]
153 pub version: u16,
154 pub cpu_some_avg10_milli: SourcedEvidenceV2<u64>,
156 pub cpu_full_avg10_milli: SourcedEvidenceV2<u64>,
158 pub memory_some_avg10_milli: SourcedEvidenceV2<u64>,
160 pub io_some_avg10_milli: SourcedEvidenceV2<u64>,
162}
163
164#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
166pub struct ThermalEvidenceV2 {
167 #[serde(default = "legacy_component_version")]
168 pub version: u16,
169 pub max_zone_millicelsius: SourcedEvidenceV2<u64>,
171 pub throttle_events: SourcedEvidenceV2<u64>,
173}
174
175#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
177pub struct NetworkProcessCountersV2 {
178 #[serde(default = "legacy_component_version")]
179 pub version: u16,
180 pub read_bytes: u64,
181 pub written_bytes: u64,
182}
183
184#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
187pub struct NetworkEvidenceV2 {
188 #[serde(default = "legacy_component_version")]
189 pub version: u16,
190 pub process_counters: SourcedEvidenceV2<NetworkProcessCountersV2>,
191 pub retry_annotations: u64,
193}
194
195#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
197pub struct DecodeRetentionEvidenceV2 {
198 #[serde(default = "legacy_component_version")]
199 pub version: u16,
200 pub expansion_ratio_milli: Evidence<u64>,
202 pub retained_bytes: Evidence<u64>,
204 pub retained_peak_bytes: Evidence<u64>,
206}
207
208#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
210pub struct SystemRunEvidenceV2 {
211 #[serde(default = "legacy_component_version")]
212 pub version: u16,
213 pub allocation: AllocationEvidenceV2,
214 pub faults: FaultEvidenceV2,
215 pub io: IoEvidenceV2,
216 pub memory: MemoryEvidenceV2,
217 pub pressure: PressureEvidenceV2,
218 pub thermal: ThermalEvidenceV2,
219 pub network: NetworkEvidenceV2,
220 pub decode: DecodeRetentionEvidenceV2,
221}
222
223#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
225pub struct SystemIoSampleV2 {
226 #[serde(default = "legacy_component_version")]
227 pub version: u16,
228 pub minor_faults: SourcedEvidenceV2<u64>,
229 pub major_faults: SourcedEvidenceV2<u64>,
230 pub read_bytes: SourcedEvidenceV2<u64>,
231 pub write_bytes: SourcedEvidenceV2<u64>,
232 pub read_syscalls: SourcedEvidenceV2<u64>,
233 pub write_syscalls: SourcedEvidenceV2<u64>,
234 pub cancelled_write_bytes: SourcedEvidenceV2<u64>,
235}
236
237#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
239pub struct PressureThermalSampleV2 {
240 #[serde(default = "legacy_component_version")]
241 pub version: u16,
242 pub cpu_some_avg10_milli: SourcedEvidenceV2<u64>,
243 pub cpu_full_avg10_milli: SourcedEvidenceV2<u64>,
244 pub memory_some_avg10_milli: SourcedEvidenceV2<u64>,
245 pub io_some_avg10_milli: SourcedEvidenceV2<u64>,
246 pub max_zone_millicelsius: SourcedEvidenceV2<u64>,
247 pub throttle_events: SourcedEvidenceV2<u64>,
248}
249
250pub struct SystemIoCollector {
253 capability: CollectorCapability,
254}
255
256impl SystemIoCollector {
257 pub fn new() -> Self {
258 Self {
259 capability: platform::system_io_capability(),
260 }
261 }
262}
263
264impl Default for SystemIoCollector {
265 fn default() -> Self {
266 Self::new()
267 }
268}
269
270impl SnapshotCollector for SystemIoCollector {
271 type Snapshot = SystemIoSampleV2;
272
273 fn capability(&self) -> CollectorCapability {
274 self.capability.clone()
275 }
276
277 fn sample(&mut self) -> Self::Snapshot {
278 platform::sample_system_io()
279 }
280}
281
282pub struct PressureThermalCollector {
285 capability: CollectorCapability,
286}
287
288impl PressureThermalCollector {
289 pub fn new() -> Self {
290 Self {
291 capability: platform::pressure_thermal_capability(),
292 }
293 }
294}
295
296impl Default for PressureThermalCollector {
297 fn default() -> Self {
298 Self::new()
299 }
300}
301
302impl SnapshotCollector for PressureThermalCollector {
303 type Snapshot = PressureThermalSampleV2;
304
305 fn capability(&self) -> CollectorCapability {
306 self.capability.clone()
307 }
308
309 fn sample(&mut self) -> Self::Snapshot {
310 platform::sample_pressure_thermal()
311 }
312}
313
314fn sourced_delta_u64(
315 end: &SourcedEvidenceV2<u64>,
316 start: &SourcedEvidenceV2<u64>,
317) -> SourcedEvidenceV2<u64> {
318 crate::hardware::sourced_delta(end, start)
319}
320
321pub(crate) struct SystemSession {
323 io_collector: SystemIoCollector,
324 pressure_collector: PressureThermalCollector,
325 io_start: Option<SystemIoSampleV2>,
326 allocation_start: Option<AllocationSnapshotV2>,
327 allocation_session: AllocationSessionToken,
328}
329
330impl SystemSession {
331 pub(crate) fn new() -> Self {
332 let mut io_collector = SystemIoCollector::new();
333 let io_available = matches!(
334 io_collector.capability.availability,
335 crate::collector::CollectorAvailability::Available
336 );
337 let io_start = io_available.then(|| io_collector.sample());
338 let allocation_start = crate::allocation::allocation_tracking_installed()
339 .then(crate::allocation::allocation_snapshot);
340 let allocation_session = if allocation_start.is_some() {
344 crate::allocation::enter_allocation_session()
345 } else {
346 AllocationSessionToken::inactive()
347 };
348 Self {
349 io_collector,
350 pressure_collector: PressureThermalCollector::new(),
351 io_start,
352 allocation_start,
353 allocation_session,
354 }
355 }
356
357 pub(crate) fn capabilities(&self) -> Vec<CollectorCapability> {
358 vec![
359 allocation_capability(),
360 self.io_collector.capability(),
361 self.pressure_collector.capability(),
362 ]
363 }
364
365 pub(crate) fn finish_evidence(
367 mut self,
368 runtime: &crate::Runtime,
369 finish_resources: &ResourceSnapshot,
370 input_bytes: u64,
371 derived_decoder_bytes: u64,
372 ) -> Evidence<SystemRunEvidenceV2> {
373 let io_end = self.io_collector.sample();
374 let (faults, io) = match &self.io_start {
375 Some(start) => (
376 FaultEvidenceV2 {
377 version: SYSTEM_EVIDENCE_V2_VERSION,
378 minor_faults: sourced_delta_u64(&io_end.minor_faults, &start.minor_faults),
379 major_faults: sourced_delta_u64(&io_end.major_faults, &start.major_faults),
380 },
381 IoEvidenceV2 {
382 version: SYSTEM_EVIDENCE_V2_VERSION,
383 read_bytes: sourced_delta_u64(&io_end.read_bytes, &start.read_bytes),
384 write_bytes: sourced_delta_u64(&io_end.write_bytes, &start.write_bytes),
385 read_syscalls: sourced_delta_u64(&io_end.read_syscalls, &start.read_syscalls),
386 write_syscalls: sourced_delta_u64(
387 &io_end.write_syscalls,
388 &start.write_syscalls,
389 ),
390 cancelled_write_bytes: sourced_delta_u64(
391 &io_end.cancelled_write_bytes,
392 &start.cancelled_write_bytes,
393 ),
394 },
395 ),
396 None => {
397 let reason = match io_end.minor_faults.value {
401 Evidence::Unavailable { reason } => reason,
402 Evidence::Recorded { .. } => EvidenceGap::Unavailable,
403 };
404 let io_reason = match io_end.read_bytes.value {
405 Evidence::Unavailable { reason } => reason,
406 Evidence::Recorded { .. } => EvidenceGap::Unavailable,
407 };
408 (
409 FaultEvidenceV2 {
410 version: SYSTEM_EVIDENCE_V2_VERSION,
411 minor_faults: SourcedEvidenceV2::gapped(
412 HardwareFieldSourceV2::ProcSelfStat,
413 reason,
414 ),
415 major_faults: SourcedEvidenceV2::gapped(
416 HardwareFieldSourceV2::ProcSelfStat,
417 reason,
418 ),
419 },
420 IoEvidenceV2 {
421 version: SYSTEM_EVIDENCE_V2_VERSION,
422 read_bytes: SourcedEvidenceV2::gapped(
423 HardwareFieldSourceV2::ProcSelfIo,
424 io_reason,
425 ),
426 write_bytes: SourcedEvidenceV2::gapped(
427 HardwareFieldSourceV2::ProcSelfIo,
428 io_reason,
429 ),
430 read_syscalls: SourcedEvidenceV2::gapped(
431 HardwareFieldSourceV2::ProcSelfIo,
432 io_reason,
433 ),
434 write_syscalls: SourcedEvidenceV2::gapped(
435 HardwareFieldSourceV2::ProcSelfIo,
436 io_reason,
437 ),
438 cancelled_write_bytes: SourcedEvidenceV2::gapped(
439 HardwareFieldSourceV2::ProcSelfIo,
440 io_reason,
441 ),
442 },
443 )
444 }
445 };
446 let pressure = self.pressure_collector.sample();
447 let allocation = self.allocation_evidence();
448 let memory = memory_evidence(finish_resources);
449 let retry_annotations = runtime.retry_annotation_count();
450 let network = NetworkEvidenceV2 {
451 version: SYSTEM_EVIDENCE_V2_VERSION,
452 process_counters: platform::network_process_counters(),
453 retry_annotations,
454 };
455 let decode = DecodeRetentionEvidenceV2 {
456 version: SYSTEM_EVIDENCE_V2_VERSION,
457 expansion_ratio_milli: milli_ratio(derived_decoder_bytes, input_bytes)
458 .map_or_else(|| gap(EvidenceGap::Unavailable), Evidence::recorded),
459 retained_bytes: runtime
460 .session_gauge(crate::GaugeId::RetainedBufferBytes)
461 .map_or_else(|| gap(EvidenceGap::Unavailable), Evidence::recorded),
462 retained_peak_bytes: runtime
463 .session_gauge(crate::GaugeId::RetainedBufferPeakBytes)
464 .map_or_else(|| gap(EvidenceGap::Unavailable), Evidence::recorded),
465 };
466 record_system_metrics(
467 runtime,
468 &faults,
469 &io,
470 &memory,
471 &allocation,
472 retry_annotations,
473 );
474 Evidence::recorded(SystemRunEvidenceV2 {
475 version: SYSTEM_EVIDENCE_V2_VERSION,
476 allocation,
477 faults,
478 io,
479 memory,
480 pressure: PressureEvidenceV2 {
481 version: SYSTEM_EVIDENCE_V2_VERSION,
482 cpu_some_avg10_milli: pressure.cpu_some_avg10_milli.clone(),
483 cpu_full_avg10_milli: pressure.cpu_full_avg10_milli.clone(),
484 memory_some_avg10_milli: pressure.memory_some_avg10_milli.clone(),
485 io_some_avg10_milli: pressure.io_some_avg10_milli.clone(),
486 },
487 thermal: ThermalEvidenceV2 {
488 version: SYSTEM_EVIDENCE_V2_VERSION,
489 max_zone_millicelsius: pressure.max_zone_millicelsius.clone(),
490 throttle_events: pressure.throttle_events.clone(),
491 },
492 network,
493 decode,
494 })
495 }
496
497 fn allocation_evidence(&self) -> AllocationEvidenceV2 {
498 let Some(start) = &self.allocation_start else {
499 let reason = if cfg!(feature = "allocation-tracking") {
500 EvidenceGap::Unavailable
501 } else {
502 EvidenceGap::CollectorDisabled
503 };
504 return AllocationEvidenceV2 {
505 version: SYSTEM_EVIDENCE_V2_VERSION,
506 totals: gap(reason),
507 stages: Vec::new(),
508 };
509 };
510 if !self.allocation_session.evidence_is_reliable() {
513 return AllocationEvidenceV2 {
514 version: SYSTEM_EVIDENCE_V2_VERSION,
515 totals: gap(EvidenceGap::Unavailable),
516 stages: Vec::new(),
517 };
518 }
519 let end = allocation_snapshot();
520 let totals = AllocationTotalsV2 {
521 version: SYSTEM_EVIDENCE_V2_VERSION,
522 allocations: end.allocations.saturating_sub(start.allocations),
523 deallocations: end.deallocations.saturating_sub(start.deallocations),
524 allocated_bytes: end.allocated_bytes.saturating_sub(start.allocated_bytes),
525 deallocated_bytes: end
526 .deallocated_bytes
527 .saturating_sub(start.deallocated_bytes),
528 live_bytes: end.live_bytes,
529 peak_live_bytes: end.peak_live_bytes,
530 };
531 let stages = crate::Stage::ALL
532 .into_iter()
533 .map(|stage| {
534 let start_slot = start.slot(stage);
535 let end_slot = end.slot(stage);
536 StageAllocationV2 {
537 version: SYSTEM_EVIDENCE_V2_VERSION,
538 metric_id: Some(stage.metric_id()),
539 allocations: end_slot.allocations.saturating_sub(start_slot.allocations),
540 allocated_bytes: end_slot
541 .allocated_bytes
542 .saturating_sub(start_slot.allocated_bytes),
543 live_bytes: end_slot.live_bytes,
544 peak_live_bytes: end_slot.peak_live_bytes,
545 }
546 })
547 .chain(std::iter::once(StageAllocationV2 {
548 version: SYSTEM_EVIDENCE_V2_VERSION,
549 metric_id: None,
550 allocations: end
551 .root()
552 .allocations
553 .saturating_sub(start.root().allocations),
554 allocated_bytes: end
555 .root()
556 .allocated_bytes
557 .saturating_sub(start.root().allocated_bytes),
558 live_bytes: end.root().live_bytes,
559 peak_live_bytes: end.root().peak_live_bytes,
560 }))
561 .collect();
562 AllocationEvidenceV2 {
563 version: SYSTEM_EVIDENCE_V2_VERSION,
564 totals: Evidence::recorded(totals),
565 stages,
566 }
567 }
568}
569
570fn memory_evidence(finish: &ResourceSnapshot) -> MemoryEvidenceV2 {
571 let sourced = |value: Option<u64>| match value {
572 Some(value) => SourcedEvidenceV2::recorded(value, HardwareFieldSourceV2::ProcSelfStatus),
573 None => SourcedEvidenceV2::gapped(
574 HardwareFieldSourceV2::ProcSelfStatus,
575 EvidenceGap::Unavailable,
576 ),
577 };
578 MemoryEvidenceV2 {
579 version: SYSTEM_EVIDENCE_V2_VERSION,
580 resident_bytes: sourced(finish.resident_bytes),
581 virtual_bytes: sourced(finish.virtual_bytes),
582 resident_high_water_bytes: sourced(finish.resident_high_water_bytes),
583 swap_bytes: sourced(finish.swap_bytes),
584 }
585}
586
587fn record_system_metrics(
588 runtime: &crate::Runtime,
589 faults: &FaultEvidenceV2,
590 io: &IoEvidenceV2,
591 memory: &MemoryEvidenceV2,
592 allocation: &AllocationEvidenceV2,
593 retry_annotations: u64,
594) {
595 let counters: [(&SourcedEvidenceV2<u64>, crate::CounterId); 7] = [
596 (&faults.minor_faults, crate::CounterId::MinorFaults),
597 (&faults.major_faults, crate::CounterId::MajorFaults),
598 (&io.read_bytes, crate::CounterId::IoReadBytes),
599 (&io.write_bytes, crate::CounterId::IoWriteBytes),
600 (&io.read_syscalls, crate::CounterId::IoReadSyscalls),
601 (&io.write_syscalls, crate::CounterId::IoWriteSyscalls),
602 (
603 &io.cancelled_write_bytes,
604 crate::CounterId::IoCancelledWriteBytes,
605 ),
606 ];
607 for (field, counter) in counters {
608 if let Evidence::Recorded { value } = field.value {
609 if value > 0 {
610 runtime.add_counter(counter, value);
611 }
612 }
613 }
614 if retry_annotations > 0 {
615 runtime.add_counter(crate::CounterId::NetworkRetries, retry_annotations);
616 }
617 if let Evidence::Recorded { value: hwm } = memory.resident_high_water_bytes.value {
618 runtime.set_gauge(crate::GaugeId::ResidentHighWaterBytes, hwm);
619 }
620 if let Evidence::Recorded { value: totals } = &allocation.totals {
621 if totals.allocations > 0 {
622 runtime.add_counter(crate::CounterId::AllocationCount, totals.allocations);
623 }
624 if totals.deallocations > 0 {
625 runtime.add_counter(crate::CounterId::DeallocationCount, totals.deallocations);
626 }
627 if totals.allocated_bytes > 0 {
628 runtime.add_counter(crate::CounterId::AllocationBytes, totals.allocated_bytes);
629 }
630 if totals.deallocated_bytes > 0 {
631 runtime.add_counter(
632 crate::CounterId::DeallocationBytes,
633 totals.deallocated_bytes,
634 );
635 }
636 runtime.set_gauge(crate::GaugeId::AllocationLiveBytes, totals.live_bytes);
637 runtime.set_gauge(
638 crate::GaugeId::AllocationPeakLiveBytes,
639 totals.peak_live_bytes,
640 );
641 }
642}