Dropshot-based websocket endpoint & Add Progenitor-based client

bin/propolis-cli/Cargo.toml

@@ -11,7 +11,7 @@ anyhow = "1.0"
 clap = { version = "3.2", features = ["derive"] }
 futures = "0.3"
 libc = "0.2"
-propolis-client = { path = "../../lib/propolis-client" }
+propolis-client = { path = "../../lib/propolis-client", features = ["generated"] }
 slog = "2.7"
 slog-async = "2.7"
 slog-term = "2.8"

bin/propolis-cli/src/main.rs

@@ -10,7 +10,7 @@ use std::{
 use anyhow::{anyhow, Context};
 use clap::{Parser, Subcommand};
 use futures::{future, SinkExt, StreamExt};
-use propolis_client::{
+use propolis_client::handmade::{
     api::{
         DiskRequest, InstanceEnsureRequest, InstanceMigrateInitiateRequest,
         InstanceProperties, InstanceStateRequested, MigrationState,
@@ -19,7 +19,9 @@ use propolis_client::{
 };
 use slog::{o, Drain, Level, Logger};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio_tungstenite::tungstenite::protocol::Role;
 use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::WebSocketStream;
 use uuid::Uuid;
 
 #[derive(Debug, Parser)]
@@ -317,10 +319,14 @@ async fn test_stdin_to_websockets_task() {
 }
 
 async fn serial(addr: SocketAddr) -> anyhow::Result<()> {
-    let path = format!("ws://{}/instance/serial", addr);
-    let (mut ws, _) = tokio_tungstenite::connect_async(path)
+    let upgraded = propolis_client::Client::new(&format!("http://{}", addr))
+        .instance_serial()
+        .send()
         .await
-        .with_context(|| anyhow!("failed to create serial websocket stream"))?;
+        .map_err(|e| anyhow!("Failed to upgrade connection: {}", e))?
+        .into_inner();
+    let mut ws =
+        WebSocketStream::from_raw_socket(upgraded, Role::Client, None).await;
 
     let _raw_guard = RawTermiosGuard::stdio_guard()
         .with_context(|| anyhow!("failed to set raw mode"))?;

bin/propolis-server/Cargo.toml

@@ -45,7 +45,7 @@ serde_derive = "1.0"
 serde_json = "1.0"
 slog = "2.7"
 propolis = { path = "../../lib/propolis", features = ["crucible-full", "oximeter"], default-features = false }
-propolis-client = { path = "../../lib/propolis-client" }
+propolis-client = { path = "../../lib/propolis-client", features = ["generated"] }
 propolis-server-config = { path = "../../crates/propolis-server-config" }
 rfb = { git = "https://github.com/oxidecomputer/rfb" }
 uuid = "1.0.0"
@@ -57,9 +57,10 @@ features = [ "chrono", "uuid1" ]
 
 [dev-dependencies]
 hex = "0.4.3"
-reqwest = { version = "0.11", default-features = false, features = ["rustls-tls"] }
+reqwest = { version = "0.11.12", default-features = false, features = ["rustls-tls"] }
 ring = "0.16"
 slog = { version = "2.5", features = [ "max_level_trace", "release_max_level_debug" ] }
+expectorate = "1.0.5"
 
 [features]
 default = ["dtrace-probes"]

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

@@ -4,7 +4,7 @@ use bit_field::BitField;
 use dropshot::{HttpError, RequestContext};
 use hyper::{header, Body, Method, Response, StatusCode};
 use propolis::migrate::MigrateStateError;
-use propolis_client::api::{self, MigrationState};
+use propolis_client::handmade::api::{self, MigrationState};
 use serde::{Deserialize, Serialize};
 use slog::{error, info, o};
 use thiserror::Error;

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

@@ -9,20 +9,21 @@ use std::sync::Arc;
 use std::{collections::BTreeMap, net::SocketAddr};
 
 use dropshot::{
-    endpoint, ApiDescription, HttpError, HttpResponseCreated, HttpResponseOk,
-    HttpResponseUpdatedNoContent, Path, RequestContext, TypedBody,
+    channel, endpoint, ApiDescription, HttpError, HttpResponseCreated,
+    HttpResponseOk, HttpResponseUpdatedNoContent, Path, RequestContext,
+    TypedBody, WebsocketConnection,
 };
-use hyper::StatusCode;
-use hyper::{http::header, upgrade, Body, Response};
+use hyper::{Body, Response};
 use oximeter::types::ProducerRegistry;
-use propolis_client::instance_spec;
-use propolis_client::{api, instance_spec::InstanceSpec};
+use propolis_client::{
+    handmade::api,
+    instance_spec::{self, InstanceSpec},
+};
 use propolis_server_config::Config as VmTomlConfig;
 use rfb::server::VncServer;
 use slog::{error, o, Logger};
 use thiserror::Error;
 use tokio::sync::{mpsc, oneshot, MappedMutexGuard, Mutex, MutexGuard};
-use tokio_tungstenite::tungstenite::handshake;
 use tokio_tungstenite::tungstenite::protocol::{Role, WebSocketConfig};
 use tokio_tungstenite::WebSocketStream;
 
@@ -112,9 +113,9 @@ impl VmControllerState {
     /// `VmControllerState::Destroyed`.
     pub fn take_controller(&mut self) -> Option<Arc<VmController>> {
         if let VmControllerState::Created(vm) = self {
-            let last_instance = propolis_client::api::Instance {
+            let last_instance = api::Instance {
                 properties: vm.properties().clone(),
-                state: propolis_client::api::InstanceState::Destroyed,
+                state: api::InstanceState::Destroyed,
                 disks: vec![],
                 nics: vec![],
             };
@@ -245,7 +246,7 @@ enum SpecCreationError {
 /// Creates an instance spec from an ensure request. (Both types are foreign to
 /// this crate, so implementing TryFrom for them is not allowed.)
 fn instance_spec_from_request(
-    request: &propolis_client::api::InstanceEnsureRequest,
+    request: &api::InstanceEnsureRequest,
     toml_config: &VmTomlConfig,
 ) -> Result<(InstanceSpec, BTreeMap<String, Vec<u8>>), SpecCreationError> {
     let mut in_memory_disk_contents: BTreeMap<String, Vec<u8>> =
@@ -314,11 +315,8 @@ async fn register_oximeter(
 }]
 async fn instance_ensure(
     rqctx: Arc<RequestContext<DropshotEndpointContext>>,
-    request: TypedBody<propolis_client::api::InstanceEnsureRequest>,
-) -> Result<
-    HttpResponseCreated<propolis_client::api::InstanceEnsureResponse>,
-    HttpError,
-> {
+    request: TypedBody<api::InstanceEnsureRequest>,
+) -> Result<HttpResponseCreated<api::InstanceEnsureResponse>, HttpError> {
     let server_context = rqctx.context();
     let request = request.into_inner();
 
@@ -478,9 +476,7 @@ async fn instance_ensure(
         None
     };
 
-    Ok(HttpResponseCreated(propolis_client::api::InstanceEnsureResponse {
-        migrate,
-    }))
+    Ok(HttpResponseCreated(api::InstanceEnsureResponse { migrate }))
 }
 
 #[endpoint {
@@ -489,8 +485,7 @@ async fn instance_ensure(
 }]
 async fn instance_get(
     rqctx: Arc<RequestContext<DropshotEndpointContext>>,
-) -> Result<HttpResponseOk<propolis_client::api::InstanceGetResponse>, HttpError>
-{
+) -> Result<HttpResponseOk<api::InstanceGetResponse>, HttpError> {
     let ctx = rqctx.context();
     let instance_info = match &*ctx.services.vm.lock().await {
         VmControllerState::NotCreated => {
@@ -499,7 +494,7 @@ async fn instance_get(
             ));
         }
         VmControllerState::Created(vm) => {
-            propolis_client::api::Instance {
+            api::Instance {
                 properties: vm.properties().clone(),
                 state: vm.external_instance_state(),
                 disks: vec![],
@@ -518,9 +513,7 @@ async fn instance_get(
         }
     };
 
-    Ok(HttpResponseOk(propolis_client::api::InstanceGetResponse {
-        instance: instance_info,
-    }))
+    Ok(HttpResponseOk(api::InstanceGetResponse { instance: instance_info }))
 }
 
 #[endpoint {
@@ -604,73 +597,21 @@ async fn instance_state_put(
     result
 }
 
-#[endpoint {
-    method = GET,
+#[channel {
+    protocol = WEBSOCKETS,
     path = "/instance/serial",
 }]
 async fn instance_serial(
     rqctx: Arc<RequestContext<DropshotEndpointContext>>,
-) -> Result<Response<Body>, HttpError> {
+    websock: WebsocketConnection,
+) -> dropshot::WebsocketChannelResult {
     let ctx = rqctx.context();
     let vm = ctx.vm().await?;
     let serial = vm.com1().clone();
-    let request = &mut *rqctx.request.lock().await;
-
-    if !request
-        .headers()
-        .get(header::CONNECTION)
-        .and_then(|hv| hv.to_str().ok())
-        .map(|hv| {
-            hv.split(|c| c == ',' || c == ' ')
-                .any(|vs| vs.eq_ignore_ascii_case("upgrade"))
-        })
-        .unwrap_or(false)
-    {
-        return Err(HttpError::for_bad_request(
-            None,
-            "expected connection upgrade".to_string(),
-        ));
-    }
-    if !request
-        .headers()
-        .get(header::UPGRADE)
-        .and_then(|v| v.to_str().ok())
-        .map(|v| {
-            v.split(|c| c == ',' || c == ' ')
-                .any(|v| v.eq_ignore_ascii_case("websocket"))
-        })
-        .unwrap_or(false)
-    {
-        return Err(HttpError::for_bad_request(
-            None,
-            "unexpected protocol for upgrade".to_string(),
-        ));
-    }
-    if request
-        .headers()
-        .get(header::SEC_WEBSOCKET_VERSION)
-        .map(|v| v.as_bytes())
-        != Some(b"13")
-    {
-        return Err(HttpError::for_bad_request(
-            None,
-            "missing or invalid websocket version".to_string(),
-        ));
-    }
-    let accept_key = request
-        .headers()
-        .get(header::SEC_WEBSOCKET_KEY)
-        .map(|hv| hv.as_bytes())
-        .map(handshake::derive_accept_key)
-        .ok_or_else(|| {
-            HttpError::for_bad_request(
-                None,
-                "missing websocket key".to_string(),
-            )
-        })?;
 
-    let ws_log = rqctx.log.new(o!());
-    let err_log = ws_log.clone();
+    let err_log = rqctx.log.new(o!());
+
+    // Create or get active serial task handle and channels
     let mut serial_task = ctx.services.serial_task.lock().await;
     let serial_task = serial_task.get_or_insert_with(move || {
         let (websocks_ch, websocks_recv) = mpsc::channel(1);
@@ -681,11 +622,11 @@ async fn instance_serial(
                 websocks_recv,
                 close_recv,
                 serial,
-                ws_log.clone(),
+                err_log.clone(),
             )
             .await
             {
-                error!(ws_log, "Failed to spawn instance serial task: {}", e);
+                error!(err_log, "Failed to spawn instance serial task: {}", e);
             }
         });
 
@@ -696,37 +637,21 @@ async fn instance_serial(
         }
     });
 
-    let upgrade_fut = upgrade::on(request);
     let config =
         WebSocketConfig { max_send_queue: Some(4096), ..Default::default() };
     let websocks_send = serial_task.websocks_ch.clone();
-    tokio::spawn(async move {
-        let upgraded = match upgrade_fut.await {
-            Ok(u) => u,
-            Err(e) => {
-                error!(err_log, "Serial socket upgrade failed: {}", e);
-                return;
-            }
-        };
 
-        let ws_stream = WebSocketStream::from_raw_socket(
-            upgraded,
-            Role::Server,
-            Some(config),
-        )
-        .await;
+    let ws_stream = WebSocketStream::from_raw_socket(
+        websock.into_inner(),
+        Role::Server,
+        Some(config),
+    )
+    .await;
 
-        if let Err(e) = websocks_send.send(ws_stream).await {
-            error!(err_log, "Serial socket hand-off failed: {}", e);
-        }
-    });
-
-    Ok(Response::builder()
-        .status(StatusCode::SWITCHING_PROTOCOLS)
-        .header(header::CONNECTION, "Upgrade")
-        .header(header::UPGRADE, "websocket")
-        .header(header::SEC_WEBSOCKET_ACCEPT, accept_key)
-        .body(Body::empty())?)
+    websocks_send
+        .send(ws_stream.into())
+        .await
+        .map_err(|e| format!("Serial socket hand-off failed: {}", e).into())
 }
 
 // This endpoint is meant to only be called during a migration from the destination
@@ -799,3 +724,25 @@ pub fn api() -> ApiDescription<DropshotEndpointContext> {
 
     api
 }
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn test_propolis_server_openapi() {
+        let mut buf: Vec<u8> = vec![];
+        super::api()
+            .openapi("Oxide Propolis Server API", "0.0.1")
+            .description(
+                "API for interacting with the Propolis hypervisor frontend.",
+            )
+            .contact_url("https://oxide.computer")
+            .contact_email("api@oxide.computer")
+            .write(&mut buf)
+            .unwrap();
+        let output = String::from_utf8(buf).unwrap();
+        expectorate::assert_contents(
+            "../../openapi/propolis-server.json",
+            &output,
+        );
+    }
+}

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

@@ -4,7 +4,7 @@ use std::collections::BTreeSet;
 use std::convert::TryInto;
 use std::str::FromStr;
 
-use propolis_client::api::{
+use propolis_client::handmade::api::{
     self, DiskRequest, InstanceProperties, NetworkInterfaceRequest,
 };
 use propolis_client::instance_spec::*;
@@ -567,7 +567,7 @@ impl SpecBuilder {
 mod test {
     use std::{collections::BTreeMap, path::PathBuf};
 
-    use propolis_client::api::Slot;
+    use propolis_client::handmade::api::Slot;
     use uuid::Uuid;
 
     use crate::config::{self, Config};

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

@@ -36,12 +36,13 @@ use propolis::{
     hw::{ps2::ctrl::PS2Ctrl, qemu::ramfb::RamFb, uart::LpcUart},
     Instance,
 };
-use propolis_client::{
+use propolis_client::handmade::{
     api::InstanceProperties, api::InstanceState as ApiInstanceState,
     api::InstanceStateMonitorResponse as ApiMonitoredState,
     api::InstanceStateRequested as ApiInstanceStateRequested,
-    api::MigrationState as ApiMigrationState, instance_spec::InstanceSpec,
+    api::MigrationState as ApiMigrationState,
 };
+use propolis_client::instance_spec::InstanceSpec;
 use slog::{error, info, Logger};
 use thiserror::Error;
 use tokio::task::JoinHandle as TaskJoinHandle;