RSS handoff, Nexus doing svc management, DNS integration

docs/how-to-run.adoc

@@ -147,11 +147,11 @@ When we deploy, we're effectively creating a number of different zones
 for all the components that make up Omicron (Nexus, Clickhouse, Crucible, etc).
 Since all these services run in different zones they cannot communicate with
 each other (and Sled Agent in the global zone) via `localhost`. In practice,
-we'll assign addresses as per RFD 63 as well as incorporating DNS based
+we assign addresses as per RFD 63 as well as incorporating DNS based
 service discovery.
 
-For the purposes of local development today, we specify some hardcoded IPv6
-unique local addresses in the subnet of the first Sled Agent: `fd00:1122:3344:1::/64`.
+For the purposes of local development today, we specify some hardcoded IP
+addresses.
 
 If you'd like to modify these values to suit your local network, you can modify
 them within the https://github.com/oxidecomputer/omicron/tree/main/smf[`smf/` subdirectory].
@@ -164,15 +164,6 @@ be set as a default route for the Nexus zone.
 |===================================================================================================
 | Service                    | Endpoint
 | Sled Agent: Bootstrap      | Derived from MAC address of physical data link.
-| Sled Agent: Dropshot API   | `[fd00:1122:3344:0101::1]:12345`
-| Cockroach DB               | `[fd00:1122:3344:0101::2]:32221`
-| Nexus: Internal API        | `[fd00:1122:3344:0101::3]:12221`
-| Oximeter                   | `[fd00:1122:3344:0101::4]:12223`
-| Clickhouse                 | `[fd00:1122:3344:0101::5]:8123`
-| Crucible Downstairs 1      | `[fd00:1122:3344:0101::6]:32345`
-| Crucible Downstairs 2      | `[fd00:1122:3344:0101::7]:32345`
-| Crucible Downstairs 3      | `[fd00:1122:3344:0101::8]:32345`
-| Internal DNS Service       | `[fd00:1122:3344:0001::1]:5353`
 | Nexus: External API        | `192.168.1.20:80`
 | Internet Gateway           | None, but can be set in `smf/sled-agent/config.toml`
 |===================================================================================================

nexus/src/app/rack.rs

@@ -8,7 +8,7 @@ use crate::authz;
 use crate::context::OpContext;
 use crate::db;
 use crate::db::lookup::LookupPath;
-use crate::internal_api::params::ServicePutRequest;
+use crate::internal_api::params::RackInitializationRequest;
 use omicron_common::api::external::DataPageParams;
 use omicron_common::api::external::Error;
 use omicron_common::api::external::ListResultVec;
@@ -57,12 +57,13 @@ impl super::Nexus {
         &self,
         opctx: &OpContext,
         rack_id: Uuid,
-        services: Vec<ServicePutRequest>,
+        request: RackInitializationRequest,
     ) -> Result<(), Error> {
         opctx.authorize(authz::Action::Modify, &authz::FLEET).await?;
 
         // Convert from parameter -> DB type.
-        let services: Vec<_> = services
+        let services: Vec<_> = request
+            .services
             .into_iter()
             .map(|svc| {
                 db::model::Service::new(
@@ -74,14 +75,51 @@ impl super::Nexus {
             })
             .collect();
 
-        // TODO(https://github.com/oxidecomputer/omicron/pull/1216):
-        // Actually supply datasets provided from the sled agent.
-        //
-        // This requires corresponding changes on the RSS side.
+        let datasets: Vec<_> = request
+            .datasets
+            .into_iter()
+            .map(|dataset| {
+                db::model::Dataset::new(
+                    dataset.dataset_id,
+                    dataset.zpool_id,
+                    dataset.request.address,
+                    dataset.request.kind.into(),
+                )
+            })
+            .collect();
         self.db_datastore
-            .rack_set_initialized(opctx, rack_id, services, vec![])
+            .rack_set_initialized(opctx, rack_id, services, datasets)
             .await?;
 
         Ok(())
     }
+
+    /// Awaits the initialization of the rack.
+    ///
+    /// This will occur by either:
+    /// 1. RSS invoking the internal API, handing off responsibility, or
+    /// 2. Re-reading a value from the DB, if the rack has already been
+    ///    initialized.
+    ///
+    /// See RFD 278 for additional context.
+    pub async fn await_rack_initialization(&self, opctx: &OpContext) {
+        loop {
+            let result = self.rack_lookup(&opctx, &self.rack_id).await;
+            match result {
+                Ok(rack) => {
+                    if rack.initialized {
+                        return;
+                    }
+                    info!(
+                        self.log,
+                        "Still waiting for rack initialization: {:?}", rack
+                    );
+                }
+                Err(e) => {
+                    warn!(self.log, "Cannot look up rack: {}", e);
+                }
+            }
+            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+        }
+    }
 }

nexus/src/internal_api/http_entrypoints.rs

@@ -7,8 +7,9 @@ use crate::context::OpContext;
 use crate::ServerContext;
 
 use super::params::{
-    DatasetPutRequest, DatasetPutResponse, OximeterInfo, ServicePutRequest,
-    SledAgentStartupInfo, ZpoolPutRequest, ZpoolPutResponse,
+    DatasetPutRequest, DatasetPutResponse, OximeterInfo,
+    RackInitializationRequest, SledAgentStartupInfo, ZpoolPutRequest,
+    ZpoolPutResponse,
 };
 use dropshot::endpoint;
 use dropshot::ApiDescription;
@@ -104,15 +105,15 @@ struct RackPathParam {
 async fn rack_initialization_complete(
     rqctx: Arc<RequestContext<Arc<ServerContext>>>,
     path_params: Path<RackPathParam>,
-    info: TypedBody<Vec<ServicePutRequest>>,
+    info: TypedBody<RackInitializationRequest>,
 ) -> Result<HttpResponseUpdatedNoContent, HttpError> {
     let apictx = rqctx.context();
     let nexus = &apictx.nexus;
     let path = path_params.into_inner();
-    let svcs = info.into_inner();
+    let request = info.into_inner();
     let opctx = OpContext::for_internal_api(&rqctx).await;
 
-    nexus.rack_initialize(&opctx, path.rack_id, svcs).await?;
+    nexus.rack_initialize(&opctx, path.rack_id, request).await?;
 
     Ok(HttpResponseUpdatedNoContent())
 }

nexus/src/lib.rs

@@ -66,23 +66,27 @@ pub fn run_openapi_internal() -> Result<(), String> {
         .map_err(|e| e.to_string())
 }
 
-/// Packages up a [`Nexus`], running both external and internal HTTP API servers
-/// wired up to Nexus
-pub struct Server {
+/// A partially-initialized Nexus server, which exposes an internal interface,
+/// but is not ready to receive external requests.
+pub struct InternalServer<'a> {
     /// shared state used by API request handlers
     pub apictx: Arc<ServerContext>,
-    /// dropshot server for external API
-    pub http_server_external: dropshot::HttpServer<Arc<ServerContext>>,
     /// dropshot server for internal API
     pub http_server_internal: dropshot::HttpServer<Arc<ServerContext>>,
+
+    config: &'a Config,
+    log: Logger,
 }
 
-impl Server {
-    /// Start a nexus server.
+impl<'a> InternalServer<'a> {
+    /// Creates a Nexus instance with only the internal API exposed.
+    ///
+    /// This is often used as an argument when creating a [`Server`],
+    /// which also exposes the external API.
     pub async fn start(
-        config: &Config,
+        config: &'a Config,
         log: &Logger,
-    ) -> Result<Server, String> {
+    ) -> Result<InternalServer<'a>, String> {
         let log = log.new(o!("name" => config.deployment.id.to_string()));
         info!(log, "setting up nexus server");
 
@@ -92,24 +96,55 @@ impl Server {
             ServerContext::new(config.deployment.rack_id, ctxlog, &config)
                 .await?;
 
-        let http_server_starter_external = dropshot::HttpServerStarter::new(
-            &config.deployment.dropshot_external,
-            external_api(),
-            Arc::clone(&apictx),
-            &log.new(o!("component" => "dropshot_external")),
-        )
-        .map_err(|error| format!("initializing external server: {}", error))?;
-
         let http_server_starter_internal = dropshot::HttpServerStarter::new(
             &config.deployment.dropshot_internal,
             internal_api(),
             Arc::clone(&apictx),
             &log.new(o!("component" => "dropshot_internal")),
         )
         .map_err(|error| format!("initializing internal server: {}", error))?;
+        let http_server_internal = http_server_starter_internal.start();
+
+        Ok(Self { apictx, http_server_internal, config, log })
+    }
+}
+
+/// Packages up a [`Nexus`], running both external and internal HTTP API servers
+/// wired up to Nexus
+pub struct Server {
+    /// shared state used by API request handlers
+    pub apictx: Arc<ServerContext>,
+    /// dropshot server for external API
+    pub http_server_external: dropshot::HttpServer<Arc<ServerContext>>,
+    /// dropshot server for internal API
+    pub http_server_internal: dropshot::HttpServer<Arc<ServerContext>>,
+}
 
+impl Server {
+    pub async fn start(internal: InternalServer<'_>) -> Result<Self, String> {
+        let apictx = internal.apictx;
+        let http_server_internal = internal.http_server_internal;
+        let log = internal.log;
+        let config = internal.config;
+
+        // Wait until RSS handoff completes.
+        let opctx = apictx.nexus.opctx_for_service_balancer();
+        apictx.nexus.await_rack_initialization(&opctx).await;
+
+        // With the exception of integration tests environments,
+        // we expect background tasks to be enabled.
+        if config.pkg.tunables.enable_background_tasks {
+            apictx.nexus.start_background_tasks().map_err(|e| e.to_string())?;
+        }
+
+        let http_server_starter_external = dropshot::HttpServerStarter::new(
+            &config.deployment.dropshot_external,
+            external_api(),
+            Arc::clone(&apictx),
+            &log.new(o!("component" => "dropshot_external")),
+        )
+        .map_err(|error| format!("initializing external server: {}", error))?;
         let http_server_external = http_server_starter_external.start();
-        let http_server_internal = http_server_starter_internal.start();
 
         Ok(Server { apictx, http_server_external, http_server_internal })
     }
@@ -167,7 +202,8 @@ pub async fn run_server(config: &Config) -> Result<(), String> {
     } else {
         debug!(log, "registered DTrace probes");
     }
-    let server = Server::start(config, &log).await?;
+    let internal_server = InternalServer::start(config, &log).await?;
+    let server = Server::start(internal_server).await?;
     server.register_as_producer().await;
     server.wait_for_finish().await
 }

nexus/test-utils/src/lib.rs

@@ -109,15 +109,42 @@ pub async fn test_setup_with_config(
         .expect("Tests expect to set a port of Clickhouse")
         .set_port(clickhouse.port());
 
-    let server =
-        omicron_nexus::Server::start(&config, &logctx.log).await.unwrap();
-    server
+    // Start the Nexus internal API.
+    let internal_server =
+        omicron_nexus::InternalServer::start(&config, &logctx.log)
+            .await
+            .unwrap();
+    internal_server
         .apictx
         .nexus
         .wait_for_populate()
         .await
         .expect("Nexus never loaded users");
 
+    // Perform the "handoff from RSS".
+    //
+    // However, RSS isn't running, so we'll do the handoff ourselves.
+    let opctx = internal_server.apictx.nexus.opctx_for_service_balancer();
+    internal_server
+        .apictx
+        .nexus
+        .rack_initialize(
+            &opctx,
+            config.deployment.rack_id,
+            // NOTE: In the context of this test utility, we arguably do have an
+            // instance of CRDB and Nexus running. However, as this info isn't
+            // necessary for most tests, we pass no information here.
+            omicron_nexus::internal_api::params::RackInitializationRequest {
+                services: vec![],
+                datasets: vec![],
+            },
+        )
+        .await
+        .expect("Could not initialize rack");
+
+    // Start the Nexus external API.
+    let server = omicron_nexus::Server::start(internal_server).await.unwrap();
+
     let testctx_external = ClientTestContext::new(
         server.http_server_external.local_addr(),
         logctx.log.new(o!("component" => "external client test context")),

nexus/tests/config.test.toml

@@ -34,6 +34,7 @@ address = "[::1]:0"
 [tunables]
 # Allow small subnets, so we can test IP address exhaustion easily / quickly
 max_vpc_ipv4_subnet_prefix = 29
+# Disable background tests to help with test determinism
 enable_background_tasks = false
 
 [deployment]

nexus/types/src/internal_api/params.rs

@@ -152,6 +152,19 @@ pub struct ServicePutRequest {
     pub kind: ServiceKind,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
+pub struct DatasetCreateRequest {
+    pub zpool_id: Uuid,
+    pub dataset_id: Uuid,
+    pub request: DatasetPutRequest,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
+pub struct RackInitializationRequest {
+    pub services: Vec<ServicePutRequest>,
+    pub datasets: Vec<DatasetCreateRequest>,
+}
+
 /// Message used to notify Nexus that this oximeter instance is up and running.
 #[derive(Debug, Clone, Copy, JsonSchema, Serialize, Deserialize)]
 pub struct OximeterInfo {

openapi/nexus-internal.json

@@ -255,11 +255,7 @@
           "content": {
             "application/json": {
               "schema": {
-                "title": "Array_of_ServicePutRequest",
-                "type": "array",
-                "items": {
-                  "$ref": "#/components/schemas/ServicePutRequest"
-                }
+                "$ref": "#/components/schemas/RackInitializationRequest"
               }
             }
           },
@@ -674,6 +670,27 @@
           "value"
         ]
       },
+      "DatasetCreateRequest": {
+        "type": "object",
+        "properties": {
+          "dataset_id": {
+            "type": "string",
+            "format": "uuid"
+          },
+          "request": {
+            "$ref": "#/components/schemas/DatasetPutRequest"
+          },
+          "zpool_id": {
+            "type": "string",
+            "format": "uuid"
+          }
+        },
+        "required": [
+          "dataset_id",
+          "request",
+          "zpool_id"
+        ]
+      },
       "DatasetKind": {
         "description": "Describes the purpose of the dataset.",
         "type": "string",
@@ -1711,6 +1728,27 @@
           }
         ]
       },
+      "RackInitializationRequest": {
+        "type": "object",
+        "properties": {
+          "datasets": {
+            "type": "array",
+            "items": {
+              "$ref": "#/components/schemas/DatasetCreateRequest"
+            }
+          },
+          "services": {
+            "type": "array",
+            "items": {
+              "$ref": "#/components/schemas/ServicePutRequest"
+            }
+          }
+        },
+        "required": [
+          "datasets",
+          "services"
+        ]
+      },
       "Sample": {
         "description": "A concrete type representing a single, timestamped measurement from a timeseries.",
         "type": "object",