diff --git a/crates/collector/src/publishers/kafka_yang.rs b/crates/collector/src/publishers/kafka_yang.rs index f6f7cd29..d245c661 100644 --- a/crates/collector/src/publishers/kafka_yang.rs +++ b/crates/collector/src/publishers/kafka_yang.rs @@ -126,7 +126,10 @@ fn subscription_info_tags( subscription_info: Option<&SubscriptionInfo>, ) -> Vec { let Some(subscription_info) = subscription_info else { - return Vec::new(); + return vec![opentelemetry::KeyValue::new( + "network.peer.address", + "unknown".to_string(), + )]; }; vec![ opentelemetry::KeyValue::new( diff --git a/crates/yang-push/src/cache/actor.rs b/crates/yang-push/src/cache/actor.rs index 4d3ba9d6..5c5b2866 100644 --- a/crates/yang-push/src/cache/actor.rs +++ b/crates/yang-push/src/cache/actor.rs @@ -650,10 +650,10 @@ impl CacheActor { Err(err) => { self.stats.device_fetch_failed.add( 1, - &[opentelemetry::KeyValue::new( - "error.message", - format!("{err}"), - )], + &[ + opentelemetry::KeyValue::new("network.peer.address", "unknown".to_string()), + opentelemetry::KeyValue::new("error.message", format!("{err}")), + ], ); warn!(error=%err, "cache actor worker failed to execute a task"); } @@ -910,10 +910,14 @@ impl CacheActor { error=%err, "failed to fetch yang library from device" ); - // remove the sender we just added - // since the fetch failed to start - entry.remove(entry.len() - 1); + // Fetch failed to start: drop the sender we + // queued. The entry was empty beforehand + // so remove the key and refresh the gauge + self.pending_requests.remove(&subscription_info); self.stats.device_fetch_failed.add(1, &otl_tags); + self.stats + .pending_cache_requests + .record(self.pending_requests.len() as u64, &[]); return; } }; @@ -933,7 +937,7 @@ impl CacheActor { } self.stats .pending_cache_requests - .record(self.pending_requests.len() as u64, &otl_tags); + .record(self.pending_requests.len() as u64, &[]); } } } @@ -1060,10 +1064,14 @@ impl CacheActor { error=%err, "failed to fetch yang library from device" ); - // remove the sender we just added - // since the fetch failed to start - entry.remove(entry.len() - 1); + // Fetch failed to start: drop the sender we + // queued. The entry was empty beforehand so + // remove the key and refresh the gauge + self.pending_requests.remove(&subscription_info); self.stats.device_fetch_failed.add(1, &otel_tags); + self.stats + .pending_cache_requests + .record(self.pending_requests.len() as u64, &[]); return; } }; @@ -1083,7 +1091,7 @@ impl CacheActor { } self.stats .pending_cache_requests - .record(self.pending_requests.len() as u64, &otel_tags); + .record(self.pending_requests.len() as u64, &[]); } } CacheLookupCommand::LookupBySubscriptionIdOneShot { @@ -1147,10 +1155,13 @@ impl CacheActor { content_id, tx: sender, } => { - let otl_tags = [opentelemetry::KeyValue::new( - OTL_YANG_PUSH_CACHED_CONTENT_ID_KEY, - content_id.to_string(), - )]; + let otl_tags = [ + opentelemetry::KeyValue::new("network.peer.address", "unknown".to_string()), + opentelemetry::KeyValue::new( + OTL_YANG_PUSH_CACHED_CONTENT_ID_KEY, + content_id.to_string(), + ), + ]; self.stats.requests_received.add(1, &otl_tags); debug!(content_id, "processing cache lookup request by content id"); let yang_lib_ref = self.schema_cache.get_by_content_id(&content_id); @@ -1165,10 +1176,13 @@ impl CacheActor { content_id, tx: sender, } => { - let otl_tags = [opentelemetry::KeyValue::new( - OTL_YANG_PUSH_CACHED_CONTENT_ID_KEY, - content_id.to_string(), - )]; + let otl_tags = [ + opentelemetry::KeyValue::new("network.peer.address", "unknown".to_string()), + opentelemetry::KeyValue::new( + OTL_YANG_PUSH_CACHED_CONTENT_ID_KEY, + content_id.to_string(), + ), + ]; self.stats.requests_received.add(1, &otl_tags); debug!( content_id, @@ -1229,7 +1243,6 @@ impl CacheActor { if let Ok(request) = request { self.process_request(request).await; } else { - self.stats.requests_received.add(1, &[]); warn!( pending_requests = self.pending_requests.len(), active_workers = self.workers_queue.len(), diff --git a/crates/yang-push/src/validation/mod.rs b/crates/yang-push/src/validation/mod.rs index 33384af0..88f70bd1 100644 --- a/crates/yang-push/src/validation/mod.rs +++ b/crates/yang-push/src/validation/mod.rs @@ -307,7 +307,10 @@ impl ValidationStats { .build(); let buffered = meter .u64_gauge("netcalyx.yang_push.validation.buffered") - .with_description("Number of YANG-Push messages currently buffered waiting for schemas") + .with_description( + "Number of YANG-Push messages currently buffered waiting for schemas, \ + per peer, tagged with network.peer.address. Sum gives the global total.", + ) .build(); let buffer_drained = meter .u64_counter("netcalyx.yang_push.validation.buffer.drained") @@ -458,32 +461,42 @@ impl ValidationActor { /// gate in `extract_subscription_info`), so `buffered_packets` is always /// empty whenever an entry is removed below. fn check_subscription_new(&mut self, peer: SocketAddr, subscription_info: &SubscriptionInfo) { - if let Some(cached_peer_subscriptions) = self.peer_cache.get_mut(&peer.ip()) { - let is_different = cached_peer_subscriptions - .subscriptions - .get(&subscription_info.id()) - .map(|x| x.subscription_info != *subscription_info) - .unwrap_or(true); - if is_different { - trace!( - peer=%peer, - subscription_id=subscription_info.id(), - router_content_id=subscription_info.content_id(), - target=%subscription_info.target(), - "Subscription changed, removing from cache to allow a new fetch schemas request" - ); - if let Some(removed) = cached_peer_subscriptions - .subscriptions - .remove(&subscription_info.id()) - { - cached_peer_subscriptions.total_buffered -= removed.buffered_packets.len(); - } - } - // clear peer if there are no subscriptions left - if cached_peer_subscriptions.subscriptions.is_empty() { - self.peer_cache.remove(&peer.ip()); - } + let Some(cached_peer_subscriptions) = self.peer_cache.get_mut(&peer.ip()) else { + return; + }; + let is_different = cached_peer_subscriptions + .subscriptions + .get(&subscription_info.id()) + .map(|x| x.subscription_info != *subscription_info) + .unwrap_or(true); + if !is_different { + return; } + trace!( + peer=%peer, + subscription_id=subscription_info.id(), + router_content_id=subscription_info.content_id(), + target=%subscription_info.target(), + "Subscription changed, removing from cache to allow a new fetch schemas request" + ); + if let Some(removed) = cached_peer_subscriptions + .subscriptions + .remove(&subscription_info.id()) + { + cached_peer_subscriptions.total_buffered -= removed.buffered_packets.len(); + } + // clear peer if there are no subscriptions left + if cached_peer_subscriptions.subscriptions.is_empty() { + self.peer_cache.remove(&peer.ip()); + } + // Resync the per-peer gauges after the removal (0/0 when the peer is + // gone, resetting its series). + let (total_buffered, sub_count) = self + .peer_cache + .get(&peer.ip()) + .map(|p| (p.total_buffered, p.subscriptions.len())) + .unwrap_or((0, 0)); + self.record_otel_gauges(peer.ip(), total_buffered, sub_count); } /// Get the subscription info from the cache or from the SubscriptionStarted @@ -575,6 +588,7 @@ impl ValidationActor { let message_id = packet.message_id(); let publisher_id = packet.publisher_id(); let mut peer_tags = Self::peer_tags_from_packet(peer, packet); + Self::extend_peer_tags_with_subscription_info(&subscription_info, &mut peer_tags); let subscription_id = subscription_info.id(); let peer_cache = self.peer_cache.entry(peer.ip()).or_default(); @@ -632,9 +646,9 @@ impl ValidationActor { ); subscription_cache.buffered_packets.push(message); peer_cache.total_buffered += 1; - self.stats - .buffered - .record(peer_cache.total_buffered as u64, &peer_tags); + let total_buffered = peer_cache.total_buffered; + let sub_count = peer_cache.subscriptions.len(); + self.record_otel_gauges(peer.ip(), total_buffered, sub_count); true } @@ -678,6 +692,23 @@ impl ValidationActor { )); } + /// Record the peer-scoped cache gauges together so they never drift. Call + /// after any change to `peer_cache` for `peer_ip`; pass `0`/`0` once the + /// peer is gone so its per-peer series reset to zero. + fn record_otel_gauges(&self, peer_ip: IpAddr, total_buffered: usize, sub_count: usize) { + let peer_tag = [opentelemetry::KeyValue::new( + "network.peer.address", + format!("{peer_ip}"), + )]; + self.stats.buffered.record(total_buffered as u64, &peer_tag); + self.stats + .cached_subscriptions + .record(sub_count as u64, &peer_tag); + self.stats + .cached_peers + .record(self.peer_cache.len() as u64, &[]); + } + /// Decode a raw `UdpNotifPacket` into a `UdpNotifPacketDecoded`. /// Returns `Err(())` and drops the packet on unsupported media type or /// parse failure. @@ -1342,26 +1373,16 @@ impl ValidationActor { ); self.pending_packets.push_back(message); } - self.stats.buffered.record(remaining as u64, &otl_tags); self.stats .pending .record(self.pending_packets.len() as u64, &[]); - self.stats - .cached_peers - .record(self.peer_cache.len() as u64, &[]); let peer_ip = subscription_info.peer_ip(); let peer_sub_count = self .peer_cache .get(&peer_ip) .map(|p| p.subscriptions.len()) .unwrap_or(0); - self.stats.cached_subscriptions.record( - peer_sub_count as u64, - &[opentelemetry::KeyValue::new( - "network.peer.address", - format!("{peer_ip}"), - )], - ); + self.record_otel_gauges(peer_ip, remaining, peer_sub_count); Ok(()) }