IPv6 Multicast over UDP to share addresses

sled-agent/Cargo.toml

@@ -23,6 +23,7 @@ serde = { version = "1.0", features = [ "derive" ] }
 serde_json = "1.0"
 slog = { version = "2.5", features = [ "max_level_trace", "release_max_level_debug" ] }
 smf = "0.2"
+socket2 = { version = "0.4", features = [ "all" ] }
 structopt = "0.3"
 tar = "0.4"
 tempfile = "3.2"

sled-agent/src/bootstrap/agent.rs

@@ -6,19 +6,25 @@
 
 use super::client::types as bootstrap_types;
 use super::client::Client as BootstrapClient;
+use super::discovery;
 use super::views::ShareResponse;
-use omicron_common::api::external::Error;
+use omicron_common::api::external::Error as ExternalError;
+use omicron_common::backoff::{
+    internal_service_policy, retry_notify, BackoffError,
+};
 use omicron_common::packaging::sha256_digest;
 
 use slog::Logger;
 use std::collections::HashMap;
 use std::fs::File;
 use std::io::{Seek, SeekFrom};
-use std::net::SocketAddr;
 use std::path::Path;
 use tar::Archive;
 use thiserror::Error;
 
+const UNLOCK_THRESHOLD: usize = 2;
+const BOOTSTRAP_PORT: u16 = 12346;
+
 /// Describes errors which may occur while operating the bootstrap service.
 #[derive(Error, Debug)]
 pub enum BootstrapError {
@@ -39,24 +45,35 @@ pub enum BootstrapError {
 
     #[error("Error making HTTP request")]
     Api(#[from] anyhow::Error),
+
+    #[error("Not enough peers to unlock storage")]
+    NotEnoughPeers,
+}
+
+impl From<BootstrapError> for ExternalError {
+    fn from(err: BootstrapError) -> Self {
+        Self::internal_error(&err.to_string())
+    }
 }
 
 /// The entity responsible for bootstrapping an Oxide rack.
 pub(crate) struct Agent {
     /// Debug log
     log: Logger,
+    peer_monitor: discovery::PeerMonitor,
 }
 
 impl Agent {
-    pub fn new(log: Logger) -> Self {
-        Agent { log }
+    pub fn new(log: Logger) -> Result<Self, BootstrapError> {
+        let peer_monitor = discovery::PeerMonitor::new(&log)?;
+        Ok(Agent { log, peer_monitor })
     }
 
     /// Implements the "request share" API.
     pub async fn request_share(
         &self,
         identity: Vec<u8>,
-    ) -> Result<ShareResponse, Error> {
+    ) -> Result<ShareResponse, BootstrapError> {
         // TODO-correctness: Validate identity, return whatever
         // information is necessary to establish trust quorum.
         //
@@ -66,42 +83,92 @@ impl Agent {
         Ok(ShareResponse { shared_secret: vec![] })
     }
 
-    /// Performs device initialization:
+    /// Communicates with peers, sharing secrets, until the rack has been
+    /// sufficiently unlocked.
     ///
-    /// - TODO: Communicates with other sled agents to establish a trust quorum.
-    /// - Verifies, unpacks, and launches other services.
-    pub async fn initialize(
-        &self,
-        other_agents: Vec<SocketAddr>,
-    ) -> Result<(), BootstrapError> {
-        info!(&self.log, "bootstrap service initializing");
-        // TODO-correctness:
-        // - Establish trust quorum.
-        // - Once this is done, "unlock" local storage
-        //
-        // The current implementation sends a stub request to all known
-        // sled agents, but does not actually create a quorum / unlock
-        // anything.
-        let other_agents: Vec<BootstrapClient> = other_agents
-            .into_iter()
-            .map(|addr| {
-                let addr_str = addr.to_string();
-                BootstrapClient::new(
-                    &format!("http://{}", addr_str,),
-                    self.log.new(o!(
-                        "Address" => addr_str,
-                    )),
+    /// - This method retries until [`UNLOCK_THRESHOLD`] other agents are
+    /// online, and have successfully responded to "share requests".
+    async fn establish_sled_quorum(&self) -> Result<(), BootstrapError> {
+        retry_notify(
+            internal_service_policy(),
+            || async {
+                let other_agents = self.peer_monitor.addrs().await;
+                info!(&self.log, "Bootstrap: Communicating with peers: {:?}", other_agents);
+
+                // "-1" to account for ourselves.
+                //
+                // NOTE: Clippy error exists while the compile-time unlock
+                // threshold is "1", because we basically don't require any
+                // peers to unlock.
+                #[allow(clippy::absurd_extreme_comparisons)]
+                if other_agents.len() < UNLOCK_THRESHOLD - 1 {
+                    warn!(&self.log, "Not enough peers to start establishing quorum");
+                    return Err(BackoffError::Transient(
+                        BootstrapError::NotEnoughPeers,
+                    ));
+                }
+                info!(&self.log, "Bootstrap: Enough peers to start share transfer");
+
+                // TODO-correctness:
+                // - Establish trust quorum.
+                // - Once this is done, "unlock" local storage
+                //
+                // The current implementation sends a stub request to all known sled
+                // agents, but does not actually create a quorum / unlock anything.
+                let other_agents: Vec<BootstrapClient> = other_agents
+                    .into_iter()
+                    .map(|mut addr| {
+                        addr.set_port(BOOTSTRAP_PORT);
+                        // TODO-correctness:
+                        //
+                        // Many rust crates - such as "URL" - really dislike
+                        // using scopes in IPv6 addresses. Using
+                        // "addr.to_string()" results in an IP address format
+                        // that is rejected when embedded into a URL.
+                        //
+                        // Instead, we merely use IP and port for the moment,
+                        // which loses the scope information. Longer-term, if we
+                        // use ULAs (Unique Local Addresses) the scope shouldn't
+                        // be a factor anyway.
+                        let addr_str = format!("[{}]:{}", addr.ip(), addr.port());
+                        info!(&self.log, "bootstrap: Connecting to {}", addr_str);
+                        BootstrapClient::new(
+                            &format!("http://{}", addr_str),
+                            self.log.new(o!(
+                                "Address" => addr_str,
+                            )),
+                        )
+                    })
+                    .collect();
+                for agent in &other_agents {
+                    agent
+                        .api_request_share(&bootstrap_types::ShareRequest {
+                            identity: vec![],
+                        })
+                        .await
+                        .map_err(|e| {
+                            info!(&self.log, "Bootstrap: Failed to share request with peer: {:?}", e);
+                            BackoffError::Transient(BootstrapError::Api(e))
+                        })?;
+                        info!(&self.log, "Bootstrap: Shared request with peer");
+                }
+                Ok(())
+            },
+            |error, duration| {
+                warn!(
+                    self.log,
+                    "Failed to unlock sleds (will retry after {:?}: {:#}",
+                    duration,
+                    error,
                 )
-            })
-            .collect();
-        for agent in &other_agents {
-            agent
-                .api_request_share(&bootstrap_types::ShareRequest {
-                    identity: vec![],
-                })
-                .await?;
-        }
+            },
+        )
+        .await?;
+
+        Ok(())
+    }
 
+    async fn launch_local_services(&self) -> Result<(), BootstrapError> {
         let tar_source = Path::new("/opt/oxide");
         let destination = Path::new("/opt/oxide");
         // TODO-correctness: Validation should come from ROT, not local file.
@@ -129,6 +196,19 @@ impl Agent {
         Ok(())
     }
 
+    /// Performs device initialization:
+    ///
+    /// - TODO: Communicates with other sled agents to establish a trust quorum.
+    /// - Verifies, unpacks, and launches other services.
+    pub async fn initialize(&self) -> Result<(), BootstrapError> {
+        info!(&self.log, "bootstrap service initializing");
+
+        self.establish_sled_quorum().await?;
+        self.launch_local_services().await?;
+
+        Ok(())
+    }
+
     fn launch<S, P1, P2>(
         &self,
         digests: &HashMap<String, Vec<u8>>,

sled-agent/src/bootstrap/discovery.rs

@@ -0,0 +1,93 @@
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+//! Sled announcement and discovery.
+
+use super::multicast;
+use slog::Logger;
+use std::collections::HashSet;
+use std::io;
+use std::net::{Ipv6Addr, SocketAddr, SocketAddrV6};
+use std::sync::Arc;
+use tokio::net::UdpSocket;
+use tokio::sync::Mutex;
+use tokio::task::JoinHandle;
+
+/// Manages Sled Discovery - both our announcement to other Sleds,
+/// as well as our discovery of those sleds.
+pub struct PeerMonitor {
+    sleds: Arc<Mutex<HashSet<SocketAddr>>>,
+    _worker: JoinHandle<()>,
+}
+
+async fn monitor_worker(
+    log: Logger,
+    address: SocketAddrV6,
+    sender: UdpSocket,
+    listener: UdpSocket,
+    sleds: Arc<Mutex<HashSet<SocketAddr>>>,
+) {
+    // Let this message be a reminder that this content is *not*
+    // encrypted, authenticated, or otherwise verified. We're just using
+    // it as a starting point for swapping addresses.
+    let message =
+        b"We've been trying to reach you about your car's extended warranty";
+    loop {
+        let mut buf = vec![0u8; 128];
+        tokio::select! {
+            _ = tokio::time::sleep(tokio::time::Duration::from_millis(5000)) => {
+                info!(log, "Bootstrap Peer Monitor: Broadcasting our own address: {}", address);
+                if let Err(e) = sender.try_send_to(message, address.into()) {
+                    warn!(log, "PeerMonitor failed to broadcast: {}", e);
+                }
+            }
+            result = listener.recv_from(&mut buf) => {
+                match result {
+                    Ok((_, addr)) => {
+                        info!(log, "Bootstrap Peer Monitor: Successfully received an address: {}", addr);
+                        sleds.lock().await.insert(addr);
+                    },
+                    Err(e) => warn!(log, "PeerMonitor failed to receive: {}", e),
+                }
+            }
+        }
+    }
+}
+
+impl PeerMonitor {
+    /// Creates a new [`PeerMonitor`].
+    // TODO: Address, port, interface, etc, probably should be
+    // configuration options.
+    pub fn new(log: &Logger) -> Result<Self, io::Error> {
+        let scope = multicast::Ipv6MulticastScope::LinkLocal.first_hextet();
+        let address = SocketAddrV6::new(
+            Ipv6Addr::new(scope, 0, 0, 0, 0, 0, 0, 0x1),
+            7645,
+            0,
+            0,
+        );
+        let loopback = false;
+        let interface = 0;
+        let (sender, listener) =
+            multicast::new_ipv6_udp_pair(&address, loopback, interface)?;
+
+        let sleds = Arc::new(Mutex::new(HashSet::new()));
+        let sleds_for_worker = sleds.clone();
+        let log = log.clone();
+
+        let worker = tokio::task::spawn(async move {
+            monitor_worker(log, address, sender, listener, sleds_for_worker)
+                .await
+        });
+
+        Ok(PeerMonitor { sleds, _worker: worker })
+    }
+
+    /// Returns the addresses of connected sleds.
+    ///
+    /// Note: These sleds have not yet been verified.
+    pub async fn addrs(&self) -> Vec<SocketAddr> {
+        self.sleds.lock().await.iter().map(|addr| *addr).collect()
+    }
+}

sled-agent/src/bootstrap/http_entrypoints.rs

@@ -30,6 +30,7 @@ use dropshot::HttpError;
 use dropshot::HttpResponseOk;
 use dropshot::RequestContext;
 use dropshot::TypedBody;
+use omicron_common::api::external::Error as ExternalError;
 use std::sync::Arc;
 
 use super::agent::Agent;
@@ -62,5 +63,10 @@ async fn api_request_share(
     let bootstrap_agent = rqctx.context();
 
     let request = request.into_inner();
-    Ok(HttpResponseOk(bootstrap_agent.request_share(request.identity).await?))
+    Ok(HttpResponseOk(
+        bootstrap_agent
+            .request_share(request.identity)
+            .await
+            .map_err(|e| ExternalError::from(e))?,
+    ))
 }

sled-agent/src/bootstrap/mod.rs

@@ -7,7 +7,9 @@
 pub mod agent;
 mod client;
 pub mod config;
+mod discovery;
 mod http_entrypoints;
+mod multicast;
 mod params;
 pub mod server;
 mod views;