diff --git a/assets/yang/ietf-interfaces/subscriptions-info.json b/assets/yang/ietf-interfaces/subscriptions-info.json index b7778881..69bf6c1b 100644 --- a/assets/yang/ietf-interfaces/subscriptions-info.json +++ b/assets/yang/ietf-interfaces/subscriptions-info.json @@ -1,7 +1,6 @@ [ { - "collector": "127.0.0.1:10000", - "peer": "0.0.0.0:830", + "peer_ip": "0.0.0.0", "id": 1, "target": { "ietf-yang-push:datastore": "ietf-datastores:operational", diff --git a/assets/yang/ietf-telemetry-message/subscriptions-info.json b/assets/yang/ietf-telemetry-message/subscriptions-info.json index d1081e60..72c427b4 100644 --- a/assets/yang/ietf-telemetry-message/subscriptions-info.json +++ b/assets/yang/ietf-telemetry-message/subscriptions-info.json @@ -1,6 +1,6 @@ [ { - "peer": "0.0.0.0:830", + "peer_ip": "0.0.0.0", "id": 1, "content_id": "ietf-telemetry-message", "target": { diff --git a/crates/collector/examples/kafka-yang-consumer.rs b/crates/collector/examples/kafka-yang-consumer.rs index 0928a01c..8436589a 100644 --- a/crates/collector/examples/kafka-yang-consumer.rs +++ b/crates/collector/examples/kafka-yang-consumer.rs @@ -46,7 +46,7 @@ use schema_registry_client::rest::schema_registry_client::{Client, SchemaRegistr use serde_json::json; use shadow_rs::shadow; use std::collections::{HashMap, HashSet}; -use std::net::{IpAddr, SocketAddr}; +use std::net::IpAddr; use tokio::signal; use tracing::{debug, error, info, trace, warn}; use yang5::context::Context; @@ -845,12 +845,8 @@ async fn main() -> Result<()> { let sr_client = SchemaRegistryClient::new(sr_config); // Create placeholder subscription info (reused for all schemas) - let subscription_info = SubscriptionInfo::new_empty( - SocketAddr::new(IpAddr::V4(std::net::Ipv4Addr::new(0, 0, 0, 0)), 0), - None, - SocketAddr::new(IpAddr::V4(std::net::Ipv4Addr::new(0, 0, 0, 0)), 0), - 0, - ); + let subscription_info = + SubscriptionInfo::new_empty(IpAddr::V4(std::net::Ipv4Addr::new(0, 0, 0, 0)), 0); // Create YANG context cache let mut yang_ctx_cache = YangContextCache::new(); diff --git a/crates/collector/src/lib.rs b/crates/collector/src/lib.rs index f4ae77b2..50412619 100644 --- a/crates/collector/src/lib.rs +++ b/crates/collector/src/lib.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2024-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -24,7 +25,7 @@ use crate::publishers::http::{HttpPublisherActorHandle, Message}; use crate::publishers::kafka_avro::KafkaAvroPublisherActorHandle; use crate::publishers::kafka_json::KafkaJsonPublisherActorHandle; use crate::publishers::kafka_yang::KafkaYangPublisherActorHandle; -use crate::yang_push::enrichment::YangPushEnrichmentActorHandle; +use crate::yang_push::enrichment::{EnrichedNotification, YangPushEnrichmentActorHandle}; use futures_util::StreamExt; use futures_util::stream::FuturesUnordered; @@ -36,11 +37,9 @@ use netcalyx_flow_service::flow_supervisor::FlowCollectorsSupervisorActorHandle; use netcalyx_udp_notif_pkt::raw::MediaType; use netcalyx_udp_notif_service::UdpNotifRequest; use netcalyx_udp_notif_service::supervisor::UdpNotifSupervisorHandle; -use netcalyx_yang_push::ContentId; use netcalyx_yang_push::cache::actor::CacheActorHandle; use netcalyx_yang_push::cache::fetcher::{NetconfYangLibraryFetcher, RetryConfig}; -use netcalyx_yang_push::cache::storage::SubscriptionInfo; -use netcalyx_yang_push::model::telemetry::{Manifest, TelemetryMessageWrapper}; +use netcalyx_yang_push::model::telemetry::Manifest; use netcalyx_yang_push::validation::ValidationActorHandle; use shadow_rs::shadow; use std::net::IpAddr; @@ -926,10 +925,10 @@ fn serialize_udp_notif( } fn serialize_telemetry_json( - input: (Option, SubscriptionInfo, TelemetryMessageWrapper), + input: EnrichedNotification, _writer_id: String, ) -> Result<(Option, serde_json::Value), UdpNotifSerializationError> { - let tmw = input.2; + let tmw = input.message; let ip = tmw.message().telemetry_message_metadata().export_address(); let value = serde_json::to_value(tmw)?; let key = serde_json::Value::String(ip.to_string()); @@ -1054,6 +1053,7 @@ mod tests { use super::*; use bytes::Bytes; use netcalyx_udp_notif_pkt::raw::UdpNotifPacket; + use netcalyx_udp_notif_service::SessionInfo; use std::collections::HashMap; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; @@ -1070,7 +1070,10 @@ mod tests { Bytes::from(&[0xffu8, 0xffu8][..]), ); - let request = Arc::new(UdpNotifRequest::new(collector, None, peer, pkt)); + let request = Arc::new(UdpNotifRequest::new( + SessionInfo::new(collector, None, peer), + pkt, + )); let serialized = serialize_udp_notif(request.clone(), writer_id.clone()); assert!(matches!( serialized, @@ -1112,12 +1115,13 @@ mod tests { } ); let request_invalid = Arc::new(UdpNotifRequest::new( - collector, - None, - peer, + SessionInfo::new(collector, None, peer), pkt_invalid_json, )); - let request_good = Arc::new(UdpNotifRequest::new(collector, None, peer, pkt)); + let request_good = Arc::new(UdpNotifRequest::new( + SessionInfo::new(collector, None, peer), + pkt, + )); let result_invalid = serialize_udp_notif(request_invalid, writer_id.clone()); let serialized = serialize_udp_notif(request_good, writer_id.clone()).expect("failed to serialize json"); @@ -1168,12 +1172,13 @@ mod tests { ); let request_invalid = Arc::new(UdpNotifRequest::new( - collector, - None, - peer, + SessionInfo::new(collector, None, peer), pkt_invalid_utf8, )); - let request_good = Arc::new(UdpNotifRequest::new(collector, None, peer, pkt)); + let request_good = Arc::new(UdpNotifRequest::new( + SessionInfo::new(collector, None, peer), + pkt, + )); let result_invalid = serialize_udp_notif(request_invalid, writer_id.clone()); let serialized = serialize_udp_notif(request_good, writer_id.clone()).expect("failed to serialize json"); @@ -1226,8 +1231,14 @@ mod tests { } ); - let request_invalid = Arc::new(UdpNotifRequest::new(collector, None, peer, pkt_invalid)); - let request_good = Arc::new(UdpNotifRequest::new(collector, None, peer, pkt)); + let request_invalid = Arc::new(UdpNotifRequest::new( + SessionInfo::new(collector, None, peer), + pkt_invalid, + )); + let request_good = Arc::new(UdpNotifRequest::new( + SessionInfo::new(collector, None, peer), + pkt, + )); let result_invalid = serialize_udp_notif(request_invalid, writer_id.clone()); let serialized = serialize_udp_notif(request_good, writer_id.clone()).expect("failed to serialize json"); diff --git a/crates/collector/src/publishers/kafka_yang.rs b/crates/collector/src/publishers/kafka_yang.rs index 21a86c90..61a99127 100644 --- a/crates/collector/src/publishers/kafka_yang.rs +++ b/crates/collector/src/publishers/kafka_yang.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2025-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -430,7 +431,7 @@ where return if let Some(default_schema_id) = self.default_schema_id { if let Some(subscription_info) = subscription_info { warn!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -449,7 +450,7 @@ where } else { if let Some(subscription_info) = subscription_info { warn!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -470,7 +471,7 @@ where if let Some(&schema_id) = self.schema_id_cache.get(id) { if let Some(subscription_info) = subscription_info { trace!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -495,10 +496,10 @@ where if let Err(err) = self .cache_req_tx - .send(CacheLookupCommand::LookupByContentIdOneShot( - id.to_string(), - response_tx, - )) + .send(CacheLookupCommand::LookupByContentIdOneShot { + content_id: id.to_string(), + tx: response_tx, + }) .await { warn!("Failed to request schema for content_id: {}", id); diff --git a/crates/collector/src/yang_push/config.rs b/crates/collector/src/yang_push/config.rs index 87c79e6f..39970096 100644 --- a/crates/collector/src/yang_push/config.rs +++ b/crates/collector/src/yang_push/config.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2025-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -21,9 +22,9 @@ //! JSON format. use crate::publishers::kafka_yang::YangConverter; +use crate::yang_push::enrichment::EnrichedNotification; use netcalyx_yang_push::ContentId; use netcalyx_yang_push::cache::storage::{SubscriptionInfo, YangLibraryReference}; -use netcalyx_yang_push::model::telemetry::TelemetryMessageWrapper; use serde::{Deserialize, Serialize}; #[derive(Debug, strum_macros::Display)] @@ -81,12 +82,7 @@ impl TelemetryYangConverter { } } -impl - YangConverter< - (Option, SubscriptionInfo, TelemetryMessageWrapper), - TelemetryYangConverterError, - > for TelemetryYangConverter -{ +impl YangConverter for TelemetryYangConverter { fn subject_prefix(&self) -> Option<&str> { self.subject_prefix.as_deref() } @@ -103,38 +99,27 @@ impl self.extension_yang_lib_ref.as_ref() } - fn content_id( - &self, - input: &(Option, SubscriptionInfo, TelemetryMessageWrapper), - ) -> Option { - input.0.clone() + fn content_id(&self, input: &EnrichedNotification) -> Option { + input.cached_content_id.clone() } - fn get_key( - &self, - input: &(Option, SubscriptionInfo, TelemetryMessageWrapper), - ) -> Option { - let (_, subscription_info, _) = input; - let ip = subscription_info.peer().ip(); + fn get_key(&self, input: &EnrichedNotification) -> Option { + let ip = input.subscription_info.peer_ip(); Some(serde_json::Value::String(ip.to_string())) } fn serialize_json( &self, - input: (Option, SubscriptionInfo, TelemetryMessageWrapper), + input: EnrichedNotification, ) -> Result, TelemetryYangConverterError> { - let telemetry_message_wrapper = input.2; - serde_json::to_vec(&telemetry_message_wrapper).map_err(Into::into) + serde_json::to_vec(&input.message).map_err(Into::into) } - fn subscription_info( - &self, - input: &(Option, SubscriptionInfo, TelemetryMessageWrapper), - ) -> Option { - if input.1.is_empty() { + fn subscription_info(&self, input: &EnrichedNotification) -> Option { + if input.subscription_info.is_empty() { None } else { - Some(input.1.clone()) + Some(input.subscription_info.clone()) } } } @@ -145,14 +130,13 @@ mod tests { use chrono::TimeZone; use netcalyx_netconf_proto::yang_push::identities::{Encoding, Transport}; use netcalyx_netconf_proto::yang_push::subscription::YangPushModuleVersion; + use netcalyx_udp_notif_service::SessionInfo; use netcalyx_yang_push::model::telemetry::*; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; fn create_test_subscription_info(ip: IpAddr) -> SubscriptionInfo { SubscriptionInfo::new( - SocketAddr::from(([127, 0, 0, 1], 10000)), - None, - SocketAddr::new(ip, 8080), + ip, 1, netcalyx_udp_notif_pkt::notification::Target::new_datastore( "ietf-datastores:operational".to_string(), @@ -223,7 +207,16 @@ mod tests { let sub_info = create_test_subscription_info(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))); let msg = create_test_telemetry_message_wrapper(); - let input = (content_id.clone(), sub_info, msg); + let input = EnrichedNotification { + cached_content_id: content_id.clone(), + subscription_info: sub_info, + session: SessionInfo::new( + SocketAddr::from(([127, 0, 0, 1], 0)), + None, + SocketAddr::from(([127, 0, 0, 1], 0)), + ), + message: msg, + }; assert_eq!(converter.content_id(&input), content_id); } @@ -234,7 +227,16 @@ mod tests { let sub_info = create_test_subscription_info(ip); let msg = create_test_telemetry_message_wrapper(); - let input = (None, sub_info, msg); + let input = EnrichedNotification { + cached_content_id: None, + subscription_info: sub_info, + session: SessionInfo::new( + SocketAddr::from(([127, 0, 0, 1], 0)), + None, + SocketAddr::from(([127, 0, 0, 1], 0)), + ), + message: msg, + }; let key = converter.get_key(&input).unwrap(); assert_eq!(key, serde_json::Value::String("192.168.1.1".to_string())); } @@ -246,7 +248,16 @@ mod tests { let sub_info = create_test_subscription_info(ip); let msg = create_test_telemetry_message_wrapper(); - let input = (None, sub_info, msg); + let input = EnrichedNotification { + cached_content_id: None, + subscription_info: sub_info, + session: SessionInfo::new( + SocketAddr::from(([127, 0, 0, 1], 0)), + None, + SocketAddr::from(([127, 0, 0, 1], 0)), + ), + message: msg, + }; let key = converter.get_key(&input).unwrap(); assert_eq!(key, serde_json::Value::String("2001:db8::1".to_string())); } @@ -259,7 +270,16 @@ mod tests { let expected = serde_json::to_value(&msg).unwrap(); // Call serialize_json to serialize into bytes - let input = (None, sub_info, msg); + let input = EnrichedNotification { + cached_content_id: None, + subscription_info: sub_info, + session: SessionInfo::new( + SocketAddr::from(([127, 0, 0, 1], 0)), + None, + SocketAddr::from(([127, 0, 0, 1], 0)), + ), + message: msg, + }; let result = converter.serialize_json(input); assert!(result.is_ok()); diff --git a/crates/collector/src/yang_push/enrichment.rs b/crates/collector/src/yang_push/enrichment.rs index 4586984f..cb6fffae 100644 --- a/crates/collector/src/yang_push/enrichment.rs +++ b/crates/collector/src/yang_push/enrichment.rs @@ -36,12 +36,13 @@ use crate::yang_push::{ }; use chrono::Utc; use netcalyx_udp_notif_pkt::decoded::{UdpNotifPacketDecoded, UdpNotifPayload}; -use netcalyx_udp_notif_service::OTL_UDP_NOTIF_PUBLISHER_ID_KEY; +use netcalyx_udp_notif_service::{OTL_UDP_NOTIF_PUBLISHER_ID_KEY, SessionInfo}; use netcalyx_yang_push::cache::storage::SubscriptionInfo; use netcalyx_yang_push::model::telemetry::{ EventType, Label, Manifest, NetworkOperatorMetadata, SessionProtocol, TelemetryMessage, TelemetryMessageMetadata, TelemetryMessageWrapper, YangPushSubscriptionMetadata, }; +use netcalyx_yang_push::validation::ValidatedNotification; use netcalyx_yang_push::{ ContentId, OTL_YANG_PUSH_CACHED_CONTENT_ID_KEY, OTL_YANG_PUSH_SUBSCRIPTION_ID_KEY, OTL_YANG_PUSH_SUBSCRIPTION_ROUTER_CONTENT_ID_KEY, OTL_YANG_PUSH_SUBSCRIPTION_TARGET_KEY, @@ -150,15 +151,31 @@ impl YangPushEnrichmentStats { } } +/// The output of the enrichment stage: a `TelemetryMessage` assembled from a +/// validated UDP-Notif packet, together with the subscription identity and +/// transport session that produced it. +/// +/// - `cached_content_id`: the YANG schema fingerprint used for validation, or +/// `None` when the packet was forwarded unvalidated. +/// - `subscription_info`: subscription identity (peer IP, target, modules…). +/// - `session`: transport session context (collector, interface, full peer +/// `SocketAddr`) — available for logging, routing, and Kafka key selection. +/// - `message`: the enriched telemetry message ready for publishing. +#[derive(Debug)] +pub struct EnrichedNotification { + pub cached_content_id: Option, + pub subscription_info: SubscriptionInfo, + pub session: SessionInfo, + pub message: TelemetryMessageWrapper, +} + /// Actor responsible for enriching YANG-Push notifications. /// Sends enriched TelemetryMessage objects. struct YangPushEnrichmentActor { cmd_rx: mpsc::Receiver, enrichment_rx: async_channel::Receiver, - validated_rx: - async_channel::Receiver<(Option, SubscriptionInfo, UdpNotifPacketDecoded)>, - enriched_tx: - async_channel::Sender<(Option, SubscriptionInfo, TelemetryMessageWrapper)>, + validated_rx: async_channel::Receiver, + enriched_tx: async_channel::Sender, labels: HashMap>, manifest: Manifest, stats: YangPushEnrichmentStats, @@ -168,16 +185,8 @@ impl YangPushEnrichmentActor { fn new( cmd_rx: mpsc::Receiver, enrichment_rx: async_channel::Receiver, - validated_rx: async_channel::Receiver<( - Option, - SubscriptionInfo, - UdpNotifPacketDecoded, - )>, - enriched_tx: async_channel::Sender<( - Option, - SubscriptionInfo, - TelemetryMessageWrapper, - )>, + validated_rx: async_channel::Receiver, + enriched_tx: async_channel::Sender, manifest: Manifest, stats: YangPushEnrichmentStats, ) -> Self { @@ -340,12 +349,13 @@ impl YangPushEnrichmentActor { &mut self, content_id: Option<&ContentId>, subscription_info: &SubscriptionInfo, + session: &SessionInfo, decoded_packet: &UdpNotifPacketDecoded, ) -> Result { if decoded_packet.notification_type().is_none() { return Err(YangPushEnrichmentActorError::NotificationWithoutContent); } - let peer = subscription_info.peer(); + let peer_ip = subscription_info.peer_ip(); let message_id = decoded_packet.message_id(); let publisher_id = decoded_packet.publisher_id(); let notification_type = decoded_packet @@ -355,7 +365,7 @@ impl YangPushEnrichmentActor { let labels: Option> = self .labels - .get(&peer.ip()) + .get(&peer_ip) .map(|l_map| l_map.values().cloned().map(|wl| wl.label).collect()); // Match on the wrapper and process the notification content @@ -378,17 +388,17 @@ impl YangPushEnrichmentActor { EventType::Log, None, // we don't set sequence numbers for now SessionProtocol::YangPush, // only option at the moment - peer.ip(), - Some(peer.port()), - None, - None, + peer_ip, + Some(session.peer().port()), + Some(session.collector().ip()), + Some(session.collector().port()), subscription_metadata, ); // Re-serialize the UDP-Notif payload into JSON let json_payload = serde_json::to_value(decoded_packet.payload()).map_err(|err| { error!( - peer=%peer, + peer_ip=%peer_ip, message_id, publisher_id, subscription_id=subscription_info.id(), @@ -444,17 +454,13 @@ impl YangPushEnrichmentActor { msg = self.validated_rx.recv() => { match msg { Ok(msg) => { - let (content_id, subscription_info, pkt) = msg; - let peer = subscription_info.peer(); + let ValidatedNotification { cached_content_id: content_id, subscription_info, session, packet: pkt } = msg; + let peer_ip = subscription_info.peer_ip(); let publisher_id = pkt.publisher_id(); let peer_tags = [ opentelemetry::KeyValue::new( "network.peer.address", - format!("{}", peer.ip()), - ), - opentelemetry::KeyValue::new( - "network.peer.port", - opentelemetry::Value::I64(peer.port().into()), + format!("{peer_ip}"), ), opentelemetry::KeyValue::new( OTL_UDP_NOTIF_PUBLISHER_ID_KEY, @@ -480,9 +486,14 @@ impl YangPushEnrichmentActor { self.stats.received_messages.add(1, &peer_tags); // Process the payload and send the enriched TelemetryMessage - match self.process_decoded_udp_notif_packet(content_id.as_ref(), &subscription_info, &pkt) { + match self.process_decoded_udp_notif_packet(content_id.as_ref(), &subscription_info, &session, &pkt) { Ok(telemetry_message) => { - if let Err(err) = self.enriched_tx.send((content_id, subscription_info, telemetry_message)).await { + if let Err(err) = self.enriched_tx.send(EnrichedNotification { + cached_content_id: content_id, + subscription_info, + session, + message: telemetry_message, + }).await { error!("YangPushEnrichmentActor send error: {err}"); self.stats.send_error.add(1, &peer_tags); } else { @@ -519,18 +530,13 @@ impl std::error::Error for YangPushEnrichmentActorHandleError {} pub struct YangPushEnrichmentActorHandle { cmd_send: mpsc::Sender, enrichment_tx: async_channel::Sender, - enriched_rx: - async_channel::Receiver<(Option, SubscriptionInfo, TelemetryMessageWrapper)>, + enriched_rx: async_channel::Receiver, } impl YangPushEnrichmentActorHandle { pub fn new( buffer_size: usize, - validated_rx: async_channel::Receiver<( - Option, - SubscriptionInfo, - UdpNotifPacketDecoded, - )>, + validated_rx: async_channel::Receiver, manifest: Manifest, stats: either::Either, ) -> (JoinHandle>, Self) { @@ -565,10 +571,7 @@ impl YangPushEnrichmentActorHandle { .map_err(|_| YangPushEnrichmentActorHandleError::SendError) } - pub fn subscribe( - &self, - ) -> async_channel::Receiver<(Option, SubscriptionInfo, TelemetryMessageWrapper)> - { + pub fn subscribe(&self) -> async_channel::Receiver { self.enriched_rx.clone() } } @@ -625,7 +628,7 @@ mod tests { #[allow(clippy::type_complexity)] fn create_actor_handle() -> ( - async_channel::Sender<(Option, SubscriptionInfo, UdpNotifPacketDecoded)>, + async_channel::Sender, Manifest, JoinHandle>, YangPushEnrichmentActorHandle, @@ -655,7 +658,7 @@ mod tests { } fn create_subscription_started( - peer: SocketAddr, + peer_ip: IpAddr, id: SubscriptionId, ) -> (SubscriptionInfo, serde_json::Value, UdpNotifPacketDecoded) { let payload = json!({ @@ -679,7 +682,6 @@ mod tests { } }); - let collector = SocketAddr::from(([127, 0, 0, 1], 10000)); let packet = UdpNotifPacket::new( MediaType::YangDataJson, 1234, @@ -690,9 +692,7 @@ mod tests { let decoded: UdpNotifPacketDecoded = (&packet).try_into().unwrap(); let subscription_info = SubscriptionInfo::new( - collector, - None, - peer, + peer_ip, id, Target::new_datastore( DatastoreName::Operational.to_string(), @@ -722,21 +722,30 @@ mod tests { async fn test_process_payload_empty_subscription() { // Set up the enrichment actor and input test data let (msgs_tx, test_manifest, join_handle, actor_handle) = create_actor_handle(); - let collector = SocketAddr::from(([127, 0, 0, 1], 10000)); - let peer = SocketAddr::from(([127, 0, 0, 1], 12345)); - let (_subscription_info, json_payload, decoded) = create_subscription_started(peer, 1); - let empty_subscription_info = SubscriptionInfo::new_empty(collector, None, peer, 1); + let peer_ip = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)); + let (_subscription_info, json_payload, decoded) = create_subscription_started(peer_ip, 1); + let empty_subscription_info = SubscriptionInfo::new_empty(peer_ip, 1); msgs_tx - .send(( - Some(empty_subscription_info.content_id().clone()), - empty_subscription_info.clone(), - decoded.clone(), - )) + .send(ValidatedNotification { + cached_content_id: Some(empty_subscription_info.content_id().clone()), + subscription_info: empty_subscription_info.clone(), + session: SessionInfo::new( + SocketAddr::from(([127, 0, 0, 1], 10000)), + None, + SocketAddr::new(peer_ip, 0), + ), + packet: decoded.clone(), + }) .await .expect("Failed to send message to the actor"); tokio::task::yield_now().await; - let (received_content_id, received_subscription_info, received_enriched) = actor_handle + let EnrichedNotification { + cached_content_id: received_content_id, + subscription_info: received_subscription_info, + message: received_enriched, + .. + } = actor_handle .enriched_rx .recv() .await @@ -756,9 +765,9 @@ mod tests { None, SessionProtocol::YangPush, IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), - Some(12345), - None, - None, + Some(0), + Some(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))), + Some(10000), None, ), Some(test_manifest.clone()), @@ -785,19 +794,29 @@ mod tests { async fn test_process_payload_envelope() { // Set up the enrichment actor and input test data let (msgs_tx, test_manifest, join_handle, actor_handle) = create_actor_handle(); - let peer = SocketAddr::from(([127, 0, 0, 1], 12345)); - let (subscription_info, json_payload, decoded) = create_subscription_started(peer, 1); + let peer_ip = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)); + let (subscription_info, json_payload, decoded) = create_subscription_started(peer_ip, 1); msgs_tx - .send(( - Some(subscription_info.content_id().clone()), - subscription_info.clone(), - decoded.clone(), - )) + .send(ValidatedNotification { + cached_content_id: Some(subscription_info.content_id().clone()), + subscription_info: subscription_info.clone(), + session: SessionInfo::new( + SocketAddr::from(([127, 0, 0, 1], 10000)), + None, + SocketAddr::new(peer_ip, 0), + ), + packet: decoded.clone(), + }) .await .expect("Failed to send message to the actor"); tokio::task::yield_now().await; - let (received_content_id, received_subscription_info, received_enriched) = actor_handle + let EnrichedNotification { + cached_content_id: received_content_id, + subscription_info: received_subscription_info, + message: received_enriched, + .. + } = actor_handle .enriched_rx .recv() .await @@ -818,9 +837,9 @@ mod tests { None, SessionProtocol::YangPush, IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), - Some(12345), - None, - None, + Some(0), + Some(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))), + Some(10000), Some(expected_metadata), ), Some(test_manifest.clone()), @@ -845,8 +864,7 @@ mod tests { #[test] fn test_process_payload_envelope_without_content() { let mut actor = create_actor(); - let collector = SocketAddr::from(([127, 0, 0, 1], 10000)); - let peer = SocketAddr::from(([127, 0, 0, 1], 12345)); + let peer_ip = IpAddr::from([127, 0, 0, 1]); // Create a UdpNotifPayload without content let payload = json!({ @@ -867,11 +885,17 @@ mod tests { Bytes::from(payload), ); - let subscription_info = SubscriptionInfo::new_empty(collector, None, peer, 1); + let subscription_info = SubscriptionInfo::new_empty(peer_ip, 1); // Attempt to decode the packet (should succeed) let decoded: UdpNotifPacketDecoded = (&packet).try_into().unwrap(); + let session = SessionInfo::new( + SocketAddr::from(([127, 0, 0, 1], 10000)), + None, + SocketAddr::new(peer_ip, 0), + ); - let result = actor.process_decoded_udp_notif_packet(None, &subscription_info, &decoded); + let result = + actor.process_decoded_udp_notif_packet(None, &subscription_info, &session, &decoded); assert_eq!( result, diff --git a/crates/udp-notif-service/src/actor.rs b/crates/udp-notif-service/src/actor.rs index 0e784177..788ebcf5 100644 --- a/crates/udp-notif-service/src/actor.rs +++ b/crates/udp-notif-service/src/actor.rs @@ -115,8 +115,8 @@ //! ``` use crate::{ - ActorId, OTL_UDP_NOTIF_PUBLISHER_ID_KEY, SubscriberId, Subscription, UdpNotifPacket, - UdpNotifReceiver, UdpNotifRequest, UdpNotifSender, create_udp_notif_channel, + ActorId, OTL_UDP_NOTIF_PUBLISHER_ID_KEY, SessionInfo, SubscriberId, Subscription, + UdpNotifPacket, UdpNotifReceiver, UdpNotifRequest, UdpNotifSender, create_udp_notif_channel, }; use bytes::{Bytes, BytesMut}; use futures_util::StreamExt; @@ -645,9 +645,11 @@ impl UdpNotifActor { ); let mut send_handlers = vec![]; let request = Arc::new(UdpNotifRequest::new( - self.socket_addr, - self.interface_bind.clone().map(String::into_boxed_str), - peer, + SessionInfo::new( + self.socket_addr, + self.interface_bind.clone().map(String::into_boxed_str), + peer, + ), msg, )); for (id, tx) in &self.subscribers { @@ -1322,15 +1324,11 @@ mod tests { // Create references to compare with the received ones let ref1 = Ok(Ok(Arc::new(UdpNotifRequest::new( - handle.local_addr(), - None, - local_addr1, + SessionInfo::new(handle.local_addr(), None, local_addr1), pkt1.clone(), )))); let ref2 = Ok(Ok(Arc::new(UdpNotifRequest::new( - handle.local_addr(), - None, - local_addr2, + SessionInfo::new(handle.local_addr(), None, local_addr2), pkt2.clone(), )))); diff --git a/crates/udp-notif-service/src/lib.rs b/crates/udp-notif-service/src/lib.rs index 67425346..9794dcfa 100644 --- a/crates/udp-notif-service/src/lib.rs +++ b/crates/udp-notif-service/src/lib.rs @@ -38,38 +38,65 @@ pub type ActorId = u32; /// Type alias to that YANG-Push subscription ID as defined in [RFC8641](https://datatracker.ietf.org/doc/html/rfc8641) pub type SubscriberId = u32; -/// The UDP-Notif packet and the peer [SocketAddr] that sent it. +/// Transport session context for a UDP-Notif packet. +/// +/// - **collector**: The local socket address on which the collector received +/// the packet. +/// - **interface**: The network interface or VRF the collector socket is bound +/// to, if any. +/// - **peer**: The full remote socket address (IP + source port) of the sending +/// device. +#[derive(Debug, Clone, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)] +pub struct SessionInfo { + pub collector: SocketAddr, + pub interface: Option>, + pub peer: SocketAddr, +} + +impl SessionInfo { + pub fn new(collector: SocketAddr, interface: Option>, peer: SocketAddr) -> Self { + Self { + collector, + interface, + peer, + } + } + + pub const fn collector(&self) -> SocketAddr { + self.collector + } + + pub fn interface(&self) -> Option<&str> { + self.interface.as_deref() + } + + pub const fn peer(&self) -> SocketAddr { + self.peer + } +} + +/// The UDP-Notif packet together with the transport session it arrived on. #[derive(Debug, Clone, Eq, PartialEq)] pub struct UdpNotifRequest { - collector_address: SocketAddr, - collector_interface: Option>, - peer_address: SocketAddr, + session: SessionInfo, packet: UdpNotifPacket, } impl UdpNotifRequest { - pub fn new( - collector_address: SocketAddr, - collector_interface: Option>, - peer_address: SocketAddr, - packet: UdpNotifPacket, - ) -> Self { - Self { - collector_address, - collector_interface, - peer_address, - packet, - } + pub fn new(session: SessionInfo, packet: UdpNotifPacket) -> Self { + Self { session, packet } + } + pub const fn session(&self) -> &SessionInfo { + &self.session } - pub const fn collector_address(&self) -> SocketAddr { - self.collector_address + self.session.collector } pub fn collector_interface(&self) -> Option<&str> { - self.collector_interface.as_deref() + self.session.interface.as_deref() } pub const fn peer_address(&self) -> SocketAddr { - self.peer_address + self.session.peer } pub const fn packet(&self) -> &UdpNotifPacket { &self.packet diff --git a/crates/yang-push/src/cache/actor.rs b/crates/yang-push/src/cache/actor.rs index e411090d..29b0ce66 100644 --- a/crates/yang-push/src/cache/actor.rs +++ b/crates/yang-push/src/cache/actor.rs @@ -177,7 +177,7 @@ //! let (tx, rx) = async_channel::unbounded(); //! let subscription_info = /* ... */; //! handle1.request_tx() -//! .send(CacheLookupCommand::LookupBySubscriptionInfo(subscription_info, tx)) +//! .send(CacheLookupCommand::LookupBySubscriptionInfo { subscription_info, session, tx }) //! .await?; //! //! if let Some(yang_lib_ref) = rx.recv().await? { @@ -191,7 +191,7 @@ //! let (tx, rx) = oneshot::channel(); //! let content_id = "abc123".into(); //! handle2.request_tx() -//! .send(CacheLookupCommand::LookupByContentIdOneShot(content_id, tx)) +//! .send(CacheLookupCommand::LookupByContentIdOneShot { content_id, tx }) //! .await?; //! //! if let Some(yang_lib_ref) = rx.await? { @@ -224,8 +224,8 @@ use crate::{ use futures_util::StreamExt; use futures_util::stream::FuturesUnordered; use netcalyx_netconf_proto::yang_push::types::SubscriptionId; +use netcalyx_udp_notif_service::SessionInfo; use rustc_hash::FxHashMap; -use std::net::SocketAddr; use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; @@ -305,77 +305,96 @@ pub enum CacheActorCommand { #[derive(Debug)] pub enum CacheLookupCommand { - LookupBySubscriptionInfo(SubscriptionInfo, async_channel::Sender), + LookupBySubscriptionInfo { + subscription_info: SubscriptionInfo, + session: SessionInfo, + tx: async_channel::Sender, + }, - LookupBySubscriptionInfoOneShot(SubscriptionInfo, oneshot::Sender), + LookupBySubscriptionInfoOneShot { + subscription_info: SubscriptionInfo, + session: SessionInfo, + tx: oneshot::Sender, + }, LookupBySubscriptionId { - collector: SocketAddr, - interface: Option, - peer: SocketAddr, subscription_id: SubscriptionId, + session: SessionInfo, tx: async_channel::Sender, }, LookupBySubscriptionIdOneShot { - collector: SocketAddr, - interface: Option, - peer: SocketAddr, subscription_id: SubscriptionId, + session: SessionInfo, tx: oneshot::Sender, }, - LookupByContentId( - ContentId, - async_channel::Sender<(ContentId, Option>)>, - ), + LookupByContentId { + content_id: ContentId, + tx: async_channel::Sender<(ContentId, Option>)>, + }, - LookupByContentIdOneShot( - ContentId, - oneshot::Sender<(ContentId, Option>)>, - ), + LookupByContentIdOneShot { + content_id: ContentId, + tx: oneshot::Sender<(ContentId, Option>)>, + }, } impl std::fmt::Display for CacheLookupCommand { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - Self::LookupBySubscriptionInfo(subscription_info, _) => { - write!(f, "lookup by subscription info {subscription_info}") + Self::LookupBySubscriptionInfo { + subscription_info, + session, + tx: _, + } => { + write!( + f, + "lookup by subscription info {subscription_info} from peer {}", + session.peer() + ) } - Self::LookupBySubscriptionInfoOneShot(subscription_info, _) => { + Self::LookupBySubscriptionInfoOneShot { + subscription_info, + session, + tx: _, + } => { write!( f, - "lookup by subscription info {subscription_info} (one shot)" + "lookup by subscription info {subscription_info} from peer {} (one shot)", + session.peer() ) } Self::LookupBySubscriptionId { - collector, - interface, - peer, subscription_id, + session, tx: _tx, } => { write!( f, - "lookup by subscription id {subscription_id} from peer {peer}, collector {collector}, interface {interface:?}", + "lookup by subscription id {subscription_id} from peer {}, collector {}, interface {:?}", + session.peer(), + session.collector(), + session.interface(), ) } Self::LookupBySubscriptionIdOneShot { - collector, - interface, - peer, subscription_id, + session, tx: _tx, } => { write!( f, - "lookup by subscription id {subscription_id} from peer {peer}, collector {collector}, interface {interface:?} (one shot)", + "lookup by subscription id {subscription_id} from peer {}, collector {}, interface {:?} (one shot)", + session.peer(), + session.collector(), + session.interface(), ) } - Self::LookupByContentId(content_id, _) => { + Self::LookupByContentId { content_id, tx: _ } => { write!(f, "lookup by content id {content_id}") } - Self::LookupByContentIdOneShot(content_id, _) => { + Self::LookupByContentIdOneShot { content_id, tx: _ } => { write!(f, "lookup by content id {content_id} (one shot)") } } @@ -455,7 +474,7 @@ impl CacheActor { match sender.send(response).await { Ok(_) => { debug!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -466,7 +485,7 @@ impl CacheActor { } Err(err) => { warn!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -525,7 +544,7 @@ impl CacheActor { match sender.send(response) { Ok(_) => { debug!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -536,7 +555,7 @@ impl CacheActor { } Err(_err) => { warn!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -588,11 +607,7 @@ impl CacheActor { &[ opentelemetry::KeyValue::new( "network.peer.address", - format!("{}", subscription_info.peer().ip()), - ), - opentelemetry::KeyValue::new( - "network.peer.port", - opentelemetry::Value::I64(subscription_info.peer().port().into()), + format!("{}", subscription_info.peer_ip()), ), opentelemetry::KeyValue::new( OTL_YANG_PUSH_SUBSCRIPTION_ID_KEY, @@ -610,7 +625,7 @@ impl CacheActor { ], ); warn!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -632,7 +647,7 @@ impl CacheActor { .await .map_err(|error| { warn!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -641,7 +656,7 @@ impl CacheActor { ); }); debug!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -655,11 +670,7 @@ impl CacheActor { let otl_tags = [ opentelemetry::KeyValue::new( "network.peer.address", - format!("{}", subscription_info.peer().ip()), - ), - opentelemetry::KeyValue::new( - "network.peer.port", - opentelemetry::Value::I64(subscription_info.peer().port().into()), + format!("{}", subscription_info.peer_ip()), ), opentelemetry::KeyValue::new( OTL_YANG_PUSH_SUBSCRIPTION_ID_KEY, @@ -695,7 +706,7 @@ impl CacheActor { let yang_lib_ref = match result { Ok(yang_lib_ref) => { info!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -705,7 +716,7 @@ impl CacheActor { } Err(err) => { warn!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -718,9 +729,7 @@ impl CacheActor { // First, remove all pending requests for this subscription info that are // requested with full subscription info let empty = SubscriptionInfo::new_empty( - subscription_info.collector(), - subscription_info.interface(), - subscription_info.peer(), + subscription_info.peer_ip(), subscription_info.id(), ); let mut pending_senders = self @@ -755,11 +764,7 @@ impl CacheActor { Vec::from([ opentelemetry::KeyValue::new( "network.peer.address", - format!("{}", subscription_info.peer().ip()), - ), - opentelemetry::KeyValue::new( - "network.peer.port", - opentelemetry::Value::I64(subscription_info.peer().port().into()), + format!("{}", subscription_info.peer_ip()), ), opentelemetry::KeyValue::new( OTL_YANG_PUSH_SUBSCRIPTION_ID_KEY, @@ -779,11 +784,15 @@ impl CacheActor { async fn process_request(&mut self, request: CacheLookupCommand) { match request { - CacheLookupCommand::LookupBySubscriptionInfo(subscription_info, sender) => { + CacheLookupCommand::LookupBySubscriptionInfo { + subscription_info, + session, + tx: sender, + } => { let otl_tags = Self::otl_tags_from_subscription_inf(&subscription_info); self.stats.requests_received.add(1, otl_tags.as_ref()); debug!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -795,7 +804,7 @@ impl CacheActor { match yang_lib_ref { Some(yang_lib_ref) => { info!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -819,7 +828,7 @@ impl CacheActor { if should_fetch { info!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -828,14 +837,15 @@ impl CacheActor { self.stats.device_fetch_request.add(1, &otl_tags); let job_result = tokio::time::timeout( self.fetcher_timeout, - self.fetcher.fetch(subscription_info.clone()), + self.fetcher + .fetch(subscription_info.clone(), session.clone()), ) .await; let job = match job_result { Ok(worker_result) => worker_result, Err(err) => { warn!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -855,7 +865,7 @@ impl CacheActor { .record(self.workers_queue.len() as u64, &[]); } else { debug!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -869,11 +879,15 @@ impl CacheActor { } } } - CacheLookupCommand::LookupBySubscriptionInfoOneShot(subscription_info, sender) => { + CacheLookupCommand::LookupBySubscriptionInfoOneShot { + subscription_info, + session, + tx: sender, + } => { let mut otl_tags = Self::otl_tags_from_subscription_inf(&subscription_info); self.stats.requests_received.add(1, otl_tags.as_ref()); debug!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -885,7 +899,7 @@ impl CacheActor { .get_by_subscription_info(&subscription_info) { info!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -898,7 +912,8 @@ impl CacheActor { self.stats.device_fetch_request.add(1, &otl_tags); let worker_result = tokio::time::timeout( self.fetcher_timeout, - self.fetcher.fetch_blocking(subscription_info.clone()), + self.fetcher + .fetch_blocking(subscription_info.clone(), session.clone()), ) .await; @@ -924,18 +939,13 @@ impl CacheActor { Self::send_yang_lib_ref_oneshot(&subscription_info, yang_lib_ref, sender); } CacheLookupCommand::LookupBySubscriptionId { - collector, - interface, - peer, subscription_id, + session, tx, } => { + let peer_ip = session.peer().ip(); let otel_tags = [ - opentelemetry::KeyValue::new("network.peer.address", format!("{}", peer.ip())), - opentelemetry::KeyValue::new( - "network.peer.port", - opentelemetry::Value::I64(peer.port().into()), - ), + opentelemetry::KeyValue::new("network.peer.address", format!("{peer_ip}")), opentelemetry::KeyValue::new( OTL_YANG_PUSH_SUBSCRIPTION_ID_KEY, opentelemetry::Value::I64(subscription_id.into()), @@ -943,29 +953,24 @@ impl CacheActor { ]; self.stats.requests_received.add(1, &otel_tags); debug!( - peer=%peer, + peer_ip=%peer_ip, subscription_id, "processing cache lookup by subscription id request" ); let response = self .schema_cache - .get_by_subscription_id(peer.ip(), subscription_id); + .get_by_subscription_id(peer_ip, subscription_id); if let Some((subscription_info, yang_lib_ref)) = response { self.stats.cache_hits.add(1, &otel_tags); Self::send_yang_lib_ref(&subscription_info, yang_lib_ref, tx).await; } else { warn!( - peer=%peer, + peer_ip=%peer_ip, subscription_id, "cache miss: subscription id not found in cache" ); self.stats.cache_misses.add(1, &otel_tags); - let subscription_info = SubscriptionInfo::new_empty( - collector, - interface.clone(), - peer, - subscription_id, - ); + let subscription_info = SubscriptionInfo::new_empty(peer_ip, subscription_id); let entry = self .pending_requests .entry(subscription_info.clone()) @@ -975,26 +980,22 @@ impl CacheActor { if should_fetch { info!( - peer=%peer, + peer_ip=%peer_ip, subscription_id, "cache miss: starting fetch from device by subscription id" ); self.stats.device_fetch_request.add(1, &otel_tags); let job_result = tokio::time::timeout( self.fetcher_timeout, - self.fetcher.fetch_by_subscription_id( - collector, - interface.clone().map(String::into_boxed_str), - peer, - subscription_id, - ), + self.fetcher + .fetch_by_subscription_id(session.clone(), subscription_id), ) .await; let job = match job_result { Ok(worker_result) => worker_result, Err(err) => { warn!( - peer=%peer, + peer_ip=%peer_ip, subscription_id, router_content_id=subscription_info.clone().content_id().clone(), target=%subscription_info.target(), @@ -1014,7 +1015,7 @@ impl CacheActor { .record(self.workers_queue.len() as u64, &[]); } else { debug!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -1028,18 +1029,13 @@ impl CacheActor { } } CacheLookupCommand::LookupBySubscriptionIdOneShot { - collector, - interface, - peer, subscription_id, + session, tx, } => { + let peer_ip = session.peer().ip(); let mut otel_tags = vec![ - opentelemetry::KeyValue::new("network.peer.address", format!("{}", peer.ip())), - opentelemetry::KeyValue::new( - "network.peer.port", - opentelemetry::Value::I64(peer.port().into()), - ), + opentelemetry::KeyValue::new("network.peer.address", format!("{peer_ip}")), opentelemetry::KeyValue::new( OTL_YANG_PUSH_SUBSCRIPTION_ID_KEY, opentelemetry::Value::I64(subscription_id.into()), @@ -1047,18 +1043,18 @@ impl CacheActor { ]; self.stats.requests_received.add(1, &otel_tags); debug!( - peer=%peer, + peer_ip=%peer_ip, subscription_id, "processing cache lookup by subscription id request (one shot)"); let response = self .schema_cache - .get_by_subscription_id(peer.ip(), subscription_id); + .get_by_subscription_id(peer_ip, subscription_id); if let Some((subscription_info, yang_lib_ref)) = response { self.stats.cache_hits.add(1, &otel_tags); Self::send_yang_lib_ref_oneshot(&subscription_info, yang_lib_ref, tx); } else { warn!( - peer=%peer, + peer_ip=%peer_ip, subscription_id, "cache miss: subscription id not found in cache" ); @@ -1067,12 +1063,8 @@ impl CacheActor { let worker_result = tokio::time::timeout( self.fetcher_timeout, - self.fetcher.fetch_by_subscription_id_blocking( - collector, - interface.clone().map(String::into_boxed_str), - peer, - subscription_id, - ), + self.fetcher + .fetch_by_subscription_id_blocking(session.clone(), subscription_id), ) .await; let worker_result = match worker_result { @@ -1086,19 +1078,17 @@ impl CacheActor { format!("{err}"), )); self.stats.device_fetch_failed.add(1, &otel_tags); - let empty = SubscriptionInfo::new_empty( - collector, - interface, - peer, - subscription_id, - ); + let empty = SubscriptionInfo::new_empty(peer_ip, subscription_id); Err(Box::new((empty.clone(), err.into()))) } }; self.process_worker_result(Ok(worker_result)).await; } } - CacheLookupCommand::LookupByContentId(content_id, sender) => { + CacheLookupCommand::LookupByContentId { + content_id, + tx: sender, + } => { let otl_tags = [opentelemetry::KeyValue::new( OTL_YANG_PUSH_CACHED_CONTENT_ID_KEY, content_id.to_string(), @@ -1113,7 +1103,10 @@ impl CacheActor { } Self::send_yang_lib_ref_content_id(&content_id, yang_lib_ref, sender).await; } - CacheLookupCommand::LookupByContentIdOneShot(content_id, sender) => { + CacheLookupCommand::LookupByContentIdOneShot { + content_id, + tx: sender, + } => { let otl_tags = [opentelemetry::KeyValue::new( OTL_YANG_PUSH_CACHED_CONTENT_ID_KEY, content_id.to_string(), @@ -1295,11 +1288,17 @@ pub(crate) mod tests { use std::path::Path; use std::time::Duration; - pub(crate) fn test_subscription_info() -> SubscriptionInfo { - SubscriptionInfo::new( + pub(crate) fn test_session_info() -> SessionInfo { + SessionInfo::new( SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100)), 10000), None, - SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 200)), 830), + SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 200)), 50000), + ) + } + + pub(crate) fn test_subscription_info() -> SubscriptionInfo { + SubscriptionInfo::new( + IpAddr::V4(Ipv4Addr::new(192, 168, 1, 200)), 1, Target::new_datastore( "ds:operational".to_string(), @@ -1435,10 +1434,11 @@ pub(crate) mod tests { let (tx, rx) = async_channel::unbounded(); handle .request_tx() - .send(CacheLookupCommand::LookupBySubscriptionInfo( - subscription_info.clone(), + .send(CacheLookupCommand::LookupBySubscriptionInfo { + subscription_info: subscription_info.clone(), + session: test_session_info(), tx, - )) + }) .await .unwrap(); @@ -1487,10 +1487,11 @@ pub(crate) mod tests { let (tx, rx) = async_channel::unbounded(); handle .request_tx() - .send(CacheLookupCommand::LookupBySubscriptionInfo( - subscription_info.clone(), + .send(CacheLookupCommand::LookupBySubscriptionInfo { + subscription_info: subscription_info.clone(), + session: test_session_info(), tx, - )) + }) .await .unwrap(); @@ -1515,10 +1516,11 @@ pub(crate) mod tests { let (tx, rx) = async_channel::unbounded(); handle .request_tx() - .send(CacheLookupCommand::LookupBySubscriptionInfo( - subscription_info.clone(), + .send(CacheLookupCommand::LookupBySubscriptionInfo { + subscription_info: subscription_info.clone(), + session: test_session_info(), tx, - )) + }) .await .unwrap(); @@ -1563,7 +1565,11 @@ pub(crate) mod tests { tasks.push_back(tokio::spawn(async move { let (tx, rx) = async_channel::unbounded(); h.request_tx() - .send(CacheLookupCommand::LookupBySubscriptionInfo(sub, tx)) + .send(CacheLookupCommand::LookupBySubscriptionInfo { + subscription_info: sub, + session: test_session_info(), + tx, + }) .await .unwrap(); tokio::time::timeout(Duration::from_secs(1), rx.recv()) @@ -1612,10 +1618,8 @@ pub(crate) mod tests { handle .request_tx() .send(CacheLookupCommand::LookupBySubscriptionId { - collector: subscription_info.collector(), - interface: subscription_info.interface(), - peer: subscription_info.peer(), subscription_id: subscription_info.id(), + session: test_session_info(), tx, }) .await @@ -1629,12 +1633,7 @@ pub(crate) mod tests { .expect("failed to receive response"); assert_eq!( response.subscription_info(), - &SubscriptionInfo::new_empty( - subscription_info.collector(), - subscription_info.interface(), - subscription_info.peer(), - subscription_info.id(), - ) + &SubscriptionInfo::new_empty(subscription_info.peer_ip(), subscription_info.id(),) ); assert_eq!(response.yang_lib_ref(), None); @@ -1647,9 +1646,7 @@ pub(crate) mod tests { assert_eq!(hits_counts.len(), 1); assert_eq!( hits_counts.get(&SubscriptionInfo::new_empty( - subscription_info.collector(), - subscription_info.interface(), - subscription_info.peer(), + subscription_info.peer_ip(), subscription_info.id() )), Some(&1) @@ -1684,10 +1681,12 @@ pub(crate) mod tests { handle .request_tx() .send(CacheLookupCommand::LookupBySubscriptionId { - collector: subscription_info.collector(), - interface: subscription_info.interface(), - peer: subscription_info.peer(), subscription_id: subscription_info.id(), + session: SessionInfo::new( + SocketAddr::from(([127, 0, 0, 1], 10000)), + None, + SocketAddr::new(subscription_info.peer_ip(), 0), + ), tx, }) .await @@ -1714,10 +1713,11 @@ pub(crate) mod tests { let (tx, rx) = async_channel::unbounded(); handle .request_tx() - .send(CacheLookupCommand::LookupBySubscriptionInfo( - subscription_info.clone(), + .send(CacheLookupCommand::LookupBySubscriptionInfo { + subscription_info: subscription_info.clone(), + session: test_session_info(), tx, - )) + }) .await .unwrap(); @@ -1763,10 +1763,12 @@ pub(crate) mod tests { let (tx, rx) = async_channel::unbounded(); h.request_tx() .send(CacheLookupCommand::LookupBySubscriptionId { - collector: sub.collector(), - interface: sub.interface(), - peer: sub.peer(), subscription_id: sub.id(), + session: SessionInfo::new( + SocketAddr::from(([127, 0, 0, 1], 10000)), + None, + SocketAddr::new(sub.peer_ip(), 0), + ), tx, }) .await diff --git a/crates/yang-push/src/cache/fetcher.rs b/crates/yang-push/src/cache/fetcher.rs index f6bca478..6c30570e 100644 --- a/crates/yang-push/src/cache/fetcher.rs +++ b/crates/yang-push/src/cache/fetcher.rs @@ -35,13 +35,14 @@ use netcalyx_netconf_proto::yang_push::subscription::{ }; use netcalyx_netconf_proto::yang_push::types::SubscriptionId; use netcalyx_netconf_proto::yanglib::{DatastoreName, PermissiveVersionChecker, YangLibrary}; +use netcalyx_udp_notif_service::SessionInfo; use rand::RngExt; use std::collections::{HashMap, HashSet}; use std::future::Future; -use std::net::SocketAddr; +use std::net::{IpAddr, SocketAddr}; use std::sync::Arc; use tokio::task::JoinHandle; -use tracing::{error, info, trace, warn}; +use tracing::{debug, error, info, trace, warn}; pub type FetcherResult = Result< (SubscriptionInfo, YangLibrary, HashMap, Box>), @@ -55,27 +56,25 @@ pub trait YangLibraryFetcher { fn fetch( &self, subscription_info: SubscriptionInfo, + session: SessionInfo, ) -> impl Future> + Send; /// A blocking version which returns directly the YANG library and schemas. fn fetch_blocking( &self, subscription_info: SubscriptionInfo, + session: SessionInfo, ) -> impl Future + Send; fn fetch_by_subscription_id( &self, - collector: SocketAddr, - interface: Option>, - peer: SocketAddr, + session: SessionInfo, subscription_id: SubscriptionId, ) -> impl Future> + Send; fn fetch_by_subscription_id_blocking( &self, - collector: SocketAddr, - interface: Option>, - peer: SocketAddr, + session: SessionInfo, subscription_id: SubscriptionId, ) -> impl Future + Send; } @@ -146,17 +145,18 @@ impl NetconfYangLibraryFetcher { async fn fetch_from_device( cfg: &FetchConfig, subscription_info: SubscriptionInfo, + session: SessionInfo, ) -> FetcherResult { - let collector = subscription_info.collector(); - let interface = subscription_info.interface(); - let peer = subscription_info.peer(); + let collector = session.collector(); + let interface = session.interface(); + let peer_ip = subscription_info.peer_ip(); let subscription_id = subscription_info.id(); - let host = SocketAddr::new(peer.ip(), cfg.default_port); + let host = SocketAddr::new(peer_ip, cfg.default_port); info!( host=%host, collector=%collector, interface, - peer=%peer, + peer_ip=%peer_ip, subscription_id, "starting fetching YANG Library from device", ); @@ -170,7 +170,7 @@ impl NetconfYangLibraryFetcher { auth, host, None, - interface, + interface.map(str::to_string), announce_caps, ssh_handler, Arc::clone(&cfg.client_config), @@ -208,7 +208,7 @@ impl NetconfYangLibraryFetcher { } info!( host=%host, - peer=%peer, + peer_ip=%peer_ip, subscription_id, cached_content_id=yang_lib.content_id(), schema_count=schemas.len(), @@ -219,17 +219,18 @@ impl NetconfYangLibraryFetcher { async fn fetch_from_device_by_id( cfg: &FetchConfig, - collector: SocketAddr, - interface: Option>, - peer: SocketAddr, + session: SessionInfo, subscription_id: SubscriptionId, ) -> FetcherResult { - let host = SocketAddr::new(peer.ip(), cfg.default_port); + let collector = session.collector(); + let interface = session.interface(); + let peer_ip = session.peer().ip(); + let host = SocketAddr::new(peer_ip, cfg.default_port); info!( host=%host, collector=%collector, interface, - peer=%peer, + peer_ip=%peer_ip, subscription_id, "starting fetching YANG Library from device", ); @@ -243,19 +244,14 @@ impl NetconfYangLibraryFetcher { auth, host, None, - interface.clone().map(String::from), + interface.map(String::from), announce_caps, ssh_handler, Arc::clone(&cfg.client_config), ); // Empty subscription info returned in case of errors to keep track of peer and // subscription ID - let empty = SubscriptionInfo::new_empty( - collector, - interface.clone().map(String::from), - peer, - subscription_id, - ); + let empty = SubscriptionInfo::new_empty(peer_ip, subscription_id); let mut client = match tokio::time::timeout(cfg.timeout, connect(config)).await { Ok(Ok(c)) => c, Ok(Err(err)) => { @@ -278,6 +274,12 @@ impl NetconfYangLibraryFetcher { .map_err(|err| Box::new((empty.clone(), err.into())))?; let modules = if let Some(modules) = &subscription.module_version { + debug!( + peer_ip=%peer_ip, + subscription_id, + modules=?modules, + "using module-version reported by device for subscription", + ); modules.clone().to_vec() } else { let (ds_name, namespaces) = match &subscription.target { @@ -312,15 +314,49 @@ impl NetconfYangLibraryFetcher { } }, }; + debug!( + peer_ip=%peer_ip, + subscription_id, + ds_name=?ds_name, + namespaces=?namespaces, + target=?subscription.target, + "no module-version reported by device, resolving target namespaces against YANG Library instead", + ); let mut ret = Vec::with_capacity(namespaces.len()); - for (_prefix, namespace) in namespaces { - let module = router_yang_library.find_module_by_datastore_and_ns(&ds_name, namespace).ok_or_else(|| Box::new((empty.clone(), YangLibraryCacheError::IoError(std::io::Error::other(format!("module with namespace {namespace} not found in YANG Library for datastore {ds_name}"))))))?; + for (prefix, namespace) in namespaces { + let module = router_yang_library.find_module_by_datastore_and_ns(&ds_name, namespace).ok_or_else(|| { + warn!( + peer_ip=%peer_ip, + subscription_id, + ds_name=?ds_name, + prefix, + namespace, + "module with namespace not found in YANG Library for datastore", + ); + Box::new((empty.clone(), YangLibraryCacheError::IoError(std::io::Error::other(format!("module with namespace {namespace} not found in YANG Library for datastore {ds_name}"))))) + })?; + trace!( + peer_ip=%peer_ip, + subscription_id, + prefix, + namespace, + module_name=module.name(), + "resolved xpath-filter prefix to module via namespace", + ); ret.push(YangPushModuleVersion::new( module.name().into(), module.revision().map(|x| x.into()), None, )); } + if ret.is_empty() { + warn!( + peer_ip=%peer_ip, + subscription_id, + target=?subscription.target, + "target namespaces resolution produced no modules; the target's YANG module(s) will not be fetched", + ); + } ret }; @@ -328,6 +364,12 @@ impl NetconfYangLibraryFetcher { if !module_names.contains(&"ietf-subscribed-notifications") { module_names.push("ietf-subscribed-notifications"); } + debug!( + peer_ip=%peer_ip, + subscription_id, + module_names=?module_names, + "final module list requested from device for subscription", + ); // TODO: add timeout to loading YANG Library from the device let (yang_lib, schemas) = client .load_from_modules(&module_names, &PermissiveVersionChecker) @@ -351,9 +393,7 @@ impl NetconfYangLibraryFetcher { )) })?; let subscription_info = SubscriptionInfo::new( - collector, - interface.map(String::from), - peer, + peer_ip, subscription_id, subscription_target, subscription.stop_time, @@ -366,7 +406,7 @@ impl NetconfYangLibraryFetcher { ); info!( host=%host, - peer=%peer, + peer_ip=%peer_ip, subscription_id, router_content_id=yang_lib.content_id(), target=%subscription_info.target(), @@ -382,7 +422,7 @@ impl NetconfYangLibraryFetcher { /// `operation` is called up to `retry.max_retries + 1` times. Each failed /// attempt waits `base * 2^(attempt-1)` (capped at `retry.max_backoff`) /// with equal jitter before the next try. - async fn with_retry(peer: SocketAddr, retry: RetryConfig, operation: F) -> FetcherResult + async fn with_retry(peer_ip: IpAddr, retry: RetryConfig, operation: F) -> FetcherResult where F: Fn() -> Fut, Fut: Future, @@ -396,7 +436,7 @@ impl NetconfYangLibraryFetcher { let jitter = rand::rng().random_range(0.0..=half); let delay = std::time::Duration::from_secs_f64(half + jitter); trace!( - %peer, + %peer_ip, attempt, delay_ms = delay.as_millis() as u64, "retrying YANG Library fetch after backoff", @@ -413,42 +453,43 @@ impl NetconfYangLibraryFetcher { } impl YangLibraryFetcher for NetconfYangLibraryFetcher { - async fn fetch(&self, subscription_info: SubscriptionInfo) -> JoinHandle { + async fn fetch( + &self, + subscription_info: SubscriptionInfo, + session: SessionInfo, + ) -> JoinHandle { let fetch_cfg = self.fetch_cfg.clone(); let retry_cfg = self.retry_cfg; tokio::spawn(async move { - Self::with_retry(subscription_info.peer(), retry_cfg, || { - Self::fetch_from_device(&fetch_cfg, subscription_info.clone()) + Self::with_retry(subscription_info.peer_ip(), retry_cfg, || { + Self::fetch_from_device(&fetch_cfg, subscription_info.clone(), session.clone()) }) .await }) } - async fn fetch_blocking(&self, subscription_info: SubscriptionInfo) -> FetcherResult { - Self::with_retry(subscription_info.peer(), self.retry_cfg, || { - Self::fetch_from_device(&self.fetch_cfg, subscription_info.clone()) + async fn fetch_blocking( + &self, + subscription_info: SubscriptionInfo, + session: SessionInfo, + ) -> FetcherResult { + Self::with_retry(subscription_info.peer_ip(), self.retry_cfg, || { + Self::fetch_from_device(&self.fetch_cfg, subscription_info.clone(), session.clone()) }) .await } async fn fetch_by_subscription_id( &self, - collector: SocketAddr, - interface: Option>, - peer: SocketAddr, + session: SessionInfo, subscription_id: SubscriptionId, ) -> JoinHandle { let fetch_cfg = self.fetch_cfg.clone(); let retry_cfg = self.retry_cfg; + let peer_ip = session.peer().ip(); tokio::spawn(async move { - Self::with_retry(peer, retry_cfg, || { - Self::fetch_from_device_by_id( - &fetch_cfg, - collector, - interface.clone(), - peer, - subscription_id, - ) + Self::with_retry(peer_ip, retry_cfg, || { + Self::fetch_from_device_by_id(&fetch_cfg, session.clone(), subscription_id) }) .await }) @@ -456,19 +497,12 @@ impl YangLibraryFetcher for NetconfYangLibraryFetcher { async fn fetch_by_subscription_id_blocking( &self, - collector: SocketAddr, - interface: Option>, - peer: SocketAddr, + session: SessionInfo, subscription_id: SubscriptionId, ) -> FetcherResult { - Self::with_retry(peer, self.retry_cfg, || { - Self::fetch_from_device_by_id( - &self.fetch_cfg, - collector, - interface.clone(), - peer, - subscription_id, - ) + let peer_ip = session.peer().ip(); + Self::with_retry(peer_ip, self.retry_cfg, || { + Self::fetch_from_device_by_id(&self.fetch_cfg, session.clone(), subscription_id) }) .await } @@ -493,7 +527,7 @@ pub(crate) mod tests { ) -> Self { for (subscription_info, (yang_lib, _schemas)) in &yang_libs { info!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -509,7 +543,7 @@ pub(crate) mod tests { fn get_from_cache(&self, subscription_info: SubscriptionInfo) -> FetcherResult { info!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -527,7 +561,7 @@ pub(crate) mod tests { .cloned() .ok_or_else(|| { info!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -542,23 +576,21 @@ pub(crate) mod tests { } fn get_from_cache_by_id(&self, subscription_info: SubscriptionInfo) -> FetcherResult { - let collector = subscription_info.collector(); - let interface = subscription_info.interface(); - let peer = subscription_info.peer(); + let peer_ip = subscription_info.peer_ip(); let subscription_id = subscription_info.id(); info!( - peer=%peer, + peer_ip=%peer_ip, subscription_id, "fetching from device by id" ); let subscription_info = self .yang_libs .keys() - .find(|x| x.id() == subscription_id && x.peer().ip() == peer.ip()); + .find(|x| x.id() == subscription_id && x.peer_ip() == peer_ip); let subscription_info = if let Some(subscription_info) = subscription_info { subscription_info.clone() } else { - SubscriptionInfo::new_empty(collector, interface, peer, subscription_id) + SubscriptionInfo::new_empty(peer_ip, subscription_id) }; // Increment counter in the instance state { @@ -577,7 +609,7 @@ pub(crate) mod tests { .cloned() .ok_or_else(|| { info!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -593,45 +625,41 @@ pub(crate) mod tests { } impl YangLibraryFetcher for TestYangLibFetcher { - async fn fetch(&self, subscription_info: SubscriptionInfo) -> JoinHandle { + async fn fetch( + &self, + subscription_info: SubscriptionInfo, + _session: SessionInfo, + ) -> JoinHandle { let result = self.get_from_cache(subscription_info); tokio::spawn(async move { result }) } - async fn fetch_blocking(&self, subscription_info: SubscriptionInfo) -> FetcherResult { + async fn fetch_blocking( + &self, + subscription_info: SubscriptionInfo, + _session: SessionInfo, + ) -> FetcherResult { self.get_from_cache(subscription_info) } async fn fetch_by_subscription_id( &self, - collector: SocketAddr, - interface: Option>, - peer: SocketAddr, + session: SessionInfo, subscription_id: SubscriptionId, ) -> JoinHandle { - let subscription_info = SubscriptionInfo::new_empty( - collector, - interface.map(String::from), - peer, - subscription_id, - ); + let subscription_info = + SubscriptionInfo::new_empty(session.peer().ip(), subscription_id); let result = self.get_from_cache_by_id(subscription_info); tokio::spawn(async move { result }) } async fn fetch_by_subscription_id_blocking( &self, - collector: SocketAddr, - interface: Option>, - peer: SocketAddr, + session: SessionInfo, subscription_id: SubscriptionId, ) -> FetcherResult { - let subscription_info = SubscriptionInfo::new_empty( - collector, - interface.map(String::from), - peer, - subscription_id, - ); + let subscription_info = + SubscriptionInfo::new_empty(session.peer().ip(), subscription_id); self.get_from_cache_by_id(subscription_info) } } @@ -647,22 +675,18 @@ mod retry_tests { RetryConfig::new(max_retries, std::time::Duration::from_millis(1)) } - fn collector() -> SocketAddr { - SocketAddr::from(([127, 0, 0, 1], 10000)) - } - - fn dummy_peer() -> SocketAddr { - "127.0.0.1:0".parse().unwrap() + fn dummy_peer_ip() -> IpAddr { + "127.0.0.1".parse().unwrap() } fn make_ok() -> FetcherResult { - let info = SubscriptionInfo::new_empty(collector(), None, dummy_peer(), 1); + let info = SubscriptionInfo::new_empty(dummy_peer_ip(), 1); let yang_lib = YangLibrary::new("test-content-id".into(), vec![], vec![], vec![]); Ok((info, yang_lib, HashMap::new())) } fn make_err(msg: &'static str) -> FetcherResult { - let info = SubscriptionInfo::new_empty(collector(), None, dummy_peer(), 1); + let info = SubscriptionInfo::new_empty(dummy_peer_ip(), 1); Err(Box::new(( info, YangLibraryCacheError::IoError(std::io::Error::other(msg)), @@ -676,7 +700,7 @@ mod retry_tests { let call_count = Arc::new(AtomicU32::new(0)); let cc = Arc::clone(&call_count); - let result = NetconfYangLibraryFetcher::with_retry(dummy_peer(), retry_cfg(5), || { + let result = NetconfYangLibraryFetcher::with_retry(dummy_peer_ip(), retry_cfg(5), || { let cc = Arc::clone(&cc); async move { cc.fetch_add(1, Ordering::SeqCst); @@ -705,7 +729,7 @@ mod retry_tests { let call_count = Arc::new(AtomicU32::new(0)); let cc = Arc::clone(&call_count); - let result = NetconfYangLibraryFetcher::with_retry(dummy_peer(), retry_cfg(0), || { + let result = NetconfYangLibraryFetcher::with_retry(dummy_peer_ip(), retry_cfg(0), || { let cc = Arc::clone(&cc); async move { cc.fetch_add(1, Ordering::SeqCst); @@ -736,7 +760,7 @@ mod retry_tests { const MAX_RETRIES: u32 = 3; let result = - NetconfYangLibraryFetcher::with_retry(dummy_peer(), retry_cfg(MAX_RETRIES), || { + NetconfYangLibraryFetcher::with_retry(dummy_peer_ip(), retry_cfg(MAX_RETRIES), || { let cc = Arc::clone(&cc); async move { let n = cc.fetch_add(1, Ordering::SeqCst); @@ -769,7 +793,7 @@ mod retry_tests { let cc = Arc::clone(&call_count); const FAIL_FIRST: u32 = 2; // fail twice, succeed on the 3rd call - let result = NetconfYangLibraryFetcher::with_retry(dummy_peer(), retry_cfg(5), || { + let result = NetconfYangLibraryFetcher::with_retry(dummy_peer_ip(), retry_cfg(5), || { let cc = Arc::clone(&cc); async move { let n = cc.fetch_add(1, Ordering::SeqCst); diff --git a/crates/yang-push/src/cache/storage.rs b/crates/yang-push/src/cache/storage.rs index d758623e..c00ad316 100644 --- a/crates/yang-push/src/cache/storage.rs +++ b/crates/yang-push/src/cache/storage.rs @@ -106,7 +106,7 @@ use rustc_hash::FxHashMap; use serde::{Deserialize, Serialize}; use std::collections::{HashMap, HashSet}; use std::io::Write; -use std::net::{IpAddr, SocketAddr}; +use std::net::IpAddr; use std::path::{Path, PathBuf}; use std::sync::Arc; use tracing::{debug, info, trace, warn}; @@ -727,15 +727,8 @@ impl YangLibraryReference { /// /// # Fields /// -/// - **collector**: The socket address (IP and port) of the YANG-Push collector -/// that received the subscription. -/// -/// - **interface**: The network interface name (if available) on which the -/// subscription was received. -/// -/// - **peer**: The socket address (IP and port) of the remote device that -/// established the subscription. This identifies which network element sent -/// the subscription-started notification. +/// - **peer_ip**: The IP address of the remote device that established the +/// subscription. /// /// - **content_id**: The YANG Library content identifier associated with this /// subscription. This links the subscription to a specific version of the @@ -761,31 +754,36 @@ impl YangLibraryReference { /// /// # Example /// -/// ```rust,ignore -/// use std::net::SocketAddr; +/// ```rust +/// use netcalyx_udp_notif_pkt::notification::Target; +/// use netcalyx_yang_push::cache::storage::SubscriptionInfo; +/// use std::net::IpAddr; /// /// let subscription_info = SubscriptionInfo::new( -/// "192.168.1.100:830".parse().unwrap(), -/// 1, -/// ContentId::from("2024-01-15-content-id"), +/// "192.168.1.1".parse::().unwrap(), // peer_ip +/// 1, // subscription id /// Target::new_datastore( /// "ds:operational".to_string(), /// either::Right("/ietf-interfaces:interfaces/ietf-interfaces/statistics".to_string()), /// ), -/// vec!["ietf-interfaces".to_string(), "ietf-ip".to_string()], +/// None, +/// None, +/// None, +/// None, +/// None, +/// Box::new([]), +/// "2024-01-15-content-id".to_string(), /// ); /// /// // Access subscription details -/// println!("Peer: {}", subscription_info.peer()); +/// println!("Peer IP: {}", subscription_info.peer_ip()); /// println!("Content ID: {}", subscription_info.content_id()); /// println!("Subscription Target: {}", subscription_info.target()); /// println!("Models: {:?}", subscription_info.models()); /// ``` #[derive(Clone, Debug, Eq, Hash, PartialEq, serde::Serialize, serde::Deserialize)] pub struct SubscriptionInfo { - pub collector: SocketAddr, - pub interface: Option, - pub peer: SocketAddr, + pub peer_ip: IpAddr, pub id: SubscriptionId, pub target: Target, @@ -806,15 +804,14 @@ pub struct SubscriptionInfo { pub models: Box<[YangPushModuleVersion]>, + /// router content-id for the advertised YANG Library pub content_id: ContentId, } impl SubscriptionInfo { #[allow(clippy::too_many_arguments)] pub fn new( - collector: SocketAddr, - interface: Option, - peer: SocketAddr, + peer_ip: IpAddr, id: SubscriptionId, target: Target, stop_time: Option>, @@ -826,9 +823,7 @@ impl SubscriptionInfo { content_id: ContentId, ) -> Self { Self { - collector, - interface, - peer, + peer_ip, id, target, stop_time, @@ -844,16 +839,9 @@ impl SubscriptionInfo { /// Create an empty subscription info placeholder. /// This can be used when no subscription info is available. /// Or to indicate that no yang library is associated with the subscription. - pub fn new_empty( - collector: SocketAddr, - interface: Option, - peer: SocketAddr, - id: SubscriptionId, - ) -> Self { + pub fn new_empty(peer_ip: IpAddr, id: SubscriptionId) -> Self { Self { - collector, - interface, - peer, + peer_ip, id, target: Target::new_datastore("EMPTY".to_string(), either::Right("EMPTY".to_string())), stop_time: None, @@ -870,20 +858,9 @@ impl SubscriptionInfo { self.content_id == "EMPTY" } - /// The peer address of the device who sent the subscription started - /// message. - pub const fn peer(&self) -> SocketAddr { - self.peer - } - - /// The collector address of the subscription. - pub const fn collector(&self) -> SocketAddr { - self.collector - } - - /// The interface name if available. - pub fn interface(&self) -> Option { - self.interface.clone() + /// The IP address of the device who sent the subscription started message. + pub const fn peer_ip(&self) -> IpAddr { + self.peer_ip } /// The subscription ID associated with the subscription. This is a unique @@ -933,8 +910,8 @@ impl std::fmt::Display for SubscriptionInfo { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!( f, - "SubscriptionInfo {{ peer: {}, content_id: {}, target: {}, models: {:?} }}", - self.peer, self.content_id, self.target, self.models + "SubscriptionInfo {{ peer_ip: {}, content_id: {}, target: {}, models: {:?} }}", + self.peer_ip, self.content_id, self.target, self.models ) } } @@ -1024,7 +1001,7 @@ impl YangLibraryCache { cache_by_content_id.insert(content_id, Arc::clone(&yang_lib_ref)); for subscription_info in subscriptions_info { cache_by_subscription_id - .entry(subscription_info.peer().ip()) + .entry(subscription_info.peer_ip()) .or_default() .insert(subscription_info.id(), subscription_info.clone()); cache_by_subscription_info.insert(subscription_info, Arc::clone(&yang_lib_ref)); @@ -1065,7 +1042,7 @@ impl YangLibraryCache { self.cache_by_subscription_info .insert(subscription_info.clone(), Arc::clone(existing_ref)); self.cache_by_subscription_id - .entry(subscription_info.peer().ip()) + .entry(subscription_info.peer_ip()) .or_default() .insert(subscription_info.id(), subscription_info); } @@ -1086,7 +1063,7 @@ impl YangLibraryCache { self.cache_by_content_id .insert(content_id, Arc::clone(&yang_lib_ref)); self.cache_by_subscription_id - .entry(subscription_info.peer().ip()) + .entry(subscription_info.peer_ip()) .or_default() .insert(subscription_info.id(), subscription_info); Ok(yang_lib_ref) @@ -1132,12 +1109,12 @@ impl YangLibraryCache { // Step 2: Remove from (peer_ip, sub_id) -> SubscriptionInfo mapping if let Some(sub_map) = self .cache_by_subscription_id - .get_mut(&subscription_info.peer().ip()) + .get_mut(&subscription_info.peer_ip()) { sub_map.remove(&subscription_info.id()); if sub_map.is_empty() { self.cache_by_subscription_id - .remove(&subscription_info.peer().ip()); + .remove(&subscription_info.peer_ip()); } } @@ -1185,7 +1162,7 @@ impl YangLibraryCache { .get(subscription_info) .map(Arc::clone); debug!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -1244,9 +1221,7 @@ mod tests { fn create_test_subscription_info(content_id: &str) -> SubscriptionInfo { SubscriptionInfo::new( - SocketAddr::from(([192, 168, 1, 100], 10000)), - None, - SocketAddr::from(([192, 168, 1, 101], 830)), + IpAddr::from([192, 168, 1, 101]), 1, Target::new_datastore( "ds:operational".to_string(), @@ -1454,8 +1429,7 @@ mod tests { #[test] #[tracing_test::traced_test] fn test_subscription_info_new() { - let collector = SocketAddr::from(([192, 168, 1, 100], 10000)); - let peer = SocketAddr::from(([192, 168, 1, 101], 12345)); + let peer_ip = IpAddr::from([192, 168, 1, 101]); let content_id = ContentId::from("content-123".to_string()); let target = Target::new_datastore( "ds:operational".to_string(), @@ -1467,9 +1441,7 @@ mod tests { YangPushModuleVersion::new("model2".into(), None, None), ]); let info = SubscriptionInfo::new( - collector, - None, - peer, + peer_ip, 1, Target::new_datastore( "ds:operational".to_string(), @@ -1488,7 +1460,7 @@ mod tests { content_id.clone(), ); - assert_eq!(info.peer(), peer); + assert_eq!(info.peer_ip(), peer_ip); assert_eq!(info.content_id(), &content_id); assert_eq!(info.target(), &target); assert_eq!(info.models(), models.as_slice()); @@ -1500,7 +1472,7 @@ mod tests { let info = create_test_subscription_info("test-id"); let display = format!("{info}"); - assert!(display.contains("192.168.1.101:830")); + assert!(display.contains("192.168.1.101")); assert!(display.contains("test-id")); assert!(display.contains("/ietf-interfaces:interfaces/ietf-interfaces:interface[ietf-interfaces:name='eth0']/statistics")); assert!(display.contains("ietf-interfaces")); @@ -1620,7 +1592,7 @@ mod tests { let cache = YangLibraryCache::from_disk(temp_dir.path().to_path_buf()).unwrap(); let subscription_info = create_test_subscription_info(content_id); - let peer_ip = subscription_info.peer().ip(); + let peer_ip = subscription_info.peer_ip(); let subscription_id = subscription_info.id(); // Test successful lookup @@ -1726,9 +1698,7 @@ mod tests { let subscription_info1 = create_test_subscription_info("content_id"); let subscription_info2 = SubscriptionInfo::new( - SocketAddr::from(([192, 168, 1, 100], 10000)), - None, - SocketAddr::from(([192, 168, 1, 102], 830)), + IpAddr::from([192, 168, 1, 102]), 2, Target::new_datastore( "ds:operational".to_string(), @@ -1751,9 +1721,7 @@ mod tests { ContentId::from("content_id2".to_string()), ); let subscription_info3 = SubscriptionInfo::new( - SocketAddr::from(([192, 168, 1, 100], 10000)), - None, - SocketAddr::from(([192, 168, 1, 103], 830)), + IpAddr::from([192, 168, 1, 103]), 2, Target::new_datastore( "ds:operational".to_string(), @@ -1834,9 +1802,7 @@ mod tests { // Create different subscription info with the same content_id let subscription_info2 = SubscriptionInfo::new( - SocketAddr::from(([192, 168, 1, 100], 10000)), - None, - SocketAddr::from(([192, 168, 1, 102], 830)), + IpAddr::from([192, 168, 1, 102]), 2, Target::new_datastore( "ds:operational".to_string(), @@ -1891,9 +1857,7 @@ mod tests { // Create first subscription info let subscription_info1 = SubscriptionInfo::new( - SocketAddr::from(([192, 168, 1, 100], 10000)), - None, - SocketAddr::from(([192, 168, 1, 101], 830)), + IpAddr::from([192, 168, 1, 101]), 1, Target::new_datastore( "ds:operational".to_string(), @@ -1944,9 +1908,7 @@ mod tests { // Create second subscription info with different peer and target but same // content_id let subscription_info2 = SubscriptionInfo::new( - SocketAddr::from(([192, 168, 1, 100], 10000)), - None, - SocketAddr::from(([192, 168, 1, 102], 830)), + IpAddr::from([192, 168, 1, 102]), 2, Target::new_datastore( "ds:operational".to_string(), @@ -1995,13 +1957,13 @@ mod tests { // Verify both subscription IDs can be retrieved let (sub1_info, sub1_ref) = cache - .get_by_subscription_id(subscription_info1.peer().ip(), subscription_info1.id()) + .get_by_subscription_id(subscription_info1.peer_ip(), subscription_info1.id()) .unwrap(); assert_eq!(sub1_info, subscription_info1); assert!(sub1_ref.is_some()); let (sub2_info, sub2_ref) = cache - .get_by_subscription_id(subscription_info2.peer().ip(), subscription_info2.id()) + .get_by_subscription_id(subscription_info2.peer_ip(), subscription_info2.id()) .unwrap(); assert_eq!(sub2_info, subscription_info2); assert!(sub2_ref.is_some()); diff --git a/crates/yang-push/src/validation/mod.rs b/crates/yang-push/src/validation/mod.rs index 3a460b7a..cb20530e 100644 --- a/crates/yang-push/src/validation/mod.rs +++ b/crates/yang-push/src/validation/mod.rs @@ -130,7 +130,7 @@ use netcalyx_netconf_proto::yang_push::types::SubscriptionId; use netcalyx_udp_notif_pkt::decoded::{UdpNotifPacketDecoded, UdpNotifPayload}; use netcalyx_udp_notif_pkt::notification::{NotificationVariant, SubscriptionStartedModified}; use netcalyx_udp_notif_pkt::raw::UdpNotifPacket; -use netcalyx_udp_notif_service::{OTL_UDP_NOTIF_PUBLISHER_ID_KEY, UdpNotifRequest}; +use netcalyx_udp_notif_service::{OTL_UDP_NOTIF_PUBLISHER_ID_KEY, SessionInfo, UdpNotifRequest}; use rustc_hash::FxHashMap; use std::collections::VecDeque; use std::net::{IpAddr, SocketAddr}; @@ -385,13 +385,34 @@ enum ValidationActorCommand { Shutdown, } +/// The output of the validation stage: a decoded UDP-Notif packet together +/// with the subscription identity, transport session, and the locally-computed +/// schema fingerprint used for validation. +/// +/// - `cached_content_id`: the SHA-256 fingerprint of the YANG library that was +/// used to validate the packet, or `None` when the packet was forwarded +/// without validation (schema unavailable or fetch failed). +/// - `subscription_info`: subscription identity — what this data stream is +/// about. +/// - `session`: transport session context — collector Socket Address, +/// interface/VRF and peer Socket Address +/// - `packet`: the decoded UDP-Notif payload ready for enrichment and +/// publishing. +#[derive(Debug)] +pub struct ValidatedNotification { + pub cached_content_id: Option, + pub subscription_info: SubscriptionInfo, + pub session: SessionInfo, + pub packet: UdpNotifPacketDecoded, +} + struct ValidationActor { max_buffered_packets_per_peer: usize, max_buffered_packets_per_subscription: usize, peer_cache: FxHashMap, cmd_rx: mpsc::Receiver, rx: async_channel::Receiver>, - tx: async_channel::Sender<(Option, SubscriptionInfo, UdpNotifPacketDecoded)>, + tx: async_channel::Sender, cache_cmd_tx: async_channel::Sender, cache_tx: async_channel::Sender, cache_rx: async_channel::Receiver, @@ -446,8 +467,6 @@ impl ValidationActor { fn get_subscription_info( &mut self, peer: SocketAddr, - collector: SocketAddr, - interface: Option, decoded: &UdpNotifPacketDecoded, ) -> Option<(SubscriptionInfo, Option>)> { let message_id = decoded.message_id(); @@ -471,14 +490,9 @@ impl ValidationActor { subscription_started, ) = notif_contents { - let subscription_info = if let Some(subscription_info) = self.build_subscription_info( - peer, - collector, - interface, - message_id, - publisher_id, - subscription_started, - ) { + let subscription_info = if let Some(subscription_info) = + self.build_subscription_info(peer, message_id, publisher_id, subscription_started) + { subscription_info } else { warn!( @@ -708,6 +722,7 @@ impl ValidationActor { ) -> Result<(), ValidationActorError> { let peer = message.peer_address(); let packet = message.packet(); + let session = message.session().clone(); // Step 1: decode the raw UDP-Notif payload. let decoded = match self.decode_message(peer, packet, !is_reprocessed) { @@ -796,11 +811,12 @@ impl ValidationActor { // Step 4: forward to the enrichment actor. self.tx - .send(( - cached_content_id.clone(), - subscription_info.clone(), - decoded, - )) + .send(ValidatedNotification { + cached_content_id: cached_content_id.clone(), + subscription_info: subscription_info.clone(), + session, + packet: decoded, + }) .await .map_err(|_| { warn!( @@ -950,8 +966,6 @@ impl ValidationActor { peer: SocketAddr, decoded: &UdpNotifPacketDecoded, ) -> Result, ValidationActorError> { - let collector = message.collector_address(); - let interface = message.collector_interface(); let packet = message.packet(); let mut peer_tags = Self::peer_tags_from_packet(peer, packet); let message_id = decoded.message_id(); @@ -961,7 +975,7 @@ impl ValidationActor { .map(|x| x.to_string()) .unwrap_or("UNKNOWN".to_string()); - match self.get_subscription_info(peer, collector, interface.map(String::from), decoded) { + match self.get_subscription_info(peer, decoded) { Some((subscription_info, cached_content_id)) => { Self::extend_peer_tags_with_subscription_info(&subscription_info, &mut peer_tags); @@ -1027,10 +1041,11 @@ impl ValidationActor { .or_insert_with(|| CachedSubscription::new(subscription_info.clone())) .schema_fetch_pending = true; self.cache_cmd_tx - .send(CacheLookupCommand::LookupBySubscriptionInfo( - subscription_info.clone(), - self.cache_tx.clone(), - )) + .send(CacheLookupCommand::LookupBySubscriptionInfo { + subscription_info: subscription_info.clone(), + session: message.session().clone(), + tx: self.cache_tx.clone(), + }) .await .map_err(|error| { warn!( @@ -1088,12 +1103,7 @@ impl ValidationActor { <&str>::from(CacheLookupBy::SubscriptionId), )); self.stats.cache_lookups.add(1, &peer_tags); - let subscription_info = SubscriptionInfo::new_empty( - collector, - interface.map(String::from), - peer, - subscription_id, - ); + let subscription_info = SubscriptionInfo::new_empty(peer.ip(), subscription_id); // Mark the fetch as in-flight self.peer_cache @@ -1105,9 +1115,7 @@ impl ValidationActor { .schema_fetch_pending = true; self.cache_cmd_tx .send(CacheLookupCommand::LookupBySubscriptionId { - collector, - interface: interface.map(String::from), - peer, + session: message.session().clone(), subscription_id, tx: self.cache_tx.clone(), }) @@ -1152,24 +1160,18 @@ impl ValidationActor { response: CacheResponse, ) -> Result<(), ValidationActorError> { let (cached_content_id, subscription_info, yang_lib_ref) = response.into(); - let mut otl_tags = Vec::from([ - opentelemetry::KeyValue::new( - "network.peer.address", - format!("{}", subscription_info.peer().ip()), - ), - opentelemetry::KeyValue::new( - "network.peer.port", - opentelemetry::Value::I64(subscription_info.peer().port().into()), - ), - ]); + let mut otl_tags = Vec::from([opentelemetry::KeyValue::new( + "network.peer.address", + format!("{}", subscription_info.peer_ip()), + )]); Self::extend_peer_tags_with_subscription_info(&subscription_info, &mut otl_tags); let peer_cache = if let Some(peer_cache) = - self.peer_cache.get_mut(&subscription_info.peer().ip()) + self.peer_cache.get_mut(&subscription_info.peer_ip()) { peer_cache } else { warn!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -1185,7 +1187,7 @@ impl ValidationActor { subscription_cache } else { warn!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), target=%subscription_info.target(), @@ -1212,7 +1214,7 @@ impl ValidationActor { Err(err) => { self.stats.yang_context_failed.add(1, &otl_tags); warn!( - peer=%subscription_info.peer(), + peer_ip=%subscription_info.peer_ip(), subscription_id=subscription_info.id(), router_content_id=subscription_info.content_id(), cached_content_id=yang_lib_ref.content_id(), @@ -1261,7 +1263,7 @@ impl ValidationActor { self.stats .cached_peers .record(self.peer_cache.len() as u64, &[]); - let peer_ip = subscription_info.peer().ip(); + let peer_ip = subscription_info.peer_ip(); let peer_sub_count = self .peer_cache .get(&peer_ip) @@ -1282,8 +1284,6 @@ impl ValidationActor { fn build_subscription_info( &self, peer: SocketAddr, - collector: SocketAddr, - interface: Option, message_id: u32, publisher_id: u32, sub_started: &SubscriptionStartedModified, @@ -1312,9 +1312,7 @@ impl ValidationActor { }; Some(SubscriptionInfo::new( - collector, - interface, - peer, + peer.ip(), sub_started.id(), sub_started.target().clone(), sub_started.stop_time().cloned(), @@ -1423,7 +1421,7 @@ impl ValidationActorHandle { max_buffered_packets_per_peer: usize, max_buffered_packets_per_subscription: usize, rx: async_channel::Receiver>, - tx: async_channel::Sender<(Option, SubscriptionInfo, UdpNotifPacketDecoded)>, + tx: async_channel::Sender, cache_cmd_tx: async_channel::Sender, stats: either::Either, ) -> Result< @@ -1485,7 +1483,7 @@ mod tests { SubscriptionInfo, Arc>>, async_channel::Sender>, - async_channel::Receiver<(Option, SubscriptionInfo, UdpNotifPacketDecoded)>, + async_channel::Receiver, ValidationActorHandle, ) { let (caching_join_handle, caching_handle, subscription_info, fetcher_count) = @@ -1520,11 +1518,7 @@ mod tests { /// be validated against the loaded context. async fn setup_and_load_schema( udp_notif_tx: &async_channel::Sender>, - validated_rx: &async_channel::Receiver<( - Option, - SubscriptionInfo, - UdpNotifPacketDecoded, - )>, + validated_rx: &async_channel::Receiver, peer: SocketAddr, ) { let payload = serde_json::json!({ @@ -1560,9 +1554,7 @@ mod tests { let bytes = serde_json::to_vec(&payload).unwrap(); udp_notif_tx .send(Arc::new(UdpNotifRequest::new( - SocketAddr::from(([127, 0, 0, 1], 10000)), - None, - peer, + SessionInfo::new(SocketAddr::from(([127, 0, 0, 1], 10000)), None, peer), UdpNotifPacket::new( MediaType::YangDataJson, 10, @@ -1576,7 +1568,10 @@ mod tests { // Draining the validated SubscriptionStarted also serves as the // synchronisation point: by the time it is forwarded the YANG context // is fully loaded and ready for subsequent push-update packets. - let (content_id, _, _) = tokio::time::timeout(Duration::from_secs(2), validated_rx.recv()) + let ValidatedNotification { + cached_content_id: content_id, + .. + } = tokio::time::timeout(Duration::from_secs(2), validated_rx.recv()) .await .expect("timeout waiting for SubscriptionStarted to be validated") .unwrap(); @@ -1600,7 +1595,7 @@ mod tests { ) = setup_validation_actor(); assert_eq!(fetcher_count.lock().unwrap().len(), 0); - let peer = subscription_info.peer(); + let peer = SocketAddr::new(subscription_info.peer_ip(), 0); let payload = serde_json::json!( { "ietf-yp-notification:envelope": { @@ -1645,9 +1640,7 @@ mod tests { // Send SubscriptionStarted packet udp_notif_tx .send(Arc::new(UdpNotifRequest::new( - SocketAddr::from(([127, 0, 0, 1], 10000)), - None, - peer, + SessionInfo::new(SocketAddr::from(([127, 0, 0, 1], 10000)), None, peer), subscription_started_packet, ))) .await @@ -1657,11 +1650,14 @@ mod tests { tokio::time::sleep(Duration::from_millis(50)).await; // Verify packet is validated - let (content_id, sub_info, _validated) = - tokio::time::timeout(Duration::from_secs(1), validated_rx.recv()) - .await - .expect("timeout waiting for response") - .unwrap(); + let ValidatedNotification { + cached_content_id: content_id, + subscription_info: sub_info, + .. + } = tokio::time::timeout(Duration::from_secs(1), validated_rx.recv()) + .await + .expect("timeout waiting for response") + .unwrap(); assert!(content_id.is_some()); assert!(!sub_info.is_empty()); @@ -1694,7 +1690,7 @@ mod tests { ) = setup_validation_actor(); assert_eq!(fetcher_count.lock().unwrap().len(), 0); - let peer = subscription_info.peer(); + let peer = SocketAddr::new(subscription_info.peer_ip(), 0); let payload = serde_json::json!( { "ietf-yp-notification:envelope": { @@ -1739,9 +1735,7 @@ mod tests { // Send SubscriptionStarted packet udp_notif_tx .send(Arc::new(UdpNotifRequest::new( - SocketAddr::from(([127, 0, 0, 1], 10000)), - None, - peer, + SessionInfo::new(SocketAddr::from(([127, 0, 0, 1], 10000)), None, peer), subscription_started_packet, ))) .await @@ -1751,11 +1745,14 @@ mod tests { tokio::time::sleep(Duration::from_millis(50)).await; // Verify packet is not validated - let (content_id, sub_info, _validated) = - tokio::time::timeout(Duration::from_secs(1), validated_rx.recv()) - .await - .expect("timeout waiting for response") - .unwrap(); + let ValidatedNotification { + cached_content_id: content_id, + subscription_info: sub_info, + .. + } = tokio::time::timeout(Duration::from_secs(1), validated_rx.recv()) + .await + .expect("timeout waiting for response") + .unwrap(); assert!(content_id.is_none()); assert!(!sub_info.is_empty()); @@ -1802,7 +1799,7 @@ mod tests { ) .expect("Failed to spawn validation actor"); - let peer = subscription_info.peer(); + let peer = SocketAddr::new(subscription_info.peer_ip(), 0); let payload = serde_json::json!({ "ietf-yp-notification:envelope": { "event-time": "2026-04-21T13:33:31.007Z", @@ -1841,9 +1838,7 @@ mod tests { for i in 0..N { udp_notif_tx .send(Arc::new(UdpNotifRequest::new( - SocketAddr::from(([127, 0, 0, 1], 10000)), - None, - peer, + SessionInfo::new(SocketAddr::from(([127, 0, 0, 1], 10000)), None, peer), UdpNotifPacket::new( MediaType::YangDataJson, 10, @@ -1895,7 +1890,7 @@ mod tests { // SubscriptionStarted WITHOUT module-version → build_subscription_info // returns None → must be dropped permanently. - let peer = subscription_info.peer(); + let peer = SocketAddr::new(subscription_info.peer_ip(), 0); let payload = serde_json::json!({ "ietf-yp-notification:envelope": { "event-time": "2025-09-23T14:12:16.024Z", @@ -1916,9 +1911,7 @@ mod tests { let bytes = serde_json::to_vec(&payload).unwrap(); udp_notif_tx .send(Arc::new(UdpNotifRequest::new( - SocketAddr::from(([127, 0, 0, 1], 10000)), - None, - peer, + SessionInfo::new(SocketAddr::from(([127, 0, 0, 1], 10000)), None, peer), UdpNotifPacket::new( MediaType::YangDataJson, 10, @@ -1958,7 +1951,7 @@ mod tests { handle, ) = setup_validation_actor(); - let peer = subscription_info.peer(); + let peer = SocketAddr::new(subscription_info.peer_ip(), 0); setup_and_load_schema(&udp_notif_tx, &validated_rx, peer).await; // Send a push-update with ietf-interfaces data. @@ -2007,9 +2000,7 @@ mod tests { let bytes = serde_json::to_vec(&push_update_payload).unwrap(); udp_notif_tx .send(Arc::new(UdpNotifRequest::new( - SocketAddr::from(([127, 0, 0, 1], 10000)), - None, - peer, + SessionInfo::new(SocketAddr::from(([127, 0, 0, 1], 10000)), None, peer), UdpNotifPacket::new( MediaType::YangDataJson, 10, @@ -2021,11 +2012,14 @@ mod tests { .await .unwrap(); - let (content_id, sub_info, _decoded) = - tokio::time::timeout(Duration::from_secs(1), validated_rx.recv()) - .await - .expect("timeout: valid push-update was not forwarded") - .unwrap(); + let ValidatedNotification { + cached_content_id: content_id, + subscription_info: sub_info, + .. + } = tokio::time::timeout(Duration::from_secs(1), validated_rx.recv()) + .await + .expect("timeout: valid push-update was not forwarded") + .unwrap(); assert!( content_id.is_some(), "valid push-update must pass strict YANG validation" @@ -2056,9 +2050,11 @@ mod tests { // UnsupportedMediaType. udp_notif_tx .send(Arc::new(UdpNotifRequest::new( - SocketAddr::from(([127, 0, 0, 1], 10000)), - None, - subscription_info.peer(), + SessionInfo::new( + SocketAddr::from(([127, 0, 0, 1], 10000)), + None, + SocketAddr::new(subscription_info.peer_ip(), 0), + ), UdpNotifPacket::new( MediaType::YangDataXml, 10, @@ -2100,9 +2096,11 @@ mod tests { // YangDataJson with bytes that are not valid JSON → serde_json parse error. udp_notif_tx .send(Arc::new(UdpNotifRequest::new( - SocketAddr::from(([127, 0, 0, 1], 10000)), - None, - subscription_info.peer(), + SessionInfo::new( + SocketAddr::from(([127, 0, 0, 1], 10000)), + None, + SocketAddr::new(subscription_info.peer_ip(), 0), + ), UdpNotifPacket::new( MediaType::YangDataJson, 10, @@ -2141,7 +2139,7 @@ mod tests { handle, ) = setup_validation_actor(); - let peer = subscription_info.peer(); + let peer = SocketAddr::new(subscription_info.peer_ip(), 0); setup_and_load_schema(&udp_notif_tx, &validated_rx, peer).await; // Push-update with "enabelled" (typo for "enabled"): an unknown YANG node @@ -2177,9 +2175,7 @@ mod tests { let bytes = serde_json::to_vec(&invalid_push_update_payload).unwrap(); udp_notif_tx .send(Arc::new(UdpNotifRequest::new( - SocketAddr::from(([127, 0, 0, 1], 10000)), - None, - peer, + SessionInfo::new(SocketAddr::from(([127, 0, 0, 1], 10000)), None, peer), UdpNotifPacket::new( MediaType::YangDataJson, 10, @@ -2221,7 +2217,7 @@ mod tests { handle, ) = setup_validation_actor(); - let peer = subscription_info.peer(); + let peer = SocketAddr::new(subscription_info.peer_ip(), 0); setup_and_load_schema(&udp_notif_tx, &validated_rx, peer).await; // Push-update with the mandatory `type` leaf absent from the interface @@ -2256,9 +2252,7 @@ mod tests { let bytes = serde_json::to_vec(&missing_type_payload).unwrap(); udp_notif_tx .send(Arc::new(UdpNotifRequest::new( - SocketAddr::from(([127, 0, 0, 1], 10000)), - None, - peer, + SessionInfo::new(SocketAddr::from(([127, 0, 0, 1], 10000)), None, peer), UdpNotifPacket::new( MediaType::YangDataJson, 10, @@ -2278,7 +2272,11 @@ mod tests { "libyang limitation apparently addressed: mandatory nodes inside anydata are now \ enforced; flip this test to assert `res.is_err()` + `logs_contain(\"Failed to validate\")`" ); - let (content_id, sub_info, _decoded) = res.unwrap().unwrap(); + let ValidatedNotification { + cached_content_id: content_id, + subscription_info: sub_info, + .. + } = res.unwrap().unwrap(); assert!(content_id.is_some()); assert!(!sub_info.is_empty()); @@ -2304,7 +2302,7 @@ mod tests { validated_rx, handle, ) = setup_validation_actor(); - let peer = subscription_info.peer(); + let peer = SocketAddr::new(subscription_info.peer_ip(), 0); let payload = serde_json::json!({ "ietf-yp-notification:envelope": { @@ -2340,9 +2338,7 @@ mod tests { let make_packet = |msg_id: u32| { Arc::new(UdpNotifRequest::new( - SocketAddr::from(([127, 0, 0, 1], 10000)), - None, - peer, + SessionInfo::new(SocketAddr::from(([127, 0, 0, 1], 10000)), None, peer), UdpNotifPacket::new( MediaType::YangDataJson, 10, @@ -2367,11 +2363,14 @@ mod tests { // Regardless of whether the duplicate arrived before or after the cache // responded, content_id must be Some (never forwarded unvalidated). for i in 1..=2u32 { - let (content_id, sub_info, _) = - tokio::time::timeout(Duration::from_secs(3), validated_rx.recv()) - .await - .unwrap_or_else(|_| panic!("timeout waiting for packet {i}")) - .unwrap(); + let ValidatedNotification { + cached_content_id: content_id, + subscription_info: sub_info, + .. + } = tokio::time::timeout(Duration::from_secs(3), validated_rx.recv()) + .await + .unwrap_or_else(|_| panic!("timeout waiting for packet {i}")) + .unwrap(); assert!( content_id.is_some(), "packet {i}: duplicate SubscriptionStarted must be validated, not forwarded unvalidated" @@ -2406,7 +2405,7 @@ mod tests { validated_rx, handle, ) = setup_validation_actor(); - let peer = subscription_info.peer(); + let peer = SocketAddr::new(subscription_info.peer_ip(), 0); // Load the schema via the first SubscriptionStarted and drain the result. setup_and_load_schema(&udp_notif_tx, &validated_rx, peer).await; @@ -2450,9 +2449,7 @@ mod tests { let bytes = serde_json::to_vec(&payload).unwrap(); udp_notif_tx .send(Arc::new(UdpNotifRequest::new( - SocketAddr::from(([127, 0, 0, 1], 10000)), - None, - peer, + SessionInfo::new(SocketAddr::from(([127, 0, 0, 1], 10000)), None, peer), UdpNotifPacket::new( MediaType::YangDataJson, 10, @@ -2465,11 +2462,14 @@ mod tests { .unwrap(); // Must be validated immediately using the cached schema. - let (content_id, sub_info, _) = - tokio::time::timeout(Duration::from_secs(1), validated_rx.recv()) - .await - .expect("timeout: duplicate SubscriptionStarted was not forwarded") - .unwrap(); + let ValidatedNotification { + cached_content_id: content_id, + subscription_info: sub_info, + .. + } = tokio::time::timeout(Duration::from_secs(1), validated_rx.recv()) + .await + .expect("timeout: duplicate SubscriptionStarted was not forwarded") + .unwrap(); assert!( content_id.is_some(), "duplicate SubscriptionStarted after schema loaded must be validated" @@ -2488,6 +2488,110 @@ mod tests { caching_join_handle.await.unwrap().unwrap(); } + /// YANG-Push subscription state must be keyed by IP alone. + /// A device reconnecting from a new ephemeral UDP source port + /// (same IP, same subscription id) must reuse the already-cached schema. + /// `SessionInfo` forwarded downstream still reflects the new port. + #[tokio::test] + #[tracing_test::traced_test] + async fn test_validation_actor_reconnect_new_source_port_reuses_schema() { + let ( + caching_join_handle, + caching_handle, + subscription_info, + fetcher_count, + udp_notif_tx, + validated_rx, + handle, + ) = setup_validation_actor(); + + let peer_ip = subscription_info.peer_ip(); + let peer_port_a = SocketAddr::new(peer_ip, 12345); + let peer_port_b = SocketAddr::new(peer_ip, 12346); + + // Load the schema using the first source port. + setup_and_load_schema(&udp_notif_tx, &validated_rx, peer_port_a).await; + assert_eq!( + fetcher_count.lock().unwrap().len(), + 1, + "initial fetch must have fired once" + ); + + // Send a push-update for the same subscription id from a different + // source port, simulating the device reconnecting with a new + // ephemeral UDP port. + let push_update_payload = serde_json::json!({ + "ietf-yp-notification:envelope": { + "event-time": "2026-04-21T13:33:31.007Z", + "hostname": "test-router-01", + "sequence-number": 1, + "contents": { + "ietf-yang-push:push-update": { + "id": 1, + "datastore-contents": { + "ietf-interfaces:interfaces": { + "interface": [ + { + "name": "GigabitEthernet0/0/0", + "type": "iana-if-type:ethernetCsmacd", + "enabled": true, + "admin-status": "up", + "oper-status": "up", + "if-index": 1, + "speed": "1000000000" + } + ] + } + }, + "ietf-distributed-notif:message-publisher-id": 16974839 + } + } + } + }); + let bytes = serde_json::to_vec(&push_update_payload).unwrap(); + udp_notif_tx + .send(Arc::new(UdpNotifRequest::new( + SessionInfo::new(SocketAddr::from(([127, 0, 0, 1], 10000)), None, peer_port_b), + UdpNotifPacket::new( + MediaType::YangDataJson, + 10, + 2, + HashMap::new(), + Bytes::from(bytes), + ), + ))) + .await + .unwrap(); + + let notification = tokio::time::timeout(Duration::from_secs(1), validated_rx.recv()) + .await + .expect("timeout: push-update from new source port was not forwarded") + .unwrap(); + assert!( + notification.cached_content_id.is_some(), + "packet from a new source port for an already-known peer/subscription must reuse \ + the cached schema, not be treated as an unknown subscription" + ); + assert!(!notification.subscription_info.is_empty()); + assert_eq!( + notification.session.peer(), + peer_port_b, + "forwarded SessionInfo must retain the new source port, not the original one" + ); + + // No additional cache fetch must have been triggered: the peer/subscription + // must be recognized from the IP alone, regardless of source port. + assert_eq!( + fetcher_count.lock().unwrap().len(), + 1, + "reconnecting from a new source port must not trigger a second cache fetch" + ); + + handle.shutdown().await.unwrap(); + caching_handle.shutdown().await.unwrap(); + caching_join_handle.await.unwrap().unwrap(); + } + /// When a SubscriptionStarted with changed params (same id, updated /// yang-library-content-id) arrives after the schema is already loaded: /// 1. The validation actor clears its local cache entry, buffers the @@ -2515,7 +2619,7 @@ mod tests { validated_rx, handle, ) = setup_validation_actor(); - let peer = subscription_info.peer(); + let peer = SocketAddr::new(subscription_info.peer_ip(), 0); // Load the schema for the initial subscription (content-id = // "test-content-id-1"). @@ -2555,9 +2659,7 @@ mod tests { let bytes = serde_json::to_vec(&changed_payload).unwrap(); udp_notif_tx .send(Arc::new(UdpNotifRequest::new( - SocketAddr::from(([127, 0, 0, 1], 10000)), - None, - peer, + SessionInfo::new(SocketAddr::from(([127, 0, 0, 1], 10000)), None, peer), UdpNotifPacket::new( MediaType::YangDataJson, 10, @@ -2574,11 +2676,14 @@ mod tests { // buffered during the fetch attempt; after the fetch fails it is forwarded // unvalidated. In production the fetch would succeed and content_id would be // Some. - let (content_id, sub_info, _) = - tokio::time::timeout(Duration::from_secs(3), validated_rx.recv()) - .await - .expect("timeout: changed SubscriptionStarted was not forwarded") - .unwrap(); + let ValidatedNotification { + cached_content_id: content_id, + subscription_info: sub_info, + .. + } = tokio::time::timeout(Duration::from_secs(3), validated_rx.recv()) + .await + .expect("timeout: changed SubscriptionStarted was not forwarded") + .unwrap(); assert!( content_id.is_none(), "device fetch failed for new content-id → packet must be forwarded unvalidated"