1use std::collections::BTreeMap;
9use std::sync::Arc;
10use std::time::{Instant, SystemTime, UNIX_EPOCH};
11
12use harn_vm::VmValue;
13use serde::{Deserialize, Serialize};
14#[cfg(target_os = "linux")]
15use std::path::Path;
16
17use crate::error::HostlibError;
18use crate::registry::{BuiltinRegistry, HostlibCapability, RegisteredBuiltin, SyncHandler};
19use crate::tools::args::{build_dict, dict_arg, str_value};
20
21pub const HOST_CONDITIONS_SCHEMA_VERSION: u32 = 1;
23
24const SAMPLE_BUILTIN: &str = "hostlib_host_conditions_sample";
25
26#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
28#[serde(rename_all = "snake_case")]
29pub enum HostContentionQuestion {
30 PromisedCpu,
32 NominalSpeed,
34 AcceleratorShared,
36 MemoryOrIoContended,
38}
39
40impl HostContentionQuestion {
41 pub const ALL: [Self; 4] = [
43 Self::PromisedCpu,
44 Self::NominalSpeed,
45 Self::AcceleratorShared,
46 Self::MemoryOrIoContended,
47 ];
48
49 pub const fn as_str(self) -> &'static str {
51 match self {
52 Self::PromisedCpu => "promised_cpu",
53 Self::NominalSpeed => "nominal_speed",
54 Self::AcceleratorShared => "accelerator_shared",
55 Self::MemoryOrIoContended => "memory_or_io_contended",
56 }
57 }
58}
59
60#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
62#[serde(rename_all = "snake_case")]
63pub enum HostConditionStatus {
64 Observed,
66 Unavailable,
68 NotObservable,
70}
71
72impl HostConditionStatus {
73 const fn as_str(self) -> &'static str {
74 match self {
75 Self::Observed => "observed",
76 Self::Unavailable => "unavailable",
77 Self::NotObservable => "not_observable",
78 }
79 }
80}
81
82#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
84#[serde(deny_unknown_fields)]
85pub struct HostConditionObservation {
86 pub question: HostContentionQuestion,
88 pub status: HostConditionStatus,
90 pub contention: Option<f64>,
92 pub reason: Option<String>,
94}
95
96impl HostConditionObservation {
97 pub fn observed(question: HostContentionQuestion, contention: f64) -> Self {
99 Self {
100 question,
101 status: HostConditionStatus::Observed,
102 contention: Some(contention.clamp(0.0, 1.0)),
103 reason: None,
104 }
105 }
106
107 pub fn unavailable(question: HostContentionQuestion, reason: impl Into<String>) -> Self {
109 Self {
110 question,
111 status: HostConditionStatus::Unavailable,
112 contention: None,
113 reason: Some(reason.into()),
114 }
115 }
116
117 pub fn not_observable(question: HostContentionQuestion, reason: impl Into<String>) -> Self {
119 Self {
120 question,
121 status: HostConditionStatus::NotObservable,
122 contention: None,
123 reason: Some(reason.into()),
124 }
125 }
126}
127
128#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
130#[serde(rename_all = "snake_case")]
131pub enum HostEnvironment {
132 BareMetal,
134 Virtualized,
136 Containerized,
138 Unknown,
140}
141
142impl HostEnvironment {
143 const fn as_str(self) -> &'static str {
144 match self {
145 Self::BareMetal => "bare_metal",
146 Self::Virtualized => "virtualized",
147 Self::Containerized => "containerized",
148 Self::Unknown => "unknown",
149 }
150 }
151}
152
153#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
155#[serde(deny_unknown_fields)]
156pub struct HostConditionsSnapshot {
157 pub schema_version: u32,
159 pub observed_at_ms: i64,
161 pub environment: HostEnvironment,
163 pub sample_cost_us: u64,
165 pub questions: Vec<HostConditionObservation>,
167}
168
169impl HostConditionsSnapshot {
170 fn normalize(&mut self) -> Result<(), String> {
171 if self.schema_version != HOST_CONDITIONS_SCHEMA_VERSION {
172 return Err(format!(
173 "unsupported response schema_version {}; expected {}",
174 self.schema_version, HOST_CONDITIONS_SCHEMA_VERSION
175 ));
176 }
177 let mut by_question = BTreeMap::new();
178 for observation in &self.questions {
179 if by_question
180 .insert(observation.question, observation)
181 .is_some()
182 {
183 return Err(format!(
184 "duplicate answer for {}",
185 observation.question.as_str()
186 ));
187 }
188 match observation.status {
189 HostConditionStatus::Observed => {
190 let Some(value) = observation.contention else {
191 return Err(format!(
192 "{} is observed but has no contention value",
193 observation.question.as_str()
194 ));
195 };
196 if !value.is_finite() || !(0.0..=1.0).contains(&value) {
197 return Err(format!(
198 "{} contention must be finite and between 0 and 1",
199 observation.question.as_str()
200 ));
201 }
202 if observation.reason.is_some() {
203 return Err(format!(
204 "{} is observed but includes an absence reason",
205 observation.question.as_str()
206 ));
207 }
208 }
209 HostConditionStatus::Unavailable | HostConditionStatus::NotObservable => {
210 if observation.contention.is_some() {
211 return Err(format!(
212 "{} is not observed but includes a contention value",
213 observation.question.as_str()
214 ));
215 }
216 if observation
217 .reason
218 .as_deref()
219 .is_none_or(|reason| reason.trim().is_empty())
220 {
221 return Err(format!(
222 "{} is not observed but has no reason",
223 observation.question.as_str()
224 ));
225 }
226 }
227 }
228 }
229 for question in HostContentionQuestion::ALL {
230 if !by_question.contains_key(&question) {
231 return Err(format!("missing answer for {}", question.as_str()));
232 }
233 }
234 drop(by_question);
235 self.questions
236 .sort_by_key(|observation| observation.question);
237 Ok(())
238 }
239}
240
241pub trait HostConditionsSource: Send + Sync + 'static {
243 fn sample(&self) -> Result<HostConditionsSnapshot, String>;
245}
246
247#[derive(Clone)]
249pub struct InjectedHostConditionsSource {
250 snapshot: HostConditionsSnapshot,
251}
252
253impl InjectedHostConditionsSource {
254 pub fn new(snapshot: HostConditionsSnapshot) -> Self {
256 Self { snapshot }
257 }
258}
259
260impl HostConditionsSource for InjectedHostConditionsSource {
261 fn sample(&self) -> Result<HostConditionsSnapshot, String> {
262 Ok(self.snapshot.clone())
263 }
264}
265
266#[derive(Default)]
268pub struct LocalHostConditionsSource;
269
270impl HostConditionsSource for LocalHostConditionsSource {
271 fn sample(&self) -> Result<HostConditionsSnapshot, String> {
272 let started = Instant::now();
273 let (environment, questions) = local_questions();
274 Ok(HostConditionsSnapshot {
275 schema_version: HOST_CONDITIONS_SCHEMA_VERSION,
276 observed_at_ms: unix_time_ms(),
277 environment,
278 sample_cost_us: started.elapsed().as_micros().try_into().unwrap_or(u64::MAX),
279 questions,
280 })
281 }
282}
283
284#[derive(Clone)]
286pub struct HostConditionsCapability {
287 source: Arc<dyn HostConditionsSource>,
288}
289
290impl Default for HostConditionsCapability {
291 fn default() -> Self {
292 Self::with_source(Arc::new(LocalHostConditionsSource))
293 }
294}
295
296impl HostConditionsCapability {
297 pub fn with_source(source: Arc<dyn HostConditionsSource>) -> Self {
299 Self { source }
300 }
301
302 fn sample_builtin(&self, args: &[VmValue]) -> Result<VmValue, HostlibError> {
303 let request = dict_arg(SAMPLE_BUILTIN, args)?;
304 let version = match request.get("schema_version") {
305 Some(VmValue::Int(version)) if *version > 0 => {
306 u32::try_from(*version).map_err(|_| HostlibError::InvalidParameter {
307 builtin: SAMPLE_BUILTIN,
308 param: "schema_version",
309 message: "must fit in an unsigned 32-bit integer".to_string(),
310 })?
311 }
312 None => {
313 return Err(HostlibError::MissingParameter {
314 builtin: SAMPLE_BUILTIN,
315 param: "schema_version",
316 });
317 }
318 _ => {
319 return Err(HostlibError::InvalidParameter {
320 builtin: SAMPLE_BUILTIN,
321 param: "schema_version",
322 message: "must be a positive integer".to_string(),
323 });
324 }
325 };
326 if version != HOST_CONDITIONS_SCHEMA_VERSION {
327 return Err(HostlibError::InvalidParameter {
328 builtin: SAMPLE_BUILTIN,
329 param: "schema_version",
330 message: format!(
331 "unsupported version {version}; expected {HOST_CONDITIONS_SCHEMA_VERSION}"
332 ),
333 });
334 }
335 let mut snapshot = self
336 .source
337 .sample()
338 .map_err(|message| HostlibError::Backend {
339 builtin: SAMPLE_BUILTIN,
340 message,
341 })?;
342 snapshot
343 .normalize()
344 .map_err(|message| HostlibError::Backend {
345 builtin: SAMPLE_BUILTIN,
346 message: format!("source returned an invalid host-conditions snapshot: {message}"),
347 })?;
348 snapshot_to_value(&snapshot)
349 }
350}
351
352impl HostlibCapability for HostConditionsCapability {
353 fn module_name(&self) -> &'static str {
354 "host_conditions"
355 }
356
357 fn register_builtins(&self, registry: &mut BuiltinRegistry) {
358 let capability = self.clone();
359 let handler: SyncHandler = Arc::new(move |args| capability.sample_builtin(args));
360 registry.register(RegisteredBuiltin {
361 name: SAMPLE_BUILTIN,
362 module: "host_conditions",
363 method: "sample",
364 handler,
365 });
366 }
367}
368
369fn snapshot_to_value(snapshot: &HostConditionsSnapshot) -> Result<VmValue, HostlibError> {
370 let questions = snapshot
371 .questions
372 .iter()
373 .map(|observation| {
374 build_dict([
375 ("question", str_value(observation.question.as_str())),
376 ("status", str_value(observation.status.as_str())),
377 (
378 "contention",
379 observation
380 .contention
381 .map(VmValue::Float)
382 .unwrap_or(VmValue::Nil),
383 ),
384 (
385 "reason",
386 observation
387 .reason
388 .as_deref()
389 .map(str_value)
390 .unwrap_or(VmValue::Nil),
391 ),
392 ])
393 })
394 .collect();
395 Ok(build_dict([
396 (
397 "schema_version",
398 VmValue::Int(i64::from(snapshot.schema_version)),
399 ),
400 ("observed_at_ms", VmValue::Int(snapshot.observed_at_ms)),
401 ("environment", str_value(snapshot.environment.as_str())),
402 (
403 "sample_cost_us",
404 VmValue::Int(snapshot.sample_cost_us.try_into().map_err(|_| {
405 HostlibError::Backend {
406 builtin: SAMPLE_BUILTIN,
407 message: "sample cost exceeds Harn integer range".to_string(),
408 }
409 })?),
410 ),
411 ("questions", VmValue::List(Arc::new(questions))),
412 ]))
413}
414
415fn unix_time_ms() -> i64 {
416 SystemTime::now()
417 .duration_since(UNIX_EPOCH)
418 .unwrap_or_default()
419 .as_millis()
420 .try_into()
421 .unwrap_or(i64::MAX)
422}
423
424fn local_questions() -> (HostEnvironment, Vec<HostConditionObservation>) {
425 #[cfg(target_os = "linux")]
426 {
427 let input = LinuxProbeInput::read();
428 linux_questions(&input)
429 }
430 #[cfg(not(target_os = "linux"))]
431 {
432 let environment = local_non_linux_environment();
433 #[cfg(target_os = "macos")]
434 let promised_cpu = {
435 let load = sysinfo::System::load_average().one;
436 let cores = std::thread::available_parallelism()
437 .map(usize::from)
438 .unwrap_or(1);
439 HostConditionObservation::observed(
440 HostContentionQuestion::PromisedCpu,
441 load / cores as f64,
442 )
443 };
444 #[cfg(not(target_os = "macos"))]
445 let promised_cpu = HostConditionObservation::unavailable(
446 HostContentionQuestion::PromisedCpu,
447 "native CPU contention counters are unavailable on this platform",
448 );
449 (
450 environment,
451 vec![
452 promised_cpu,
453 if environment == HostEnvironment::Virtualized {
454 HostConditionObservation::not_observable(
455 HostContentionQuestion::NominalSpeed,
456 "guest cannot observe hypervisor speed caps or host thermal throttling",
457 )
458 } else {
459 HostConditionObservation::unavailable(
460 HostContentionQuestion::NominalSpeed,
461 "native nominal-speed probe is unavailable on this platform",
462 )
463 },
464 HostConditionObservation::not_observable(
465 HostContentionQuestion::AcceleratorShared,
466 "local allocation metadata does not expose accelerator sharing",
467 ),
468 HostConditionObservation::unavailable(
469 HostContentionQuestion::MemoryOrIoContended,
470 "native memory and IO pressure counters are unavailable on this platform",
471 ),
472 ],
473 )
474 }
475}
476
477#[cfg(target_os = "macos")]
478fn local_non_linux_environment() -> HostEnvironment {
479 let mut present: libc::c_int = 0;
480 let mut len = std::mem::size_of::<libc::c_int>();
481 let name = c"kern.hv_vmm_present";
482 let result = unsafe {
485 libc::sysctlbyname(
486 name.as_ptr(),
487 (&raw mut present).cast(),
488 &raw mut len,
489 std::ptr::null_mut(),
490 0,
491 )
492 };
493 if result == 0 && present == 1 {
494 HostEnvironment::Virtualized
495 } else {
496 HostEnvironment::BareMetal
497 }
498}
499
500#[cfg(not(any(target_os = "linux", target_os = "macos")))]
501fn local_non_linux_environment() -> HostEnvironment {
502 HostEnvironment::Unknown
503}
504
505#[cfg(any(target_os = "linux", test))]
506#[derive(Default)]
507struct LinuxProbeInput {
508 container_marker: bool,
509 cgroup: Option<String>,
510 product_name: Option<String>,
511 proc_stat: Option<String>,
512 cpu_stat: Option<String>,
513 cpu_max: Option<String>,
514 memory_pressure: Option<String>,
515 io_pressure: Option<String>,
516 load_one: f64,
517 cores: usize,
518 current_frequency_khz: Option<u64>,
519 max_frequency_khz: Option<u64>,
520}
521
522#[cfg(any(target_os = "linux", test))]
523impl LinuxProbeInput {
524 #[cfg(target_os = "linux")]
525 fn read() -> Self {
526 Self {
527 container_marker: Path::new("/.dockerenv").exists(),
528 cgroup: read_to_string("/proc/1/cgroup"),
529 product_name: read_to_string("/sys/class/dmi/id/product_name"),
530 proc_stat: read_to_string("/proc/stat"),
531 cpu_stat: read_to_string("/sys/fs/cgroup/cpu.stat"),
532 cpu_max: read_to_string("/sys/fs/cgroup/cpu.max"),
533 memory_pressure: read_to_string("/proc/pressure/memory"),
534 io_pressure: read_to_string("/proc/pressure/io"),
535 load_one: sysinfo::System::load_average().one,
536 cores: std::thread::available_parallelism()
537 .map(usize::from)
538 .unwrap_or(1),
539 current_frequency_khz: read_u64(
540 "/sys/devices/system/cpu/cpu0/cpufreq/scaling_cur_freq",
541 ),
542 max_frequency_khz: read_u64("/sys/devices/system/cpu/cpu0/cpufreq/cpuinfo_max_freq"),
543 }
544 }
545}
546
547#[cfg(any(target_os = "linux", test))]
548fn linux_questions(input: &LinuxProbeInput) -> (HostEnvironment, Vec<HostConditionObservation>) {
549 let environment = classify_linux_environment(input);
550 let promised_cpu = match environment {
551 HostEnvironment::Containerized => parse_cgroup_throttle(input)
552 .map(|value| {
553 HostConditionObservation::observed(HostContentionQuestion::PromisedCpu, value)
554 })
555 .unwrap_or_else(|| {
556 HostConditionObservation::unavailable(
557 HostContentionQuestion::PromisedCpu,
558 "container CPU quota or throttling counters could not be read",
559 )
560 }),
561 HostEnvironment::Virtualized => parse_steal_fraction(input.proc_stat.as_deref())
562 .map(|value| {
563 HostConditionObservation::observed(HostContentionQuestion::PromisedCpu, value)
564 })
565 .unwrap_or_else(|| {
566 HostConditionObservation::unavailable(
567 HostContentionQuestion::PromisedCpu,
568 "guest CPU steal counters could not be read",
569 )
570 }),
571 HostEnvironment::BareMetal | HostEnvironment::Unknown => {
572 HostConditionObservation::observed(
573 HostContentionQuestion::PromisedCpu,
574 input.load_one / input.cores.max(1) as f64,
575 )
576 }
577 };
578 let nominal_speed = match environment {
579 HostEnvironment::Containerized => parse_cgroup_throttle(input)
580 .map(|value| {
581 HostConditionObservation::observed(HostContentionQuestion::NominalSpeed, value)
582 })
583 .unwrap_or_else(|| {
584 HostConditionObservation::unavailable(
585 HostContentionQuestion::NominalSpeed,
586 "container CPU throttling counters could not be read",
587 )
588 }),
589 HostEnvironment::Virtualized => HostConditionObservation::not_observable(
590 HostContentionQuestion::NominalSpeed,
591 "guest cannot observe host thermal throttling, credits, or hypervisor caps",
592 ),
593 HostEnvironment::BareMetal | HostEnvironment::Unknown => {
594 match (input.current_frequency_khz, input.max_frequency_khz) {
595 (Some(current), Some(max)) if max > 0 => HostConditionObservation::observed(
596 HostContentionQuestion::NominalSpeed,
597 1.0 - current as f64 / max as f64,
598 ),
599 _ => HostConditionObservation::unavailable(
600 HostContentionQuestion::NominalSpeed,
601 "CPU frequency counters could not be read",
602 ),
603 }
604 }
605 };
606 let pressure = match (
607 parse_psi(input.memory_pressure.as_deref()),
608 parse_psi(input.io_pressure.as_deref()),
609 ) {
610 (Some(memory), Some(io)) => HostConditionObservation::observed(
611 HostContentionQuestion::MemoryOrIoContended,
612 memory.max(io),
613 ),
614 _ => HostConditionObservation::unavailable(
615 HostContentionQuestion::MemoryOrIoContended,
616 "memory or IO pressure counters could not be read",
617 ),
618 };
619 (
620 environment,
621 vec![
622 promised_cpu,
623 nominal_speed,
624 HostConditionObservation::not_observable(
625 HostContentionQuestion::AcceleratorShared,
626 "local allocation metadata does not expose accelerator sharing",
627 ),
628 pressure,
629 ],
630 )
631}
632
633#[cfg(any(target_os = "linux", test))]
634fn classify_linux_environment(input: &LinuxProbeInput) -> HostEnvironment {
635 let cgroup = input
636 .cgroup
637 .as_deref()
638 .unwrap_or_default()
639 .to_ascii_lowercase();
640 if input.container_marker
641 || ["docker", "containerd", "kubepods", "podman", "lxc"]
642 .iter()
643 .any(|needle| cgroup.contains(needle))
644 {
645 return HostEnvironment::Containerized;
646 }
647 let product = input
648 .product_name
649 .as_deref()
650 .unwrap_or_default()
651 .to_ascii_lowercase();
652 if [
653 "kvm",
654 "qemu",
655 "vmware",
656 "virtualbox",
657 "virtual machine",
658 "xen",
659 "amazon ec2",
660 "google compute",
661 ]
662 .iter()
663 .any(|needle| product.contains(needle))
664 {
665 HostEnvironment::Virtualized
666 } else {
667 HostEnvironment::BareMetal
668 }
669}
670
671#[cfg(any(target_os = "linux", test))]
672fn parse_cgroup_throttle(input: &LinuxProbeInput) -> Option<f64> {
673 let stat = input.cpu_stat.as_deref()?;
674 let quota = input.cpu_max.as_deref()?.split_whitespace().next()?;
675 if quota == "max" {
676 return None;
677 }
678 let fields: BTreeMap<_, _> = stat
679 .lines()
680 .filter_map(|line| line.split_once(' '))
681 .collect();
682 let usage: f64 = fields.get("usage_usec")?.parse().ok()?;
683 let throttled: f64 = fields.get("throttled_usec")?.parse().ok()?;
684 let total = usage + throttled;
685 Some(if total > 0.0 { throttled / total } else { 0.0 })
686}
687
688#[cfg(any(target_os = "linux", test))]
689fn parse_steal_fraction(proc_stat: Option<&str>) -> Option<f64> {
690 let cpu = proc_stat?.lines().find(|line| line.starts_with("cpu "))?;
691 let values: Vec<f64> = cpu
692 .split_whitespace()
693 .skip(1)
694 .map(str::parse)
695 .collect::<Result<_, _>>()
696 .ok()?;
697 let steal = *values.get(7)?;
698 let total: f64 = values.iter().sum();
699 Some(if total > 0.0 { steal / total } else { 0.0 })
700}
701
702#[cfg(any(target_os = "linux", test))]
703fn parse_psi(value: Option<&str>) -> Option<f64> {
704 let some = value?.lines().find(|line| line.starts_with("some "))?;
705 let avg10 = some
706 .split_whitespace()
707 .find_map(|field| field.strip_prefix("avg10="))?
708 .parse::<f64>()
709 .ok()?;
710 Some(avg10 / 100.0)
711}
712
713#[cfg(target_os = "linux")]
714fn read_to_string(path: impl AsRef<Path>) -> Option<String> {
715 std::fs::read_to_string(path).ok()
716}
717
718#[cfg(target_os = "linux")]
719fn read_u64(path: impl AsRef<Path>) -> Option<u64> {
720 read_to_string(path)?.trim().parse().ok()
721}
722
723#[cfg(test)]
724mod tests {
725 use super::*;
726
727 #[test]
728 fn container_uses_quota_throttling_and_pressure_questions() {
729 let input = LinuxProbeInput {
730 container_marker: true,
731 cpu_stat: Some("usage_usec 900\nthrottled_usec 100\n".to_string()),
732 cpu_max: Some("200000 100000\n".to_string()),
733 memory_pressure: Some("some avg10=2.50 avg60=0.00 total=1\n".to_string()),
734 io_pressure: Some("some avg10=4.00 avg60=0.00 total=1\n".to_string()),
735 cores: 2,
736 ..LinuxProbeInput::default()
737 };
738 let (environment, answers) = linux_questions(&input);
739 assert_eq!(environment, HostEnvironment::Containerized);
740 assert_eq!(answers[0].contention, Some(0.1));
741 assert_eq!(answers[1].contention, Some(0.1));
742 assert_eq!(answers[3].contention, Some(0.04));
743 }
744
745 #[test]
746 fn virtual_guest_uses_steal_and_never_calls_thermal_absence_quiet() {
747 let input = LinuxProbeInput {
748 product_name: Some("KVM Virtual Machine".to_string()),
749 proc_stat: Some("cpu 10 0 10 70 0 0 0 10 0 0\n".to_string()),
750 memory_pressure: Some("some avg10=0.00 avg60=0.00 total=0\n".to_string()),
751 io_pressure: Some("some avg10=0.00 avg60=0.00 total=0\n".to_string()),
752 cores: 2,
753 ..LinuxProbeInput::default()
754 };
755 let (environment, answers) = linux_questions(&input);
756 assert_eq!(environment, HostEnvironment::Virtualized);
757 assert_eq!(answers[0].contention, Some(0.1));
758 assert_eq!(answers[1].status, HostConditionStatus::NotObservable);
759 assert_eq!(answers[1].contention, None);
760 }
761
762 #[test]
763 fn snapshot_rejects_missing_or_incoherent_answers() {
764 let mut snapshot = HostConditionsSnapshot {
765 schema_version: HOST_CONDITIONS_SCHEMA_VERSION,
766 observed_at_ms: 1,
767 environment: HostEnvironment::BareMetal,
768 sample_cost_us: 10,
769 questions: vec![HostConditionObservation {
770 question: HostContentionQuestion::PromisedCpu,
771 status: HostConditionStatus::Unavailable,
772 contention: Some(0.0),
773 reason: None,
774 }],
775 };
776 assert!(snapshot.normalize().is_err());
777 }
778}