Skip to content

fix(metrics): always tag peer-scoped metrics with network.peer.address #63

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
riccardo-negri merged 4 commits into from
Oct 14, 2026
Merged

fix(metrics): always tag peer-scoped metrics with network.peer.address #63

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
24eaf6c
fix(metrics): always tag peer-scoped metrics with network.peer.address
riccardo-negri Oct 12, 2026
0a9696e
fix(yang-cache): drop empty pending entry on fetch-start timeout
riccardo-negri Oct 13, 2026
a4e443a
fix(metrics): keep validation cache gauges consistent and complete
riccardo-negri Oct 13, 2026
587fb88
style(metrics): address cargo fmt for comments
riccardo-negri Oct 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion crates/collector/src/publishers/kafka_yang.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,10 @@ fn subscription_info_tags(
subscription_info: Option<&SubscriptionInfo>,
) -> Vec<opentelemetry::KeyValue> {
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(
Expand Down
55 changes: 34 additions & 21 deletions crates/yang-push/src/cache/actor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -650,10 +650,10 @@ impl<F: YangLibraryFetcher> CacheActor<F> {
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");
}
Expand Down Expand Up @@ -910,10 +910,14 @@ impl<F: YangLibraryFetcher> CacheActor<F> {
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;
}
};
Expand All @@ -933,7 +937,7 @@ impl<F: YangLibraryFetcher> CacheActor<F> {
}
self.stats
.pending_cache_requests
.record(self.pending_requests.len() as u64, &otl_tags);
.record(self.pending_requests.len() as u64, &[]);
Comment thread
riccardo-negri marked this conversation as resolved.
}
}
}
Expand Down Expand Up @@ -1060,10 +1064,14 @@ impl<F: YangLibraryFetcher> CacheActor<F> {
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;
}
};
Expand All @@ -1083,7 +1091,7 @@ impl<F: YangLibraryFetcher> CacheActor<F> {
}
self.stats
.pending_cache_requests
.record(self.pending_requests.len() as u64, &otel_tags);
.record(self.pending_requests.len() as u64, &[]);
}
}
CacheLookupCommand::LookupBySubscriptionIdOneShot {
Expand Down Expand Up @@ -1147,10 +1155,13 @@ impl<F: YangLibraryFetcher> CacheActor<F> {
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);
Expand All @@ -1165,10 +1176,13 @@ impl<F: YangLibraryFetcher> CacheActor<F> {
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,
Expand Down Expand Up @@ -1229,7 +1243,6 @@ impl<F: YangLibraryFetcher> CacheActor<F> {
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(),
Expand Down
101 changes: 61 additions & 40 deletions crates/yang-push/src/validation/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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();

Expand Down Expand Up @@ -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
}

Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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(())
}

Expand Down
Loading