Add migration endpoints to propolis-server (#69)

This adds the endpoints described in [RFD 71](https://rfd.shared.oxide.computer/rfd/0071) along with a first pass for some of the setup.

**New Endpoints:**
1. `/instances/{src-uuid}/migrate/start` — This is the endpoint called by the new (destination) propolis-server instance when its ready to begin the migration process. This request requires no body as we'll perform an HTTP upgrade to reuse the underlying TCP socket as a bidirectional channel to perform the migration over.
2. `/instances/{src-uuid}/migrate/status` — This endpoint allows querying either the source or destination on the current progress of a migration.

**Modified Endpoints:**
1. `/instances/{dst-uuid}` — This is the existing "ensure" endpoint, which has been extended with an optional blob describing a source VM from which a new VM will be populated.

cli/src/main.rs

@@ -4,14 +4,15 @@ use std::path::{Path, PathBuf};
 use std::{
     net::{IpAddr, SocketAddr, ToSocketAddrs},
     os::unix::prelude::AsRawFd,
+    time::Duration,
 };
 
 use anyhow::{anyhow, Context};
-use futures::{SinkExt, StreamExt};
+use futures::{future, SinkExt, StreamExt};
 use propolis_client::{
     api::{
-        DiskRequest, InstanceEnsureRequest, InstanceProperties,
-        InstanceStateRequested,
+        DiskRequest, InstanceEnsureRequest, InstanceMigrateInitiateRequest,
+        InstanceProperties, InstanceStateRequested, MigrationState,
     },
     Client,
 };
@@ -50,6 +51,10 @@ enum Command {
         /// Instance name
         name: String,
 
+        /// Instance uuid (if specified)
+        #[structopt(short = "u")]
+        uuid: Option<Uuid>,
+
         /// Number of vCPUs allocated to instance
         #[structopt(short = "c", default_value = "4")]
         vcpus: u8,
@@ -84,6 +89,24 @@ enum Command {
         /// Instance name
         name: String,
     },
+
+    /// Migrate instance to new propolis-server
+    Migrate {
+        /// Instance name
+        name: String,
+
+        /// Destination propolis-server address
+        #[structopt(parse(try_from_str = resolve_host))]
+        dst_server: IpAddr,
+
+        /// Destination propolis-server port
+        #[structopt(short = "p", default_value = "12400")]
+        dst_port: u16,
+
+        /// Uuid for the destination instance
+        #[structopt(short = "u")]
+        dst_uuid: Option<Uuid>,
+    },
 }
 
 fn parse_state(state: &str) -> anyhow::Result<InstanceStateRequested> {
@@ -128,13 +151,11 @@ fn create_logger(opt: &Opt) -> Logger {
 async fn new_instance(
     client: &Client,
     name: String,
+    id: Uuid,
     vcpus: u8,
     memory: u64,
     disks: Vec<DiskRequest>,
 ) -> anyhow::Result<()> {
-    // Generate a UUID for the new instance
-    let id = Uuid::new_v4();
-
     let properties = InstanceProperties {
         id,
         name,
@@ -151,7 +172,8 @@ async fn new_instance(
         properties,
         // TODO: Allow specifying NICs
         nics: vec![],
-        disks: disks.to_vec(),
+        disks,
+        migrate: None,
     };
 
     // Try to create the instance
@@ -261,6 +283,72 @@ async fn serial(
     Ok(())
 }
 
+async fn migrate_instance(
+    src_client: Client,
+    dst_client: Client,
+    src_name: String,
+    src_addr: SocketAddr,
+    dst_uuid: Uuid,
+) -> anyhow::Result<()> {
+    // Grab the src instance UUID
+    let src_uuid = src_client
+        .instance_get_uuid(&src_name)
+        .await
+        .with_context(|| anyhow!("failed to get src instance UUID"))?;
+
+    // Grab the instance details
+    let src_instance = src_client
+        .instance_get(src_uuid)
+        .await
+        .with_context(|| anyhow!("failed to get src instance properties"))?;
+
+    let request = InstanceEnsureRequest {
+        properties: InstanceProperties {
+            // Use a new ID for the destination instance we're creating
+            id: dst_uuid,
+            ..src_instance.instance.properties
+        },
+        // TODO: Handle migrating NICs & disks
+        nics: vec![],
+        disks: vec![],
+        migrate: Some(InstanceMigrateInitiateRequest { src_addr, src_uuid }),
+    };
+
+    // Initiate the migration via the destination instance
+    let migration_id = dst_client
+        .instance_ensure(&request)
+        .await?
+        .migrate
+        .ok_or_else(|| anyhow!("no migrate id on response"))?
+        .migration_id;
+
+    // Wait for the migration to complete by polling both source and destination
+    // TODO: replace with into_iter method call after edition upgrade
+    let handles = IntoIterator::into_iter([
+        ("src", src_client, src_uuid),
+        ("dst", dst_client, dst_uuid),
+    ])
+    .map(|(role, client, id)| {
+        tokio::spawn(async move {
+            loop {
+                let state = client
+                    .instance_migrate_status(id, migration_id)
+                    .await?
+                    .state;
+                println!("{}({}) migration state={:?}", role, id, state);
+                if state == MigrationState::Finish {
+                    return Ok::<_, anyhow::Error>(());
+                }
+                tokio::time::sleep(Duration::from_secs(1)).await;
+            }
+        })
+    });
+
+    future::join_all(handles).await;
+
+    Ok(())
+}
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     let opt = Opt::from_args();
@@ -270,20 +358,33 @@ async fn main() -> anyhow::Result<()> {
     let client = Client::new(addr, log.new(o!()));
 
     match opt.cmd {
-        Command::New { name, vcpus, memory, crucible_disks } => {
+        Command::New { name, uuid, vcpus, memory, crucible_disks } => {
             let disks = if let Some(crucible_disks) = crucible_disks {
                 parse_crucible_disks(&crucible_disks)?
             } else {
                 vec![]
             };
-            new_instance(&client, name.to_string(), vcpus, memory, disks)
-                .await?
+            new_instance(
+                &client,
+                name.to_string(),
+                uuid.unwrap_or_else(Uuid::new_v4),
+                vcpus,
+                memory,
+                disks,
+            )
+            .await?
         }
         Command::Get { name } => get_instance(&client, name).await?,
         Command::State { name, state } => {
             put_instance(&client, name, state).await?
         }
         Command::Serial { name } => serial(&client, addr, name).await?,
+        Command::Migrate { name, dst_server, dst_port, dst_uuid } => {
+            let dst_addr = SocketAddr::new(dst_server, dst_port);
+            let dst_client = Client::new(dst_addr, log.clone());
+            let dst_uuid = dst_uuid.unwrap_or_else(Uuid::new_v4);
+            migrate_instance(client, dst_client, name, addr, dst_uuid).await?
+        }
     }
 
     Ok(())

client/src/api.rs

@@ -28,10 +28,63 @@ pub struct InstanceEnsureRequest {
 
     #[serde(default)]
     pub disks: Vec<DiskRequest>,
+
+    pub migrate: Option<InstanceMigrateInitiateRequest>,
 }
 
 #[derive(Clone, Deserialize, Serialize, JsonSchema)]
-pub struct InstanceEnsureResponse {}
+pub struct InstanceEnsureResponse {
+    pub migrate: Option<InstanceMigrateInitiateResponse>,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema)]
+pub struct InstanceMigrateInitiateRequest {
+    pub src_addr: SocketAddr,
+    pub src_uuid: Uuid,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema)]
+pub struct InstanceMigrateInitiateResponse {
+    pub migration_id: Uuid,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema)]
+pub struct InstanceMigrateStartRequest {
+    pub migration_id: Uuid,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema)]
+pub struct InstanceMigrateStatusRequest {
+    pub migration_id: Uuid,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema)]
+pub struct InstanceMigrateStatusResponse {
+    pub state: MigrationState,
+}
+
+#[derive(
+    Clone,
+    Copy,
+    Debug,
+    Deserialize,
+    PartialEq,
+    Eq,
+    PartialOrd,
+    Ord,
+    Serialize,
+    JsonSchema,
+)]
+pub enum MigrationState {
+    Sync,
+    Ram,
+    Pause,
+    RamDirty,
+    Device,
+    Arch,
+    Resume,
+    Finish,
+}
 
 #[derive(Clone, Deserialize, Serialize, JsonSchema)]
 pub struct InstanceGetResponse {

client/src/lib.rs

@@ -157,4 +157,21 @@ impl Client {
         let body = Body::from(serde_json::to_string(&state).unwrap());
         self.put_no_response(path, Some(body)).await
     }
+
+    /// Get the status of an ongoing migration
+    pub async fn instance_migrate_status(
+        &self,
+        id: Uuid,
+        migration_id: Uuid,
+    ) -> Result<api::InstanceMigrateStatusResponse, Error> {
+        let path =
+            format!("http://{}/instances/{}/migrate/status", self.address, id);
+        let body = Body::from(
+            serde_json::to_string(&api::InstanceMigrateStatusRequest {
+                migration_id,
+            })
+            .unwrap(),
+        );
+        self.get(path, Some(body)).await
+    }
 }

server/Cargo.toml

@@ -18,6 +18,7 @@ doc = false
 
 [dependencies]
 anyhow = "1.0"
+const_format = "0.2"
 # dropshot = "0.6"
 dropshot = { git = "https://github.com/oxidecomputer/dropshot", branch = "main" }
 futures = "0.3"
@@ -28,6 +29,7 @@ tokio-tungstenite = "0.14"
 toml = "0.5"
 serde = "1.0"
 serde_derive = "1.0"
+serde_json = "1.0"
 slog = "2.7"
 structopt = { version = "0.3", default-features = false }
 propolis = { path = "../propolis", features = ["crucible"], default-features = false }

server/src/lib/lib.rs

@@ -2,5 +2,6 @@
 
 pub mod config;
 mod initializer;
+mod migrate;
 mod serial;
 pub mod server;