vmm: add migration-progress HTTP endpoint

This prepares the HTTP endpoint to export ongoing VM live-migration
progress.

This works because of the following fundamental prerequisites:
- internal API was made async
- http thread was made async

This way, one can send requests to fetch the latest state without
blocking anywhere.

Signed-off-by: Philipp Schuster <[email protected]>
On-behalf-of: SAP [email protected]

Author: phip1611

vm-migration/src/lib.rs

@@ -10,6 +10,7 @@ use thiserror::Error;
 use crate::protocol::MemoryRangeTable;
 
 mod bitpos_iterator;
+pub mod progress;
 pub mod protocol;
 pub mod tls;
 

vm-migration/src/progress.rs

@@ -0,0 +1,346 @@
+// Copyright © 2025 Cyberus Technology GmbH
+//
+// SPDX-License-Identifier: Apache-2.0
+
+//! Module for reporting of the live-migration progress.
+//!
+//! The main export is [`MigrationProgressAndStatus`].
+
+use std::error::Error;
+use std::num::NonZeroU32;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub enum TransportationMode {
+    Local,
+    Tcp { connections: NonZeroU32, tls: bool },
+}
+
+/// Carries information about the transmission of the VM's memory.
+#[derive(Clone, Copy, Debug, Default, serde::Serialize, serde::Deserialize)]
+pub struct MemoryTransmissionInfo {
+    /// The memory iteration (only in precopy mode).
+    pub memory_iteration: u64,
+    /// Memory bytes per second.
+    pub memory_transmission_bps: u64,
+    /// The total size of the VMs memory in bytes.
+    pub memory_bytes_total: u64,
+    /// The total size of transmitted bytes.
+    pub memory_bytes_transmitted: u64,
+    /// The amount of remaining bytes for this iteration.
+    pub memory_bytes_remaining_iteration: u64,
+    /// The amount of transmitted 4k pages.
+    pub memory_pages_4k_transmitted: u64,
+    /// The amount of remaining 4k pages for this iteration.
+    pub memory_pages_4k_remaining_iteration: u64,
+    /// The amount of zero pages for that we could take a shortcut
+    /// as all bytes have on fixed value (e.g., a zero page).
+    pub memory_pages_constant_count: u64,
+    /// Current memory dirty rate in pages per seconds.
+    pub memory_dirty_rate_pps: u64,
+}
+
+/// The different phases of an ongoing migration (good case).
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub enum MigrationPhase {
+    /// The migration starts. Handshake and transfer of VM config.
+    Starting,
+    /// Transfer of memory FDs.
+    ///
+    /// Only used for local migrations.
+    MemoryFds,
+    /// Transfer of VM memory in precopy mode.
+    ///
+    /// Not used for local migrations.
+    MemoryPrecopy,
+    /*/// Transfer of VM memory in postcopy mode.
+    ///
+    /// This follows after a precopy phase.
+    ///
+    /// Not used for local migrations.
+    MemoryPostcopy,*/
+    /// The VM migration is completing. This means the last chunks of memory
+    /// are transmitted as well as the final VM state (vCPUs, devices).
+    Completing,
+}
+
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub enum MigrationProgressState {
+    /// The migration has been cancelled.
+    Cancelled {
+        /// The latest memory transmission info, if any.
+        memory_transmission_info: MemoryTransmissionInfo,
+    },
+    /// The migration has failed.
+    Failed {
+        /// The last memory transmission info, if any.
+        memory_transmission_info: MemoryTransmissionInfo,
+        /// Stringified error.
+        error_msg: String,
+        /// Debug-stringified error.
+        error_msg_debug: String,
+        // TODO this is very tricky because I need clone()
+        // error: Box<dyn Error>,
+    },
+    /// The migration has finished successfully.
+    Finished {
+        /// The last memory transmission info, if any.
+        memory_transmission_info: MemoryTransmissionInfo,
+    },
+    /// The migration is ongoing.
+    Ongoing {
+        phase: MigrationPhase,
+        memory_transmission_info: MemoryTransmissionInfo,
+        /// Percent in range `0..=100`.
+        vcpu_throttle_percent: u8,
+    },
+}
+
+impl MigrationProgressState {
+    fn memory_transmission_info(&self) -> MemoryTransmissionInfo {
+        match self {
+            MigrationProgressState::Cancelled {
+                memory_transmission_info,
+                ..
+            } => *memory_transmission_info,
+            MigrationProgressState::Failed {
+                memory_transmission_info,
+                ..
+            } => *memory_transmission_info,
+            MigrationProgressState::Finished {
+                memory_transmission_info,
+                ..
+            } => *memory_transmission_info,
+            MigrationProgressState::Ongoing {
+                memory_transmission_info,
+                ..
+            } => *memory_transmission_info,
+        }
+    }
+
+    fn state_name(&self) -> &'static str {
+        match self {
+            MigrationProgressState::Cancelled { .. } => "cancelled",
+            MigrationProgressState::Failed { .. } => "failed",
+            MigrationProgressState::Finished { .. } => "finished",
+            MigrationProgressState::Ongoing { .. } => "ongoing",
+        }
+    }
+
+    fn cpu_throttle_percent(&self) -> Option<u8> {
+        match self {
+            MigrationProgressState::Ongoing {
+                vcpu_throttle_percent,
+                ..
+            } => Some(*vcpu_throttle_percent),
+            _ => None,
+        }
+    }
+}
+
+/// Returns the current UNIX timestamp in ms.
+fn current_unix_timestamp_ms() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap()
+        .as_millis() as u64
+}
+
+/// Type holding a current snapshot about the progress and status information
+/// of an ongoing live migration.
+///
+/// The states correspond to the [live-migration protocol]. This type was
+/// specifically crafted with easy yet clear semantics for API users in mind.
+///
+/// [live-migration protocol]: super::protocol
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub struct MigrationProgressAndStatus {
+    /// UNIX timestamp of the start of the live-migration process.
+    pub timestamp_begin: u64,
+    /// UNIX timestamp of the current snapshot.
+    pub timestamp_snapshot: u64,
+    /// Configured target downtime.
+    pub downtime_ms_target: u64,
+    /// The currently expected downtime.
+    pub downtime_ms_expected: Option<u64>,
+    /// The requested transportation mode.
+    pub transportation_mode: TransportationMode,
+    /// Snapshot of the current phase.
+    pub state: MigrationProgressState,
+}
+
+impl MigrationProgressAndStatus {
+    pub fn new(transportation_mode: TransportationMode, target_downtime: Duration) -> Self {
+        let timestamp = current_unix_timestamp_ms();
+        Self {
+            timestamp_begin: timestamp,
+            timestamp_snapshot: timestamp,
+            downtime_ms_target: target_downtime.as_millis() as u64,
+            downtime_ms_expected: None,
+            transportation_mode,
+            state: MigrationProgressState::Ongoing {
+                phase: MigrationPhase::Starting,
+                memory_transmission_info: MemoryTransmissionInfo::default(),
+                vcpu_throttle_percent: 42,
+            },
+        }
+    }
+
+    /// Updates the state of an ongoing migration.
+    pub fn update_ongoing_migration_state(
+        &mut self,
+        phase: MigrationPhase,
+        latest_memory_transmission_info: Option<MemoryTransmissionInfo>,
+        latest_cpu_throttle_percent: Option<u8>,
+    ) {
+        if !matches!(self.state, MigrationProgressState::Ongoing { .. }) {
+            panic!(
+                "illegal state transition: {} -> ongoing",
+                self.state.state_name()
+            );
+        }
+
+        if let Some(cpu_throttle_percent) = latest_cpu_throttle_percent {
+            assert!(cpu_throttle_percent <= 100);
+        }
+
+        self.state = MigrationProgressState::Ongoing {
+            phase,
+            memory_transmission_info: latest_memory_transmission_info
+                .unwrap_or_else(|| self.state.memory_transmission_info()),
+            vcpu_throttle_percent: latest_cpu_throttle_percent
+                .or_else(|| self.state.cpu_throttle_percent())
+                .unwrap_or(0),
+        };
+    }
+
+    /// Sets the underlying state to [`MigrationProgressState::Cancelled`] and
+    /// updates all corresponding metadata.
+    ///
+    /// After this state change, the object is supposed to be handled as immutable.
+    pub fn mark_as_cancelled(&mut self) {
+        if !matches!(self.state, MigrationProgressState::Ongoing { .. }) {
+            panic!(
+                "illegal state transition: {} -> cancelled",
+                self.state.state_name()
+            );
+        }
+        self.timestamp_snapshot = current_unix_timestamp_ms();
+        self.timestamp_snapshot = current_unix_timestamp_ms();
+        self.state = MigrationProgressState::Cancelled {
+            memory_transmission_info: self.state.memory_transmission_info(),
+        };
+    }
+
+    /// Sets the underlying state to [`MigrationProgressState::Failed`] and
+    /// updates all corresponding metadata.
+    ///
+    /// After this state change, the object is supposed to be handled as immutable.
+    pub fn mark_as_failed(&mut self, error: &dyn Error) {
+        if !matches!(self.state, MigrationProgressState::Ongoing { .. }) {
+            panic!(
+                "illegal state transition: {} -> failed",
+                self.state.state_name()
+            );
+        }
+        self.timestamp_snapshot = current_unix_timestamp_ms();
+        self.state = MigrationProgressState::Failed {
+            memory_transmission_info: self.state.memory_transmission_info(),
+            error_msg: format!("{error}",),
+            error_msg_debug: format!("{error:?}",),
+        };
+    }
+
+    /// Sets the underlying state to [`MigrationProgressState::Finished`] and
+    /// updates all corresponding metadata.
+    ///
+    /// After this state change, the object is supposed to be handled as immutable.
+    pub fn mark_as_finished(&mut self) {
+        if !matches!(self.state, MigrationProgressState::Ongoing { .. }) {
+            panic!(
+                "illegal state transition: {} -> finished",
+                self.state.state_name()
+            );
+        }
+        self.timestamp_snapshot = current_unix_timestamp_ms();
+        self.state = MigrationProgressState::Finished {
+            memory_transmission_info: self.state.memory_transmission_info(),
+        };
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use anyhow::anyhow;
+
+    use super::*;
+
+    // Helpful to see what the API will look like.
+    #[test]
+    fn print_json() {
+        let starting = MigrationProgressAndStatus::new(
+            TransportationMode::Tcp {
+                connections: NonZeroU32::new(1).unwrap(),
+                tls: false,
+            },
+            Duration::from_millis(100),
+        );
+        let memory_precopy = {
+            let mut state = starting.clone();
+            state.update_ongoing_migration_state(
+                MigrationPhase::MemoryPrecopy,
+                Some(MemoryTransmissionInfo {
+                    memory_iteration: 7,
+                    memory_transmission_bps: 0,
+                    memory_bytes_total: 0x1337,
+                    memory_bytes_transmitted: 0x1337,
+                    memory_pages_4k_transmitted: 42,
+                    memory_pages_4k_remaining_iteration: 42,
+                    memory_bytes_remaining_iteration: 124,
+                    memory_dirty_rate_pps: 42,
+                    memory_pages_constant_count: 0,
+                }),
+                Some(42),
+            );
+            state
+        };
+        let completing = {
+            let mut state = memory_precopy.clone();
+            state.update_ongoing_migration_state(MigrationPhase::Completing, None, Some(99));
+            state
+        };
+        let completed = {
+            let mut state = completing.clone();
+            state.mark_as_finished();
+            state
+        };
+        let failed = {
+            let mut state = completing.clone();
+            let error = anyhow!("Some very bad error".to_string());
+            let error: &dyn Error = error.as_ref();
+            state.mark_as_failed(error);
+            state
+        };
+        let cancelled = {
+            let mut state = completing.clone();
+            state.mark_as_cancelled();
+            state
+        };
+
+        let vals = [
+            starting,
+            memory_precopy,
+            completing,
+            completed,
+            failed,
+            cancelled,
+        ];
+        for val in vals {
+            println!(
+                "{:?}:\n{}\n\n",
+                val,
+                serde_json::to_string_pretty(&val).unwrap()
+            );
+        }
+    }
+}