diff --git a/README.md b/README.md index 6e5a31d..0bcff3d 100644 --- a/README.md +++ b/README.md @@ -219,6 +219,8 @@ let parser = NetflowParser::builder() - Templates are keyed by template ID (per source) - Native and V9-style IPFIX data and Options templates share one Template ID namespace; each accepted definition becomes the sole owner of its ID +- NetFlow v9 ordinary and Options templates share one Template ID namespace; + the last valid definition replaces the previous owner - Each parser instance maintains its own template cache - For multi-source deployments, use `RouterScopedParser` (see Template Management section) diff --git a/src/variable_versions/v9/parser.rs b/src/variable_versions/v9/parser.rs index 404b772..cfd6209 100644 --- a/src/variable_versions/v9/parser.rs +++ b/src/variable_versions/v9/parser.rs @@ -39,6 +39,9 @@ use std::num::NonZeroUsize; use std::sync::Arc; /// Stateful NetFlow V9 parser with LRU template caching and optional pending flow support. +/// +/// Ordinary and Options templates share one protocol-level Template ID namespace. +/// The last valid definition received for an ID is its sole logical owner. #[derive(Debug)] pub struct V9Parser { pub(crate) templates: LazyLruCache>>, @@ -193,8 +196,8 @@ impl V9Parser { } } - /// Best-effort removal of an LRU-evicted entry from the secondary store. - fn evict_template_from_store(&mut self, kind: TemplateKind, template_id: u16) { + /// Best-effort removal of a superseded or evicted entry from the secondary store. + fn remove_template_from_store(&mut self, kind: TemplateKind, template_id: u16) { let Some(store) = self.template_store.as_ref() else { return; }; @@ -205,6 +208,87 @@ impl V9Parser { } } + /// Remove one physical cache entry and report whether it was still a live owner. + fn remove_cached_owner( + cache: &mut LazyLruCache>>, + template_id: TemplateId, + ttl_config: &Option, + metrics: &mut CacheMetricsInner, + ) -> bool { + let Some(entry) = cache.pop(&template_id) else { + return false; + }; + + if ttl_config.as_ref().is_some_and(|ttl| entry.is_expired(ttl)) { + metrics.record_expiration(); + false + } else { + true + } + } + + /// Install an ordinary template as the sole owner of its Template ID. + fn install_template(&mut self, template: &Template) { + let template_id = template.template_id; + if Self::remove_cached_owner( + &mut self.options_templates, + template_id, + &self.ttl_config, + &mut self.metrics, + ) { + self.metrics.record_collision(); + } + self.remove_template_from_store(TemplateKind::V9Options, template_id); + + if let Some(existing) = self.templates.peek(&template_id) + && existing.template.as_ref() != template + { + self.metrics.record_collision(); + } + + let wrapped = + TemplateWithTtl::new(Arc::new(template.clone()), self.ttl_config.is_some()); + if let Some((evicted_key, _)) = self.templates.push(template_id, wrapped) + && evicted_key != template_id + { + self.metrics.record_eviction(); + self.remove_template_from_store(TemplateKind::V9Data, evicted_key); + } + self.metrics.record_insertion(); + self.store_template(template); + } + + /// Install an Options template as the sole owner of its Template ID. + fn install_options_template(&mut self, template: &OptionsTemplate) { + let template_id = template.template_id; + if Self::remove_cached_owner( + &mut self.templates, + template_id, + &self.ttl_config, + &mut self.metrics, + ) { + self.metrics.record_collision(); + } + self.remove_template_from_store(TemplateKind::V9Data, template_id); + + if let Some(existing) = self.options_templates.peek(&template_id) + && existing.template.as_ref() != template + { + self.metrics.record_collision(); + } + + let wrapped = + TemplateWithTtl::new(Arc::new(template.clone()), self.ttl_config.is_some()); + if let Some((evicted_key, _)) = self.options_templates.push(template_id, wrapped) + && evicted_key != template_id + { + self.metrics.record_eviction(); + self.remove_template_from_store(TemplateKind::V9Options, evicted_key); + } + self.metrics.record_insertion(); + self.store_options_template(template); + } + /// Insert a read-through-recovered template into the in-process LRU, /// mirroring eviction back to the secondary store and tracking the /// `Restored` event for hook firing after this packet completes. @@ -707,29 +791,8 @@ impl FlowSetBody { nom::error::ErrorKind::Verify, ))); } - let ttl_enabled = parser.ttl_config.is_some(); for template in &valid_templates { - let arc_template = Arc::new(template.clone()); - let wrapped = TemplateWithTtl::new(arc_template, ttl_enabled); - // Check for collision (same ID, different definition) - // Use peek() to avoid affecting LRU ordering - if let Some(existing) = parser.templates.peek(&template.template_id) - && existing.template.as_ref() != template - { - parser.metrics.record_collision(); - } - // push() returns Some in two cases: (1) a different key was LRU-evicted - // to make room, or (2) the same key existed and its value was replaced. - // Only count case (1) as an eviction. - if let Some((evicted_key, _evicted)) = - parser.templates.push(template.template_id, wrapped) - && evicted_key != template.template_id - { - parser.metrics.record_eviction(); - parser.evict_template_from_store(TemplateKind::V9Data, evicted_key); - } - parser.metrics.record_insertion(); - parser.store_template(template); + parser.install_template(template); } let result = Templates { templates: valid_templates, @@ -752,30 +815,8 @@ impl FlowSetBody { nom::error::ErrorKind::Verify, ))); } - // Store templates efficiently using Arc for zero-cost sharing - let ttl_enabled = parser.ttl_config.is_some(); for template in &valid_templates { - let arc_template = Arc::new(template.clone()); - let wrapped = TemplateWithTtl::new(arc_template, ttl_enabled); - // Check for collision (same ID, different definition) - // Use peek() to avoid affecting LRU ordering - if let Some(existing) = parser.options_templates.peek(&template.template_id) - && existing.template.as_ref() != template - { - parser.metrics.record_collision(); - } - // push() returns Some in two cases: (1) a different key was LRU-evicted - // to make room, or (2) the same key existed and its value was replaced. - // Only count case (1) as an eviction. - if let Some((evicted_key, _evicted)) = - parser.options_templates.push(template.template_id, wrapped) - && evicted_key != template.template_id - { - parser.metrics.record_eviction(); - parser.evict_template_from_store(TemplateKind::V9Options, evicted_key); - } - parser.metrics.record_insertion(); - parser.store_options_template(template); + parser.install_options_template(template); } let result = OptionsTemplates { templates: valid_templates, diff --git a/tests/v9_template_id_ownership.rs b/tests/v9_template_id_ownership.rs new file mode 100644 index 0000000..5d8c50a --- /dev/null +++ b/tests/v9_template_id_ownership.rs @@ -0,0 +1,165 @@ +use netflow_parser::variable_versions::v9::FlowSetBody; +use netflow_parser::{ + InMemoryTemplateStore, NetflowPacket, NetflowParser, TemplateKind, TemplateStore, + TemplateStoreKey, TtlConfig, +}; +use std::sync::Arc; +use std::thread; +use std::time::Duration; + +const TEMPLATE_ID: u16 = 256; +const STORE_SCOPE: &str = "v9-template-owner"; + +fn v9_header(count: u16) -> Vec { + let mut packet = Vec::with_capacity(20); + packet.extend_from_slice(&9u16.to_be_bytes()); + packet.extend_from_slice(&count.to_be_bytes()); + packet.extend_from_slice(&0u32.to_be_bytes()); + packet.extend_from_slice(&0u32.to_be_bytes()); + packet.extend_from_slice(&0u32.to_be_bytes()); + packet.extend_from_slice(&42u32.to_be_bytes()); + packet +} + +fn flowset(id: u16, body: &[u8]) -> Vec { + let mut bytes = Vec::with_capacity(4 + body.len()); + bytes.extend_from_slice(&id.to_be_bytes()); + bytes.extend_from_slice(&u16::try_from(4 + body.len()).unwrap().to_be_bytes()); + bytes.extend_from_slice(body); + bytes +} + +fn data_template() -> Vec { + let mut body = Vec::with_capacity(8); + body.extend_from_slice(&TEMPLATE_ID.to_be_bytes()); + body.extend_from_slice(&1u16.to_be_bytes()); + body.extend_from_slice(&1u16.to_be_bytes()); + body.extend_from_slice(&4u16.to_be_bytes()); + flowset(0, &body) +} + +fn options_template() -> Vec { + let mut body = Vec::with_capacity(16); + body.extend_from_slice(&TEMPLATE_ID.to_be_bytes()); + body.extend_from_slice(&4u16.to_be_bytes()); + body.extend_from_slice(&4u16.to_be_bytes()); + body.extend_from_slice(&1u16.to_be_bytes()); + body.extend_from_slice(&4u16.to_be_bytes()); + body.extend_from_slice(&2u16.to_be_bytes()); + body.extend_from_slice(&4u16.to_be_bytes()); + flowset(1, &body) +} + +fn data_flowset() -> Vec { + flowset(TEMPLATE_ID, &[0, 0, 0, 1, 0, 0, 0, 2]) +} + +fn packet(flowsets: &[&[u8]]) -> Vec { + let mut packet = v9_header(u16::try_from(flowsets.len()).unwrap()); + for flowset in flowsets { + packet.extend_from_slice(flowset); + } + packet +} + +fn parsed_v9( + result: &netflow_parser::ParseResult, +) -> &netflow_parser::variable_versions::v9::V9 { + assert!(result.error.is_none(), "{:?}", result.error); + assert_eq!(result.packets.len(), 1); + let NetflowPacket::V9(packet) = &result.packets[0] else { + panic!("expected NetFlow v9 packet"); + }; + packet +} + +#[test] +fn last_template_kind_owns_the_id_in_wire_order() { + for options_last in [true, false] { + let ordinary = data_template(); + let options = options_template(); + let data = data_flowset(); + let definitions = if options_last { + [&ordinary[..], &options[..], &data[..]] + } else { + [&options[..], &ordinary[..], &data[..]] + }; + + let mut parser = NetflowParser::default(); + let result = parser.parse_bytes(&packet(&definitions)); + let v9 = parsed_v9(&result); + + if options_last { + assert!(matches!(v9.flowsets[2].body, FlowSetBody::OptionsData(_))); + } else { + assert!(matches!(v9.flowsets[2].body, FlowSetBody::Data(_))); + } + + let cache = parser.v9_cache_info(); + assert_eq!(cache.current_size, 1); + assert_eq!(cache.num_caches, 2); + assert_eq!(cache.metrics.insertions, 2); + assert_eq!(cache.metrics.collisions, 1); + assert_eq!(cache.metrics.evictions, 0); + } +} + +#[test] +fn template_store_retains_only_the_last_kind() { + for options_last in [true, false] { + let store = Arc::new(InMemoryTemplateStore::new()); + let builder = NetflowParser::builder() + .with_template_store(store.clone()) + .with_template_store_scope(STORE_SCOPE); + let mut writer = builder.clone().build().unwrap(); + + let ordinary = data_template(); + let options = options_template(); + let definitions = if options_last { + [&ordinary[..], &options[..]] + } else { + [&options[..], &ordinary[..]] + }; + let result = writer.parse_bytes(&packet(&definitions)); + parsed_v9(&result); + + let data_key = TemplateStoreKey::new(STORE_SCOPE, TemplateKind::V9Data, TEMPLATE_ID); + let options_key = + TemplateStoreKey::new(STORE_SCOPE, TemplateKind::V9Options, TEMPLATE_ID); + assert_eq!(store.get(&data_key).unwrap().is_some(), !options_last); + assert_eq!(store.get(&options_key).unwrap().is_some(), options_last); + + let mut reader = builder.build().unwrap(); + let data = data_flowset(); + let result = reader.parse_bytes(&packet(&[&data])); + let v9 = parsed_v9(&result); + if options_last { + assert!(matches!(v9.flowsets[0].body, FlowSetBody::OptionsData(_))); + } else { + assert!(matches!(v9.flowsets[0].body, FlowSetBody::Data(_))); + } + assert_eq!(reader.v9_cache_info().current_size, 1); + } +} + +#[test] +fn expired_opposite_kind_is_not_counted_as_a_collision() { + let mut parser = NetflowParser::builder() + .with_v9_ttl(TtlConfig::new(Duration::from_millis(1))) + .build() + .unwrap(); + + let ordinary = data_template(); + parsed_v9(&parser.parse_bytes(&packet(&[&ordinary]))); + thread::sleep(Duration::from_millis(5)); + + let options = options_template(); + parsed_v9(&parser.parse_bytes(&packet(&[&options]))); + + let cache = parser.v9_cache_info(); + assert_eq!(cache.current_size, 1); + assert_eq!(cache.metrics.insertions, 2); + assert_eq!(cache.metrics.expired, 1); + assert_eq!(cache.metrics.collisions, 0); + assert_eq!(cache.metrics.evictions, 0); +}