server: import/export history

Author: gjcolombo

bin/propolis-server/src/lib/migrate/destination.rs

@@ -6,7 +6,8 @@ use bitvec::prelude as bv;
 use futures::{SinkExt, StreamExt};
 use propolis::common::{GuestAddr, Lifecycle, PAGE_SIZE};
 use propolis::migrate::{
-    MigrateCtx, MigrateStateError, Migrator, PayloadOffer, PayloadOffers,
+    MigrateCtx, MigrateSingle, MigrateStateError, Migrator, PayloadOffer,
+    PayloadOffers,
 };
 use propolis::vmm;
 use propolis_api_types::instance_spec::SpecKey;
@@ -22,10 +23,10 @@ use tokio_tungstenite::tungstenite::protocol::CloseFrame;
 use tokio_tungstenite::{tungstenite, WebSocketStream};
 use uuid::Uuid;
 
-use crate::migrate::codec;
 use crate::migrate::memx;
 use crate::migrate::preamble::Preamble;
 use crate::migrate::probes;
+use crate::migrate::{codec, DevicePayload};
 use crate::migrate::{
     Device, MigrateError, MigratePhase, MigrateRole, MigrationState, PageIter,
 };
@@ -511,6 +512,21 @@ impl<T: MigrateConn> RonV0<T> {
                 return Err(MigrateError::UnexpectedMessage);
             }
         };
+
+        let com1_payload: DevicePayload = match self.read_msg().await? {
+            codec::Message::Serialized(encoded) => {
+                ron::de::from_reader(encoded.as_bytes())
+                    .map_err(codec::ProtocolError::from)?
+            }
+            msg => {
+                error!(
+                    self.log(),
+                    "device_state: unexpected COM1 history message: {msg:?}"
+                );
+                return Err(MigrateError::UnexpectedMessage);
+            }
+        };
+
         self.read_ok().await?;
 
         info!(self.log(), "Devices: {devices:#?}");
@@ -529,6 +545,18 @@ impl<T: MigrateConn> RonV0<T> {
                     })?;
                 self.import_device(&target, &device, &migrate_ctx)?;
             }
+
+            let com1_data =
+                &mut ron::Deserializer::from_str(&com1_payload.data)
+                    .map_err(codec::ProtocolError::from)?;
+            let com1_offer = PayloadOffer {
+                kind: &com1_payload.kind,
+                version: com1_payload.version,
+                payload: Box::new(<dyn erased_serde::Deserializer>::erase(
+                    com1_data,
+                )),
+            };
+            vm_objects.com1().import(com1_offer, &migrate_ctx)?;
         }
 
         self.send_msg(codec::Message::Okay).await
@@ -762,24 +790,8 @@ impl<T: MigrateConn> RonV0<T> {
                 .map_err(codec::ProtocolError::from)?,
         ))
         .await?;
-        let com1_history = match self.read_msg().await? {
-            codec::Message::Serialized(encoded) => encoded,
-            msg => {
-                error!(self.log(), "server_state: unexpected message: {msg:?}");
-                return Err(MigrateError::UnexpectedMessage);
-            }
-        };
-
-        ensure_ctx
-            .vm_objects()
-            .lock_shared()
-            .await
-            .com1()
-            .import(&com1_history)
-            .await
-            .map_err(|e| MigrateError::Codec(e.to_string()))?;
 
-        self.send_msg(codec::Message::Okay).await
+        Ok(())
     }
 
     async fn finish(

bin/propolis-server/src/lib/migrate/source.rs

@@ -6,7 +6,7 @@ use bitvec::prelude::{BitSlice, Lsb0};
 use futures::{SinkExt, StreamExt};
 use propolis::common::{GuestAddr, GuestData, PAGE_SIZE};
 use propolis::migrate::{
-    MigrateCtx, MigrateStateError, Migrator, PayloadOutputs,
+    MigrateCtx, MigrateSingle, MigrateStateError, Migrator, PayloadOutputs,
 };
 use propolis::vmm;
 use propolis_api_types::instance_spec::VersionedInstanceSpec;
@@ -685,6 +685,7 @@ impl<T: MigrateConn> RonV0Runner<'_, T> {
     async fn device_state(&mut self) -> Result<(), MigrateError> {
         self.update_state(MigrationState::Device);
         let mut device_states = vec![];
+        let com1_payload;
         {
             let objects = self.vm.lock_shared().await;
             let migrate_ctx =
@@ -733,6 +734,14 @@ impl<T: MigrateConn> RonV0Runner<'_, T> {
                 }
                 Ok(())
             })?;
+
+            let com1_state = objects.com1().export(&migrate_ctx)?;
+            com1_payload = DevicePayload {
+                kind: com1_state.kind.to_owned(),
+                version: com1_state.version,
+                data: ron::ser::to_string(&com1_state.payload)
+                    .map_err(codec::ProtocolError::from)?,
+            };
         }
 
         info!(self.log(), "Device States: {device_states:#?}");
@@ -743,7 +752,14 @@ impl<T: MigrateConn> RonV0Runner<'_, T> {
         ))
         .await?;
 
+        self.send_msg(codec::Message::Serialized(
+            ron::ser::to_string(&com1_payload)
+                .map_err(codec::ProtocolError::from)?,
+        ))
+        .await?;
+
         self.send_msg(codec::Message::Okay).await?;
+
         self.read_ok().await
     }
 
@@ -787,15 +803,15 @@ impl<T: MigrateConn> RonV0Runner<'_, T> {
             }
             _ => return Err(MigrateError::UnexpectedMessage),
         };
-        let com1_history = self
-            .vm
-            .lock_shared()
-            .await
-            .com1()
-            .export_history(remote_addr)
-            .await?;
-        self.send_msg(codec::Message::Serialized(com1_history)).await?;
-        self.read_ok().await
+
+        {
+            let mgr = self.vm_services.serial_mgr.lock().await;
+            if let Some(mgr) = mgr.as_ref() {
+                mgr.notify_migration(remote_addr).await;
+            }
+        }
+
+        Ok(())
     }
 
     async fn finish(&mut self) -> Result<(), MigrateError> {

bin/propolis-server/src/lib/serial/backend.rs

@@ -25,9 +25,6 @@ use super::history_buffer::{HistoryBuffer, SerialHistoryOffset};
 #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
 struct ClientId(u64);
 
-trait ConsoleDevice: Source + Sink {}
-impl<T: Source + Sink> ConsoleDevice for T {}
-
 /// A client's rights when accessing a character backend.
 #[derive(Clone, Copy)]
 pub(super) enum Permissions {
@@ -188,7 +185,6 @@ impl Inner {
 /// access a single serial device.
 pub struct ConsoleBackend {
     inner: Arc<Mutex<Inner>>,
-    dev: Arc<dyn ConsoleDevice>,
     sink: Arc<dyn Sink>,
     sink_buffer: Arc<SinkBuffer>,
     done_tx: oneshot::Sender<()>,
@@ -211,7 +207,6 @@ impl ConsoleBackend {
 
         let this = Arc::new(Self {
             inner: Arc::new(Mutex::new(Inner::new(history_bytes))),
-            dev: device,
             sink,
             sink_buffer,
             done_tx,
@@ -268,6 +263,36 @@ impl Drop for ConsoleBackend {
     }
 }
 
+mod migrate {
+    use propolis::migrate::{
+        MigrateCtx, MigrateSingle, MigrateStateError, PayloadOffer,
+    };
+
+    use crate::serial::history_buffer::migrate::HistoryBufferContentsV1;
+
+    use super::ConsoleBackend;
+
+    impl MigrateSingle for ConsoleBackend {
+        fn export(
+            &self,
+            _ctx: &MigrateCtx,
+        ) -> Result<propolis::migrate::PayloadOutput, MigrateStateError>
+        {
+            Ok(self.inner.lock().unwrap().buffer.export().into())
+        }
+
+        fn import(
+            &self,
+            mut offer: PayloadOffer,
+            _ctx: &MigrateCtx,
+        ) -> Result<(), MigrateStateError> {
+            let contents: HistoryBufferContentsV1 = offer.parse()?;
+            self.inner.lock().unwrap().buffer.import(contents);
+            Ok(())
+        }
+    }
+}
+
 async fn read_task(
     inner: Arc<Mutex<Inner>>,
     source: Arc<dyn Source>,

bin/propolis-server/src/lib/serial/history_buffer.rs

@@ -7,7 +7,6 @@
 
 use dropshot::HttpError;
 use propolis_api_types as api;
-use serde::{Deserialize, Serialize};
 use std::collections::VecDeque;
 use std::convert::TryFrom;
 
@@ -29,7 +28,7 @@ const DEFAULT_MAX_LENGTH: isize = 16 * 1024;
 /// An abstraction for storing the contents of the instance's serial console
 /// output, intended for retrieval by the web console or other monitoring or
 /// troubleshooting tools.
-#[derive(Deserialize, Serialize, Clone)]
+#[derive(Clone)]
 pub(crate) struct HistoryBuffer {
     beginning: Vec<u8>,
     rolling: VecDeque<u8>,
@@ -223,6 +222,48 @@ impl HistoryBuffer {
     pub fn bytes_from_start(&self) -> usize {
         self.total_bytes
     }
+
+    pub(super) fn export(&self) -> migrate::HistoryBufferContentsV1 {
+        let slices = self.rolling.as_slices();
+        let mut most_recent = Vec::with_capacity(self.rolling.len());
+        most_recent.extend(slices.0);
+        most_recent.extend(slices.1);
+        migrate::HistoryBufferContentsV1 {
+            beginning: self.beginning.clone(),
+            most_recent,
+            total_bytes: self.total_bytes,
+            buffer_size: self.buffer_size,
+        }
+    }
+
+    pub(super) fn import(
+        &mut self,
+        contents: migrate::HistoryBufferContentsV1,
+    ) {
+        self.beginning = contents.beginning;
+        self.rolling.clear();
+        self.rolling.extend(contents.most_recent.iter());
+        self.total_bytes = contents.total_bytes;
+        self.buffer_size = contents.buffer_size;
+    }
+}
+
+pub(crate) mod migrate {
+    use serde::{Deserialize, Serialize};
+
+    #[derive(Deserialize, Serialize, Clone)]
+    pub(crate) struct HistoryBufferContentsV1 {
+        pub(super) beginning: Vec<u8>,
+        pub(super) most_recent: Vec<u8>,
+        pub(super) total_bytes: usize,
+        pub(super) buffer_size: usize,
+    }
+
+    impl propolis::migrate::Schema<'_> for HistoryBufferContentsV1 {
+        fn id() -> propolis::migrate::SchemaId {
+            ("serial-console-backend", 1)
+        }
+    }
 }
 
 #[cfg(test)]

bin/propolis-server/src/lib/serial/mod.rs

@@ -145,48 +145,52 @@ impl SerialConsoleManager {
             ),
         };
 
-        let mut client_tasks = self.client_tasks.lock().unwrap();
-        let client_id = client_tasks.next_id();
-        let prev_rw_task = if kind == ClientKind::ReadWrite {
-            client_tasks.remove_rw_client()
-        } else {
-            None
-        };
+        let prev_rw_task;
+        {
+            let mut client_tasks = self.client_tasks.lock().unwrap();
+            let client_id = client_tasks.next_id();
+            prev_rw_task = if kind == ClientKind::ReadWrite {
+                client_tasks.remove_rw_client()
+            } else {
+                None
+            };
 
-        let backend_hdl =
-            self.backend.attach_client(console_tx, permissions, discipline);
-
-        let ctx = SerialTaskContext {
-            log: self.log.clone(),
-            ws,
-            backend_hdl,
-            console_rx,
-            control_rx,
-            start_rx: task_start_rx,
-            done_rx: task_done_rx,
-            client_tasks: self.client_tasks.clone(),
-            client_id,
-        };
+            let backend_hdl =
+                self.backend.attach_client(console_tx, permissions, discipline);
+
+            let ctx = SerialTaskContext {
+                log: self.log.clone(),
+                ws,
+                backend_hdl,
+                console_rx,
+                control_rx,
+                start_rx: task_start_rx,
+                done_rx: task_done_rx,
+                client_tasks: self.client_tasks.clone(),
+                client_id,
+            };
 
-        let task = ClientTask {
-            hdl: tokio::spawn(async move { serial_task(ctx).await }),
-            control_tx,
-            done_tx: task_done_tx,
-        };
+            let task = ClientTask {
+                hdl: tokio::spawn(async move { serial_task(ctx).await }),
+                control_tx,
+                done_tx: task_done_tx,
+            };
 
-        client_tasks.tasks.insert(client_id, task);
-        if kind == ClientKind::ReadWrite {
-            assert!(client_tasks.rw_client_id.is_none());
-            client_tasks.rw_client_id = Some(client_id);
+            client_tasks.tasks.insert(client_id, task);
+            if kind == ClientKind::ReadWrite {
+                assert!(client_tasks.rw_client_id.is_none());
+                client_tasks.rw_client_id = Some(client_id);
+            }
         }
 
-        drop(client_tasks);
         if let Some(task) = prev_rw_task {
-            task.done_tx.send(());
-            task.hdl.await;
+            let _ = task.done_tx.send(());
+            let _ = task.hdl.await;
         }
 
-        task_start_tx.send(());
+        task_start_tx
+            .send(())
+            .expect("new serial task shouldn't exit before starting");
     }
 
     pub async fn notify_migration(&self, destination: SocketAddr) {
@@ -468,5 +472,5 @@ async fn serial_task(
         close(&log, client_id, sink, stream, close_reason).await;
     }
 
-    client_tasks.lock().unwrap().tasks.remove(&client_id);
+    client_tasks.lock().unwrap().remove_by_id(client_id);
 }

bin/propolis-server/src/lib/server.rs

@@ -459,7 +459,8 @@ async fn instance_serial(
     serial_mgr
         .as_ref()
         .ok_or("Instance has no serial console manager")?
-        .connect(ws_stream, crate::serial::ClientKind::ReadWrite);
+        .connect(ws_stream, crate::serial::ClientKind::ReadWrite)
+        .await;
 
     Ok(())
 }