[feat] Multicast groups

Introduces end-to-end multicast group support across control plane and sled-agent, integrated with IP pool extensions required
for supporting multicast workflows. This work enables project-scoped multicast groups with lifecycle-driven dataplane programming
and exposes an API for operating multicast groups over instances.

Highlights:
  - DB: new multicast_group tables; member lifecycle management
  - API: multicast group/member CRUD; source IP validation; VPC/project hierarchy integration with default VNI fallback
  - Control plane: RPW reconcilers for groups/members; sagas for dataplane updates atomically at the group level; instance lifecycle hooks and piggybacking
  - Dataplane: Dendrite DPD switch programming via trait abstraction; DPD client used in tests
  - Sled agent: multicast-aware instance management; network interface configuration for multicast traffic; cross-version testing; OPTE stubs present
  - Tests: comprehensive integration suites under nexus/tests/integration_tests/multicast/

Components:
  - Database schema: external and underlay multicast groups; member/instance association tables
  - Control plane modules: multicast group management, member lifecycle, dataplane abstraction; RPW reconcilers to ensure convergence
  - API layer: endpoints and validation; default-VNI semantics when VPC not provided
  - Sled agent: OPTE stubs and compatibility shims for older agents

Workflows Implemented:
  1. Instance lifecycle integration:

     - "Create" -> resolve VPC/VNI (or default), validate source IPs, create memberships, enqueue group ensure RPW
     - "Start" -> program dataplane via ensure/update sagas; activate member flows after switch ack
     - "Stop" -> deactivate dataplane membership; retain DB membership for fast restart
     - "Delete" -> remove instance memberships; group deletion is explicit
     - "Migrate" -> deactivate on source sled; activate on target; idempotent with ordering guarantees
     - Restart/recovery -> RPWs reconcile desired state; compensations clean up partial programming

  2. RPW reconciliation:

     - ensure dataplane switches match database state
     - handle sled migrations and state transitions
     - Eventual consistency with retry logic

Migrations:
  - Apply schema changes in schema/crdb/multicast-support/up01.sql (and update dbinit.sql)
  - Bump schema versions accordingly

API/Compatibility:
  - OpenAPI updated: openapi/nexus.json, openapi/sled-agent/sled-agent-5.0.0-89f1f7.json
  - Regenerate clients where applicable

References:
  - RFD 488: https://rfd.shared.oxide.computer/rfd/488
  - IP Pool extensions: #9084
  - Dendrite PRs (based on recency):
    * oxidecomputer/dendrite#132
    * oxidecomputer/dendrite#109
    * oxidecomputer/dendrite#14

Follow-ups include:
  - OPTE integration
  - commtest extension
  - omdb commands are tracked in issues
  - pool and group stats

Author: zeeshanlakhani

Cargo.lock

@@ -952,6 +952,8 @@ pub enum ResourceType {
     LldpLinkConfig,
     LoopbackAddress,
     MetricProducer,
+    MulticastGroup,
+    MulticastGroupMember,
     NatEntry,
     Oximeter,
     PhysicalDisk,
@@ -2510,6 +2512,12 @@ impl Vni {
     /// The VNI for the builtin services VPC.
     pub const SERVICES_VNI: Self = Self(100);
 
+    /// VNI default if no VPC is provided for a multicast group.
+    ///
+    /// This is a low-numbered VNI, to avoid colliding with user VNIs.
+    /// However, it is not in the Oxide-reserved yet.
+    pub const DEFAULT_MULTICAST_VNI: Self = Self(77);
+
     /// Oxide reserves a slice of initial VNIs for its own use.
     pub const MIN_GUEST_VNI: u32 = 1024;
 

common/src/api/external/mod.rs

@@ -124,6 +124,10 @@ task: "metrics_producer_gc"
     unregisters Oximeter metrics producers that have not renewed their lease
 
 
+task: "multicast_group_reconciler"
+    reconciles multicast group state with dendrite switch configuration
+
+
 task: "nat_garbage_collector"
     prunes soft-deleted NAT entries from nat_entry table based on a
     predetermined retention policy
@@ -332,6 +336,10 @@ task: "metrics_producer_gc"
     unregisters Oximeter metrics producers that have not renewed their lease
 
 
+task: "multicast_group_reconciler"
+    reconciles multicast group state with dendrite switch configuration
+
+
 task: "nat_garbage_collector"
     prunes soft-deleted NAT entries from nat_entry table based on a
     predetermined retention policy
@@ -527,6 +535,10 @@ task: "metrics_producer_gc"
     unregisters Oximeter metrics producers that have not renewed their lease
 
 
+task: "multicast_group_reconciler"
+    reconciles multicast group state with dendrite switch configuration
+
+
 task: "nat_garbage_collector"
     prunes soft-deleted NAT entries from nat_entry table based on a
     predetermined retention policy

dev-tools/omdb/tests/env.out

@@ -349,6 +349,10 @@ task: "metrics_producer_gc"
     unregisters Oximeter metrics producers that have not renewed their lease
 
 
+task: "multicast_group_reconciler"
+    reconciles multicast group state with dendrite switch configuration
+
+
 task: "nat_garbage_collector"
     prunes soft-deleted NAT entries from nat_entry table based on a
     predetermined retention policy
@@ -648,6 +652,12 @@ task: "metrics_producer_gc"
     started at <REDACTED_TIMESTAMP> (<REDACTED DURATION>s ago) and ran for <REDACTED DURATION>ms
 warning: unknown background task: "metrics_producer_gc" (don't know how to interpret details: Object {"expiration": String("<REDACTED_TIMESTAMP>"), "pruned": Array []})
 
+task: "multicast_group_reconciler"
+  configured period: every <REDACTED_DURATION>m
+  last completed activation: <REDACTED ITERATIONS>, triggered by <TRIGGERED_BY_REDACTED>
+    started at <REDACTED_TIMESTAMP> (<REDACTED DURATION>s ago) and ran for <REDACTED DURATION>ms
+warning: unknown background task: "multicast_group_reconciler" (don't know how to interpret details: Object {"errors": Array [String("failed to create multicast dataplane client: Internal Error: failed to build DPD clients")], "groups_created": Number(0), "groups_deleted": Number(0), "groups_verified": Number(0), "members_deleted": Number(0), "members_processed": Number(0)})
+
 task: "phantom_disks"
   configured period: every <REDACTED_DURATION>s
   last completed activation: <REDACTED ITERATIONS>, triggered by <TRIGGERED_BY_REDACTED>
@@ -1166,6 +1176,12 @@ task: "metrics_producer_gc"
     started at <REDACTED_TIMESTAMP> (<REDACTED DURATION>s ago) and ran for <REDACTED DURATION>ms
 warning: unknown background task: "metrics_producer_gc" (don't know how to interpret details: Object {"expiration": String("<REDACTED_TIMESTAMP>"), "pruned": Array []})
 
+task: "multicast_group_reconciler"
+  configured period: every <REDACTED_DURATION>m
+  last completed activation: <REDACTED ITERATIONS>, triggered by <TRIGGERED_BY_REDACTED>
+    started at <REDACTED_TIMESTAMP> (<REDACTED DURATION>s ago) and ran for <REDACTED DURATION>ms
+warning: unknown background task: "multicast_group_reconciler" (don't know how to interpret details: Object {"errors": Array [String("failed to create multicast dataplane client: Internal Error: failed to build DPD clients")], "groups_created": Number(0), "groups_deleted": Number(0), "groups_verified": Number(0), "members_deleted": Number(0), "members_processed": Number(0)})
+
 task: "phantom_disks"
   configured period: every <REDACTED_DURATION>s
   last completed activation: <REDACTED ITERATIONS>, triggered by <TRIGGERED_BY_REDACTED>

dev-tools/omdb/tests/successes.out

@@ -80,6 +80,7 @@ async fn instance_launch() -> Result<()> {
             auto_restart_policy: Default::default(),
             anti_affinity_groups: Vec::new(),
             cpu_platform: None,
+            multicast_groups: Vec::new(),
         })
         .send()
         .await?;

end-to-end-tests/src/instance_launch.rs

@@ -52,6 +52,11 @@ pub enum Error {
     #[error("Tried to update external IPs on non-existent port ({0}, {1:?})")]
     ExternalIpUpdateMissingPort(uuid::Uuid, NetworkInterfaceKind),
 
+    #[error(
+        "Tried to update multicast groups on non-existent port ({0}, {1:?})"
+    )]
+    MulticastUpdateMissingPort(uuid::Uuid, NetworkInterfaceKind),
+
     #[error("Could not find Primary NIC")]
     NoPrimaryNic,
 

illumos-utils/src/opte/illumos.rs

@@ -31,6 +31,7 @@ use oxide_vpc::api::RouterTarget;
 pub use oxide_vpc::api::Vni;
 use oxnet::IpNet;
 pub use port::Port;
+pub use port_manager::MulticastGroupCfg;
 pub use port_manager::PortCreateParams;
 pub use port_manager::PortManager;
 pub use port_manager::PortTicket;
@@ -71,7 +72,7 @@ impl Gateway {
     }
 }
 
-/// Convert a nexus `IpNet` to an OPTE `IpCidr`.
+/// Convert a nexus [IpNet] to an OPTE [IpCidr].
 fn net_to_cidr(net: IpNet) -> IpCidr {
     match net {
         IpNet::V4(net) => IpCidr::Ip4(Ipv4Cidr::new(
@@ -85,9 +86,10 @@ fn net_to_cidr(net: IpNet) -> IpCidr {
     }
 }
 
-/// Convert a nexus `RouterTarget` to an OPTE `RouterTarget`.
+/// Convert a nexus [shared::RouterTarget] to an OPTE [RouterTarget].
 ///
-/// This is effectively a `From` impl, but defined for two out-of-crate types.
+/// This is effectively a [`From`] impl, but defined for two
+/// out-of-crate types.
 /// We map internet gateways that target the (single) "system" VPC IG to
 /// `InternetGateway(None)`. Everything else is mapped directly, translating IP
 /// address types as needed.

illumos-utils/src/opte/mod.rs

@@ -46,6 +46,11 @@ pub enum Error {
     #[error("Tried to update external IPs on non-existent port ({0}, {1:?})")]
     ExternalIpUpdateMissingPort(uuid::Uuid, NetworkInterfaceKind),
 
+    #[error(
+        "Tried to update multicast groups on non-existent port ({0}, {1:?})"
+    )]
+    MulticastUpdateMissingPort(uuid::Uuid, NetworkInterfaceKind),
+
     #[error("Could not find Primary NIC")]
     NoPrimaryNic,
 

illumos-utils/src/opte/non_illumos.rs

@@ -62,6 +62,18 @@ use std::sync::atomic::AtomicU64;
 use std::sync::atomic::Ordering;
 use uuid::Uuid;
 
+/// IPv4 multicast address range (224.0.0.0/4).
+/// See RFC 5771 (IPv4 Multicast Address Assignments):
+/// <https://www.rfc-editor.org/rfc/rfc5771>
+#[allow(dead_code)]
+const IPV4_MULTICAST_RANGE: &str = "224.0.0.0/4";
+
+/// IPv6 multicast address range (ff00::/8).
+/// See RFC 4291 (IPv6 Addressing Architecture):
+/// <https://www.rfc-editor.org/rfc/rfc4291>
+#[allow(dead_code)]
+const IPV6_MULTICAST_RANGE: &str = "ff00::/8";
+
 /// Stored routes (and usage count) for a given VPC/subnet.
 #[derive(Debug, Default, Clone)]
 struct RouteSet {
@@ -70,6 +82,21 @@ struct RouteSet {
     active_ports: usize,
 }
 
+/// Configuration for multicast groups on an OPTE port.
+///
+/// TODO: This type should be moved to [oxide_vpc::api] when OPTE dependencies
+/// are updated, following the same pattern as other VPC configuration types
+/// like [ExternalIpCfg], [IpCfg], etc.
+///
+/// TODO: Eventually remove.
+#[derive(Debug, Clone, PartialEq)]
+pub struct MulticastGroupCfg {
+    /// The multicast group IP address (IPv4 or IPv6).
+    pub group_ip: IpAddr,
+    /// For Source-Specific Multicast (SSM), list of source addresses.
+    pub sources: Vec<IpAddr>,
+}
+
 #[derive(Debug)]
 struct PortManagerInner {
     log: Logger,
@@ -595,7 +622,7 @@ impl PortManager {
     }
 
     /// Set Internet Gateway mappings for all external IPs in use
-    /// by attached `NetworkInterface`s.
+    /// by attached [NetworkInterface]s.
     ///
     /// Returns whether the internal mappings were changed.
     pub fn set_eip_gateways(&self, mappings: ExternalIpGatewayMap) -> bool {
@@ -751,6 +778,68 @@ impl PortManager {
         Ok(())
     }
 
+    /// Validate multicast group memberships for an OPTE port.
+    ///
+    /// This method validates multicast group configurations but does not yet
+    /// configure OPTE port-level multicast group membership. The actual
+    /// multicast forwarding is currently handled by the reconciler + DPD
+    /// at the dataplane switch level.
+    ///
+    /// TODO: Once OPTE kernel module supports multicast group APIs, this method
+    /// should be updated accordingly to configure the port for specific
+    /// multicast group memberships.
+    pub fn multicast_groups_ensure(
+        &self,
+        nic_id: Uuid,
+        nic_kind: NetworkInterfaceKind,
+        multicast_groups: &[MulticastGroupCfg],
+    ) -> Result<(), Error> {
+        let ports = self.inner.ports.lock().unwrap();
+        let port = ports.get(&(nic_id, nic_kind)).ok_or_else(|| {
+            Error::MulticastUpdateMissingPort(nic_id, nic_kind)
+        })?;
+
+        debug!(
+            self.inner.log,
+            "Validating multicast group configuration for OPTE port";
+            "port_name" => port.name(),
+            "nic_id" => ?nic_id,
+            "groups" => ?multicast_groups,
+        );
+
+        // Validate multicast group configurations
+        for group in multicast_groups {
+            if !group.group_ip.is_multicast() {
+                error!(
+                    self.inner.log,
+                    "Invalid multicast IP address";
+                    "group_ip" => %group.group_ip,
+                    "port_name" => port.name(),
+                );
+                return Err(Error::InvalidPortIpConfig);
+            }
+        }
+
+        // TODO: Configure firewall rules to allow multicast traffic.
+        // Add exceptions in source/dest MAC/L3 addr checking for multicast
+        // addreses matching known groups, only doing cidr-checking on the
+        // multicasst destination side.
+
+        info!(
+            self.inner.log,
+            "OPTE port configured for multicast traffic";
+            "port_name" => port.name(),
+            "ipv4_range" => IPV4_MULTICAST_RANGE,
+            "ipv6_range" => IPV6_MULTICAST_RANGE,
+            "multicast_groups" => multicast_groups.len(),
+        );
+
+        // TODO: Configure OPTE port for specific multicast group membership
+        // once APIs are available.
+
+        Ok(())
+    }
+
     pub fn firewall_rules_ensure(
         &self,
         vni: external::Vni,

illumos-utils/src/opte/port_manager.rs

@@ -439,6 +439,8 @@ pub struct BackgroundTaskConfig {
     pub webhook_deliverator: WebhookDeliveratorConfig,
     /// configuration for SP ereport ingester task
     pub sp_ereport_ingester: SpEreportIngesterConfig,
+    /// configuration for multicast group reconciler task
+    pub multicast_group_reconciler: MulticastGroupReconcilerConfig,
 }
 
 #[serde_as]
@@ -836,6 +838,21 @@ impl Default for SpEreportIngesterConfig {
     }
 }
 
+#[serde_as]
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct MulticastGroupReconcilerConfig {
+    /// period (in seconds) for periodic activations of the background task that
+    /// reconciles multicast group state with dendrite switch configuration
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub period_secs: Duration,
+}
+
+impl Default for MulticastGroupReconcilerConfig {
+    fn default() -> Self {
+        Self { period_secs: Duration::from_secs(60) }
+    }
+}
+
 /// Configuration for a nexus server
 #[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 pub struct PackageConfig {
@@ -1126,6 +1143,7 @@ mod test {
             webhook_deliverator.first_retry_backoff_secs = 45
             webhook_deliverator.second_retry_backoff_secs = 46
             sp_ereport_ingester.period_secs = 47
+            multicast_group_reconciler.period_secs = 60
             [default_region_allocation_strategy]
             type = "random"
             seed = 0
@@ -1359,6 +1377,10 @@ mod test {
                             period_secs: Duration::from_secs(47),
                             disable: false,
                         },
+                        multicast_group_reconciler:
+                            MulticastGroupReconcilerConfig {
+                                period_secs: Duration::from_secs(60),
+                            },
                     },
                     default_region_allocation_strategy:
                         crate::nexus_config::RegionAllocationStrategy::Random {
@@ -1453,6 +1475,7 @@ mod test {
             alert_dispatcher.period_secs = 42
             webhook_deliverator.period_secs = 43
             sp_ereport_ingester.period_secs = 44
+            multicast_group_reconciler.period_secs = 60
 
             [default_region_allocation_strategy]
             type = "random"