The neighbor list API should return paginated results #17

Author: Nieuwejaar

Cargo.lock

@@ -41,3 +41,4 @@ slog-term = "2.6"
 structopt = "0.3"
 thiserror = "1.0"
 tokio = { version = "1.38.1", features = ["full"]}
+uuid = { version = "1.10.0" }

Cargo.toml

@@ -8,6 +8,7 @@ edition = "2021"
 [dependencies]
 anyhow.workspace = true
 chrono.workspace = true
+futures.workspace = true
 lldpd-client.workspace = true
 structopt.workspace = true
 slog.workspace = true

adm/Cargo.toml

@@ -9,6 +9,7 @@ use std::net::IpAddr;
 use anyhow::Context;
 use chrono::DateTime;
 use chrono::Utc;
+use futures::stream::TryStreamExt;
 use structopt::*;
 
 use lldpd_client::default_port;
@@ -466,10 +467,20 @@ async fn main() -> anyhow::Result<()> {
                 .context("failed to remove interface"),
         },
         Commands::Neighbors => {
-            let neighbors = client.get_neighbors().await?;
-            for n in neighbors.into_inner() {
-                println!("On interface {}", n.port);
-                display_neighbor(&n);
+            for iface in client
+                .interface_list()
+                .await
+                .context("failed to get interface list")?
+                .iter()
+            {
+                let neighbors: Vec<types::Neighbor> = client
+                    .get_neighbors_stream(&iface.port, None)
+                    .try_collect()
+                    .await?;
+                for n in neighbors {
+                    println!("On interface {}", n.port);
+                    display_neighbor(&n);
+                }
             }
             Ok(())
         }

adm/src/main.rs

@@ -7,9 +7,11 @@ description = "Client library for the Oxide LLDP daemon"
 [dependencies]
 chrono.workspace = true
 common.workspace = true
+futures.workspace = true
 progenitor.workspace = true
 reqwest.workspace = true
 schemars.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 slog.workspace = true
+uuid.workspace = true

lldpd-client/Cargo.toml

@@ -21,6 +21,7 @@ slog.workspace = true
 structopt.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
+uuid.workspace = true
 
 [target.'cfg(target_os = "illumos")'.dependencies]
 dlpi = { git = "https://github.com/oxidecomputer/dlpi-sys" }

lldpd/Cargo.toml

@@ -15,14 +15,19 @@ use std::sync::Arc;
 use chrono::DateTime;
 use chrono::Utc;
 use dropshot::endpoint;
+use dropshot::EmptyScanParams;
 use dropshot::HttpError;
 use dropshot::HttpResponseCreated;
 use dropshot::HttpResponseDeleted;
 use dropshot::HttpResponseOk;
 use dropshot::HttpResponseUpdatedNoContent;
+use dropshot::PaginationParams;
 use dropshot::Path;
+use dropshot::Query;
 use dropshot::RequestContext;
+use dropshot::ResultsPage;
 use dropshot::TypedBody;
+use dropshot::WhichPage;
 use schemars::JsonSchema;
 use serde::Deserialize;
 use serde::Serialize;
@@ -681,47 +686,81 @@ async fn interface_clear_management_addr(
         .map(|_| HttpResponseDeleted())
 }
 
+/**
+ * Represents a cursor into a paginated request for the contents of the neighbor
+ * list.
+ */
+#[derive(Deserialize, Serialize, JsonSchema)]
+struct NeighborToken {
+    id: types::NeighborId,
+}
+
 /// A remote system that has been discovered on one of our configured interfaces
 #[derive(Clone, Debug, Deserialize, JsonSchema, Serialize)]
 pub struct Neighbor {
+    /// The port on which the neighbor was seen
     pub port: String,
+    /// An ID that uniquely identifies the neighbor.  Note: this ID is assigned
+    /// when we first see a neighbor we are currently tracking.  If a neighbor
+    /// goes offline long enough to be forgotten, it will be assigned a new ID
+    /// if and when it comes back online.
+    pub id: uuid::Uuid,
+    /// When was the first beacon received from this neighbor.
     pub first_seen: DateTime<Utc>,
+    /// When was the latest beacon received from this neighbor.
     pub last_seen: DateTime<Utc>,
+    /// When was the last time this neighbor's beaconed LLDPDU contents changed.
     pub last_changed: DateTime<Utc>,
+    /// Contents of the neighbor's LLDPDU beacon.
     pub system_info: types::SystemInfo,
 }
 /// Return a list of the active neighbors
 #[endpoint {
     method = GET,
-    path = "/neighbors",
+    path = "/interface/{iface}/neighbors",
 }]
 async fn get_neighbors(
     rqctx: RequestContext<Arc<Global>>,
-) -> Result<HttpResponseOk<Vec<Neighbor>>, HttpError> {
+    path: Path<InterfacePathParams>,
+    query: Query<PaginationParams<EmptyScanParams, NeighborToken>>,
+) -> Result<HttpResponseOk<ResultsPage<Neighbor>>, HttpError> {
     let global: &Global = rqctx.context();
-    Ok(HttpResponseOk(
-        global
-            .interfaces
-            .lock()
-            .unwrap()
-            .iter()
-            .flat_map(|(name, iface)| {
-                iface
-                    .lock()
-                    .unwrap()
-                    .neighbors
-                    .values()
+    let pag_params = query.into_inner();
+    let iface = path.into_inner().iface;
+    let max = rqctx.page_limit(&pag_params)?.get();
+
+    let previous = match &pag_params.page {
+        WhichPage::First(..) => None,
+        WhichPage::Next(NeighborToken { id }) => Some(id.clone()),
+    };
+
+    let neighbors: Vec<Neighbor> =
+        interfaces::get_neighbors(global, &iface, previous, max)
+            .await
+            .map_err(HttpError::from)
+            .map(|neighbors| {
+                neighbors
+                    .iter()
                     .map(|n| Neighbor {
-                        port: name.clone(),
+                        port: iface.clone(),
+                        id: n.id,
                         first_seen: n.first_seen,
                         last_seen: n.last_seen,
                         last_changed: n.last_changed,
                         system_info: (&n.lldpdu).into(),
                     })
-                    .collect::<Vec<Neighbor>>()
-            })
-            .collect::<Vec<Neighbor>>(),
-    ))
+                    .collect()
+            })?;
+
+    ResultsPage::new(neighbors, &EmptyScanParams {}, |n: &Neighbor, _| {
+        NeighborToken {
+            id: types::NeighborId {
+                chassis_id: n.system_info.chassis_id.clone(),
+                port_id: n.system_info.port_id.clone(),
+            },
+        }
+    })
+    .map(HttpResponseOk)
 }
 
 /// Return detailed build information about the `dpd` server itself.

lldpd/src/api_server.rs

@@ -8,6 +8,7 @@ use std::collections::btree_map::Entry;
 use std::collections::BTreeMap;
 use std::collections::BTreeSet;
 use std::net::IpAddr;
+use std::ops::Bound;
 use std::sync::Arc;
 use std::sync::Mutex;
 use std::time::Duration;
@@ -347,7 +348,6 @@ fn handle_packet(
             return error_accounting(iface, e);
         }
     };
-    let id = types::NeighborId::new(&lldpdu);
 
     // If one of our own LLDPDUs was reflected back to us, drop it
     if match &iface.chassis_id {
@@ -357,12 +357,13 @@ fn handle_packet(
         return;
     }
 
+    let id = types::NeighborId::new(&lldpdu);
     // Is this is new neighbor, an update from an old neighbor, or just a
     // periodic re-advertisement of the same old stuff?
     match iface.neighbors.entry(id.clone()) {
         Entry::Vacant(e) => {
             let sysinfo: types::SystemInfo = (&lldpdu).into();
-            let neighbor = types::Neighbor::from_lldpdu(&lldpdu);
+            let neighbor = types::Neighbor::from_lldpdu(&lldpdu, None);
             info!(iface.log, "new neighbor {:?}: {}", id, &sysinfo);
             e.insert(neighbor);
         }
@@ -385,7 +386,7 @@ fn handle_packet(
                 trace!(iface.log, "refresh neighbor {:?}", id);
             }
         }
-    };
+    }
 }
 
 // Construct and transmit an LLDPDU on this interface.  Return the number of
@@ -943,3 +944,28 @@ pub async fn addr_delete_all(g: &Global, name: &String) -> LldpdResult<()> {
     })
     .await
 }
+
+pub async fn get_neighbors(
+    g: &Global,
+    iface_name: &str,
+    prev: Option<types::NeighborId>,
+    max: u32,
+) -> LldpdResult<Vec<types::Neighbor>> {
+    let iface_lock = get_interface(g, iface_name)?;
+    let iface = iface_lock.lock().unwrap();
+
+    let max = usize::try_from(max)
+        .map_err(|e| LldpdError::Invalid(format!("invalid max size: {e:?}")))?;
+
+    let lower_bound = match prev {
+        Some(p) => Bound::Excluded(p),
+        None => Bound::Unbounded,
+    };
+
+    Ok(iface
+        .neighbors
+        .range((lower_bound, Bound::Unbounded))
+        .take(max)
+        .map(|(_key, value)| value.clone())
+        .collect())
+}

lldpd/src/interfaces.rs

@@ -89,6 +89,8 @@ impl From<&protocol::Lldpdu> for SystemInfo {
 /// advertisements we have received.
 #[derive(Clone, Debug)]
 pub struct Neighbor {
+    /// Uuid assigned when the neighbor was first seen
+    pub id: uuid::Uuid,
     /// When the neighbor was first seen.  Note: this is reset if the
     /// neighbor's TTL expires and we subsequently rediscover it.
     pub first_seen: DateTime<Utc>,
@@ -104,11 +106,16 @@ pub struct Neighbor {
 }
 
 impl Neighbor {
-    pub fn from_lldpdu(lldpdu: &protocol::Lldpdu) -> Self {
+    pub fn from_lldpdu(
+        lldpdu: &protocol::Lldpdu,
+        uuid: Option<uuid::Uuid>,
+    ) -> Self {
         let now = Utc::now();
 
+        let id = uuid.unwrap_or(uuid::Uuid::new_v4());
         let ttl = std::time::Duration::from_secs(lldpdu.ttl as u64);
         Neighbor {
+            id,
             first_seen: now,
             last_seen: now,
             last_changed: now,
@@ -118,7 +125,18 @@ impl Neighbor {
     }
 }
 
-#[derive(Debug, Clone, Hash, PartialEq, Eq, PartialOrd, Ord)]
+#[derive(
+    Debug,
+    Clone,
+    Deserialize,
+    JsonSchema,
+    Serialize,
+    Hash,
+    PartialEq,
+    Eq,
+    PartialOrd,
+    Ord,
+)]
 pub struct NeighborId {
     pub chassis_id: protocol::ChassisId,
     pub port_id: protocol::PortId,