diff --git a/Cargo.lock b/Cargo.lock index 1ef80514..5f266a4f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1794,9 +1794,11 @@ version = "0.1.0" dependencies = [ "chrono", "dropshot 0.17.1", + "heapless 0.9.3", "omicron-common", "oxnet", "rand 0.9.3", + "regex", "schemars 0.8.22", "serde", "thiserror 2.0.18", @@ -2749,6 +2751,17 @@ dependencies = [ "stable_deref_trait", ] +[[package]] +name = "heapless" +version = "0.9.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25ba4bd83f9415b58b4ed8dc5714c76e626a105be4646c02630ad730ad3b5aa4" +dependencies = [ + "hash32", + "serde_core", + "stable_deref_trait", +] + [[package]] name = "heck" version = "0.4.1" @@ -7108,7 +7121,7 @@ dependencies = [ "byteorder", "cfg-if", "defmt 0.3.100", - "heapless", + "heapless 0.8.0", "managed", ] diff --git a/Cargo.toml b/Cargo.toml index a8f11cfc..f915aee1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -76,6 +76,7 @@ dropshot-api-manager = "0.7.2" dropshot-api-manager-types = "0.7.2" expectorate = { version = "1.3.0", features = ["predicates"] } futures = "0.3" +heapless = { version = "0.9.3", features = [ "serde" ] } http = "1.4.2" humantime = "2.3" iddqd = "0.4.5" diff --git a/dpd-api/src/lib.rs b/dpd-api/src/lib.rs index 7441178e..8128f6a9 100644 --- a/dpd-api/src/lib.rs +++ b/dpd-api/src/lib.rs @@ -29,6 +29,7 @@ api_versions!([ // | example for the next person. // v // (next_int, IDENT), + (14, RESOURCE_TAGS), (13, ALLOW_DDM_TRAFFIC), (12, PRBS_ERROR_TRACKING), (11, WALLCLOCK_HISTORY), @@ -1041,12 +1042,48 @@ pub trait DpdApi { #[endpoint { method = GET, path = "/ports/{port_id}/links/{link_id}/ipv4", + versions = ..VERSION_RESOURCE_TAGS, + operation_id = "link_ipv4_list", }] - async fn link_ipv4_list( + async fn link_ipv4_list_v1( rqctx: RequestContext, path: Path, query: Query>, - ) -> Result>, HttpError>; + ) -> Result>, HttpError> + { + let results = Self::link_ipv4_list( + rqctx, + path.map(latest::misc::TagScope::Any), + query, + ) + .await?; + + let results = results.map(|page| ResultsPage { + items: page + .items + .into_iter() + .map(|addr| v1::port::Ipv4Entry { + tag: String::default(), + addr, + }) + .collect(), + next_page: page.next_page, + }); + + Ok(results) + } + + /// List the IPv4 addresses associated with a link. + #[endpoint { + method = GET, + path = "/ports/{port_id}/links/{link_id}/ipv4", + versions = VERSION_RESOURCE_TAGS.., + }] + async fn link_ipv4_list( + rqctx: RequestContext, + path: Path>, + query: Query>, + ) -> Result>, HttpError>; /// Add an IPv4 address to a link. #[endpoint { @@ -1657,15 +1694,39 @@ pub trait DpdApi { */ #[endpoint { method = POST, - versions = VERSION_ALLOW_DDM_TRAFFIC.., - path = "/port/{port_id}/settings" + versions = ..VERSION_ALLOW_DDM_TRAFFIC, + path = "/port/{port_id}/settings", + operation_id = "port_settings_apply", }] - async fn port_settings_apply( + async fn port_settings_apply_v1( + rqctx: RequestContext, + path: Path, + query: Query, + body: TypedBody, + ) -> Result, HttpError> { + Self::port_settings_apply_v2(rqctx, path, query, body.map(Into::into)) + .await + .map(|resp| resp.map(Into::into)) + } + + #[endpoint { + method = POST, + versions = VERSION_ALLOW_DDM_TRAFFIC..VERSION_RESOURCE_TAGS, + path = "/port/{port_id}/settings", + operation_id = "port_settings_apply", + }] + async fn port_settings_apply_v2( rqctx: RequestContext, path: Path, query: Query, body: TypedBody, - ) -> Result, HttpError>; + ) -> Result, HttpError> { + let path = path.map(|field| latest::misc::Tagged { + tag: query.into_inner().tag.unwrap_or_default(), + field, + }); + Self::port_settings_apply(rqctz, path, body).await + } /** * Apply port settings atomically. @@ -1678,20 +1739,14 @@ pub trait DpdApi { */ #[endpoint { method = POST, - versions = ..VERSION_ALLOW_DDM_TRAFFIC, - path = "/port/{port_id}/settings", - operation_id = "port_settings_apply", + versions = VERSION_RESOURCE_TAGS.., + path = "/port/{port_id}/{tag}/settings" }] - async fn port_settings_apply_v1( + async fn port_settings_apply( rqctx: RequestContext, - path: Path, - query: Query, - body: TypedBody, - ) -> Result, HttpError> { - Self::port_settings_apply(rqctx, path, query, body.map(Into::into)) - .await - .map(|resp| resp.map(Into::into)) - } + path: Path>, + body: TypedBody, + ) -> Result, HttpError>; /** * Clear port settings atomically. diff --git a/dpd-client/tests/chaos_tests/port_settings.rs b/dpd-client/tests/chaos_tests/port_settings.rs index bd781779..702406f4 100644 --- a/dpd-client/tests/chaos_tests/port_settings.rs +++ b/dpd-client/tests/chaos_tests/port_settings.rs @@ -4,6 +4,9 @@ // // Copyright 2026 Oxide Computer Company +use crate::chaos_tests::harness; +use crate::chaos_tests::util::IpRng; + use super::harness::{ expect_chaos, expect_not_found, expect_random_chaos, init_harness, new_dpd_client, run_dpd, @@ -13,7 +16,8 @@ use asic::chaos::{AsicConfig, Chaos, TableChaos}; use asic::table_chaos; use common::table::TableType; use dpd_client::types::{ - LinkCreate, LinkId, LinkSettings, PortFec, PortId, PortSettings, PortSpeed, + Ipv4Entry, Ipv6Entry, LinkCreate, LinkId, LinkSettings, PortFec, PortId, + PortSettings, PortSpeed, }; use dpd_client::{Client, ROLLBACK_FAILURE_ERROR_CODE}; use http::status::StatusCode; @@ -32,6 +36,17 @@ const TESTING_RADIX: usize = 33; const RETRY_INTERVAL: Duration = Duration::from_millis(200); const RETRY_MAX: Duration = Duration::from_secs(5); +/// A `LinkCreate` config with common defaults. +const LINK_CREATE: LinkCreate = LinkCreate { + lane: None, + autoneg: false, + kr: false, + speed: PortSpeed::Speed100G, + fec: Some(PortFec::None), + tx_eq: None, + allow_ddm_traffic: false, +}; + #[cfg(test)] mod retry { use std::future::Future; @@ -83,18 +98,7 @@ async fn test_basic_autoneg_chaos() -> anyhow::Result<()> { let (_guard, client) = init_harness("autoneg", &config); let err = client - .link_create( - &"qsfp0".parse().unwrap(), - &LinkCreate { - lane: None, - autoneg: false, - kr: false, - speed: PortSpeed::Speed100G, - fec: Some(PortFec::None), - tx_eq: None, - allow_ddm_traffic: false, - }, - ) + .link_create(&"qsfp0".parse().unwrap(), &LINK_CREATE) .await .expect_err("Expected error on create"); @@ -122,15 +126,7 @@ async fn test_port_settings_addr_fail_1() -> anyhow::Result<()> { settings.links.insert( "0".into(), LinkSettings { - params: LinkCreate { - lane: None, - autoneg: false, - kr: false, - fec: Some(PortFec::None), - speed: PortSpeed::Speed100G, - tx_eq: None, - allow_ddm_traffic: false, - }, + params: LINK_CREATE, addrs: vec!["203.0.113.47".parse().unwrap()], }, ); @@ -164,15 +160,7 @@ async fn test_port_settings_addr_success_1() -> anyhow::Result<()> { settings.links.insert( "0".into(), LinkSettings { - params: LinkCreate { - lane: None, - autoneg: false, - kr: true, - fec: Some(PortFec::None), - speed: PortSpeed::Speed100G, - tx_eq: None, - allow_ddm_traffic: false, - }, + params: LinkCreate { kr: true, ..LINK_CREATE }, addrs: vec!["203.0.113.47".parse().unwrap()], }, ); @@ -204,15 +192,7 @@ async fn test_port_settings_addr_success_multi() -> anyhow::Result<()> { settings.links.insert( "0".into(), LinkSettings { - params: LinkCreate { - lane: None, - autoneg: false, - kr: true, - fec: Some(PortFec::None), - speed: PortSpeed::Speed100G, - tx_eq: None, - allow_ddm_traffic: false, - }, + params: LinkCreate { kr: true, ..LINK_CREATE }, addrs: vec!["203.0.113.47".parse().unwrap()], }, ); @@ -234,15 +214,7 @@ async fn test_port_settings_addr_success_multi() -> anyhow::Result<()> { settings.links.insert( "0".into(), LinkSettings { - params: LinkCreate { - lane: None, - autoneg: false, - kr: true, - fec: Some(PortFec::None), - speed: PortSpeed::Speed100G, - tx_eq: None, - allow_ddm_traffic: false, - }, + params: LinkCreate { kr: true, ..LINK_CREATE }, addrs: vec![ "203.0.113.46".parse().unwrap(), "203.0.113.48".parse().unwrap(), @@ -275,15 +247,7 @@ async fn test_port_settings_addr_success_multi() -> anyhow::Result<()> { settings.links.insert( "0".into(), LinkSettings { - params: LinkCreate { - lane: None, - autoneg: false, - kr: true, - fec: Some(PortFec::None), - speed: PortSpeed::Speed100G, - tx_eq: None, - allow_ddm_traffic: false, - }, + params: LinkCreate { kr: true, ..LINK_CREATE }, addrs: vec![ "203.0.113.47".parse().unwrap(), "fd00:1701::d".parse().unwrap(), @@ -552,3 +516,180 @@ fn random_port_settings() -> PortSettings { )]), } } + +const TAG1: &str = "chaos1"; +const TAG2: &str = "chaos2"; + +/// Validates address registration namespaces for the following sequence: +/// +/// - Create link +/// - Add some addresses to the link manually under tag1 +/// - Port settings apply some different addresses under tag2 +/// - All addresses should be on the link +/// - Port settings apply away the tag2 addresses +/// - The tag1 addresses should still be on the link +#[ignore] +#[tokio::test] +async fn addr_ns_persistent_create() -> anyhow::Result<()> { + let no_failures = AsicConfig::uniform_set(TESTING_RADIX, 0.); + let (_guard, client) = + harness::init_harness("addr_ns_persistent_create", &no_failures); + + let mut rng = IpRng::new(12345); + + let tag1_v4 = Ipv4Entry { addr: rng.unique_ipv4(), tag: TAG1.to_string() }; + let tag1_v6 = Ipv6Entry { addr: rng.unique_ipv6(), tag: TAG1.to_string() }; + let tag2_v4 = Ipv4Entry { addr: rng.unique_ipv4(), tag: TAG2.to_string() }; + let tag2_v6 = Ipv6Entry { addr: rng.unique_ipv6(), tag: TAG2.to_string() }; + + let port_id: PortId = "qsfp0".parse()?; + + let link_id = + client.link_create(&port_id, &LINK_CREATE).await?.into_inner(); + + client.link_ipv4_create(&port_id, &link_id, &tag1_v4).await?; + client.link_ipv6_create(&port_id, &link_id, &tag1_v6).await?; + + client + .port_settings_apply( + &port_id, + Some(TAG2), + &PortSettings { + links: HashMap::from([( + link_id.to_string(), + LinkSettings { + params: LINK_CREATE, + addrs: vec![tag2_v4.addr.into(), tag2_v6.addr.into()], + }, + )]), + }, + ) + .await?; + + let v4_addrs = client + .link_ipv4_list(&port_id, &link_id, None, None) + .await? + .into_inner(); + + let v6_addrs = client + .link_ipv6_list(&port_id, &link_id, None, None) + .await? + .into_inner(); + + assert_eq!(v4_addrs.items.len(), 2); + assert!(v4_addrs.items.contains(&tag1_v4)); + assert!(v4_addrs.items.contains(&tag2_v4)); + + assert_eq!(v6_addrs.items.len(), 2); + assert!(v6_addrs.items.contains(&tag1_v6)); + assert!(v6_addrs.items.contains(&tag2_v6)); + + client + .port_settings_apply( + &port_id, + Some(TAG2), + &PortSettings { + links: HashMap::from([( + link_id.to_string(), + LinkSettings { params: LINK_CREATE, addrs: Vec::new() }, + )]), + }, + ) + .await?; + + let v4_addrs = client + .link_ipv4_list(&port_id, &link_id, None, None) + .await? + .into_inner(); + + let v6_addrs = client + .link_ipv6_list(&port_id, &link_id, None, None) + .await? + .into_inner(); + + assert_eq!(&v4_addrs.items, &[tag1_v4]); + assert_eq!(&v6_addrs.items, &[tag1_v6]); + + Ok(()) +} + +/// Validates address registration namespaces for the following sequence: +/// +/// - Declare link via port settings apply with ome addresses under tag2 +/// - Manually add THE SAME addresses manually under tag1 +/// - Link get should yield the addresses +/// - Manually delete the addresses under tag1 +/// - TODO::cory: deletion isn't tagged yet, so this will obviously fail. +/// - The settings apply addresses under tag2 should still exist +#[test] +async fn addr_ns_spot_delete() -> anyhow::Result<()> { + let no_failures = AsicConfig::uniform_set(TESTING_RADIX, 0.); + let (_guard, client) = + harness::init_harness("addr_ns_spot_delete", &no_failures); + + let mut rng = IpRng::new(54321); + + let tag1_v4 = Ipv4Entry { addr: rng.unique_ipv4(), tag: TAG1.to_string() }; + let tag1_v6 = Ipv6Entry { addr: rng.unique_ipv6(), tag: TAG1.to_string() }; + let tag2_v4 = Ipv4Entry { addr: tag1_v4.addr, tag: TAG2.to_string() }; + let tag2_v6 = Ipv6Entry { addr: tag1_v6.addr, tag: TAG2.to_string() }; + + let port_id: PortId = "qsfp0".parse()?; + let link_id = LinkId(0); + + client + .port_settings_apply( + &port_id, + Some(TAG2), + &PortSettings { + links: HashMap::from([( + link_id.to_string(), + LinkSettings { + params: LINK_CREATE, + addrs: vec![tag2_v4.addr.into(), tag2_v6.addr.into()], + }, + )]), + }, + ) + .await?; + + client.link_ipv4_create(&port_id, &link_id, &tag1_v4).await?; + client.link_ipv6_create(&port_id, &link_id, &tag1_v6).await?; + + let v4_addrs = client + .link_ipv4_list(&port_id, &link_id, None, None) + .await? + .into_inner(); + + let v6_addrs = client + .link_ipv6_list(&port_id, &link_id, None, None) + .await? + .into_inner(); + + // Two addresses * two tags + assert_eq!(v4_addrs.items.len(), 2); + assert!(v4_addrs.items.contains(&tag1_v4)); + assert!(v4_addrs.items.contains(&tag2_v4)); + + assert_eq!(v6_addrs.items.len(), 2); + assert!(v6_addrs.items.contains(&tag1_v6)); + assert!(v6_addrs.items.contains(&tag2_v6)); + + client.link_ipv4_delete(&port_id, &link_id, &tag1_v4.addr)?; + client.link_ipv6_delete(&port_id, &link_id, &tag1_v6.addr)?; + + let v4_addrs = client + .link_ipv4_list(&port_id, &link_id, None, None) + .await? + .into_inner(); + + let v6_addrs = client + .link_ipv6_list(&port_id, &link_id, None, None) + .await? + .into_inner(); + + assert_eq!(&v4_addrs.items, &[tag2_v4]); + assert_eq!(&v6_addrs.items, &[tag2_v6]); + + Ok(()) +} diff --git a/dpd-client/tests/chaos_tests/util.rs b/dpd-client/tests/chaos_tests/util.rs index 22bc5c91..baf060cf 100644 --- a/dpd-client/tests/chaos_tests/util.rs +++ b/dpd-client/tests/chaos_tests/util.rs @@ -4,9 +4,14 @@ // // Copyright 2025 Oxide Computer Company +use std::collections::HashSet; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; + use dpd_client::Client; use dpd_client::types::{Ipv4Entry, Ipv6Entry}; use futures::TryStreamExt; +use rand::rngs::StdRng; +use rand::{Rng, SeedableRng}; pub(crate) async fn link_list_ipv4( client: &Client, @@ -37,3 +42,64 @@ pub(crate) async fn link_list_ipv6( .try_collect::>() .await } + +/// A random IP address generator. +pub struct IpRng { + rng: StdRng, + claimed: HashSet, +} + +impl IpRng { + /// Creates a new ip address generator from the given seed. + pub fn new(seed: u64) -> Self { + Self { rng: StdRng::seed_from_u64(seed), claimed: HashSet::default() } + } + + /// Returns a random IPv4 address that this instance + /// has never created before. + pub fn unique_ipv4(&mut self) -> Ipv4Addr { + Self::roll_unique(&mut self.claimed, || { + Ipv4Addr::from_bits(self.rng.random()) + }) + } + + /// Returns a random IPv6 address that this instance + /// has never created before. + pub fn unique_ipv6(&mut self) -> Ipv6Addr { + Self::roll_unique(&mut self.claimed, || { + Ipv6Addr::from_bits(self.rng.random()) + }) + } + + fn roll_unique( + tracker: &mut HashSet, + mut random_addr: impl FnMut() -> T, + ) -> T + where + T: Into + Copy, + { + loop { + // Executes infinitely if we've already generated the entire + // IPv4 or IPv6 address space, in which case the offending test + // has earned a more bespoke solution :) + let addr = random_addr(); + if tracker.insert(addr.into()) { + return addr; + } + } + } +} + +#[cfg(test)] +mod util_tests { + use crate::chaos_tests::util::IpRng; + + /// Basic check on unique ipv6 address generation. + #[test] + fn unique_v6() { + let mut rng = IpRng::new(7); + let one = rng.unique_ipv6(); + let two = rng.unique_ipv6(); + assert_ne!(one, two); + } +} diff --git a/dpd-types/versions/Cargo.toml b/dpd-types/versions/Cargo.toml index 70dcaeec..7fd65f96 100644 --- a/dpd-types/versions/Cargo.toml +++ b/dpd-types/versions/Cargo.toml @@ -6,9 +6,11 @@ edition = "2024" [dependencies] chrono.workspace = true dropshot.workspace = true +heapless.workspace = true omicron-common.workspace = true oxnet.workspace = true rand.workspace = true +regex.workspace = true schemars.workspace = true serde.workspace = true thiserror.workspace = true diff --git a/dpd-types/versions/src/latest.rs b/dpd-types/versions/src/latest.rs index 3b56eb39..c2f02296 100644 --- a/dpd-types/versions/src/latest.rs +++ b/dpd-types/versions/src/latest.rs @@ -95,7 +95,9 @@ pub mod mcast { pub mod misc { pub use crate::v1::misc::BuildInfo; - pub use crate::v1::misc::TagPath; + pub use crate::v14::Tag; + pub use crate::v14::TagScope; + pub use crate::v14::Tagged; } pub mod nat { @@ -129,8 +131,6 @@ pub mod oxstats { pub mod port { pub use crate::v1::port::FreeChannels; pub use crate::v1::port::InternalPort; - pub use crate::v1::port::Ipv4Entry; - pub use crate::v1::port::Ipv6Entry; pub use crate::v1::port::PORT_COUNT_INTERNAL; pub use crate::v1::port::PORT_COUNT_QSFP; pub use crate::v1::port::PORT_COUNT_REAR; diff --git a/dpd-types/versions/src/lib.rs b/dpd-types/versions/src/lib.rs index 617a58c7..fa9775e3 100644 --- a/dpd-types/versions/src/lib.rs +++ b/dpd-types/versions/src/lib.rs @@ -43,6 +43,8 @@ pub mod v11; pub mod v12; #[path = "allow_ddm_traffic/mod.rs"] pub mod v13; +#[path = "resource_tags/mod.rs"] +pub mod v14; #[path = "attached_subnets/mod.rs"] pub mod v3; #[path = "v4_over_v6_routes/mod.rs"] diff --git a/dpd-types/versions/src/resource_tags/mod.rs b/dpd-types/versions/src/resource_tags/mod.rs new file mode 100644 index 00000000..a35fcb8d --- /dev/null +++ b/dpd-types/versions/src/resource_tags/mod.rs @@ -0,0 +1,168 @@ +use std::{ + fmt::Display, + net::{Ipv4Addr, Ipv6Addr}, + str::FromStr, + sync::LazyLock, +}; + +use regex::Regex; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +use crate::v1::{link::LinkId, port::PortId}; + +/// The maximum length in bytes of a tag. Since tags are ascii, this +/// is also the maximum length in characters. +/// +/// This was informed by the pre-existing CRDB limit for dpd +/// multicast groups in omicron: +/// +/// +const TAG_CAPACITY: usize = 63; + +/// Defines the acceptable character range of a tag. +/// +/// This particular pattern mirrors that already used for multicast +/// resource tags. +static TAG_PATTERN: LazyLock = + LazyLock::new(|| Regex::new(r"^[a-zA-Z0-9_.:-]+$").unwrap()); + +#[derive(thiserror::Error, Debug)] +pub enum Error { + #[error( + "Input text {given:?} failed to match the validation regex {pattern:?}" + )] + Pattern { given: String, pattern: String }, + + #[error( + "Tag length must be in the range {range:?}. Found {0}.", range = 1..=TAG_CAPACITY + )] + Size(usize), +} + +/// An ID for namespacing network resources. Tags allow +/// different parties to CRUD resources without affecting each other. +/// +/// Tags are an internal mechanism for categorization. They don't +/// enforce authentication, and most will probably be hardcoded strings. +// +// Implementation notes: +// +// `heapless::String` is annoying for a variety of reasons. +// +// 1. It doesn't implement `Copy`, which is conceptually the whole +// point of a stack allocated type. +// 2. It doesn't optimize size. For example, a 63 byte array on +// a 64-bit machine should take up 64 bytes in total. Wasting a full +// usize when we need no more than a u6 is extravagant dissipation imho. +// 3. It doesn't expose any non-trivial const constructor, which feels like +// a missed opportunity. +// +// But a bespoke stack string isn't warranted scope for this initial PR, +// so that's an optimization for another day. And it can live behind +// this API anyway. +#[repr(transparent)] +#[derive( + serde::Serialize, serde::Deserialize, Debug, PartialEq, Eq, Clone, Hash, +)] +pub struct Tag(heapless::String); +const _: () = + assert!(std::mem::size_of::() == 72, "this could be 64 if we tried"); + +// TODO::cory: tests are warranted + +impl Tag { + pub fn as_str(&self) -> &str { + self.as_ref() + } + + pub fn coerce(tag: &str) -> Self { + const FILLER: &str = "_"; + + let mut good = heapless::String::::new(); + for ch in tag.chars() { + let mut buf = [0u8; 4]; + let mut ch = &*ch.encode_utf8(&mut buf); + + if !TAG_PATTERN.is_match(ch) { + ch = FILLER; + } + + if good.push_str(ch).is_err() { + break; + } + } + + if good.is_empty() { + good.push_str(FILLER) + .expect("Empty buf must have space for a char"); + } + + good.as_str().parse().expect("coerced tag should always be valid") + } +} + +impl FromStr for Tag { + type Err = self::Error; + + fn from_str(s: &str) -> Result { + if !TAG_PATTERN.is_match(s) { + return Err(self::Error::Pattern { + given: s.to_string(), + pattern: TAG_PATTERN.to_string(), + }); + } + + heapless::String::from_str(s) + .map_or_else(|_| Err(self::Error::Size(s.len())), |s| Ok(Self(s))) + } +} + +impl JsonSchema for Tag { + fn schema_name() -> String { + String::from(stringify!(Tag)) + } + + fn json_schema( + generator: &mut schemars::r#gen::SchemaGenerator, + ) -> schemars::schema::Schema { + let mut schema = String::json_schema(generator); + let schemars::schema::Schema::Object(object) = &mut schema else { + unreachable!(); + }; + object.metadata().description.replace(String::from( + "A text label for namespacing network resources and actions", + )); + object.string().min_length = Some(1); + object.string().max_length = Some(TAG_CAPACITY as u32); + object.string().pattern = Some(TAG_PATTERN.to_string()); + schema + } +} + +impl AsRef for Tag { + fn as_ref(&self) -> &str { + self.0.as_str() + } +} + +impl Display for Tag { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.as_ref().fmt(f) + } +} + +/// A command or request scoped to a specific owner tag. +pub struct Tagged { + pub tag: Tag, + pub field: T, +} + +/// Defines the scope of a command or query on tagged assets. +pub enum TagScope { + /// Applies to all resources regardless of tag. + Any(T), + + /// Applies only to this specific tag. + Single(Tagged), +} diff --git a/dpd/src/api_server.rs b/dpd/src/api_server.rs index 5e108a04..b9aeed43 100644 --- a/dpd/src/api_server.rs +++ b/dpd/src/api_server.rs @@ -40,7 +40,8 @@ use dpd_types::mcast::{ MulticastGroupUpdateUnderlayEntry, MulticastTagPath, MulticastUnderlayGroupIpParam, }; -use dpd_types::misc::{BuildInfo, TagPath}; +use dpd_types::misc::BuildInfo; +use dpd_types::misc::Tag; use dpd_types::nat::{ NatIpv4Path, NatIpv4PortPath, NatIpv4RangePath, NatIpv6Path, NatIpv6PortPath, NatIpv6RangePath, NatToken, @@ -1128,14 +1129,12 @@ impl DpdApi for DpdApiImpl { WhichPage::First(..) => None, WhichPage::Next(Ipv4Token { ip }) => Some(*ip), }; - let entries = - switch.list_ipv4_addresses(port_id, link_id, addr, limit)?; - ResultsPage::new( - entries, - &EmptyScanParams {}, - |entry: &Ipv4Entry, _| Ipv4Token { ip: entry.addr }, - ) - .map(HttpResponseOk) + let entries = switch + .list_ipv4_addresses(port_id, link_id, addr, limit)? + .copied() + .collect::>(); + ResultsPage::new(entries, &EmptyScanParams {}, |ip, _| Ipv4Token { ip }) + .map(HttpResponseOk) } async fn link_ipv4_create( @@ -1698,19 +1697,19 @@ impl DpdApi for DpdApiImpl { async fn reset_all_tagged( rqctx: RequestContext>, - path: Path, + path: Path, ) -> Result { let switch: &Switch = rqctx.context(); - let tag = path.into_inner().tag; + let tag = path.into_inner(); - debug!(switch.log, "resetting settings tagged with {}", tag); + debug!(switch.log, "resetting settings tagged with {tag}"); - arp::reset_ipv4_tag(switch, &tag); - arp::reset_ipv6_tag(switch, &tag); - route::reset_ipv4_tag(switch, &tag).await; - route::reset_ipv6_tag(switch, &tag).await; + arp::reset_ipv4_tag(switch, tag.as_str()); + arp::reset_ipv6_tag(switch, tag.as_str()); + route::reset_ipv4_tag(switch, tag.as_str()).await; + route::reset_ipv6_tag(switch, tag.as_str()).await; switch - .clear_link_addresses(Some(&tag)) + .clear_tagged_addrs(&tag) .map(|_| HttpResponseUpdatedNoContent()) .map_err(|e| e.into()) } diff --git a/dpd/src/link.rs b/dpd/src/link.rs index ae60baf4..e7cba80f 100644 --- a/dpd/src/link.rs +++ b/dpd/src/link.rs @@ -43,6 +43,7 @@ use dpd_types::link::LinkUpCounter; use dpd_types::link::LinkView; use dpd_types::link::TfportData; use dpd_types::serdes::Ber; +use dpd_types_versions::v14::Tag; use slog::debug; use slog::error; use slog::info; @@ -50,9 +51,13 @@ use slog::o; use slog::warn; use std::collections::BTreeMap; use std::collections::BTreeSet; +use std::collections::HashMap; +use std::collections::btree_map; use std::collections::btree_map::Entry; use std::net::Ipv4Addr; use std::net::Ipv6Addr; +use std::ops::Bound; +use std::sync; use std::sync::Arc; use std::sync::Mutex; use std::time::Duration; @@ -236,10 +241,6 @@ pub struct Link { pub link_state: LinkState, /// The kind of media in the link. pub media: PortMedia, - /// A list of IPv4 addresses assigned to this link. - pub ipv4: BTreeSet, - /// A list of IPv6 addresses assigned to this link. - pub ipv6: BTreeSet, /// Tracks the history of linkup/linkdown transitions, allowing us to /// detect flapping links. pub linkup_tracker: LinkUpTracker, @@ -247,6 +248,9 @@ pub struct Link { /// finite state machine, so we can detect and diagnose linkup failures. pub autoneg_tracker: AutonegTracker, + /// Resources whose CRUD is namespaced across different owner tags. + pub tagged: TaggedConfigs, + /// The configuration of the link as requested by the user / sled-agent pub config: LinkConfig, @@ -302,6 +306,315 @@ impl From for TfportData { } } +/// TODO::cory docs +/// +/// INVARIANT: If a key is in the strong map, then it's in the weak map. +/// +/// Implied: If a key is in the weak map, then its ref count is nonzero. +/// +/// This may be temporarily violated during the execution of +/// methods, but it must be true by the end of every pub fn. +#[derive(Debug)] +pub struct WeakMap { + map: BTreeMap>, +} + +impl Default for WeakMap { + fn default() -> Self { + Self { map: BTreeMap::new() } + } +} + +impl WeakMap +where + K: Ord + Clone, +{ + const INVARIANT: &str = "INVARIANT: if an entry was in the strong map, then it was in the weak map."; + + /// Adds this key to the strong and weak maps. + /// + /// Returns true if the strong map previously did not have + /// this key. + /// + /// Equivalently, returns false if this key was already + /// in the strong map. + fn insert(&mut self, key: K, strong: &mut BTreeMap>) -> bool { + let ct = match self.map.entry(key.clone()) { + btree_map::Entry::Vacant(slot) => { + let ct = Arc::new(()); + slot.insert(Arc::downgrade(&ct)); + ct + } + btree_map::Entry::Occupied(full) => full + .get() + .upgrade() + .expect("INVARIANT: ref ct after edit must always be nonzero"), + }; + + strong.insert(key, ct).is_none() + } + + /// Removes the key if it previously existed in the strong collection. + /// + /// Returns + /// - None if the key did not exist in the strong collection. + /// - Some(ct) where ct is the new reference count. If ct == 0, then + /// this entry was dropped from the weak map. + fn delete( + &mut self, + key: &K, + strong: &mut BTreeMap>, + ) -> Option { + let Some(ct) = strong.remove(&key) else { + return None; + }; + drop(ct); + + Some(self.check_delete(key).expect(Self::INVARIANT)) + } + + fn check_delete(&mut self, key: &K) -> Option { + let entry = self.map.get(key)?; + + let refs = entry.strong_count(); + if refs == 0 { + self.map.remove(key); + } + + Some(refs) + } + + fn drain_strong( + &mut self, + strong: &mut BTreeMap>, + ) -> impl Iterator { + std::mem::take(strong).into_iter().filter_map(|(key, ct)| { + drop(ct); + if self.check_delete(&key).expect(Self::INVARIANT) == 0 { + return Some(key); + } + None + }) + } + + // TODO::cory: scary + fn drain_all(&mut self) -> impl Iterator { + std::mem::take(&mut self.map).into_keys() + } +} + +/// In some cases, dpd is managed by multiple mutually-unaware +/// controllers. These controllers provide a namespace [`Tag`] +/// that restricts sensitive operations to only those resources +/// with the same tag. +/// +/// Only the resources within this collection respect tag isolation. +/// +/// TODO::cory docs +#[derive(Debug, Default)] +pub struct TaggedConfigs { + configs: HashMap, + ipv4_all: WeakMap, + ipv6_all: WeakMap, +} + +impl TaggedConfigs { + pub fn ipv4(&self) -> impl Iterator { + self.ipv4_all.map.keys() + } + + pub fn ipv6(&self) -> impl Iterator { + self.ipv6_all.map.keys() + } + + pub fn ipv4_range( + &self, + start: Bound, + end: Bound, + ) -> impl Iterator { + self.ipv4_all.map.range((start, end)).map(|(key, _value)| key) + } + + pub fn contains_tagged_ipv4(&mut self, tag: &Tag, addr: &Ipv4Addr) -> bool { + self.configs.get(tag).is_some_and(|conf| conf.ipv4.contains_key(addr)) + } + + pub fn insert_ipv4(&mut self, tag: Tag, addr: Ipv4Addr) -> bool { + let strong = self.configs.entry(tag).or_default(); + self.ipv4_all.insert(addr, &mut strong.ipv4) + } + + pub fn drain_ipv4_tag( + &mut self, + tag: &Tag, + ) -> Option> { + let strong = self.configs.get_mut(tag)?; + Some(self.ipv4_all.drain_strong(&mut strong.ipv4)) + } + + pub fn drain_ipv6_tag( + &mut self, + tag: &Tag, + ) -> Option> { + let strong = self.configs.get_mut(tag)?; + Some(self.ipv6_all.drain_strong(&mut strong.ipv6)) + } + + pub fn drain_ipv4(&mut self) -> impl Iterator { + for conf in self.configs.values_mut() { + std::mem::take(&mut conf.ipv4); + } + + self.ipv4_all.drain_all() + } + + pub fn drain_ipv6(&mut self) -> impl Iterator { + for conf in self.configs.values_mut() { + std::mem::take(&mut conf.ipv6); + } + + self.ipv6_all.drain_all() + } + + // pub fn insert_owned_v4( + // &mut self, + // tag: &Tag, + // addr: Ipv4Addr, + // ) -> DpdResult<()> { + // if self.configs.get(tag).is_some_and(|entry| entry.ipv4.contains(&addr)) + // { + // return Ok(()); + // } + + // if self.combined.ipv4.contains(&addr) { + // return Err(DpdError::Exists(format!( + // "Cannot insert owned IP {addr:?} for {tag} because it's already in use by another tag." + // ))); + // } + + // self.configs.entry(tag.clone()).or_default().ipv4.insert(addr); + // self.recombine(); + + // Ok(()) + // } + + // pub fn insert_ipv4(&mut self, tag: Tag, addr: Ipv4Addr) { + // self.configs.entry(tag).or_default().ipv4.insert(addr, ct); + // } + + // fn check_delete_v4(&mut self, addr: &Ipv4Addr) {} + + // pub fn delete_tag(&mut self, tag: &Tag) -> Option { + // let conf = self.configs.remove(tag); + // self.recombine(); + // conf + // } + + // pub fn drain_v4(&mut self) -> impl Iterator { + // for set in self.configs.values_mut() { + // set.ipv4.clear(); + // } + // self.combined.ipv4.drain() + // } + + // pub fn drain_v6(&mut self) -> impl Iterator { + // for set in self.configs.values_mut() { + // set.ipv6.clear(); + // } + // self.combined.ipv6.drain() + // } + + // pub fn delete_v4(&mut self, tag: &Tag, addr: &Ipv4Addr) { + // if let Some(conf) = self.configs.get_mut(tag) { + // conf.ipv4.remove(addr); + // } + // self.recombine(); + // } + + // pub fn delete_v6(&mut self, tag: &Tag, addr: &Ipv6Addr) { + // if let Some(conf) = self.configs.get_mut(tag) { + // conf.ipv6.remove(addr); + // } + // self.recombine(); + // } + + // pub fn tag_ipv4(&self, tag: &Tag) -> Option<&HashSet> { + // self.configs.get(tag).map(|conf| &conf.ipv4) + // } + + // pub fn ipv4(&self) -> &HashSet { + // &self.combined.ipv4 + // } + + // pub fn ipv6(&self) -> &HashSet { + // &self.combined.ipv6 + // } + + // fn recombine(&mut self) { + // self.combined.ipv4.clear(); + // self.combined.ipv4.extend( + // self.configs.values().flat_map(|conf| conf.ipv4.iter().copied()), + // ); + + // self.combined.ipv6.clear(); + // self.combined.ipv6.extend( + // self.configs.values().flat_map(|conf| conf.ipv6.iter().copied()), + // ); + // } +} + +enum WeakDrain<'a, K> +where + K: Ord + Clone, +{ + Full { strong: &'a mut BTreeMap>, weak: &'a mut WeakMap }, + Empty, +} + +impl<'a, K> Iterator for WeakDrain<'a, K> +where + K: Ord + Clone, +{ + type Item = K; + + fn next(&mut self) -> Option { + let (strong, weak) = match self { + Self::Full { strong, weak } => (strong, weak), + Self::Empty => return None, + }; + + let key = strong.first_key_value().map(|(addr, _)| addr)?.clone(); + weak.delete(&key, strong).expect("Key is guaranteed to exist"); + + Some(key) + } + + fn size_hint(&self) -> (usize, Option) { + match self { + Self::Full { strong, .. } => (strong.len(), Some(strong.len())), + Self::Empty => (0, Some(0)), + } + } +} + +impl<'a, K> Drop for WeakDrain<'a, K> +where + K: Ord + Clone, +{ + fn drop(&mut self) { + while self.next().is_some() {} + } +} + +#[derive(Debug, Default)] +struct TaggedConfig { + /// Registered IPv4 addresses for this link. + ipv4: BTreeMap>, + + /// Registered IPv6 addresses for this link. + ipv6: BTreeMap>, +} + // This struct represents the configuration of the link requested by the // user/sled-agent #[derive(Debug)] @@ -475,8 +788,7 @@ impl Link { fsm_state: asic::PortFsmState::default(), link_state: LinkState::Unknown, media: PortMedia::None, - ipv4: BTreeSet::new(), - ipv6: BTreeSet::new(), + tagged: TaggedConfigs::default(), linkup_tracker: LinkUpTracker::default(), autoneg_tracker: AutonegTracker::default(), @@ -486,11 +798,10 @@ impl Link { } /// Return the link-local address for this link, if one has been added. + /// + /// If multiple have been added, this returns the first that is found. pub fn link_local(&self) -> Option { - self.ipv6 - .iter() - .find(|entry| (entry.addr.segments()[0] & 0xffc0) == 0xfe80) - .map(|entry| entry.addr) + self.tagged.ipv6().copied().find(Ipv6Addr::is_unicast_link_local) } /// Return the FEC scheme in use for this link. If the link has not yet @@ -710,18 +1021,18 @@ impl Switch { let mut link = link_lock.lock().unwrap(); // Delete all addresses in the switch tables for this link. - if !link.ipv4.is_empty() { - let to_delete = std::mem::take(&mut link.ipv4) - .into_iter() - .map(|entry| entry.addr); - port_ip::ipv4_delete_many(self, link.asic_port_id, to_delete)?; - } - if !link.ipv6.is_empty() { - let to_delete = std::mem::take(&mut link.ipv6) - .into_iter() - .map(|entry| entry.addr); - port_ip::ipv6_delete_many(self, link.asic_port_id, to_delete)?; - } + + port_ip::ipv4_delete_many( + self, + link.asic_port_id, + link.tagged.drain_ipv4(), + )?; + + port_ip::ipv6_delete_many( + self, + link.asic_port_id, + link.tagged.drain_ipv6(), + )?; // Notify the reconciliation task that this link's ASIC resources need // to be released. @@ -736,78 +1047,39 @@ impl Switch { let links = self.links.lock().unwrap(); for link_lock in links.0.values() { let mut link = link_lock.lock().unwrap(); - // Clear all IP addresses. - // - // Swap out an empty map with the existing one, so that we can - // retain an iterable for calling `ipv{4,6}_delete_many`. - if !link.ipv4.is_empty() { - let to_delete = std::mem::take(&mut link.ipv4) - .into_iter() - .map(|entry| entry.addr); - port_ip::ipv4_delete_many(self, link.asic_port_id, to_delete)?; - } - if !link.ipv6.is_empty() { - let to_delete = std::mem::take(&mut link.ipv6) - .into_iter() - .map(|entry| entry.addr); - port_ip::ipv6_delete_many(self, link.asic_port_id, to_delete)?; - } + + port_ip::ipv4_delete_many( + self, + link.asic_port_id, + link.tagged.drain_ipv4(), + )?; + + port_ip::ipv6_delete_many( + self, + link.asic_port_id, + link.tagged.drain_ipv6(), + )?; } Ok(()) } - /// Clear any IP addresses associated with all links, optionally restricted - /// to a specified string `tag`. - pub fn clear_link_addresses(&self, tag: Option<&str>) -> DpdResult<()> { - if let Some(tag) = tag { - let links = self.links.lock().unwrap(); - for link_lock in links.0.values() { - let mut link = link_lock.lock().unwrap(); - self.clear_link_addresses_locked(&mut link, tag); - } - Ok(()) - } else { - self.clear_link_state() - } - } + /// Clears all IP addresses associated only with the given + /// tag from all links. + pub fn clear_tagged_addrs(&self, tag: &Tag) -> DpdResult<()> { + for link_lock in self.links.lock().unwrap().0.values() { + let mut link = link_lock.lock().unwrap(); + let port_id = link.asic_port_id; - fn clear_link_addresses_locked(&self, link: &mut Link, tag: &str) { - // Remove all entries from the set with the provided tag. - // - // TODO-cleanup: It'd be nice to use `drain_filter` here, - // but that is unstable. - let mut to_remove = Vec::new(); - link.ipv4.retain(|entry| { - if entry.tag == tag { - to_remove.push(entry.addr); - false - } else { - true + if let Some(tagged) = link.tagged.drain_ipv4_tag(tag) { + let _ = port_ip::ipv4_delete_many(self, port_id, tagged); } - }); - // Delete the entries from the ASIC tables. - let _ = port_ip::ipv4_delete_many( - self, - link.asic_port_id, - to_remove.into_iter(), - ); - - // TODO-cleanup: See note above about `drain_filter`. - let mut to_remove = Vec::new(); - link.ipv6.retain(|entry| { - if entry.tag == tag { - to_remove.push(entry.addr); - false - } else { - true + if let Some(tagged) = link.tagged.drain_ipv6_tag(tag) { + let _ = port_ip::ipv6_delete_many(self, port_id, tagged); } - }); - let _ = port_ip::ipv6_delete_many( - self, - link.asic_port_id, - to_remove.into_iter(), - ); + } + + Ok(()) } // Update the state of a link with a closure. @@ -1049,16 +1321,16 @@ impl Switch { link: &mut Link, entry: Ipv4Entry, ) -> DpdResult<()> { - if link.ipv4.contains(&entry) { - Err(DpdError::Exists(format!( - "IP address {} already exists", - entry.addr - ))) - } else { - port_ip::ipv4_add(self, link.asic_port_id, entry.addr)?; - link.ipv4.insert(entry); - Ok(()) + if !link.tagged.contains_tagged_ipv4(&entry.tag, entry.addr) { + return Err(DpdError::Exists(format!( + "Tagged address already exists: {entry:?}" + ))); } + + port_ip::ipv4_add(self, link.asic_port_id, entry.addr)?; + link.tagged.insert_ipv4(entry.tag, entry.addr); + + Ok(()) } /// Add an IPv4 address to the specified link. @@ -1080,21 +1352,11 @@ impl Switch { link_id: LinkId, last_address: Option, limit: usize, - ) -> DpdResult> { + ) -> DpdResult> { self.link_fetch(port_id, link_id, |link| { - if let Some(addr) = last_address { - // Equality only considers the address, so create an entry - // with an empty tag. - use std::ops::Bound; - let entry = Ipv4Entry { tag: String::new(), addr }; - link.ipv4 - .range((Bound::Excluded(entry), Bound::Unbounded)) - .take(limit) - .cloned() - .collect() - } else { - link.ipv4.iter().take(limit).cloned().collect() - } + let left_bound = + last_address.map_or_else(|| Bound::Unbounded, Bound::Excluded); + Ok(link.tagged.ipv4_range(left_bound, Bound::Unbounded).take(limit)) }) } diff --git a/dpd/src/loopback.rs b/dpd/src/loopback.rs index f0a4665c..31437124 100644 --- a/dpd/src/loopback.rs +++ b/dpd/src/loopback.rs @@ -49,6 +49,7 @@ pub fn add_loopback_ipv4(switch: &Switch, addr: &Ipv4Entry) -> DpdResult<()> { /// Delete a loopback IPv4 address from the switch. pub fn delete_loopback_ipv4(switch: &Switch, addr: &Ipv4Addr) -> DpdResult<()> { + // TODO::cory: loopback too? let mut loopback_data = switch.loopback.lock().unwrap(); let entry = Ipv4Entry { addr: *addr, tag: "".into() }; if !loopback_data.v4_addrs.contains(&entry) { diff --git a/dpd/src/port_settings.rs b/dpd/src/port_settings.rs index cfcea772..8b21882e 100644 --- a/dpd/src/port_settings.rs +++ b/dpd/src/port_settings.rs @@ -114,8 +114,16 @@ impl From<&Link> for LinkSpec { kr: p.config.kr, tx_eq: p.tx_eq, delete_me: p.config.delete_me, - ipv4: p.ipv4.iter().map(|x| x.addr).collect(), - ipv6: p.ipv6.iter().map(|x| x.addr).collect(), + ipv4: p + .tagged + .values() + .flat_map(|conf| conf.ipv4.iter().copied()) + .collect(), + ipv6: p + .tagged + .values() + .flat_map(|conf| conf.ipv6.iter().copied()) + .collect(), allow_ddm_traffic: p.config.allow_ddm_traffic, } }