petri: unblock commands and use com3 for hyper-v kmsg logs when possible (#1176)

Using COM3 instead of diag_client allows us to get OpenHCL logs if it
crashes early in boot. Currently this only works on our TDX and SNP
runners, since our nested runners have a build of windows that is too
old and additional com ports are not yet supported on ARM. To make this
work, the powershell and hvc commands needed to be run in a separate
process (using the `unblock` function) to avoid blocking the process
doing the logging. Also fixes guest serial logs that fail to reconnect after
os error 231 (ubuntu 22.04 tests).

Author: tjones60

Cargo.lock

@@ -5443,6 +5443,7 @@ version = "0.0.0"
 dependencies = [
  "anyhow",
  "async-trait",
+ "blocking",
  "chipset_resources",
  "clap",
  "diag_client",
@@ -5514,6 +5515,7 @@ dependencies = [
  "vmotherboard",
  "vmsocket",
  "vtl2_settings_proto",
+ "windows-version",
 ]
 
 [[package]]
@@ -9642,6 +9644,15 @@ dependencies = [
  "windows_x86_64_msvc 0.53.0",
 ]
 
+[[package]]
+name = "windows-version"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e04a5c6627e310a23ad2358483286c7df260c964eb2d003d8efd6d0f4e79265c"
+dependencies = [
+ "windows-link",
+]
+
 [[package]]
 name = "windows_aarch64_gnullvm"
 version = "0.48.5"

Cargo.toml

@@ -537,6 +537,7 @@ winapi = "0.3"
 windows = "0.59"
 windows-service = "0.8"
 windows-sys = "0.59"
+windows-version = "0.1.4"
 xshell = "=0.2.2" # pin to 0.2.2 to work around https://github.com/matklad/xshell/issues/63
 xshell-macros = "0.2"
 # We add the derive feature here since the vast majority of our crates use it.

flowey/flowey_lib_hvlite/src/install_vmm_tests_deps.rs

@@ -155,7 +155,10 @@ Otherwise, press `ctrl-c` to cancel the run.
                         // Select required reg keys
                         let mut reg_keys_to_set = BTreeSet::new();
                         if hyperv {
+                            // Allow loading IGVM from file (to run custom OpenHCL firmware)
                             reg_keys_to_set.insert("AllowFirmwareLoadFromFile");
+                            // Enable COM3 and COM4 for Hyper-V VMs so we can get the OpenHCL KMSG logs over serial
+                            reg_keys_to_set.insert("EnableAdditionalComPorts");
 
                             if hardware_isolation {
                                 reg_keys_to_set.insert("EnableHardwareIsolation");

petri/Cargo.toml

@@ -63,6 +63,7 @@ sparse_mmap.workspace = true
 
 anyhow.workspace = true
 async-trait.workspace = true
+blocking.workspace = true
 clap.workspace = true
 fatfs = { workspace = true, features = ["std", "alloc"] }
 fs-err.workspace = true
@@ -84,5 +85,8 @@ thiserror.workspace = true
 tracing.workspace = true
 tracing-subscriber.workspace = true
 
+[target.'cfg(windows)'.dependencies]
+windows-version.workspace = true
+
 [lints]
 workspace = true

petri/src/tracing.rs

@@ -1,7 +1,6 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
-use diag_client::kmsg_stream::KmsgStream;
 use fs_err::File;
 use fs_err::PathExt;
 use futures::AsyncBufReadExt;
@@ -305,24 +304,41 @@ impl<'a> MakeWriter<'a> for PetriWriter {
 }
 
 /// Logs lines from `reader` into `log_file`.
-pub async fn log_stream(
+///
+/// Attempts to parse lines as `SyslogParsedEntry`, extracting the log level.
+/// Passes through any non-conforming logs.
+pub async fn log_task(
     log_file: PetriLogFile,
     reader: impl AsyncRead + Unpin + Send + 'static,
+    name: &str,
 ) -> anyhow::Result<()> {
+    tracing::info!("connected to {name}");
     let mut buf = Vec::new();
     let mut reader = BufReader::new(reader);
     loop {
         buf.clear();
-        let n = (&mut reader).take(256).read_until(b'\n', &mut buf).await?;
-        if n == 0 {
-            break;
+        match (&mut reader).take(256).read_until(b'\n', &mut buf).await {
+            Ok(0) => {
+                tracing::info!("disconnected from {name}: EOF");
+                return Ok(());
+            }
+            Err(e) => {
+                tracing::info!("disconnected from {name}: error: {e:#}");
+                return Err(e.into());
+            }
+            _ => {}
         }
 
         let string_buf = String::from_utf8_lossy(&buf);
         let string_buf_trimmed = string_buf.trim_end();
-        log_file.write_entry(string_buf_trimmed);
+
+        if let Some(message) = kmsg::SyslogParsedEntry::new(string_buf_trimmed) {
+            let level = kernel_level_to_tracing_level(message.level);
+            log_file.write_entry_fmt(None, level, format_args!("{}", message.display(false)));
+        } else {
+            log_file.write_entry(string_buf_trimmed);
+        }
     }
-    Ok(())
 }
 
 /// Maps kernel log levels to tracing levels.
@@ -339,23 +355,30 @@ fn kernel_level_to_tracing_level(kernel_level: u8) -> Level {
 /// read from the kmsg stream and write entries to the log
 pub async fn kmsg_log_task(
     log_file: PetriLogFile,
-    mut file_stream: KmsgStream,
+    diag_client: diag_client::DiagClient,
 ) -> anyhow::Result<()> {
-    while let Some(data) = file_stream.next().await {
-        match data {
-            Ok(data) => {
-                let message = KmsgParsedEntry::new(&data).unwrap();
-                let level = kernel_level_to_tracing_level(message.level);
-                log_file.write_entry_fmt(None, level, format_args!("{}", message.display(false)));
-            }
-            Err(err) => {
-                tracing::info!("kmsg disconnected: {err:?}");
-                break;
+    loop {
+        diag_client.wait_for_server().await?;
+        let mut kmsg = diag_client.kmsg(true).await?;
+        tracing::info!("kmsg connected");
+        while let Some(data) = kmsg.next().await {
+            match data {
+                Ok(data) => {
+                    let message = KmsgParsedEntry::new(&data).unwrap();
+                    let level = kernel_level_to_tracing_level(message.level);
+                    log_file.write_entry_fmt(
+                        None,
+                        level,
+                        format_args!("{}", message.display(false)),
+                    );
+                }
+                Err(err) => {
+                    tracing::info!("kmsg disconnected: {err:#}");
+                    break;
+                }
             }
         }
     }
-
-    Ok(())
 }
 
 #[cfg(test)]

petri/src/vm/hyperv/hvc.rs

@@ -7,24 +7,34 @@ use super::vm::CommandError;
 use anyhow::Context;
 use guid::Guid;
 
-pub fn hvc_start(vmid: &Guid) -> Result<(), CommandError> {
-    hvc_output(|cmd| cmd.arg("start").arg(vmid.to_string())).map(|_| ())
+pub async fn hvc_start(vmid: &Guid) -> Result<(), CommandError> {
+    hvc_output(|cmd| cmd.arg("start").arg(vmid.to_string()))
+        .await
+        .map(|_| ())
 }
 
-pub fn hvc_stop(vmid: &Guid) -> Result<(), CommandError> {
-    hvc_output(|cmd| cmd.arg("stop").arg(vmid.to_string())).map(|_| ())
+pub async fn hvc_stop(vmid: &Guid) -> Result<(), CommandError> {
+    hvc_output(|cmd| cmd.arg("stop").arg(vmid.to_string()))
+        .await
+        .map(|_| ())
 }
 
-pub fn hvc_kill(vmid: &Guid) -> Result<(), CommandError> {
-    hvc_output(|cmd| cmd.arg("kill").arg(vmid.to_string())).map(|_| ())
+pub async fn hvc_kill(vmid: &Guid) -> Result<(), CommandError> {
+    hvc_output(|cmd| cmd.arg("kill").arg(vmid.to_string()))
+        .await
+        .map(|_| ())
 }
 
-pub fn hvc_restart(vmid: &Guid) -> Result<(), CommandError> {
-    hvc_output(|cmd| cmd.arg("restart").arg(vmid.to_string())).map(|_| ())
+pub async fn hvc_restart(vmid: &Guid) -> Result<(), CommandError> {
+    hvc_output(|cmd| cmd.arg("restart").arg(vmid.to_string()))
+        .await
+        .map(|_| ())
 }
 
-pub fn hvc_reset(vmid: &Guid) -> Result<(), CommandError> {
-    hvc_output(|cmd| cmd.arg("reset").arg(vmid.to_string())).map(|_| ())
+pub async fn hvc_reset(vmid: &Guid) -> Result<(), CommandError> {
+    hvc_output(|cmd| cmd.arg("reset").arg(vmid.to_string()))
+        .await
+        .map(|_| ())
 }
 
 /// HyperV VM state as reported by hvc
@@ -52,9 +62,10 @@ pub enum VmState {
     Resuming,
 }
 
-pub fn hvc_state(vmid: &Guid) -> anyhow::Result<VmState> {
+pub async fn hvc_state(vmid: &Guid) -> anyhow::Result<VmState> {
     Ok(
         match hvc_output(|cmd| cmd.arg("state").arg(vmid.to_string()))
+            .await
             .context("hvc_state")?
             .as_str()
         {
@@ -73,12 +84,12 @@ pub fn hvc_state(vmid: &Guid) -> anyhow::Result<VmState> {
     )
 }
 
-pub fn hvc_ensure_off(vmid: &Guid) -> anyhow::Result<()> {
+pub async fn hvc_ensure_off(vmid: &Guid) -> anyhow::Result<()> {
     for _ in 0..5 {
-        if matches!(hvc_state(vmid)?, VmState::Off) {
+        if matches!(hvc_state(vmid).await?, VmState::Off) {
             return Ok(());
         }
-        if let Err(e) = hvc_kill(vmid) {
+        if let Err(e) = hvc_kill(vmid).await {
             tracing::warn!("hvc_kill attempt failed: {e}")
         }
         std::thread::sleep(std::time::Duration::from_secs(1));
@@ -88,11 +99,11 @@ pub fn hvc_ensure_off(vmid: &Guid) -> anyhow::Result<()> {
 }
 
 /// Runs hvc with the given arguments and returns the output.
-fn hvc_output(
+async fn hvc_output(
     f: impl FnOnce(&mut std::process::Command) -> &mut std::process::Command,
 ) -> Result<String, CommandError> {
     let mut cmd = std::process::Command::new("hvc.exe");
     f(&mut cmd);
 
-    super::vm::run_cmd(cmd)
+    super::vm::run_cmd(cmd).await
 }