petri/vm/openvmm/
runtime.rs

1// Copyright (c) Microsoft Corporation.
2// Licensed under the MIT License.
3
4//! Methods to interact with a running [`PetriVmOpenVmm`].
5
6use super::PetriVmResourcesOpenVmm;
7use crate::OpenHclServicingFlags;
8use crate::PetriHaltReason;
9use crate::PetriHaltReasonDetail;
10use crate::PetriVmFramebufferAccess;
11use crate::PetriVmInspector;
12use crate::PetriVmRuntime;
13use crate::ShutdownKind;
14use crate::VmScreenshotMeta;
15use crate::openhcl_diag::OpenHclDiagHandler;
16use crate::worker::Worker;
17use anyhow::Context;
18use async_trait::async_trait;
19use framebuffer::View;
20use futures::FutureExt;
21use futures_concurrency::future::Race;
22use get_resources::ged::FirmwareEvent;
23use get_resources::ged::IpmiSelEvent;
24use hyperv_ic_resources::shutdown::ShutdownRpc;
25use mesh::CancelContext;
26use mesh::Receiver;
27use mesh::RecvError;
28use mesh::rpc::RpcError;
29use mesh::rpc::RpcSend;
30use mesh_process::Mesh;
31use openvmm_defs::rpc::PulseSaveRestoreError;
32use pal_async::socket::PolledSocket;
33use petri_artifacts_core::ResolvedArtifact;
34#[cfg(target_os = "linux")]
35use pipette_client::PIPETTE_PORT;
36use pipette_client::PipetteClient;
37use std::future::Future;
38use std::path::Path;
39use std::sync::Arc;
40use std::time::Duration;
41use vmm_core_defs::HaltReason;
42#[cfg(target_os = "linux")]
43use vmsocket::VmAddress;
44#[cfg(target_os = "linux")]
45use vmsocket::VmSocket;
46use vtl2_settings_proto::Vtl2Settings;
47
48/// A running VM that tests can interact with.
49// DEVNOTE: Really the PetriVmInner is the actual VM and channels that we interact
50// with. This struct exists as a wrapper to provide error handling, such as not
51// hanging indefinitely when waiting on certain channels if the VM crashes.
52pub struct PetriVmOpenVmm {
53    inner: PetriVmInner,
54    halt: PetriVmHaltReceiver,
55}
56
57#[async_trait]
58impl PetriVmRuntime for PetriVmOpenVmm {
59    type VmInspector = OpenVmmInspector;
60    type VmFramebufferAccess = OpenVmmFramebufferAccess;
61
62    async fn teardown(self) -> anyhow::Result<()> {
63        tracing::info!("waiting for worker");
64        let worker = Arc::into_inner(self.inner.worker)
65            .context("all references to the OpenVMM worker have not been closed")?;
66        worker.shutdown().await?;
67
68        tracing::info!("Worker quit, waiting for mesh");
69        self.inner.mesh.shutdown().await;
70
71        tracing::info!("Mesh shutdown, waiting for logging tasks");
72        for t in self.inner.resources.log_stream_tasks {
73            t.await?;
74        }
75
76        Ok(())
77    }
78
79    async fn wait_for_halt(&mut self, allow_reset: bool) -> anyhow::Result<PetriHaltReasonDetail> {
80        let halt_reason = if let Some(already) = self.halt.already_received.take() {
81            already.map_err(anyhow::Error::from)
82        } else {
83            self.halt
84                .halt_notif
85                .recv()
86                .await
87                .context("Failed to get halt reason")
88        }?;
89
90        tracing::info!(?halt_reason, "Got halt reason");
91
92        let reason = match halt_reason {
93            HaltReason::PowerOff => PetriHaltReason::PowerOff,
94            HaltReason::Reset => PetriHaltReason::Reset,
95            HaltReason::Hibernate => PetriHaltReason::Hibernate,
96            HaltReason::TripleFault { .. } => PetriHaltReason::TripleFault,
97            _ => PetriHaltReason::Other,
98        };
99
100        if allow_reset && reason == PetriHaltReason::Reset {
101            self.reset().await?
102        }
103
104        Ok(PetriHaltReasonDetail {
105            reason,
106            detail: format!("{halt_reason:?}"),
107        })
108    }
109
110    async fn wait_for_agent(&mut self, set_high_vtl: bool) -> anyhow::Result<PipetteClient> {
111        Self::wait_for_agent(self, set_high_vtl).await
112    }
113
114    fn openhcl_diag(&self) -> Option<OpenHclDiagHandler> {
115        self.inner.resources.vtl2_vsock_path.as_ref().map(|path| {
116            OpenHclDiagHandler::new(diag_client::DiagClient::from_hybrid_vsock(
117                self.inner.resources.driver.clone(),
118                path,
119            ))
120        })
121    }
122
123    async fn wait_for_boot_event(
124        &mut self,
125        timeout: Option<Duration>,
126    ) -> anyhow::Result<Option<FirmwareEvent>> {
127        // The event arrives on a channel, so cancelling the wait can't discard
128        // one that has already been delivered.
129        CancelContext::new()
130            .with_timeout(timeout.unwrap_or(Duration::MAX))
131            .until_cancelled(Self::wait_for_boot_event(self))
132            .await
133            .ok()
134            .transpose()
135    }
136
137    async fn wait_for_enlightened_shutdown_ready(&mut self) -> anyhow::Result<()> {
138        Self::wait_for_enlightened_shutdown_ready(self)
139            .await
140            .map(|_| ())
141    }
142
143    async fn send_enlightened_shutdown(&mut self, kind: ShutdownKind) -> anyhow::Result<()> {
144        Self::send_enlightened_shutdown(self, kind).await
145    }
146
147    async fn restart_openhcl(
148        &mut self,
149        new_openhcl: &ResolvedArtifact,
150        flags: OpenHclServicingFlags,
151    ) -> anyhow::Result<()> {
152        Self::save_openhcl(self, new_openhcl, flags).await?;
153        Self::restore_openhcl(self).await
154    }
155
156    async fn save_openhcl(
157        &mut self,
158        new_openhcl: &ResolvedArtifact,
159        flags: OpenHclServicingFlags,
160    ) -> anyhow::Result<()> {
161        Self::save_openhcl(self, new_openhcl, flags).await
162    }
163
164    async fn restore_openhcl(&mut self) -> anyhow::Result<()> {
165        Self::restore_openhcl(self).await
166    }
167
168    async fn update_command_line(&mut self, command_line: &str) -> anyhow::Result<()> {
169        Self::update_command_line(self, command_line).await
170    }
171
172    fn inspector(&self) -> Option<OpenVmmInspector> {
173        Some(OpenVmmInspector {
174            worker: self.inner.worker.clone(),
175        })
176    }
177
178    fn take_framebuffer_access(&mut self) -> Option<OpenVmmFramebufferAccess> {
179        self.inner
180            .framebuffer_view
181            .take()
182            .map(|view| OpenVmmFramebufferAccess { view })
183    }
184
185    async fn reset(&mut self) -> anyhow::Result<()> {
186        Self::reset(self).await
187    }
188
189    async fn set_vtl2_settings(&mut self, settings: &Vtl2Settings) -> anyhow::Result<()> {
190        Self::set_vtl2_settings(self, settings).await
191    }
192
193    async fn set_vmbus_drive(
194        &mut self,
195        _disk: &crate::Drive,
196        _controller_id: &guid::Guid,
197        _controller_location: u32,
198    ) -> anyhow::Result<()> {
199        todo!("openvmm set vmbus drive")
200    }
201
202    async fn add_pcie_device(
203        &mut self,
204        port_name: String,
205        resource: vm_resource::Resource<vm_resource::kind::PciDeviceHandleKind>,
206    ) -> anyhow::Result<()> {
207        Self::add_pcie_device(self, port_name, resource).await
208    }
209
210    async fn remove_pcie_device(&mut self, port_name: String) -> anyhow::Result<()> {
211        Self::remove_pcie_device(self, port_name).await
212    }
213}
214
215pub(super) struct PetriVmInner {
216    pub(super) resources: PetriVmResourcesOpenVmm,
217    pub(super) mesh: Mesh,
218    pub(super) worker: Arc<Worker>,
219    pub(super) framebuffer_view: Option<View>,
220    /// Whether CIDATA has already been mounted inside the guest.
221    /// Used to skip re-mounting after save/restore (where guest state is
222    /// preserved) while still mounting after a full reset/reboot.
223    pub(super) cidata_mounted: bool,
224    /// Resolved TCP pipette port for no-vmbus Windows guests. Set once
225    /// during startup and reused across reconnections (e.g. after reset).
226    pub(super) tcp_pipette_port: Option<u16>,
227    pub(super) pid: i32,
228}
229
230struct PetriVmHaltReceiver {
231    halt_notif: Receiver<HaltReason>,
232    already_received: Option<Result<HaltReason, RecvError>>,
233}
234
235// Wrap a PetriVmInner function in [`PetriVmOpenVmm::wait_for_halt_or_internal`] to
236// provide better error handling.
237macro_rules! petri_vm_fn {
238    ($(#[$($attrss:tt)*])* $vis:vis async fn $fn_name:ident (&mut self $(,$arg:ident: $ty:ty)*) $(-> $ret:ty)?) => {
239        $(#[$($attrss)*])*
240        $vis async fn $fn_name(&mut self, $($arg:$ty,)*) $(-> $ret)? {
241            Self::wait_for_halt_or_internal(&mut self.halt, self.inner.$fn_name($($arg,)*)).await
242        }
243    };
244}
245
246// TODO: Add all runtime functions that are not backend specific
247// to the `PetriVmRuntime` trait
248impl PetriVmOpenVmm {
249    pub(super) fn new(inner: PetriVmInner, halt_notif: Receiver<HaltReason>) -> Self {
250        Self {
251            inner,
252            halt: PetriVmHaltReceiver {
253                halt_notif,
254                already_received: None,
255            },
256        }
257    }
258
259    /// Get the path to the VTL 2 vsock socket, if the VM is configured with OpenHCL.
260    pub fn vtl2_vsock_path(&self) -> anyhow::Result<&Path> {
261        self.inner
262            .resources
263            .vtl2_vsock_path
264            .as_deref()
265            .context("VM is not configured with OpenHCL")
266    }
267
268    /// Get the PID of the openvmm child process.
269    pub fn pid(&self) -> i32 {
270        self.inner.pid
271    }
272
273    petri_vm_fn!(
274        /// Waits for an event emitted by the firmware about its boot status, and
275        /// returns that status.
276        pub async fn wait_for_boot_event(&mut self) -> anyhow::Result<FirmwareEvent>
277    );
278    petri_vm_fn!(
279        /// Waits for an IPMI SEL notification received from OpenHCL.
280        pub async fn wait_for_ipmi_sel(&mut self) -> anyhow::Result<IpmiSelEvent>
281    );
282    petri_vm_fn!(
283        /// Waits for the Hyper-V shutdown IC to be ready, returning a receiver
284        /// that will be closed when it is no longer ready. Returns `None` if
285        /// the shutdown IC is not configured.
286        pub async fn wait_for_enlightened_shutdown_ready(&mut self) -> anyhow::Result<Option<mesh::OneshotReceiver<()>>>
287    );
288    petri_vm_fn!(
289        /// Instruct the guest to shutdown via the Hyper-V shutdown IC.
290        pub async fn send_enlightened_shutdown(&mut self, kind: ShutdownKind) -> anyhow::Result<()>
291    );
292    petri_vm_fn!(
293        /// Waits for the KVP IC to be ready, returning a sender that can be used
294        /// to send requests to it.
295        pub async fn wait_for_kvp(&mut self) -> anyhow::Result<mesh::Sender<hyperv_ic_resources::kvp::KvpRpc>>
296    );
297    petri_vm_fn!(
298        /// Stages the new OpenHCL file and saves the existing state.
299        pub async fn save_openhcl(
300            &mut self,
301            new_openhcl: &ResolvedArtifact,
302            flags: OpenHclServicingFlags
303        ) -> anyhow::Result<()>
304    );
305    petri_vm_fn!(
306        /// Restores OpenHCL from a previously saved state.
307        pub async fn restore_openhcl(
308            &mut self
309        ) -> anyhow::Result<()>
310    );
311    petri_vm_fn!(
312        /// Updates the command line parameters of the running VM.
313        pub async fn update_command_line(
314            &mut self,
315            command_line: &str
316        ) -> anyhow::Result<()>
317    );
318
319    petri_vm_fn!(
320        /// Hot-add a PCIe device to a named port at runtime.
321        pub async fn add_pcie_device(
322            &mut self,
323            port_name: String,
324            resource: vm_resource::Resource<vm_resource::kind::PciDeviceHandleKind>
325        ) -> anyhow::Result<()>
326    );
327    petri_vm_fn!(
328        /// Hot-remove a PCIe device from a named port at runtime.
329        pub async fn remove_pcie_device(
330            &mut self,
331            port_name: String
332        ) -> anyhow::Result<()>
333    );
334    petri_vm_fn!(
335        /// Resets the hardware state of the VM, simulating a power cycle.
336        pub async fn reset(&mut self) -> anyhow::Result<()>
337    );
338    petri_vm_fn!(
339        /// Dumps the VM's processor and memory state to a `.vmrs` file at `path`.
340        pub async fn dump_state(&mut self, path: &Path) -> anyhow::Result<()>
341    );
342    petri_vm_fn!(
343        /// Wait for a connection from a pipette agent
344        pub async fn wait_for_agent(&mut self, set_high_vtl: bool) -> anyhow::Result<PipetteClient>
345    );
346    petri_vm_fn!(
347        /// Set the OpenHCL VTL2 settings.
348        pub async fn set_vtl2_settings(&mut self, settings: &Vtl2Settings) -> anyhow::Result<()>
349    );
350
351    petri_vm_fn!(
352        /// Pause the VM. Call [`resume`](Self::resume) to continue execution.
353        pub async fn pause(&mut self) -> anyhow::Result<()>
354    );
355    petri_vm_fn!(
356        /// Save the VM's device and processor state, returning the serialized
357        /// bytes. The VM should be paused before calling this.
358        pub async fn save_state(&mut self) -> anyhow::Result<Vec<u8>>
359    );
360    petri_vm_fn!(
361        /// Resume a paused VM.
362        pub async fn resume(&mut self) -> anyhow::Result<()>
363    );
364    petri_vm_fn!(
365        /// Perform a pulse save/restore cycle: pause the VM, save all state,
366        /// reset, restore, and resume. Useful for verifying that device state
367        /// survives a save/restore round-trip.
368        pub async fn verify_save_restore(&mut self) -> anyhow::Result<()>
369    );
370    petri_vm_fn!(pub(crate) async fn launch_linux_direct_pipette(&mut self) -> anyhow::Result<()>);
371
372    /// Wrap the provided future in a race with the worker process's halt
373    /// notification channel. This is useful for preventing a future from
374    /// waiting indefinitely if the VM dies for any reason. If the worker
375    /// process crashes the halt notification channel will return an error, and
376    /// if the VM halts for any other reason the future will complete with that
377    /// reason.
378    pub async fn wait_for_halt_or<T, F: Future<Output = anyhow::Result<T>>>(
379        &mut self,
380        future: F,
381    ) -> anyhow::Result<T> {
382        Self::wait_for_halt_or_internal(&mut self.halt, future).await
383    }
384
385    async fn wait_for_halt_or_internal<T, F: Future<Output = anyhow::Result<T>>>(
386        halt: &mut PetriVmHaltReceiver,
387        future: F,
388    ) -> anyhow::Result<T> {
389        let future = &mut std::pin::pin!(future);
390        enum Either<T> {
391            Future(anyhow::Result<T>),
392            Halt(Result<HaltReason, RecvError>),
393        }
394        let res = (
395            future.map(Either::Future),
396            halt.halt_notif.recv().map(Either::Halt),
397        )
398            .race()
399            .await;
400
401        match res {
402            Either::Future(Ok(success)) => Ok(success),
403            Either::Future(Err(e)) => {
404                tracing::warn!(
405                    ?e,
406                    "Future returned with an error, sleeping for 5 seconds to let outstanding work finish"
407                );
408                let mut c = CancelContext::new().with_timeout(Duration::from_secs(5));
409                c.cancelled().await;
410                Err(e)
411            }
412            Either::Halt(halt_result) => {
413                tracing::warn!(
414                    halt_result = format_args!("{:x?}", halt_result),
415                    "Halt channel returned while waiting for other future, sleeping for 5 seconds to let outstanding work finish"
416                );
417                let mut c = CancelContext::new().with_timeout(Duration::from_secs(5));
418                let try_again = c.until_cancelled(future).await;
419
420                match try_again {
421                    Ok(fut_result) => {
422                        halt.already_received = Some(halt_result);
423                        if let Err(e) = &fut_result {
424                            tracing::warn!(
425                                ?e,
426                                "Future returned with an error, sleeping for 5 seconds to let outstanding work finish"
427                            );
428                            let mut c = CancelContext::new().with_timeout(Duration::from_secs(5));
429                            c.cancelled().await;
430                        }
431                        fut_result
432                    }
433                    Err(_cancel) => match halt_result {
434                        Ok(halt_reason) => Err(anyhow::anyhow!("VM halted: {:x?}", halt_reason)),
435                        Err(e) => Err(e).context("VM disappeared"),
436                    },
437                }
438            }
439        }
440    }
441}
442
443impl PetriVmInner {
444    async fn wait_for_boot_event(&mut self) -> anyhow::Result<FirmwareEvent> {
445        self.resources
446            .firmware_event_recv
447            .recv()
448            .await
449            .context("Failed to get firmware boot event")
450    }
451
452    async fn wait_for_ipmi_sel(&mut self) -> anyhow::Result<IpmiSelEvent> {
453        CancelContext::new()
454            .with_timeout(Duration::from_secs(30))
455            .until_cancelled(self.resources.ipmi_sel_event_recv.recv())
456            .await
457            .context("timed out waiting for an IPMI SEL host notification")?
458            .context("IPMI SEL host notification channel closed")
459    }
460
461    async fn wait_for_enlightened_shutdown_ready(
462        &mut self,
463    ) -> anyhow::Result<Option<mesh::OneshotReceiver<()>>> {
464        let Some(send) = self.resources.shutdown_ic_send.as_ref() else {
465            return Ok(None);
466        };
467        let recv = send
468            .call(ShutdownRpc::WaitReady, ())
469            .await
470            .context("waiting for shutdown IC to be ready")?;
471        Ok(Some(recv))
472    }
473
474    async fn send_enlightened_shutdown(&mut self, kind: ShutdownKind) -> anyhow::Result<()> {
475        let send = self
476            .resources
477            .shutdown_ic_send
478            .as_ref()
479            .context("shutdown IC not configured")?;
480        let shutdown_result = send
481            .call(
482                ShutdownRpc::Shutdown,
483                hyperv_ic_resources::shutdown::ShutdownParams {
484                    shutdown_type: match kind {
485                        ShutdownKind::Shutdown => {
486                            hyperv_ic_resources::shutdown::ShutdownType::PowerOff
487                        }
488                        ShutdownKind::Reboot => hyperv_ic_resources::shutdown::ShutdownType::Reboot,
489                        ShutdownKind::Hibernate => {
490                            hyperv_ic_resources::shutdown::ShutdownType::Hibernate
491                        }
492                    },
493                    force: false,
494                },
495            )
496            .await?;
497
498        tracing::info!(?shutdown_result, "Shutdown sent");
499        anyhow::ensure!(
500            shutdown_result == hyperv_ic_resources::shutdown::ShutdownResult::Ok,
501            "Got non-Ok shutdown response"
502        );
503
504        Ok(())
505    }
506
507    async fn wait_for_kvp(
508        &mut self,
509    ) -> anyhow::Result<mesh::Sender<hyperv_ic_resources::kvp::KvpRpc>> {
510        tracing::info!("Waiting for KVP IC");
511        let send = self
512            .resources
513            .kvp_ic_send
514            .as_ref()
515            .context("KVP IC not configured")?;
516        let (send, _) = send
517            .call_failable(hyperv_ic_resources::kvp::KvpConnectRpc::WaitForGuest, ())
518            .await
519            .context("failed to connect to KVP IC")?;
520
521        Ok(send)
522    }
523
524    async fn save_openhcl(
525        &self,
526        new_openhcl: &ResolvedArtifact,
527        flags: OpenHclServicingFlags,
528    ) -> anyhow::Result<()> {
529        let ged_send = self
530            .resources
531            .ged_send
532            .as_ref()
533            .context("openhcl not configured")?;
534
535        let igvm_file = fs_err::File::open(new_openhcl).context("failed to open igvm file")?;
536        self.worker
537            .save_openhcl(ged_send, flags, igvm_file.into())
538            .await
539    }
540
541    async fn update_command_line(&mut self, command_line: &str) -> anyhow::Result<()> {
542        self.worker.update_command_line(command_line).await
543    }
544
545    async fn add_pcie_device(
546        &mut self,
547        port_name: String,
548        resource: vm_resource::Resource<vm_resource::kind::PciDeviceHandleKind>,
549    ) -> anyhow::Result<()> {
550        self.worker.add_pcie_device(port_name, resource).await
551    }
552
553    async fn remove_pcie_device(&mut self, port_name: String) -> anyhow::Result<()> {
554        self.worker.remove_pcie_device(port_name).await
555    }
556    async fn dump_state(&mut self, path: &Path) -> anyhow::Result<()> {
557        self.worker.dump_state(path).await
558    }
559    async fn restore_openhcl(&self) -> anyhow::Result<()> {
560        let ged_send = self
561            .resources
562            .ged_send
563            .as_ref()
564            .context("openhcl not configured")?;
565
566        self.worker.restore_openhcl(ged_send).await
567    }
568
569    async fn set_vtl2_settings(&self, settings: &Vtl2Settings) -> anyhow::Result<()> {
570        let ged_send = self
571            .resources
572            .ged_send
573            .as_ref()
574            .context("openhcl not configured")?;
575
576        ged_send
577            .call_failable(
578                get_resources::ged::GuestEmulationRequest::ModifyVtl2Settings,
579                prost::Message::encode_to_vec(settings),
580            )
581            .await?;
582
583        Ok(())
584    }
585
586    async fn reset(&mut self) -> anyhow::Result<()> {
587        tracing::info!("Resetting VM");
588        self.worker.reset().await?;
589        // Discard any firmware events from the boot that was abandoned, so
590        // that they aren't mistaken for the results of the new boot.
591        while self.resources.firmware_event_recv.try_recv().is_ok() {}
592        // Guest state is lost on reset, so CIDATA needs to be remounted.
593        self.cidata_mounted = false;
594        // Restart Petri-managed Linux-direct agents over serial. Image-owned
595        // and initrd-injected agents restart through their own boot path.
596        if let Some(agent) = self.resources.linux_direct_serial_agent.as_mut() {
597            agent.reset();
598
599            if self.resources.properties.using_vtl0_pipette {
600                self.launch_linux_direct_pipette().await?;
601            }
602        }
603        Ok(())
604    }
605
606    async fn wait_for_agent(&mut self, set_high_vtl: bool) -> anyhow::Result<PipetteClient> {
607        #[cfg(target_os = "linux")]
608        if let Some(guest_cid) = self.resources.properties.vhost_vsock_guest_cid {
609            assert!(
610                !set_high_vtl,
611                "kernel vhost-vsock pipette transport does not support VTL2"
612            );
613            return self.wait_for_agent_vhost_vsock(guest_cid).await;
614        }
615
616        // Use TCP transport if configured (Windows no-vmbus guests).
617        if let Some(port) = self.tcp_pipette_port {
618            assert!(!set_high_vtl, "TCP pipette transport does not support VTL2");
619            return self.wait_for_agent_tcp(port).await;
620        }
621
622        let listener = if set_high_vtl {
623            self.resources
624                .vtl2_pipette_listener
625                .as_mut()
626                .context("VM is not configured with VTL 2")?
627        } else {
628            &mut self.resources.pipette_listener
629        };
630
631        tracing::info!(set_high_vtl, "listening for pipette connection");
632        let client = loop {
633            let (conn, _) = listener
634                .accept()
635                .await
636                .context("failed to accept pipette connection")?;
637            tracing::info!(set_high_vtl, "handshaking with pipette");
638            let socket = PolledSocket::new(&self.resources.driver, conn)?;
639            match PipetteClient::new(&self.resources.driver, socket, &self.resources.output_dir)
640                .await
641            {
642                Ok(client) => break client,
643                Err(e) => {
644                    // During save/restore cycles, stale connections from
645                    // previous hvsock relay sessions can accumulate in the
646                    // listener backlog. These are already-closed sockets
647                    // that fail during the mesh handshake. Drain them and
648                    // retry until we get a live connection.
649                    tracing::warn!(
650                        error = e.as_ref() as &dyn std::error::Error,
651                        "pipette connection not live, retrying"
652                    );
653                }
654            }
655        };
656        tracing::info!(set_high_vtl, "completed pipette handshake");
657
658        // When pipette runs as PID 1 init and a CIDATA agent disk is
659        // attached, mount it so test files are available at /cidata.
660        // Skip if already mounted (e.g. reconnecting after save/restore
661        // where guest state is preserved).
662        if !set_high_vtl
663            && self.resources.properties.uses_pipette_as_init
664            && self.resources.properties.has_agent_disk
665            && !self.cidata_mounted
666        {
667            tracing::info!("mounting CIDATA agent disk via pipette");
668            client
669                .unix_shell()
670                .cmd("mkdir")
671                .arg("-p")
672                .arg("/cidata")
673                .run()
674                .await
675                .context("failed to create /cidata mount point")?;
676            client
677                .unix_shell()
678                .cmd("mount")
679                .arg("LABEL=cidata")
680                .arg("/cidata")
681                .run()
682                .await
683                .context("failed to mount CIDATA disk")?;
684            self.cidata_mounted = true;
685        }
686
687        Ok(client)
688    }
689
690    /// Connect to pipette directly through the host's AF_VSOCK namespace.
691    #[cfg(target_os = "linux")]
692    async fn wait_for_agent_vhost_vsock(
693        &mut self,
694        guest_cid: u32,
695    ) -> anyhow::Result<PipetteClient> {
696        tracing::info!(
697            guest_cid,
698            port = PIPETTE_PORT,
699            "connecting to pipette via kernel vhost-vsock"
700        );
701        let socket = loop {
702            let connect = async {
703                let socket = VmSocket::new().context("failed to create AF_VSOCK socket")?;
704                socket
705                    .set_connect_timeout(Duration::from_secs(5))
706                    .context("failed to set AF_VSOCK connect timeout")?;
707                let mut socket = PolledSocket::new(&self.resources.driver, socket)
708                    .context("failed to create polled AF_VSOCK socket")?
709                    .convert();
710                socket
711                    .connect(&VmAddress::vsock(guest_cid, PIPETTE_PORT).into())
712                    .await
713                    .context("failed to connect to guest AF_VSOCK listener")?;
714                Ok::<_, anyhow::Error>(socket)
715            };
716
717            match connect.await {
718                Ok(socket) => break socket,
719                Err(error) => {
720                    tracing::trace!(
721                        error = error.as_ref() as &dyn std::error::Error,
722                        "AF_VSOCK connect failed, guest not ready yet"
723                    );
724                }
725            }
726
727            pal_async::timer::PolledTimer::new(&self.resources.driver)
728                .sleep(Duration::from_secs(1))
729                .await;
730        };
731        tracing::info!("AF_VSOCK connected, handshaking with pipette");
732        let client = PipetteClient::new(&self.resources.driver, socket, &self.resources.output_dir)
733            .await
734            .context("pipette AF_VSOCK handshake failed")?;
735        tracing::info!("completed pipette AF_VSOCK handshake");
736        Ok(client)
737    }
738
739    /// Connect to pipette via TCP through consomme port forwarding.
740    ///
741    /// The guest pipette agent listens on `0.0.0.0:{port}` and consomme
742    /// forwards connections from `localhost:{port}` on the host into the
743    /// guest. We retry until the guest's network stack and pipette are up.
744    async fn wait_for_agent_tcp(&mut self, port: u16) -> anyhow::Result<PipetteClient> {
745        tracing::info!(port, "connecting to pipette via TCP");
746        let addr = std::net::SocketAddr::from((std::net::Ipv4Addr::LOCALHOST, port));
747        let client = loop {
748            match PolledSocket::connect_tcp(&self.resources.driver, addr).await {
749                Ok(socket) => {
750                    socket
751                        .get()
752                        .set_nodelay(true)
753                        .context("failed to set TCP_NODELAY")?;
754                    tracing::info!("TCP connected, handshaking with pipette");
755                    match PipetteClient::new(
756                        &self.resources.driver,
757                        socket,
758                        &self.resources.output_dir,
759                    )
760                    .await
761                    {
762                        Ok(client) => break client,
763                        Err(e) => {
764                            tracing::warn!(
765                                error = e.as_ref() as &dyn std::error::Error,
766                                "pipette TCP connection failed, retrying"
767                            );
768                        }
769                    }
770                }
771                Err(e) => {
772                    tracing::trace!(
773                        error = &e as &dyn std::error::Error,
774                        "TCP connect failed, guest not ready yet"
775                    );
776                }
777            }
778            // Wait before retrying — guest network stack may not be up yet.
779            pal_async::timer::PolledTimer::new(&self.resources.driver)
780                .sleep(Duration::from_secs(1))
781                .await;
782        };
783        tracing::info!("completed pipette TCP handshake");
784        Ok(client)
785    }
786
787    async fn pause(&self) -> anyhow::Result<()> {
788        self.worker.pause().await?;
789        Ok(())
790    }
791
792    async fn save_state(&self) -> anyhow::Result<Vec<u8>> {
793        let state_msg = self.worker.save().await?;
794        Ok(mesh::payload::encode(state_msg))
795    }
796
797    async fn resume(&self) -> anyhow::Result<()> {
798        self.worker.resume().await?;
799        Ok(())
800    }
801
802    async fn verify_save_restore(&self) -> anyhow::Result<()> {
803        for i in 0..2 {
804            let result = self.worker.pulse_save_restore().await;
805            match result {
806                Ok(()) => {}
807                Err(RpcError::Channel(err)) => return Err(err.into()),
808                Err(RpcError::Call(PulseSaveRestoreError::ResetNotSupported)) => {
809                    tracing::warn!("Reset not supported, could not test save + restore.");
810                    break;
811                }
812                Err(RpcError::Call(PulseSaveRestoreError::Other(err))) => {
813                    return Err(anyhow::Error::from(err))
814                        .context(format!("Save + restore {i} failed."));
815                }
816            }
817        }
818
819        Ok(())
820    }
821
822    async fn launch_linux_direct_pipette(&mut self) -> anyhow::Result<()> {
823        // Start pipette through serial on linux direct.
824        self.resources
825            .linux_direct_serial_agent
826            .as_mut()
827            .unwrap()
828            .run_command("mkdir /cidata && mount LABEL=cidata /cidata && sh -c '/cidata/pipette &'")
829            .await?;
830        Ok(())
831    }
832}
833
834/// Interface for inspecting OpenVMM
835pub struct OpenVmmInspector {
836    worker: Arc<Worker>,
837}
838
839#[async_trait]
840impl PetriVmInspector for OpenVmmInspector {
841    async fn inspect(&self, path: &str) -> anyhow::Result<inspect::Node> {
842        Ok(self.worker.inspect(path).await)
843    }
844}
845
846/// Interface to the OpenVMM framebuffer
847pub struct OpenVmmFramebufferAccess {
848    view: View,
849}
850
851#[async_trait]
852impl PetriVmFramebufferAccess for OpenVmmFramebufferAccess {
853    async fn screenshot(
854        &mut self,
855        image: &mut Vec<u8>,
856    ) -> anyhow::Result<Option<VmScreenshotMeta>> {
857        // Our framebuffer uses 4 bytes per pixel, approximating an
858        // BGRA image, however it only actually contains BGR data.
859        // The fourth byte is effectively noise. We can set the 'alpha'
860        // value to 0xFF to make the image opaque.
861        const BYTES_PER_PIXEL: usize = 4;
862        let (width, height) = self.view.resolution();
863        let (widthsize, heightsize) = (width as usize, height as usize);
864        let len = widthsize * heightsize * BYTES_PER_PIXEL;
865
866        image.resize(len, 0);
867        for (i, line) in (0..height).zip(image.chunks_exact_mut(widthsize * BYTES_PER_PIXEL)) {
868            self.view.read_line(i, line);
869            for pixel in line.chunks_exact_mut(BYTES_PER_PIXEL) {
870                pixel.swap(0, 2);
871                pixel[3] = 0xFF;
872            }
873        }
874
875        Ok(Some(VmScreenshotMeta {
876            color: image::ExtendedColorType::Rgba8,
877            width,
878            height,
879        }))
880    }
881}
petri/vm/openvmm/runtime.rs

petri/vm/openvmm/
runtime.rs
