put versions on all the things

Author: hawkw

dev-tools/omdb/src/bin/omdb/db/alert.rs

@@ -1012,6 +1012,7 @@ async fn cmd_db_alert_info(
         class,
         payload,
         num_dispatched,
+        schema_version,
     } = alert;
 
     const CLASS: &str = "class";
@@ -1029,7 +1030,7 @@ async fn cmd_db_alert_info(
 
     println!("\n{:=<80}", "== ALERT ");
     println!("    {ID:>WIDTH$}: {id:?}");
-    println!("    {CLASS:>WIDTH$}: {class}");
+    println!("    {CLASS:>WIDTH$}: {class} (v{schema_version})");
     println!("    {TIME_CREATED:>WIDTH$}: {time_created}");
     println!("    {TIME_MODIFIED:>WIDTH$}: {time_modified}");
     println!();

nexus/db-model/src/alert.rs

@@ -3,6 +3,7 @@
 // file, You can obtain one at https://mozilla.org/MPL/2.0/.
 
 use crate::AlertClass;
+use crate::SqlU32;
 use chrono::{DateTime, Utc};
 use db_macros::Asset;
 use nexus_db_schema::schema::alert;
@@ -40,6 +41,9 @@ pub struct Alert {
     pub payload: serde_json::Value,
 
     pub num_dispatched: i64,
+
+    /// The version of the JSON schema for `payload`.
+    pub schema_version: SqlU32,
 }
 
 impl Alert {

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

@@ -26,6 +26,7 @@ impl DataStore {
         opctx: &OpContext,
         id: AlertUuid,
         class: AlertClass,
+        version: u32,
         payload: serde_json::Value,
     ) -> CreateResult<Alert> {
         let conn = self.pool_connection_authorized(&opctx).await?;
@@ -36,6 +37,7 @@ impl DataStore {
                 class,
                 payload,
                 num_dispatched: 0,
+                schema_version: version.into(),
             })
             .returning(Alert::as_returning())
             .get_result_async(&*conn)

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

@@ -1208,7 +1208,7 @@ mod test {
     ) -> (authz::Alert, crate::db::model::Alert) {
         let id = AlertUuid::new_v4();
         datastore
-            .alert_create(opctx, id, alert_class, serde_json::json!({}))
+            .alert_create(opctx, id, alert_class, 1, serde_json::json!({}))
             .await
             .expect("cant create ye event");
         LookupPath::new(opctx, datastore).alert_id(id).fetch().await.expect(

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

@@ -52,15 +52,6 @@ pub struct DeliveryConfig {
     pub lease_timeout: TimeDelta,
 }
 
-/// A record from the [`WebhookDelivery`] table along with the event class and
-/// data of the corresponding [`Alert`] record.
-#[derive(Debug, Clone)]
-pub struct DeliveryAndEvent {
-    pub delivery: WebhookDelivery,
-    pub alert_class: AlertClass,
-    pub event: serde_json::Value,
-}
-
 impl DataStore {
     pub async fn webhook_delivery_create_batch(
         &self,
@@ -191,12 +182,11 @@ impl DataStore {
         opctx: &OpContext,
         rx_id: &AlertReceiverUuid,
         cfg: &DeliveryConfig,
-    ) -> Result<impl ExactSizeIterator<Item = DeliveryAndEvent> + 'static, Error>
-    {
+    ) -> ListResultVec<(WebhookDelivery, Alert)> {
         let conn = self.pool_connection_authorized(opctx).await?;
         let now =
             diesel::dsl::now.into_sql::<diesel::pg::sql_types::Timestamptz>();
-        let rows = dsl::webhook_delivery
+        dsl::webhook_delivery
             // Filter out deliveries triggered by probe requests, as those are
             // executed synchronously by the probe endpoint, rather than by the
             // webhook deliverator.
@@ -232,20 +222,14 @@ impl DataStore {
                     )),
             )
             .order_by(dsl::time_created.asc())
-            // Join with the `webhook_event` table to get the event class, which
-            // is necessary to construct delivery requests.
+            // Join with the `alert` table to get the alert, including its
+            // class, payload version, and payload (which are necessary to
+            // construct delivery requests).let rows = 
             .inner_join(alert_dsl::alert.on(alert_dsl::id.eq(dsl::alert_id)))
-            .select((
-                WebhookDelivery::as_select(),
-                alert_dsl::alert_class,
-                alert_dsl::payload,
-            ))
+            .select((WebhookDelivery::as_select(), Alert::as_select()))
             .load_async(&*conn)
             .await
-            .map_err(|e| public_error_from_diesel(e, ErrorHandler::Server))?;
-        Ok(rows.into_iter().map(|(delivery, alert_class, event)| {
-            DeliveryAndEvent { delivery, alert_class, event }
-        }))
+            .map_err(|e| public_error_from_diesel(e, ErrorHandler::Server))
     }
 
     pub async fn webhook_delivery_start_attempt(
@@ -488,6 +472,7 @@ mod test {
                 &opctx,
                 alert_id,
                 AlertClass::TestFoo,
+                1,
                 serde_json::json!({
                     "answer": 42,
                 }),

nexus/db-schema/src/schema.rs

@@ -2362,6 +2362,7 @@ table! {
         payload -> Jsonb,
         time_dispatched -> Nullable<Timestamptz>,
         num_dispatched -> Int8,
+        schema_version -> Int8,
     }
 }
 

nexus/src/app/alert.rs

@@ -150,7 +150,7 @@ use nexus_db_lookup::LookupPath;
 use nexus_db_lookup::lookup;
 use nexus_db_queries::authz;
 use nexus_db_queries::context::OpContext;
-use nexus_db_queries::db::model::Alert;
+use nexus_db_queries::db::model::Alert as DbAlert;
 use nexus_db_queries::db::model::AlertClass;
 use nexus_db_queries::db::model::AlertDeliveryState;
 use nexus_db_queries::db::model::AlertDeliveryTrigger;
@@ -182,12 +182,12 @@ impl nexus_alerts::PublishAlert for Nexus {
         opctx: &OpContext,
         id: AlertUuid,
         alert: A,
-    ) -> Result<Alert, Error> {
+    ) -> Result<DbAlert, Error> {
         self.alert_publish(opctx, id, alert).await
     }
 }
 
-pub(crate) fn alert_schemas() -> AlertSchemaRegistry {
+pub(crate) fn schemas() -> AlertSchemaRegistry {
     let mut registry = AlertSchemaRegistry::new();
 
     #[cfg(debug_assertions)]
@@ -217,7 +217,7 @@ impl Nexus {
         opctx: &OpContext,
         id: AlertUuid,
         alert: A,
-    ) -> Result<Alert, Error> {
+    ) -> Result<DbAlert, Error> {
         #[cfg(debug_assertions)]
         {
             // In test builds, assert that this is a schema that we know about.
@@ -229,8 +229,8 @@ impl Nexus {
                 None => panic!(
                     "You have attempted to publish an alert type whose class \
                      was not added to the alert schema registry in \
-                     `nexus::app::alert::alert_schemas()`! This means that \
-                     the alert type's schema will not be included in the \
+                     `nexus::app::alert::schemas()`! This means that the \
+                     alert type's schema will not be included in the \
                      alert JSON schema. This is probably a mistake. Since I \
                      am a test build, I will now panic!\n    \
                          alert class: {}\n  \
@@ -244,7 +244,7 @@ impl Nexus {
                 panic!(
                     "You have attempted to publish an alert type whose schema \
                      version is not present in the alert schema registry in \
-                     `nexus::app::alert::alert_schemas()`! This is probably a \
+                     `nexus::app::alert::schemas()`! This is probably a \
                      mistake. Since I am a test build, I will  now panic!\n    \
                          alert class: {}\n  \
                        alert version: {}",
@@ -255,7 +255,7 @@ impl Nexus {
         }
 
         let payload =
-            serde_json::to_value(event).map_err(|e| Error::InternalError {
+            serde_json::to_value(&alert).map_err(|e| Error::InternalError {
                 internal_message: format!(
                     "failed to convert {} (class: {} v{}) to JSON: {e}",
                     std::any::type_name::<A>(),
@@ -264,8 +264,10 @@ impl Nexus {
                 ),
             })?;
 
-        let alert =
-            self.datastore().alert_create(opctx, id, class, payload).await?;
+        let alert = self
+            .datastore()
+            .alert_create(opctx, id, A::CLASS, A::VERSION, payload)
+            .await?;
 
         slog::debug!(
             &opctx.log,

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

@@ -454,7 +454,8 @@ mod test {
                 &opctx,
                 alert_id,
                 db::model::AlertClass::TestQuuxBar,
-                serde_json::json!({"msg": "help im trapped in a webhook event factory"}),
+                1,
+                serde_json::json!({"msg": "help im trapped in an alert factory"}),
             )
             .await
             .expect("creating the event should work");

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

@@ -35,9 +35,9 @@ use crate::app::webhook::ReceiverClient;
 use futures::future::BoxFuture;
 use nexus_db_queries::context::OpContext;
 use nexus_db_queries::db::DataStore;
-use nexus_db_queries::db::datastore::webhook_delivery::DeliveryAndEvent;
 use nexus_db_queries::db::datastore::webhook_delivery::DeliveryAttemptState;
 pub use nexus_db_queries::db::datastore::webhook_delivery::DeliveryConfig;
+use nexus_db_queries::db::model::Alert;
 use nexus_db_queries::db::model::WebhookDeliveryAttemptResult;
 use nexus_db_queries::db::model::WebhookReceiverConfig;
 use nexus_db_queries::db::pagination::Paginator;
@@ -231,7 +231,9 @@ impl WebhookDeliverator {
             ..Default::default()
         };
 
-        for DeliveryAndEvent { delivery, alert_class, event } in deliveries {
+        for (delivery, alert) in deliveries {
+            let Alert { class, payload, schema_version, .. } = alert;
+            let version = schema_version.into();
             let attempt = (*delivery.attempts) + 1;
             let delivery_id = WebhookDeliveryUuid::from(delivery.id);
             match self
@@ -248,7 +250,8 @@ impl WebhookDeliverator {
                     slog::trace!(&opctx.log,
                         "webhook event delivery attempt started";
                         "alert_id" => %delivery.alert_id,
-                        "alert_class" => %alert_class,
+                        "alert_class" => %class,
+                        "alert_version" => %version,
                         "delivery_id" => %delivery_id,
                         "attempt" => ?attempt,
                     );
@@ -259,7 +262,8 @@ impl WebhookDeliverator {
                         "delivery of this webhook event was already completed \
                          at {time:?}";
                         "alert_id" => %delivery.alert_id,
-                        "alert_class" => %alert_class,
+                        "alert_class" => %class,
+                        "alert_version" => %version,
                         "delivery_id" => %delivery_id,
                         "time_completed" => ?time,
                     );
@@ -272,7 +276,8 @@ impl WebhookDeliverator {
                         "delivery of this webhook event is in progress by \
                          another Nexus";
                         "alert_id" => %delivery.alert_id,
-                        "alert_class" => %alert_class,
+                        "alert_class" => %class,
+                        "alert_version" => %version,
                         "delivery_id" => %delivery_id,
                         "nexus_id" => %nexus_id,
                         "time_started" => ?started,
@@ -286,7 +291,8 @@ impl WebhookDeliverator {
                         "unexpected database error error starting webhook \
                          delivery attempt";
                         "alert_id" => %delivery.alert_id,
-                        "alert_class" => %alert_class,
+                        "alert_class" => %class,
+                        "alert_version" => %version,
                         "delivery_id" => %delivery_id,
                         "error" => %error,
                     );
@@ -299,7 +305,9 @@ impl WebhookDeliverator {
 
             // okay, actually do the thing...
             let delivery_attempt = match client
-                .send_delivery_request(opctx, &delivery, alert_class, &event)
+                .send_delivery_request(
+                    opctx, &delivery, class, version, &payload,
+                )
                 .await
             {
                 Ok(delivery) => delivery,
@@ -326,7 +334,8 @@ impl WebhookDeliverator {
                     &opctx.log,
                     "{MSG}";
                     "alert_id" => %delivery.alert_id,
-                    "alert_class" => %alert_class,
+                    "alert_class" => %class,
+                    "alert_version" => %version,
                     "delivery_id" => %delivery_id,
                     "error" => %e,
                 );

nexus/src/app/mod.rs

@@ -269,6 +269,9 @@ pub struct Nexus {
 
     /// reports status of pending MGS-managed updates
     mgs_update_status_rx: watch::Receiver<MgsUpdateDriverStatus>,
+
+    /// Collection of JSON schemas for alert classes and versions.
+    alert_schemas: alert::AlertSchemaRegistry,
 }
 
 impl Nexus {
@@ -496,6 +499,7 @@ impl Nexus {
             )),
             tuf_artifact_replication_tx,
             mgs_update_status_rx,
+            alert_schemas: alert::schemas(),
         };
 
         // TODO-cleanup all the extra Arcs here seems wrong

nexus/src/app/webhook.rs

@@ -196,7 +196,7 @@ impl Nexus {
             LazyLock::new(|| serde_json::json!({}));
 
         let attempt = match client
-            .send_delivery_request(opctx, &delivery, CLASS, &DATA)
+            .send_delivery_request(opctx, &delivery, CLASS, 1, &DATA)
             .await
         {
             Ok(attempt) => attempt,
@@ -376,6 +376,7 @@ impl<'a> ReceiverClient<'a> {
         opctx: &OpContext,
         delivery: &WebhookDelivery,
         alert_class: AlertClass,
+        version: u32,
         data: &serde_json::Value,
     ) -> Result<WebhookDeliveryAttempt, anyhow::Error> {
         const HDR_DELIVERY_ID: HeaderName =
@@ -437,6 +438,7 @@ impl<'a> ReceiverClient<'a> {
                     "webhook {MSG}";
                     "alert_id" => %delivery.alert_id,
                     "alert_class" => %alert_class,
+                    "alert_version" => version,
                     "delivery_id" => %delivery.id,
                     "delivery_trigger" => %delivery.triggered_by,
                     "error" => %e,

nexus/tests/integration_tests/webhooks.rs

@@ -1711,11 +1711,7 @@ async fn subscription_remove_test(
     // Publish an event. This should be received, as it matches the subscription
     // we are about to delete.
     let event = nexus
-        .alert_publish(
-            &opctx,
-            id1,
-           test_alerts::FooBar { hello: "joe" }),
-        )
+        .alert_publish(&opctx, id1, test_alerts::FooBar { hello: "joe" })
         .await
         .expect("event should be published successfully");
     dbg!(event);
@@ -1742,11 +1738,7 @@ async fn subscription_remove_test(
     // Publish an event. This one should not be received, as we are no longer
     // subscribed to its event class.
     let event = nexus
-        .alert_publish(
-            &opctx,
-            id2,
-            test_alerts::FooBar { hello: "robert" }
-        )
+        .alert_publish(&opctx, id2, test_alerts::FooBar { hello: "robert" })
         .await
         .expect("event should be published successfully");
     dbg!(event);
@@ -1788,11 +1780,7 @@ async fn subscription_remove_test(
     };
 
     let event = nexus
-        .alert_publish(
-            &opctx,
-            id3,
-            test_alerts::Foo { hello_world: true }
-        )
+        .alert_publish(&opctx, id3, test_alerts::Foo { hello_world: true })
         .await
         .expect("event should be published successfully");
     dbg!(event);

schema/crdb/alert-schema-versions/up01.sql

@@ -0,0 +1,2 @@
+ALTER TABLE omicron.public.alert
+ADD COLUMN IF NOT EXISTS schema_version INT8 NOT NULL DEFAULT 1;

schema/crdb/alert-schema-versions/up02.sql

@@ -0,0 +1,3 @@
+ALTER TABLE omicron.public.alert
+ALTER COLUMN schema_version
+DROP DEFAULT;