feat(collector): add schema cache and registry metrics to kafka yang publisher

[rodonile]

Summary

Adds missing OpenTelemetry metrics to KafkaYangPublisherStats in the
Kafka YANG publisher. Previously only send/receive/delivery counters
existed; schema lookup, caching, registration, and error paths had no
observability.

Added/modified the following metrics:

• cache_hits / cache_misses: local schema_id_cache lookup counts
  (no I/O).
• cache_actor_requests_duration: histogram (unit s, 1ms–5s
  boundaries) of CacheActor round-trip latency, tagged by outcome
  (success/not_found/channel_closed/timeout).
• cache_actor_timeouts: CacheActor round-trips that timed out.
  Kept under its own cache_actor.requests.* namespace, separate from
  local schema_cache.*.
• schema_registry_registrations: schemas registered with the Schema
  Registry (startup + cache-miss lookups).
• schema_registration_errors: failures loading/extending/registering
  a schema during a cache-miss lookup — previously only logged, with
  no metric. Tagged by reason.
• schema_fallbacks: messages that fell back to the default schema or
  were sent without one, now tagged by reason instead of being one
  undifferentiated counter.

reason/outcome tags are backed by strum-derived enums
(SchemaRegistrationErrorReason, SchemaFallbackReason,
CacheActorRequestOutcome); each counter's description is generated
from the enum's VARIANTS so they can't drift apart.

[Copilot]

Pull request overview

Adds additional OpenTelemetry instrumentation to the Kafka YANG publisher so schema lookup, caching layers, Schema Registry registrations, and schema fallback behavior are observable alongside existing send/receive/delivery metrics.

Changes:

• Expanded KafkaYangPublisherStats with new counters and an up-down counter for cache/registry/fallback telemetry.
• Instrumented local schema ID cache hit/miss paths and CacheActor round-trip lifecycle (pending/timeouts).
• Counted Schema Registry registrations and schema fallback events.

Suppressed comments (1)

crates/collector/src/publishers/kafka_yang.rs:552

• cache_misses is incremented before the CacheActor request is successfully sent. If cache_req_tx.send(...) fails, the function returns early but the metric still implies the miss was forwarded to the CacheActor (per the counter description). Consider incrementing cache_misses only after the send succeeds (or adjust the metric/description if you intend to count attempted misses even when the request could not be dispatched).

        self.stats.cache_misses.add(1, &[]);

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 1 out of 1 changed files in this pull request and generated no new comments.

Suppressed comments (2)

crates/collector/src/publishers/kafka_yang.rs:148

• For consistency with other OTel tagging in this repo, include network.peer.port alongside network.peer.address when building subscription tags. Several existing components attach both (e.g. crates/collector/src/yang_push/enrichment.rs:141-147 and crates/yang-push/src/cache/actor.rs:756-763), so omitting the port here makes it harder to correlate metrics by peer and breaks established tagging patterns.

fn subscription_info_tags(
    subscription_info: Option<&SubscriptionInfo>,
) -> Vec<opentelemetry::KeyValue> {
    let Some(subscription_info) = subscription_info else {
        return Vec::new();

crates/collector/src/publishers/kafka_yang.rs:606

• The PR description says schema_fallbacks includes cases where the lookup channel is closed, but when the CacheActor request cannot be sent (cache_req_tx.send(...) returns Err), the code currently returns an error and the message is dropped (no fallback schema/no-schema send). If "channel closed" should be counted as a fallback (as described), consider falling back to default_schema_id (or None) here and increment schema_fallbacks.

            warn!("Failed to request schema for content_id: {}", id);
            return Err(err.into());
        }
        self.stats.cache_actor_pending.add(1, &tags);

[rodonile]

I addressed the feedback, and also added reason keys (same enum based approach used for the validation actor) for schema registration error reason and schema fallback reason

[Copilot]

Pull request overview

Copilot reviewed 1 out of 1 changed files in this pull request and generated no new comments.

Suppressed comments (2)

crates/collector/src/publishers/kafka_yang.rs:262

• The PR description says CacheActor metrics live under a cache_actor.requests.* namespace, but this histogram uses cache_actor.request.duration (singular) while the timeout counter uses cache_actor.requests.timeouts (plural). This inconsistency will make dashboards/metric discovery harder; consider renaming the histogram to match the requests.* namespace.

        let cache_actor_request_duration = meter
            .f64_histogram("netcalyx.collector.kafka.yang.cache_actor.request.duration")
            .with_description(format!(

crates/collector/src/publishers/kafka_yang.rs:214

• The PR description mentions a cache_actor_pending up-down counter and doc comments on all KafkaYangPublisherStats fields, but the implementation currently adds no pending/in-flight metric and the stats fields (including the new ones) still have no doc comments. Either add the pending metric (increment before sending the CacheActor request and decrement on every completion path, including send failure/timeout) and the field docs, or update the PR description to match what was actually implemented.

This issue also appears on line 260 of the same file.

    cache_hits: opentelemetry::metrics::Counter<u64>,
    cache_misses: opentelemetry::metrics::Counter<u64>,
    cache_actor_request_duration: opentelemetry::metrics::Histogram<f64>,
    cache_actor_timeouts: opentelemetry::metrics::Counter<u64>,
    schema_registry_registrations: opentelemetry::metrics::Counter<u64>,

[riccardo-negri]

All previous metrics

received: opentelemetry::metrics::Counter<u64>,
sent: opentelemetry::metrics::Counter<u64>,
send_retries: opentelemetry::metrics::Counter<u64>,
error_decode: opentelemetry::metrics::Counter<u64>,
error_send: opentelemetry::metrics::Counter<u64>,
delivered_messages: opentelemetry::metrics::Counter<u64>,
failed_delivery_messages: opentelemetry::metrics::Counter<u64>,

are exported without any tag (e.g. Box::new([])).

[rodonile]

All previous metrics

received: opentelemetry::metrics::Counter<u64>,
sent: opentelemetry::metrics::Counter<u64>,
send_retries: opentelemetry::metrics::Counter<u64>,
error_decode: opentelemetry::metrics::Counter<u64>,
error_send: opentelemetry::metrics::Counter<u64>,
delivered_messages: opentelemetry::metrics::Counter<u64>,
failed_delivery_messages: opentelemetry::metrics::Counter<u64>,

are exported without any tag (e.g. Box::new([])).

Addressed for consistency of the PR. Would be funny if they are removed right away as part of the next otel-related discussion though ;)

[Copilot]

Pull request overview

Copilot reviewed 1 out of 1 changed files in this pull request and generated no new comments.

Suppressed comments (4)

crates/collector/src/publishers/kafka_yang.rs:931

• sent counter is now tagged with subscription_info_tags(...) (subscription ID + router content ID). Those fields are typically unbounded/high-cardinality and can explode the number of metric time series and backend cost for a high-volume counter like ...kafka.yang.sent. Consider keeping this counter untagged (as before) or restricting tags to low-cardinality dimensions (e.g., peer address / target only).

            match self.producer.send(record) {
                Ok(_) => {
                    self.stats.sent.add(1, &tags);
                    return Ok(());

crates/collector/src/publishers/kafka_yang.rs:948

• send_retries is now tagged with subscription_info_tags(...) (includes subscription ID + router content ID). As a retry counter on a potentially hot path, this can introduce very high label cardinality and degrade observability backends. Consider leaving it untagged or using a reduced tag set.

                        debug!("Kafka message queue is full, sleeping for {polling_interval:?}");
                        self.stats.send_retries.add(1, &tags);
                        tokio::time::sleep(polling_interval).await;

crates/collector/src/publishers/kafka_yang.rs:995

• received counter is now tagged with subscription_info_tags(...), which includes subscription ID and router content ID. For a high-throughput ingest counter this creates high-cardinality metrics and makes received harder to use as a fleet-wide SLI. Consider keeping received untagged (as before) and using separate, explicitly-scoped metrics for per-subscription breakdowns if needed.

                    match msg {
                        Ok(msg) => {
                            let subscription_info = self.config.yang_converter.subscription_info(&msg);
                            self.stats
                                .received
                                .add(1, &subscription_info_tags(subscription_info.as_ref()));
                            if let Err(err) = self.send(msg).await {

crates/collector/src/publishers/kafka_yang.rs:686

• On cache-actor request send failure, register_schema now returns Ok(self.default_schema_id) (fallback) instead of propagating an error. This is a behavior change: messages may be published with a default/no schema when the cache actor path is unavailable, rather than failing the send. If the intent is only to add metrics/observability, consider restoring the previous error propagation; if fallback is intended, it would be good to explicitly document this semantic change (and ensure downstream expectations around schema correctness are still met).

        if let Err(err) = self
            .cache_req_tx
            .send(CacheLookupCommand::LookupByContentIdOneShot {
                content_id: id.to_string(),
                tx: response_tx,
            })
            .await
        {
            let mut error_tags = tags.clone();
            error_tags.push(opentelemetry::KeyValue::new(
                REASON_KEY,
                <&str>::from(SchemaRegistrationErrorReason::CacheLookupSendFailed),
            ));
            self.stats.schema_registration_errors.add(1, &error_tags);

            tags.push(opentelemetry::KeyValue::new(
                REASON_KEY,
                <&str>::from(SchemaFallbackReason::CacheLookupSendFailed),
            ));
            self.stats.schema_fallbacks.add(1, &tags);

            warn!(
                "Failed to request schema for content_id: {id}, fallback to using root schema \
                (id={:?}): {err}",
                self.default_schema_id
            );
            return Ok(self.default_schema_id);
        }