wip bgtask stuff

Author: hawkw

nexus/db-model/src/instance_state.rs

@@ -71,3 +71,8 @@ impl From<external::InstanceState> for InstanceState {
         Self::new(state)
     }
 }
+
+impl diesel::query_builder::QueryId for InstanceStateEnum {
+    type QueryId = ();
+    const HAS_STATIC_QUERY_ID: bool = false;
+}

nexus/db-queries/src/db/datastore/instance.rs

@@ -233,6 +233,40 @@ impl DataStore {
         .collect())
     }
 
+    /// List all instances with active VMMs in the `Destroyed` state that don't
+    /// have currently-running instance-updater sagas.
+    pub async fn find_instances_with_destroyed_active_vmms(
+        &self,
+        opctx: &OpContext,
+    ) -> ListResultVec<InstanceAndActiveVmm> {
+        use db::model::InstanceState as DbInstanceState;
+        use db::schema::instance::dsl;
+        use db::schema::vmm::dsl as vmm_dsl;
+        use omicron_common::api::external::InstanceState;
+        let destroyed = DbInstanceState::new(InstanceState::Destroyed);
+        Ok(vmm_dsl::vmm
+            .filter(vmm_dsl::time_deleted.is_not_null())
+            .filter(vmm_dsl::state.eq(destroyed))
+            .inner_join(
+                dsl::instance.on(dsl::active_propolis_id
+                    .eq(vmm_dsl::id.nullable())
+                    .and(dsl::time_deleted.is_null())
+                    .and(dsl::updater_id.is_null())),
+            )
+            .select((Instance::as_select(), Vmm::as_select()))
+            .load_async::<(Instance, Vmm)>(
+                &*self.pool_connection_authorized(opctx).await?,
+            )
+            .await
+            .map_err(|e| public_error_from_diesel(e, ErrorHandler::Server))?
+            .into_iter()
+            .map(|(instance, vmm)| InstanceAndActiveVmm {
+                instance,
+                vmm: Some(vmm),
+            })
+            .collect())
+    }
+
     /// Fetches information about an Instance that the caller has previously
     /// fetched
     ///

nexus/src/app/background/instance_updater.rs

@@ -0,0 +1,99 @@
+// 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 task for detecting instances in need of update sagas.
+//!
+//! TODO this is currently a placeholder for a future PR
+
+use super::common::BackgroundTask;
+use crate::app::sagas::SagaRequest;
+use futures::future::BoxFuture;
+use futures::FutureExt;
+use nexus_db_queries::context::OpContext;
+use nexus_db_queries::db::DataStore;
+use serde_json::json;
+use std::sync::Arc;
+use tokio::sync::mpsc::Sender;
+
+pub struct InstanceUpdater {
+    datastore: Arc<DataStore>,
+    saga_req: Sender<SagaRequest>,
+}
+
+impl InstanceUpdater {
+    pub fn new(
+        datastore: Arc<DataStore>,
+        saga_req: Sender<SagaRequest>,
+    ) -> Self {
+        InstanceUpdater { datastore, saga_req }
+    }
+
+    async fn activate2(
+        &mut self,
+        opctx: &OpContext,
+    ) -> Result<Updated, anyhow::Error> {
+        let mut updated = Updated::default();
+
+        let log = &opctx.log;
+
+        slog::debug!(
+            &log,
+            "looking for instances with destroyed active VMMs..."
+        );
+
+        let destroyed_active_vmms = self
+            .datastore
+            .find_instances_with_destroyed_active_vmms(opctx)
+            .await
+            .context("failed to find instances with destroyed active VMMs")?;
+
+        slog::info!(
+            &log,
+            "listed instances with destroyed active VMMs";
+            "count" => destroyed_active_vmms.len(),
+        );
+
+        updated.destroyed_active_vmms = destroyed_active_vmms.len();
+
+        for (instance, vmm) in destroyed_active_vmms {
+            let saga = SagaRequest::InstanceUpdate {};
+        }
+
+        Ok(updated)
+    }
+}
+
+#[derive(Default)]
+struct Updated {
+    destroyed_active_vmms: usize,
+    sagas_started: usize,
+}
+
+impl BackgroundTask for InstanceUpdater {
+    fn activate<'a>(
+        &'a mut self,
+        opctx: &'a OpContext,
+    ) -> BoxFuture<'a, serde_json::Value> {
+        async {
+            match self.activate2(opctx).await {
+                Ok(updated) => json!({
+                    "destroyed_active_vmms": updated.destroyed_active_vmms,
+                    "error": None,
+                }),
+                Err(error) => {
+                    slog::error!(
+                        opctx.log,
+                        "failed to start instance update saga(s)";
+                        "error" => ?error,
+                    );
+                    json!({
+                        "destroyed_active_vmms": 0,
+                        "error": error.to_string(),
+                    })
+                }
+            }
+        }
+        .boxed()
+    }
+}

nexus/src/app/background/mod.rs

@@ -13,6 +13,7 @@ mod dns_propagation;
 mod dns_servers;
 mod external_endpoints;
 mod init;
+mod instance_updater;
 mod instance_watcher;
 mod inventory_collection;
 mod metrics_producer_gc;