WIP: internal API endpoints

common/src/sql/dbinit.sql

@@ -2369,6 +2369,10 @@ CREATE TABLE omicron.public.ipv4_nat_entry (
     time_deleted TIMESTAMPTZ
 );
 
+CREATE UNIQUE INDEX ON omicron.public.ipv4_nat_entry (
+    external_address, first_port, last_port
+);
+
 CREATE UNIQUE INDEX ON omicron.public.ipv4_nat_entry (
     gen
 )

nexus/db-model/src/ipv4_nat_entry.rs

@@ -4,6 +4,8 @@ use crate::{
     SqlU16, SqlU32, Vni,
 };
 use chrono::{DateTime, Utc};
+use schemars::JsonSchema;
+use serde::Serialize;
 use uuid::Uuid;
 
 /// Database representation of an Ipv4 NAT Entry.
@@ -40,7 +42,7 @@ impl Ipv4NatEntry {
     }
 
     pub fn last_port(&self) -> u16 {
-        self.first_port.into()
+        self.last_port.into()
     }
 
     pub fn gen(&self) -> u32 {
@@ -55,3 +57,37 @@ pub struct Ipv4NatGen {
     pub log_cnt: SqlU32,
     pub is_called: bool,
 }
+
+/// NAT Record
+#[derive(Clone, Debug, Serialize, JsonSchema)]
+pub struct Ipv4NatEntryView {
+    pub external_address: String,
+    pub first_port: u16,
+    pub last_port: u16,
+    pub sled_address: String,
+    pub vni: u32,
+    pub mac: String,
+    pub gen: u32,
+    pub deleted: bool,
+}
+
+impl From<Ipv4NatEntry> for Ipv4NatEntryView {
+    fn from(value: Ipv4NatEntry) -> Self {
+        Self {
+            external_address: value.external_address.to_string(),
+            first_port: value.first_port(),
+            last_port: value.last_port(),
+            sled_address: value.sled_address.to_string(),
+            vni: value.vni.0.into(),
+            mac: value.mac.to_string(),
+            gen: value.gen(),
+            deleted: value.time_deleted.is_some(),
+        }
+    }
+}
+
+/// NAT Generation
+#[derive(Clone, Debug, Serialize, JsonSchema)]
+pub struct Ipv4NatGenView {
+    pub gen: u32,
+}

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

@@ -7,6 +7,7 @@ use crate::db::model::{Ipv4NatEntry, NewIpv4NatEntry};
 use async_bb8_diesel::AsyncRunQueryDsl;
 use chrono::{DateTime, Utc};
 use diesel::prelude::*;
+use nexus_db_model::Ipv4NatEntryView;
 use nexus_db_model::SqlU32;
 use omicron_common::api::external::CreateResult;
 use omicron_common::api::external::DeleteResult;
@@ -132,6 +133,7 @@ impl DataStore {
         &self,
         opctx: &OpContext,
         gen: u32,
+        limit: u32,
     ) -> ListResultVec<Ipv4NatEntry> {
         use db::schema::ipv4_nat_entry::dsl;
         let gen: SqlU32 = gen.into();
@@ -141,6 +143,7 @@ impl DataStore {
         let list = dsl::ipv4_nat_entry
             .filter(dsl::gen.gt(gen))
             .order_by(dsl::gen)
+            .limit(limit as i64)
             .select(Ipv4NatEntry::as_select())
             .load_async(self.pool_authorized(opctx).await?)
             .await
@@ -151,6 +154,19 @@ impl DataStore {
         Ok(list)
     }
 
+    pub async fn ipv4_nat_changeset(
+        &self,
+        opctx: &OpContext,
+        gen: u32,
+        limit: u32,
+    ) -> ListResultVec<Ipv4NatEntryView> {
+        let nat_entries =
+            self.ipv4_nat_list_since_gen(opctx, gen, limit).await?;
+        let nat_entries: Vec<Ipv4NatEntryView> =
+            nat_entries.iter().map(|e| e.clone().into()).collect();
+        Ok(nat_entries)
+    }
+
     pub async fn ipv4_nat_current_gen(
         &self,
         opctx: &OpContext,
@@ -219,7 +235,7 @@ mod test {
 
         // We should not have any NAT entries at this moment
         let initial_state =
-            datastore.ipv4_nat_list_since_gen(&opctx, 0).await.unwrap();
+            datastore.ipv4_nat_list_since_gen(&opctx, 0, 10).await.unwrap();
 
         assert!(initial_state.is_empty());
         assert_eq!(datastore.ipv4_nat_current_gen(&opctx).await.unwrap(), 0);
@@ -249,7 +265,7 @@ mod test {
             datastore.ipv4_nat_create(&opctx, nat1).await.unwrap();
 
         let nat_entries =
-            datastore.ipv4_nat_list_since_gen(&opctx, 0).await.unwrap();
+            datastore.ipv4_nat_list_since_gen(&opctx, 0, 10).await.unwrap();
 
         // The NAT table has undergone one change. One entry has been added,
         // none deleted, so we should be at generation 1.
@@ -273,7 +289,7 @@ mod test {
         datastore.ipv4_nat_create(&opctx, nat2).await.unwrap();
 
         let nat_entries =
-            datastore.ipv4_nat_list_since_gen(&opctx, 0).await.unwrap();
+            datastore.ipv4_nat_list_since_gen(&opctx, 0, 10).await.unwrap();
 
         // The NAT table has undergone two changes. Two entries have been
         // added, none deleted, so we should be at generation 2.
@@ -284,7 +300,7 @@ mod test {
         // Delete the first nat entry. It should show up as a later generation number.
         datastore.ipv4_nat_delete(&opctx, &first_entry).await.unwrap();
         let nat_entries =
-            datastore.ipv4_nat_list_since_gen(&opctx, 0).await.unwrap();
+            datastore.ipv4_nat_list_since_gen(&opctx, 0, 10).await.unwrap();
 
         // The NAT table has undergone three changes. Two entries have been
         // added, one deleted, so we should be at generation 3. Since the
@@ -299,7 +315,7 @@ mod test {
         // Cancel deletion of NAT entry
         datastore.ipv4_nat_cancel_delete(&opctx, nat_entry).await.unwrap();
         let nat_entries =
-            datastore.ipv4_nat_list_since_gen(&opctx, 0).await.unwrap();
+            datastore.ipv4_nat_list_since_gen(&opctx, 0, 10).await.unwrap();
 
         // The NAT table has undergone four changes.
         let nat_entry = nat_entries.last().unwrap();
@@ -317,7 +333,7 @@ mod test {
 
         // Nothing should have changed (no records currently marked for deletion)
         let nat_entries =
-            datastore.ipv4_nat_list_since_gen(&opctx, 0).await.unwrap();
+            datastore.ipv4_nat_list_since_gen(&opctx, 0, 10).await.unwrap();
 
         assert_eq!(nat_entries.len(), 2);
         assert_eq!(datastore.ipv4_nat_current_gen(&opctx).await.unwrap(), 4);
@@ -338,7 +354,7 @@ mod test {
         // Both records should still exist (soft deleted record is newer than cutoff
         // values )
         let nat_entries =
-            datastore.ipv4_nat_list_since_gen(&opctx, 0).await.unwrap();
+            datastore.ipv4_nat_list_since_gen(&opctx, 0, 10).await.unwrap();
 
         assert_eq!(nat_entries.len(), 2);
         assert_eq!(datastore.ipv4_nat_current_gen(&opctx).await.unwrap(), 5);
@@ -348,7 +364,7 @@ mod test {
 
         // Soft deleted NAT entry should be removed from the table
         let nat_entries =
-            datastore.ipv4_nat_list_since_gen(&opctx, 0).await.unwrap();
+            datastore.ipv4_nat_list_since_gen(&opctx, 0, 10).await.unwrap();
 
         assert_eq!(nat_entries.len(), 1);
 

nexus/src/internal_api/http_entrypoints.rs

@@ -24,6 +24,8 @@ use dropshot::RequestContext;
 use dropshot::ResultsPage;
 use dropshot::TypedBody;
 use hyper::Body;
+use nexus_db_model::Ipv4NatEntryView;
+use nexus_db_model::Ipv4NatGenView;
 use nexus_types::internal_api::params::SwitchPutRequest;
 use nexus_types::internal_api::params::SwitchPutResponse;
 use nexus_types::internal_api::views::to_list;
@@ -65,6 +67,8 @@ pub fn internal_api() -> NexusApiDescription {
 
         api.register(saga_list)?;
         api.register(saga_view)?;
+        api.register(ipv4_nat_changeset)?;
+        api.register(ipv4_nat_gen)?;
 
         Ok(())
     }
@@ -484,3 +488,78 @@ async fn saga_view(
     };
     apictx.external_latencies.instrument_dropshot_handler(&rqctx, handler).await
 }
+
+// NAT RPW internal APIs
+
+/// Path parameters for NAT ChangeSet
+#[derive(Deserialize, JsonSchema)]
+struct RpwNatPathParam {
+    /// which change number to start generating
+    /// the change set from
+    from_gen: u32,
+}
+
+/// Query parameters for NAT ChangeSet
+#[derive(Deserialize, JsonSchema)]
+struct RpwNatQueryParam {
+    limit: u32,
+}
+
+/// Fetch NAT ChangeSet
+///
+/// Caller provides their generation as `from_gen`, along with a query
+/// parameter for the page size (`limit`). Endpoint will return changes
+/// that have occured since the caller's generation number up to the latest
+/// change or until the `limit` is reached. If there are no changes, an
+/// empty vec is returned.
+#[endpoint {
+   method = GET,
+    path = "/rpw/nat/ipv4/changeset/{from_gen}"
+}]
+async fn ipv4_nat_changeset(
+    rqctx: RequestContext<Arc<ServerContext>>,
+    path_params: Path<RpwNatPathParam>,
+    query_params: Query<RpwNatQueryParam>,
+) -> Result<HttpResponseOk<Vec<Ipv4NatEntryView>>, HttpError> {
+    let apictx = rqctx.context();
+    let handler = async {
+        let opctx = crate::context::op_context_for_internal_api(&rqctx).await;
+        let nexus = &apictx.nexus;
+        let path = path_params.into_inner();
+        let query = query_params.into_inner();
+        let changeset = nexus
+            .datastore()
+            .ipv4_nat_changeset(&opctx, path.from_gen, query.limit)
+            .await?;
+        Ok(HttpResponseOk(changeset))
+    };
+    apictx.external_latencies.instrument_dropshot_handler(&rqctx, handler).await
+}
+
+// TODO: Remove (maybe)
+// This endpoint may be more trouble than it's worth. The reason this endpoint
+// may be a headache is because the underlying SEQUENCE in the DB will advance,
+// even on failed transactions. This is not a big deal, but it will lead to
+// callers trying to catch up to entries that don't exist if they check this
+// endpoint first. As an alternative, we could just page through the
+// changesets until we get nothing back, then revert to periodic polling
+// of the changeset endpoint.
+
+/// Fetch NAT generation
+#[endpoint {
+   method = GET,
+    path = "/rpw/nat/ipv4/gen"
+}]
+async fn ipv4_nat_gen(
+    rqctx: RequestContext<Arc<ServerContext>>,
+) -> Result<HttpResponseOk<Ipv4NatGenView>, HttpError> {
+    let apictx = rqctx.context();
+    let handler = async {
+        let opctx = crate::context::op_context_for_internal_api(&rqctx).await;
+        let nexus = &apictx.nexus;
+        let gen = nexus.datastore().ipv4_nat_current_gen(&opctx).await?;
+        let view = Ipv4NatGenView { gen };
+        Ok(HttpResponseOk(view))
+    };
+    apictx.external_latencies.instrument_dropshot_handler(&rqctx, handler).await
+}