start ereport ingestion task

Author: hawkw

Cargo.lock

@@ -11,6 +11,7 @@ workspace = true
 base64.workspace = true
 chrono.workspace = true
 daft.workspace = true
+ereport-types.workspace = true
 gateway-messages.workspace = true
 gateway-types.workspace = true
 progenitor.workspace = true
@@ -21,4 +22,5 @@ serde_json.workspace = true
 schemars.workspace = true
 slog.workspace = true
 uuid.workspace = true
+omicron-uuid-kinds.workspace = true
 omicron-workspace-hack.workspace = true

clients/gateway-client/Cargo.toml

@@ -74,7 +74,11 @@ progenitor::generate_api!(
         SpUpdateStatus = { derives = [PartialEq, Hash, Eq] },
         UpdatePreparationProgress = { derives = [PartialEq, Hash, Eq] },
     },
-    replace = { RotSlot = gateway_types::rot::RotSlot },
+    replace = {
+        RotSlot = gateway_types::rot::RotSlot,
+        Ena = ereport_types::Ena,
+        TypedUuidForEreporterRestartKind = omicron_uuid_kinds::EreporterRestartUuid,
+    },
 );
 
 // Override the impl of Ord for SpIdentifier because the default one orders the

clients/gateway-client/src/lib.rs

@@ -31,6 +31,7 @@ crucible-pantry-client.workspace = true
 crucible-common.workspace = true
 dns-service-client.workspace = true
 dpd-client.workspace = true
+ereport-types.workspace = true
 mg-admin-client.workspace = true
 dropshot.workspace = true
 fatfs.workspace = true

nexus/Cargo.toml

@@ -0,0 +1,105 @@
+// 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/.
+
+//! Background tasks for ereport ingestion
+
+use crate::app::background::BackgroundTask;
+use anyhow::Context;
+use futures::future::BoxFuture;
+use internal_dns_types::names::ServiceName;
+use nexus_db_queries::context::OpContext;
+use nexus_db_queries::db::DataStore;
+use omicron_common::backoff;
+use omicron_uuid_kinds::EreporterRestartUuid;
+use omicron_uuid_kinds::OmicronZoneUuid;
+use std::sync::Arc;
+use uuid::Uuid;
+
+pub struct SpEreportIngester {
+    pub datastore: Arc<DataStore>,
+    pub resolver: internal_dns_resolver::Resolver,
+    pub nexus_id: OmicronZoneUuid,
+    pub rack_id: Uuid,
+}
+
+impl BackgroundTask for SpEreportIngester {
+    fn activate<'a>(
+        &'a mut self,
+        opctx: &'a OpContext,
+    ) -> BoxFuture<'a, serde_json::Value> {
+        todo!()
+    }
+}
+
+impl SpEreportIngester {
+    async fn actually_activate(
+        &mut self,
+        opctx: &OpContext,
+    ) -> anyhow::Result<()> {
+        // Find MGS clients.
+        // TODO(eliza): reuse the same client across activations...
+        let mgs_clients = self
+            .resolver
+            .lookup_all_socket_v6(ServiceName::ManagementGatewayService)
+            .await
+            .context("looking up MGS addresses")?
+            .into_iter()
+            .map(|sockaddr| {
+                let url = format!("http://{}", sockaddr);
+                let log = opctx.log.new(o!("gateway_url" => url.clone()));
+                gateway_client::Client::new(&url, log)
+            })
+            .collect::<Vec<_>>();
+
+        Ok(())
+    }
+}
+
+struct Sp {
+    sp_type: gateway_client::types::SpType,
+    slot: u32,
+}
+
+struct SpState {
+    committed_ena: ereport_types::Ena,
+    restart_id: EreporterRestartUuid,
+}
+
+const LIMIT: std::num::NonZeroU32 = match std::num::NonZeroU32::new(255) {
+    None => unreachable!(),
+    Some(l) => l,
+};
+
+async fn mgs_request(
+    opctx: &OpContext,
+    clients: &[gateway_client::Client],
+    Sp { sp_type, slot }: Sp,
+    sp_state: Option<SpState>,
+) -> anyhow::Result<()> {
+    let mut idx = 0;
+    let restart_id = sp_state
+        .as_ref()
+        .map(|state| state.restart_id.clone())
+        .unwrap_or(EreporterRestartUuid::nil());
+    backoff::retry_notify_ext(
+        backoff::retry_policy_internal_service(),
+        || async {
+            let client = &clients[idx];
+            let res = client
+                .sp_ereports_ingest(
+                    sp_type,
+                    slot,
+                    sp_state.as_ref().map(|state| &state.committed_ena),
+                    LIMIT,
+                    &restart_id,
+                    sp_state.as_ref().map(|state| &state.committed_ena),
+                )
+                .await;
+
+            Ok(())
+        },
+        |err, count, duration| todo!(),
+    )
+    .await
+}

nexus/src/app/background/tasks/ereport_ingester.rs

@@ -15,6 +15,7 @@ pub mod decommissioned_disk_cleaner;
 pub mod dns_config;
 pub mod dns_propagation;
 pub mod dns_servers;
+pub mod ereport_ingester;
 pub mod external_endpoints;
 pub mod instance_reincarnation;
 pub mod instance_updater;