diff --git a/crates/collector/src/lib.rs b/crates/collector/src/lib.rs index f4ae77b2..29ce948d 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"); @@ -33,6 +34,7 @@ use netcalyx_bmp_service::supervisor::BmpSupervisorHandle; use netcalyx_flow_pkt::FlowInfo; use netcalyx_flow_service::FlowRequest; use netcalyx_flow_service::flow_supervisor::FlowCollectorsSupervisorActorHandle; +use netcalyx_netconf_proto::yang_module_cache::YangModuleCache; use netcalyx_udp_notif_pkt::raw::MediaType; use netcalyx_udp_notif_service::UdpNotifRequest; use netcalyx_udp_notif_service::supervisor::UdpNotifSupervisorHandle; @@ -557,12 +559,14 @@ pub async fn init_udp_notif_collection( // Only one schema cache is needed for all publishers let cache_location: PathBuf = udp_notif_config.cache_location.into(); - let netconf_fetcher = netconf_fetcher(&udp_notif_config.netconf)?; + let global_module_cache = YangModuleCache::new(); + let netconf_fetcher = netconf_fetcher(&udp_notif_config.netconf, global_module_cache.clone())?; let (_schema_join, schema_handle) = CacheActorHandle::new( 10000, either::Right(cache_location), netconf_fetcher, Duration::from_mins(5), + global_module_cache, either::Left(meter.clone()), )?; @@ -993,7 +997,10 @@ fn serialize_bmp( Ok((Some(key), value)) } -fn netconf_fetcher(config: &NetconfConfig) -> Result { +fn netconf_fetcher( + config: &NetconfConfig, + global_module_cache: YangModuleCache, +) -> Result { let user = &config.username; let private_key_path: PathBuf = (&config.private_key_path).into(); @@ -1022,6 +1029,7 @@ fn netconf_fetcher(config: &NetconfConfig) -> Result { announce_caps: HashSet, handler: H, config: Arc, + module_cache: YangModuleCache, } impl NetconfSshConnectConfig { - pub const fn new( + pub fn new( auth: SshAuth, peer_address: SocketAddr, local_address: Option, @@ -169,9 +171,15 @@ impl NetconfSshConnectConfig { announce_caps, handler, config, + module_cache: YangModuleCache::new(), } } + pub fn with_module_cache(mut self, module_cache: YangModuleCache) -> Self { + self.module_cache = module_cache; + self + } + pub const fn auth(&self) -> &SshAuth { &self.auth } @@ -286,7 +294,13 @@ where config.peer_address ); let stream = channel.into_stream(); - NetConfSshClient::connect(config.peer_address, stream, config.announce_caps).await + NetConfSshClient::connect( + config.peer_address, + stream, + config.announce_caps, + config.module_cache, + ) + .await } pub struct NetConfSshClient { @@ -312,6 +326,11 @@ pub struct NetConfSshClient { /// making multiple requests to the device to get the filters when /// processing multiple subscriptions yang_push_filters: Option, + + /// Global YANG module cache shared across all sessions that use the same + /// [`YangModuleCache`] instance. Populated transparently by + /// [`get_yang_module`](Self::get_yang_module); callers see no difference. + module_cache: YangModuleCache, } impl NetConfSshClient { @@ -340,6 +359,10 @@ impl NetConfSshClient { pub fn yang_library(&self) -> Option> { self.yang_library.as_ref().map(Arc::clone) } + + pub fn module_cache(&self) -> &YangModuleCache { + &self.module_cache + } } impl NetConfSshClient { @@ -393,6 +416,7 @@ impl NetConfSshClient { peer: SocketAddr, stream: T, announce_caps: HashSet, + module_cache: YangModuleCache, ) -> Result { let framed = Framed::new(stream, SshCodec::default()); let (framed, session_id, peer_caps) = Self::exchange_hello(framed, announce_caps).await?; @@ -405,6 +429,7 @@ impl NetConfSshClient { next_message_id, yang_library: None, yang_push_filters: None, + module_cache, }) } @@ -481,20 +506,37 @@ impl NetConfSshClient { Ok(()) } - /// Get YANG schema from the device - pub async fn get_schema( + /// Fetch a YANG module from the device via the NETCONF `get-schema` RPC, + /// consulting the shared module cache first. + /// + /// Caching requires a `version`: the shared cache is keyed by + /// `(name, revision)`, so only modules that carry a revision are cached. + /// When `version` is `Some`, a cache hit skips the RPC entirely and a miss + /// populates the cache for all future calls. When `version` is `None` the + /// module is fetched straight from the device on every call, bypassing the + /// cache. + pub async fn get_yang_module( &mut self, name: &str, version: Option<&str>, - ) -> Result, NetConfSshClientError> { + ) -> Result, NetConfSshClientError> { + if let Some(version) = version + && let Some(cached) = self.module_cache.get(name, version) + { + trace!( + "[{}] yang module cache hit for `{name}` revision {version}", + self.peer + ); + return Ok(cached); + } debug!( - "[{}] Getting a YANG schema with name `{name}` and version {version:?}", + "[{}] Getting a YANG module with name `{name}` and version {version:?}", self.peer ); let rpc = RpcOperation::WellKnown(WellKnownOperation::GetSchema { identifier: name.into(), version: version.map(Into::into), - format: Some(YangSchemaFormat::Yang), + format: Some(GetSchemaFormat::Yang), }); let message_id = self.rpc(rpc).await?; let rpc_reply = self.rpc_reply().await?; @@ -510,7 +552,13 @@ impl NetConfSshClient { if let RpcResponse::WellKnown(WellKnownRpcResponse::YangSchema { schema }) = rpc_response { - return Ok(schema); + let arc: Arc = Arc::from(schema.as_ref()); + // Only versioned modules are cached; the cache is keyed by + // `(name, revision)`. + if let Some(version) = version { + self.module_cache.insert(name, version, Arc::clone(&arc)); + } + return Ok(arc); } else { unreachable!() } @@ -615,7 +663,9 @@ impl NetConfSshClient { visited.insert(module.name().to_string()); // Fetch the YANG schema - let schema = self.get_schema(module.name(), module.revision()).await?; + let schema = self + .get_yang_module(module.name(), module.revision()) + .await?; // Parse dependencies from schema let deps = extract_yang_dependencies(&schema).map_err(|error| { NetConfSshClientError::YangSchemaParsingError { @@ -684,17 +734,22 @@ impl NetConfSshClient { match module { ModuleType::Full(module) => { builder - .add_module(module, schema, checker) + .add_module(module, Box::from(schema.as_ref()), checker) .map_err(NetConfSshClientError::DependencyError)?; } ModuleType::FullSubmodule(module_name, submodule) => { builder - .add_submodule_for_module(module_name.as_ref(), submodule, schema, checker) + .add_submodule_for_module( + module_name.as_ref(), + submodule, + Box::from(schema.as_ref()), + checker, + ) .map_err(NetConfSshClientError::DependencyError)?; } ModuleType::ImportOnly(module) => { builder - .add_import_only_module(module, schema, checker) + .add_import_only_module(module, Box::from(schema.as_ref()), checker) .map_err(NetConfSshClientError::DependencyError)?; } ModuleType::ImportOnlySubmodule(module_name, submodule) => { @@ -702,7 +757,7 @@ impl NetConfSshClient { .add_submodule_for_import_only_module( module_name.as_ref(), submodule, - schema, + Box::from(schema.as_ref()), checker, ) .map_err(NetConfSshClientError::DependencyError)?; diff --git a/crates/netconf-proto/src/lib.rs b/crates/netconf-proto/src/lib.rs index de62f629..9b4e34fe 100644 --- a/crates/netconf-proto/src/lib.rs +++ b/crates/netconf-proto/src/lib.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"); @@ -30,6 +31,7 @@ pub mod client; pub mod codec; pub mod protocol; pub mod xml_utils; +pub mod yang_module_cache; pub mod yang_push; pub mod yanglib; pub mod yangparser; diff --git a/crates/netconf-proto/src/protocol.rs b/crates/netconf-proto/src/protocol.rs index 0895017d..47397253 100644 --- a/crates/netconf-proto/src/protocol.rs +++ b/crates/netconf-proto/src/protocol.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"); @@ -323,7 +324,7 @@ impl XmlSerialize for Rpc { } #[derive(Eq, PartialEq, Debug, Copy, Clone, Serialize, Deserialize, strum_macros::Display)] -pub enum YangSchemaFormat { +pub enum GetSchemaFormat { #[strum(serialize = "xsd")] Xsd, @@ -340,17 +341,17 @@ pub enum YangSchemaFormat { Rnc, } -impl<'a> XmlDeserialize<'a, YangSchemaFormat> for YangSchemaFormat { +impl<'a> XmlDeserialize<'a, GetSchemaFormat> for GetSchemaFormat { fn xml_deserialize(parser: &mut XmlParser<'a, impl io::BufRead>) -> Result { parser.skip_text()?; parser.open(Some(NETCONF_MONITORING_NS), "format")?; let value_str = parser.tag_string()?; let value = match value_str.as_ref().trim() { - "xsd" => YangSchemaFormat::Xsd, - "yang" => YangSchemaFormat::Yang, - "yin" => YangSchemaFormat::Yin, - "rng" => YangSchemaFormat::Rng, - "rnc" => YangSchemaFormat::Rnc, + "xsd" => GetSchemaFormat::Xsd, + "yang" => GetSchemaFormat::Yang, + "yin" => GetSchemaFormat::Yin, + "rng" => GetSchemaFormat::Rng, + "rnc" => GetSchemaFormat::Rnc, _ => { return Err(ParsingError::InvalidValue(format!( "unknown YANG schema format `{value_str}`" @@ -362,7 +363,7 @@ impl<'a> XmlDeserialize<'a, YangSchemaFormat> for YangSchemaFormat { } } -impl XmlSerialize for YangSchemaFormat { +impl XmlSerialize for GetSchemaFormat { fn xml_serialize( &self, writer: &mut XmlWriter, @@ -841,7 +842,7 @@ pub enum WellKnownOperation { /// The data modeling language of the schema. If this parameter is not /// present, and more than one formats of the schema exists on the /// server, a 'data-not-unique' error is returned, as described above. - format: Option, + format: Option, }, } @@ -982,7 +983,7 @@ impl WellKnownOperation { None }; - let format = match YangSchemaFormat::xml_deserialize(parser) { + let format = match GetSchemaFormat::xml_deserialize(parser) { Ok(format) => Some(format), Err(ParsingError::WrongToken { expecting, .. }) if expecting == "" => None, Err(err) => return Err(err), @@ -1001,7 +1002,7 @@ impl WellKnownOperation { writer: &mut XmlWriter, identifier: &str, version: &Option>, - format: &Option, + format: &Option, ) -> Result<(), quick_xml::Error> { let mut ns_added = false; if writer.get_namespace_prefix(NETCONF_MONITORING_NS).is_none() { @@ -3120,7 +3121,7 @@ mod tests { let get_schema = WellKnownOperation::GetSchema { identifier: "foo".into(), version: Some("1.0".into()), - format: Some(YangSchemaFormat::Yang), + format: Some(GetSchemaFormat::Yang), }; test_xml_value(get_schema_str, get_schema)?; Ok(()) @@ -3140,7 +3141,7 @@ mod tests { RpcOperation::WellKnown(WellKnownOperation::GetSchema { identifier: "foo".into(), version: Some("1.0".into()), - format: Some(YangSchemaFormat::Yang), + format: Some(GetSchemaFormat::Yang), }), ); test_xml_value(get_schema_str, get_schema)?; diff --git a/crates/netconf-proto/src/yang_module_cache.rs b/crates/netconf-proto/src/yang_module_cache.rs new file mode 100644 index 00000000..8796e8b7 --- /dev/null +++ b/crates/netconf-proto/src/yang_module_cache.rs @@ -0,0 +1,262 @@ +// Copyright (C) 2026-present The NetCalyx Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or +// implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Global YANG module text cache. +//! +//! The YANG specification guarantees that a `(module_name, revision)` pair +//! identifies stable content: any published change to a module MUST add a new +//! revision date ([RFC 7950], §11). A single instance of +//! [`YangModuleCache`] can therefore be shared across all routers and all SSH +//! sessions: once a module is fetched from any device, subsequent calls to +//! [`NetConfSshClient::get_yang_module`](crate::client::NetConfSshClient::get_yang_module) +//! skip the NETCONF `get-schema` RPC entirely. +//! +//! Note: the NETCONF RPC is still called `get-schema` (per [RFC 6022]), but +//! what it returns — and what we cache — is a YANG **module** text, not a +//! schema. +//! +//! ## Metrics +//! +//! [`YangModuleCache`] exposes three plain counters via [`YangModuleCacheStats`]: +//! +//! | field | meaning | +//! |-------|---------| +//! | [`YangModuleCacheStats::hits`] | `get-schema` RPC avoided (module already cached) | +//! | [`YangModuleCacheStats::misses`] | `get-schema` RPC issued (module not yet cached) | +//! | [`YangModuleCacheStats::size`] | number of distinct modules currently cached | +//! +//! These are `AtomicU64` so they can be read from any thread without holding +//! the cache lock. Higher-level crates that own an OTel meter can poll them +//! and record gauges / counters as needed. +//! +//! ## References +//! +//! - [RFC 6022]: YANG Module for NETCONF Monitoring — defines the `get-schema` +//! operation used to fetch module texts. +//! - [RFC 7950]: The YANG 1.1 Data Modeling Language — §11 "Updating a Module" +//! (any published change MUST add a new revision date). +//! +//! [RFC 6022]: https://www.rfc-editor.org/rfc/rfc6022 +//! [RFC 7950]: https://www.rfc-editor.org/rfc/rfc7950 + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, RwLock}; + +type ModuleCacheInner = Arc>>>; + +/// Metrics counters exposed by [`YangModuleCache`]. +#[derive(Debug, Default)] +pub struct YangModuleCacheStats { + /// Number of `get-schema` RPCs avoided because the module was already + /// cached. + pub hits: AtomicU64, + /// Number of `get-schema` RPCs issued because the module was not yet + /// cached. + pub misses: AtomicU64, + /// Current number of distinct `(name, revision)` entries in the cache. + pub size: AtomicU64, +} + +/// A thread-safe, globally-shared cache of raw YANG module texts. +/// +/// Keyed by `(module_name, revision)`. The value is the raw module text as +/// returned by a NETCONF `get-schema` RPC, stored as `Arc` so that cache +/// hits — and the value handed back by +/// [`NetConfSshClient::get_yang_module`](crate::client::NetConfSshClient::get_yang_module) +/// — are cheap pointer clones rather than full string copies. (Feeding a +/// module into the `ModuleSetBuilder` still costs one copy, because the builder +/// takes an owned `Box`.) +/// +/// Clone is cheap — clones share the same backing store and stats. +#[derive(Debug, Clone, Default)] +pub struct YangModuleCache { + inner: ModuleCacheInner, + stats: Arc, +} + +impl YangModuleCache { + pub fn new() -> Self { + Self::default() + } + + pub fn stats(&self) -> &Arc { + &self.stats + } + + /// Return the cached module text for `(name, revision)`, or `None` on miss. + /// Increments the appropriate stats counter. + pub fn get(&self, name: &str, revision: &str) -> Option> { + let key = Self::make_key(name, revision); + let result = self + .inner + .read() + .expect("yang module cache lock poisoned") + .get(&key) + .cloned(); + if result.is_some() { + self.stats.hits.fetch_add(1, Ordering::Relaxed); + } else { + self.stats.misses.fetch_add(1, Ordering::Relaxed); + } + result + } + + /// Insert a module text. First writer wins: if `(name, revision)` is + /// already present the call is a no-op. This is safe because identical + /// `(name, revision)` always has identical content per the YANG spec. + pub fn insert(&self, name: &str, revision: &str, text: Arc) { + let key = Self::make_key(name, revision); + let mut map = self.inner.write().expect("yang module cache lock poisoned"); + let prev_len = map.len(); + map.entry(key).or_insert(text); + if map.len() > prev_len { + self.stats.size.fetch_add(1, Ordering::Relaxed); + } + } + + /// Number of entries currently in the cache. + pub fn len(&self) -> usize { + self.inner + .read() + .expect("yang module cache lock poisoned") + .len() + } + + pub fn is_empty(&self) -> bool { + self.inner + .read() + .expect("yang module cache lock poisoned") + .is_empty() + } + + fn make_key(name: &str, revision: &str) -> String { + format!("{name}@{revision}") + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_new_is_empty() { + let c = YangModuleCache::new(); + assert!(c.is_empty()); + assert_eq!(c.len(), 0); + assert_eq!(c.stats().size.load(Ordering::Relaxed), 0); + } + + #[test] + fn test_miss_increments_miss_counter() { + let c = YangModuleCache::new(); + assert!(c.get("ietf-interfaces", "2018-02-20").is_none()); + assert_eq!(c.stats().misses.load(Ordering::Relaxed), 1); + assert_eq!(c.stats().hits.load(Ordering::Relaxed), 0); + } + + #[test] + fn test_insert_and_hit_increments_hit_counter() { + let c = YangModuleCache::new(); + c.insert( + "ietf-interfaces", + "2018-02-20", + Arc::from("module ietf-interfaces { }"), + ); + assert_eq!(c.stats().size.load(Ordering::Relaxed), 1); + + let result = c.get("ietf-interfaces", "2018-02-20"); + assert_eq!(result.as_deref(), Some("module ietf-interfaces { }")); + assert_eq!(c.stats().hits.load(Ordering::Relaxed), 1); + assert_eq!(c.stats().misses.load(Ordering::Relaxed), 0); + } + + #[test] + fn test_first_writer_wins() { + let c = YangModuleCache::new(); + c.insert("mod", "2024-01-01", Arc::from("first")); + c.insert("mod", "2024-01-01", Arc::from("second")); + assert_eq!(c.get("mod", "2024-01-01").as_deref(), Some("first")); + assert_eq!(c.stats().size.load(Ordering::Relaxed), 1); + } + + #[test] + fn test_different_revisions_are_distinct_keys() { + let c = YangModuleCache::new(); + c.insert("mod", "2023-01-01", Arc::from("old")); + c.insert("mod", "2024-01-01", Arc::from("new")); + assert_eq!(c.get("mod", "2023-01-01").as_deref(), Some("old")); + assert_eq!(c.get("mod", "2024-01-01").as_deref(), Some("new")); + assert_eq!(c.stats().size.load(Ordering::Relaxed), 2); + } + + #[test] + fn test_clone_shares_state() { + let a = YangModuleCache::new(); + let b = a.clone(); + a.insert("mod", "2024-01-01", Arc::from("value")); + assert_eq!(b.get("mod", "2024-01-01").as_deref(), Some("value")); + // hit recorded on `b` is visible via `a.stats` (same Arc) + assert_eq!(a.stats().hits.load(Ordering::Relaxed), 1); + } + + #[test] + fn test_concurrent_insert_and_get() { + use std::thread; + + let cache = YangModuleCache::new(); + let n_threads = 8; + let n_modules = 20; + + let handles: Vec<_> = (0..n_threads) + .map(|t| { + let c = cache.clone(); + thread::spawn(move || { + for i in 0..n_modules { + let name = format!("mod-{i}"); + let rev = format!("2024-{i:02}-01"); + let text = Arc::from(format!("text-{i}").as_str()); + c.insert(&name, &rev, text); + assert!(c.get(&name, &rev).is_some()); + let _ = c.len(); + } + for i in 0..n_modules { + let name = format!("mod-{i}"); + let rev = format!("2024-{i:02}-01"); + c.insert( + &name, + &rev, + Arc::from(format!("other-text-{t}-{i}").as_str()), + ); + } + }) + }) + .collect(); + + for h in handles { + h.join().expect("thread panicked"); + } + + // All modules must be present and have the first-writer value. + assert_eq!(cache.len(), n_modules); + for i in 0..n_modules { + let name = format!("mod-{i}"); + let rev = format!("2024-{i:02}-01"); + let expected = format!("text-{i}"); + assert_eq!(cache.get(&name, &rev).as_deref(), Some(expected.as_str())); + } + assert_eq!(cache.stats().size.load(Ordering::Relaxed), n_modules as u64); + } +} diff --git a/crates/yang-push/src/cache/actor.rs b/crates/yang-push/src/cache/actor.rs index e411090d..24cbe117 100644 --- a/crates/yang-push/src/cache/actor.rs +++ b/crates/yang-push/src/cache/actor.rs @@ -223,11 +223,13 @@ use crate::{ }; use futures_util::StreamExt; use futures_util::stream::FuturesUnordered; +use netcalyx_netconf_proto::yang_module_cache::YangModuleCache; use netcalyx_netconf_proto::yang_push::types::SubscriptionId; use rustc_hash::FxHashMap; use std::net::SocketAddr; use std::path::PathBuf; use std::sync::Arc; +use std::sync::atomic::Ordering; use std::time::Duration; use tokio::sync::{mpsc, oneshot}; use tokio::task::{JoinError, JoinHandle}; @@ -245,10 +247,15 @@ pub struct CachingStats { pub device_fetch_queue: opentelemetry::metrics::Gauge, pub device_fetch_succeeded: opentelemetry::metrics::Counter, pub device_fetch_failed: opentelemetry::metrics::Counter, + // Observable instruments that read from YangModuleCache atomics. + // Held here to keep the OTel callbacks registered for the lifetime of the actor. + _yang_module_cache_hits: opentelemetry::metrics::ObservableCounter, + _yang_module_cache_misses: opentelemetry::metrics::ObservableCounter, + _yang_module_cache_size: opentelemetry::metrics::ObservableGauge, } impl CachingStats { - pub fn new(meter: opentelemetry::metrics::Meter) -> Self { + pub fn new(meter: opentelemetry::metrics::Meter, module_cache: &YangModuleCache) -> Self { let requests_received = meter .u64_counter("netcalyx.collector.yang_push.caching.requests.received") .with_description("Number of requests received by the YANG library cache actor") @@ -283,6 +290,41 @@ impl CachingStats { .u64_counter("netcalyx.collector.yang_push.caching.device.fetch.response.failed") .with_description("Number of device fetch requests initiated by the YANG library cache actor and failed") .build(); + + let stats = module_cache.stats().clone(); + let _yang_module_cache_hits = { + let s = Arc::clone(&stats); + meter + .u64_observable_counter("netcalyx.collector.yang_push.caching.yang_module_cache.hits") + .with_description("Number of get-schema RPCs avoided because the YANG module was already in the shared cache") + .with_callback(move |counter| { + counter.observe(s.hits.load(Ordering::Relaxed), &[]); + }) + .build() + }; + let _yang_module_cache_misses = { + let s = Arc::clone(&stats); + meter + .u64_observable_counter("netcalyx.collector.yang_push.caching.yang_module_cache.misses") + .with_description("Number of get-schema RPCs issued because the YANG module was not yet in the shared cache") + .with_callback(move |counter| { + counter.observe(s.misses.load(Ordering::Relaxed), &[]); + }) + .build() + }; + let _yang_module_cache_size = { + let s = Arc::clone(&stats); + meter + .u64_observable_gauge("netcalyx.collector.yang_push.caching.yang_module_cache.size") + .with_description( + "Number of distinct YANG modules currently held in the shared module cache", + ) + .with_callback(move |gauge| { + gauge.observe(s.size.load(Ordering::Relaxed), &[]); + }) + .build() + }; + Self { requests_received, pending_cache_requests, @@ -292,6 +334,9 @@ impl CachingStats { device_fetch_queue, device_fetch_succeeded, device_fetch_failed, + _yang_module_cache_hits, + _yang_module_cache_misses, + _yang_module_cache_size, } } } @@ -1225,6 +1270,7 @@ impl CacheActorHandle { schema_cache: either::Either, fetcher: F, fetcher_timeout: Duration, + global_module_cache: YangModuleCache, stats: either::Either, ) -> Result<(JoinHandle>, Self), CacheActorHandleError> { @@ -1235,7 +1281,7 @@ impl CacheActorHandle { either::Either::Right(root_path) => YangLibraryCache::from_disk(root_path)?, }; let stats = match stats { - either::Either::Left(meter) => CachingStats::new(meter), + either::Either::Left(meter) => CachingStats::new(meter, &global_module_cache), either::Either::Right(stats) => stats, }; @@ -1370,6 +1416,7 @@ pub(crate) mod tests { either::Right(cache_dir.path().to_path_buf()), fetcher, Duration::from_secs(1), + YangModuleCache::new(), either::Either::Left(opentelemetry::global::meter("test-meter")), ) .expect("Failed to create cache actor"); @@ -1412,6 +1459,7 @@ pub(crate) mod tests { either::Right(cache_dir.path().to_path_buf()), fetcher, Duration::from_secs(1), + YangModuleCache::new(), either::Either::Left(opentelemetry::global::meter("test-meter")), ) .expect("Failed to create cache actor"); diff --git a/crates/yang-push/src/cache/fetcher.rs b/crates/yang-push/src/cache/fetcher.rs index f6bca478..e18b0bc4 100644 --- a/crates/yang-push/src/cache/fetcher.rs +++ b/crates/yang-push/src/cache/fetcher.rs @@ -29,6 +29,7 @@ use crate::cache::storage::{SubscriptionInfo, YangLibraryCacheError}; use netcalyx_netconf_proto::capabilities::{Capability, NetconfVersion}; use netcalyx_netconf_proto::client::{NetconfSshConnectConfig, SshAuth, SshHandler, connect}; +use netcalyx_netconf_proto::yang_module_cache::YangModuleCache; use netcalyx_netconf_proto::yang_push::filters::StreamSelectionFilterObjects; use netcalyx_netconf_proto::yang_push::subscription::{ DatastoreSelectionFilterObjects, Target, YangPushModuleVersion, @@ -87,6 +88,7 @@ struct FetchConfig { client_config: Arc, default_port: u16, timeout: std::time::Duration, + module_cache: YangModuleCache, } #[derive(Clone, Copy)] @@ -130,6 +132,7 @@ impl NetconfYangLibraryFetcher { default_port: u16, default_timeout: std::time::Duration, retry_cfg: RetryConfig, + global_module_cache: YangModuleCache, ) -> Self { Self { fetch_cfg: FetchConfig { @@ -138,6 +141,7 @@ impl NetconfYangLibraryFetcher { client_config: Arc::new(client_config), default_port, timeout: default_timeout, + module_cache: global_module_cache, }, retry_cfg, } @@ -174,7 +178,8 @@ impl NetconfYangLibraryFetcher { announce_caps, ssh_handler, Arc::clone(&cfg.client_config), - ); + ) + .with_module_cache(cfg.module_cache.clone()); let mut client = match tokio::time::timeout(cfg.timeout, connect(config)).await { Ok(Ok(c)) => c, @@ -247,7 +252,8 @@ impl NetconfYangLibraryFetcher { announce_caps, ssh_handler, Arc::clone(&cfg.client_config), - ); + ) + .with_module_cache(cfg.module_cache.clone()); // Empty subscription info returned in case of errors to keep track of peer and // subscription ID let empty = SubscriptionInfo::new_empty(