Skip to content

Commit a9745c5

Browse files
committed
feat: pluggable TemplateStore for horizontal parser scale-out
Add a TemplateStore trait so V9 and IPFIX templates can be persisted to and re-read from an external backend (Redis, NATS KV, etc.). With a store configured, the parser writes through every learned template, consults the store on every cache miss, and propagates LRU evictions, RFC 7011 §8.1 withdrawals, and explicit clear_*_templates calls. This unblocks running multiple stateless parser replicas behind a UDP load balancer without source-IP-affinity routing. AutoScopedParser auto-derives a per-source scope so exporters using the same template ID with different layouts do not collide in the store. The trait sees opaque Vec<u8> payloads encoded with a small versioned custom binary wire format — no serde_json or other runtime serializer is added to the dependency tree. An InMemoryTemplateStore reference impl is provided for tests. New public API: - TemplateStore, TemplateStoreKey, TemplateKind, TemplateStoreError - InMemoryTemplateStore - NetflowParserBuilder::with_template_store / with_template_store_scope - NetflowParser::set_template_store_scope Eight integration tests in tests/template_store.rs cover write-through, cross-replica read-through, baseline (no-store) unchanged, clear_* propagation, IPFIX withdrawal eviction, and per-source scoping. README gains a "Pluggable Template Storage" section under the Template Management Guide; RELEASES.md notes the feature under 1.0.3. fix: address TemplateStore code-review feedback Apply all blocking and high-value items from the independent review of the TemplateStore branch. Must-fix: - Codec errors no longer collapse into silent cache misses. fetch_*_from_store now distinguishes Ok(None) vs Err vs codec rejection: backend errors are counted in metrics, codec errors are counted AND remove the corrupted key so a fresh template announce can repopulate cleanly. - README's "RedisTemplateStore" example dropped rustdoc hidden-line `#` markers (only stripped in `rust` doctests, not `rust,ignore`) so readers see valid Rust. Should-fix: - Read-through LRU push now captures the eviction return and propagates evict-from-store + record_eviction(), keeping primary and secondary tiers consistent. - V9-in-IPFIX validation extracted into Template::is_valid_with_limits and OptionsTemplate::is_valid_with_limits. Three copies of the rule in parse_templates and the IPFIX read-through helpers collapse into one. - New TemplateEvent::Restored variant fires when a template is pulled in via read-through; observability tools that count Learned can now also count Restored after a parser restart. Hooks fire from parse_bytes via per-parser drain_restored_templates(), driven by a small Vec buffer that is cleared at the top of each V9::parse / IPFix::parse. - Pending-flow replay now also runs against templates restored via the secondary store, not just templates announced in the current packet. - decode_v9_options_template rejects payloads where options_scope_length or options_length is not a multiple of 4, matching the live parse path. - TemplateStoreKey.scope is now Arc<str> (was String). Internal scope fields on Config / V9Parser / IPFixParser / NetflowParserBuilder follow. Per-key clones become refcount bumps. NetflowParserBuilder::with_template_store_scope and NetflowParser::set_template_store_scope take impl Into<Arc<str>>. - Three new metric counters: template_store_restored, template_store_codec_errors, template_store_backend_errors. Tests: - Nine new tests cover backend Err returns (FaultStore fault injection), corrupted-payload codec rejection + cleanup, LRU eviction propagating to store on a full cache, IPFIX options-template read-through, Restored event firing, pending-flow replay after read-through, duplicate-ID write-through, and set_template_store_scope retrofit. Docs: - WIRE_VERSION migration story added to template_store module docs (drain-before-upgrade or version-namespaced scope). - InMemoryTemplateStore documents its mutex-poison panic behavior. All 216 lib tests + 17 integration tests + 66 doctests pass.
1 parent 80678db commit a9745c5

13 files changed

Lines changed: 2395 additions & 23 deletions

File tree

README.md

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -789,6 +789,78 @@ let router_builder = NetflowParser::builder()
789789
let mut scoped = RouterScopedParser::<String>::try_with_builder(router_builder).expect("valid config");
790790
```
791791

792+
### Pluggable Template Storage (Horizontal Scale-Out)
793+
794+
When you scale flow ingestion across multiple parser instances behind a UDP
795+
load balancer, every replica needs to observe the templates announced to
796+
*any* replica — otherwise a data record routed to a fresh pod will be queued
797+
or dropped because that pod has never seen the template. The
798+
[`TemplateStore`] trait is the extension point that solves this.
799+
800+
With a store configured, the parser:
801+
802+
1. **Writes through** every successfully learned template to the store as
803+
opaque bytes (a small custom binary wire format — no `serde_json` or
804+
other runtime serializer is added to the dependency tree).
805+
2. **Reads through** the store on every cache miss before declaring a
806+
template unknown, repopulating the in-process LRU on hit so subsequent
807+
records are served from the hot path.
808+
3. **Propagates** LRU evictions, RFC 7011 §8.1 template withdrawals, and
809+
explicit `clear_*_templates` calls so the store stays in sync.
810+
811+
`AutoScopedParser` automatically derives a per-source scope (e.g.
812+
`v9:1.2.3.4:2055/0`, `ipfix:1.2.3.4:2055/42`) so that two exporters using
813+
the same template ID with different layouts do not collide.
814+
815+
```rust,ignore
816+
use netflow_parser::{AutoScopedParser, InMemoryTemplateStore, NetflowParser};
817+
use std::sync::Arc;
818+
819+
// Plug in your own backend: implement TemplateStore for Redis, NATS KV,
820+
// DynamoDB, etc. The reference InMemoryTemplateStore is shown here for
821+
// illustration; in production it would be replaced.
822+
let store = Arc::new(InMemoryTemplateStore::new());
823+
824+
let builder = NetflowParser::builder()
825+
.with_template_store(store.clone());
826+
827+
// Multi-source: per-exporter scoping is automatic.
828+
let mut parser = AutoScopedParser::try_with_builder(builder).expect("valid");
829+
830+
// Replica B can be brought up cold and start serving data records for
831+
// templates Replica A learned, as long as both share the same store.
832+
```
833+
834+
Implementing the trait against your backend of choice:
835+
836+
```rust,ignore
837+
use netflow_parser::{TemplateStore, TemplateStoreError, TemplateStoreKey};
838+
839+
#[derive(Debug)]
840+
struct RedisTemplateStore { /* your client */ }
841+
842+
impl TemplateStore for RedisTemplateStore {
843+
fn get(&self, key: &TemplateStoreKey) -> Result<Option<Vec<u8>>, TemplateStoreError> {
844+
// GET "{scope}/{kind:?}/{template_id}" from Redis, returning the bytes.
845+
unimplemented!()
846+
}
847+
848+
fn put(&self, key: &TemplateStoreKey, value: &[u8]) -> Result<(), TemplateStoreError> {
849+
// SET with an appropriate TTL matching your operational envelope.
850+
unimplemented!()
851+
}
852+
853+
fn remove(&self, key: &TemplateStoreKey) -> Result<(), TemplateStoreError> {
854+
// DEL — must be idempotent for absent keys.
855+
unimplemented!()
856+
}
857+
}
858+
```
859+
860+
**Single-source deployments** can still benefit from a store for surviving
861+
parser restarts without losing template state. Set the scope explicitly
862+
(or leave it empty) via `with_template_store_scope`.
863+
792864
### Template Lifecycle Management
793865

794866
#### Template Introspection

RELEASES.md

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,38 @@
11
# 1.0.3
22

3+
## Features
4+
5+
* **Pluggable secondary template storage (`TemplateStore`).** Templates can now
6+
be persisted to and re-read from an external backend such as Redis or NATS
7+
KV via a new `TemplateStore` trait. With a store configured the parser
8+
writes through every learned template, consults the store on every cache
9+
miss, and propagates LRU evictions, withdrawals, and explicit clears so the
10+
store stays in sync. This unblocks running multiple stateless parser
11+
instances behind a UDP load balancer without source-IP-affinity routing.
12+
13+
Wire up via the new builder hooks:
14+
15+
```rust
16+
let store = Arc::new(my_redis_backed_store);
17+
let parser = NetflowParser::builder()
18+
.with_template_store(store)
19+
.with_template_store_scope("collector-eu-west-1")
20+
.build()?;
21+
```
22+
23+
`AutoScopedParser` automatically derives a per-source scope so two exporters
24+
using the same template ID with different layouts do not collide in the
25+
store. The trait sees opaque `Vec<u8>` payloads encoded with a small custom
26+
binary wire format — no `serde_json` or other runtime serializer is added
27+
to the dependency tree. An `InMemoryTemplateStore` reference impl is
28+
provided for tests. See the new `template_store` module for the protocol.
29+
30+
* New public API: `TemplateStore`, `TemplateStoreKey`, `TemplateKind`,
31+
`TemplateStoreError`, `InMemoryTemplateStore`,
32+
`NetflowParserBuilder::with_template_store`,
33+
`NetflowParserBuilder::with_template_store_scope`,
34+
`NetflowParser::set_template_store_scope`.
35+
336
## Tests
437

538
* **Restored 34 snapshot-based parser tests** in a new `restored_legacy_tests` module in `src/tests.rs`. These tests had been deleted in PR #210 ("further template validation"), leaving 33 orphaned `.snap` files that were swept up later in PR #262. Coverage included real-world v9/IPFIX captures (`it_parses_v9_ipv6flowlabel`, `it_parses_v9_template_and_data_packet`, `it_parses_ipfix_scappy_example`, mixed-enterprise field templates, multi-template IPFIX, etc.), version-filter checks (`it_doesnt_allow_v5/v7/v9/ipfix`), and re-export round-trip tests (`it_parses_*_and_re_exports`).

src/lib.rs

Lines changed: 158 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ pub mod netflow_common;
66
pub mod protocol;
77
pub mod scoped_parser;
88
pub mod static_versions;
9+
pub mod template_store;
910
mod tests;
1011
pub mod variable_versions;
1112

@@ -47,6 +48,9 @@ pub use variable_versions::template_events::{
4748
};
4849

4950
// Re-export configuration and utility types for convenience
51+
pub use template_store::{
52+
InMemoryTemplateStore, TemplateKind, TemplateStore, TemplateStoreError, TemplateStoreKey,
53+
};
5054
pub use variable_versions::enterprise_registry::{EnterpriseFieldDef, EnterpriseFieldRegistry};
5155
pub use variable_versions::metrics::{CacheInfo, CacheMetrics, ParserCacheInfo};
5256
pub use variable_versions::ttl::TtlConfig;
@@ -274,6 +278,8 @@ pub struct NetflowParserBuilder {
274278
requested_versions: Option<Vec<u16>>,
275279
max_error_sample_size: usize,
276280
template_hooks: TemplateHooks,
281+
template_store: Option<Arc<dyn TemplateStore>>,
282+
template_store_scope: Arc<str>,
277283
}
278284

279285
/// Helper to create a `[bool; 11]` allowed_versions array from a set of version numbers.
@@ -304,6 +310,15 @@ impl std::fmt::Debug for NetflowParserBuilder {
304310
"template_hooks",
305311
&format!("{} hooks", self.template_hooks.len()),
306312
)
313+
.field(
314+
"template_store",
315+
&if self.template_store.is_some() {
316+
"configured"
317+
} else {
318+
"none"
319+
},
320+
)
321+
.field("template_store_scope", &self.template_store_scope)
307322
.finish()
308323
}
309324
}
@@ -317,6 +332,8 @@ impl Default for NetflowParserBuilder {
317332
requested_versions: None,
318333
max_error_sample_size: 256,
319334
template_hooks: TemplateHooks::new(),
335+
template_store: None,
336+
template_store_scope: Arc::from(""),
320337
}
321338
}
322339
}
@@ -776,6 +793,52 @@ impl NetflowParserBuilder {
776793
self
777794
}
778795

796+
/// Attach a [`TemplateStore`] to act as a secondary tier behind the parser's
797+
/// in-process LRU caches.
798+
///
799+
/// With a store configured, the parser writes through every successfully
800+
/// learned template to the store and consults the store on every cache
801+
/// miss before declaring a template unknown. This lets multiple parser
802+
/// instances behind a UDP load balancer share template state without
803+
/// requiring source-IP-affinity routing.
804+
///
805+
/// The store sees opaque `Vec<u8>` payloads encoded with a small custom
806+
/// binary wire format documented in the [`template_store`] module. See
807+
/// [`TemplateStore`] for the trait contract and threading model.
808+
///
809+
/// # Examples
810+
///
811+
/// ```rust
812+
/// use netflow_parser::{NetflowParser, InMemoryTemplateStore};
813+
/// use std::sync::Arc;
814+
///
815+
/// let store = Arc::new(InMemoryTemplateStore::new());
816+
/// let parser = NetflowParser::builder()
817+
/// .with_template_store(store)
818+
/// .build()
819+
/// .expect("Failed to build parser");
820+
/// ```
821+
#[must_use = "builder methods consume self and return a new builder; the return value must be used"]
822+
pub fn with_template_store(mut self, store: Arc<dyn TemplateStore>) -> Self {
823+
self.template_store = Some(store);
824+
self
825+
}
826+
827+
/// Set the scope string written into every [`TemplateStoreKey`].
828+
///
829+
/// Defaults to the empty string, which is appropriate for single-source
830+
/// deployments. For multi-source deployments
831+
/// [`AutoScopedParser`] automatically populates this with the source
832+
/// `SocketAddr` of each per-source parser, so callers usually do not need
833+
/// to set this directly.
834+
///
835+
/// Accepts anything `Into<Arc<str>>` — typically `&str` or `String`.
836+
#[must_use = "builder methods consume self and return a new builder; the return value must be used"]
837+
pub fn with_template_store_scope(mut self, scope: impl Into<Arc<str>>) -> Self {
838+
self.template_store_scope = scope.into();
839+
self
840+
}
841+
779842
/// Validates the builder configuration without constructing a parser.
780843
///
781844
/// This is cheaper than [`build`](Self::build) since it only checks that
@@ -822,8 +885,14 @@ impl NetflowParserBuilder {
822885
/// ```
823886
pub fn build(self) -> Result<NetflowParser, ConfigError> {
824887
self.validate()?;
825-
let v9_parser = V9Parser::try_new(self.v9_config)?;
826-
let ipfix_parser = IPFixParser::try_new(self.ipfix_config)?;
888+
let mut v9_config = self.v9_config;
889+
let mut ipfix_config = self.ipfix_config;
890+
v9_config.template_store = self.template_store.clone();
891+
v9_config.template_store_scope = Arc::clone(&self.template_store_scope);
892+
ipfix_config.template_store = self.template_store;
893+
ipfix_config.template_store_scope = self.template_store_scope;
894+
let v9_parser = V9Parser::try_new(v9_config)?;
895+
let ipfix_parser = IPFixParser::try_new(ipfix_config)?;
827896

828897
Ok(NetflowParser {
829898
v9_parser,
@@ -1092,6 +1161,21 @@ impl NetflowParser {
10921161
NetflowParserBuilder::default()
10931162
}
10941163

1164+
/// Override the scope string used for [`TemplateStore`] reads/writes by
1165+
/// the underlying V9 and IPFIX parsers.
1166+
///
1167+
/// Used by [`AutoScopedParser`] to give each per-source parser a scope
1168+
/// derived from the exporter's `SocketAddr`. Callers managing their own
1169+
/// per-source parsers may also use this to retrofit a scope after the
1170+
/// builder has produced a parser.
1171+
///
1172+
/// Accepts anything `Into<Arc<str>>` — typically `&str` or `String`.
1173+
pub fn set_template_store_scope(&mut self, scope: impl Into<Arc<str>>) {
1174+
let scope: Arc<str> = scope.into();
1175+
self.v9_parser.set_template_store_scope(Arc::clone(&scope));
1176+
self.ipfix_parser.set_template_store_scope(scope);
1177+
}
1178+
10951179
/// Returns the allowed versions array.
10961180
///
10971181
/// Indexed by version number: index 5 = V5, 7 = V7, 9 = V9, 10 = IPFIX.
@@ -1340,6 +1424,26 @@ impl NetflowParser {
13401424
/// parser.clear_v9_templates();
13411425
/// ```
13421426
pub fn clear_v9_templates(&mut self) {
1427+
// Mirror to the secondary store first (best-effort) so that
1428+
// subsequent reads do not transparently repopulate the in-process
1429+
// cache via read-through.
1430+
if let Some(store) = self.v9_parser.template_store.clone() {
1431+
let scope = self.v9_parser.template_store_scope.clone();
1432+
for (id, _) in self.v9_parser.templates.iter() {
1433+
let _ = store.remove(&template_store::TemplateStoreKey::new(
1434+
scope.clone(),
1435+
template_store::TemplateKind::V9Data,
1436+
*id,
1437+
));
1438+
}
1439+
for (id, _) in self.v9_parser.options_templates.iter() {
1440+
let _ = store.remove(&template_store::TemplateStoreKey::new(
1441+
scope.clone(),
1442+
template_store::TemplateKind::V9Options,
1443+
*id,
1444+
));
1445+
}
1446+
}
13431447
self.v9_parser.templates.clear();
13441448
self.v9_parser.options_templates.clear();
13451449
self.v9_parser.clear_pending_flows();
@@ -1358,6 +1462,37 @@ impl NetflowParser {
13581462
/// parser.clear_ipfix_templates();
13591463
/// ```
13601464
pub fn clear_ipfix_templates(&mut self) {
1465+
if let Some(store) = self.ipfix_parser.template_store.clone() {
1466+
let scope = self.ipfix_parser.template_store_scope.clone();
1467+
for (id, _) in self.ipfix_parser.templates.iter() {
1468+
let _ = store.remove(&template_store::TemplateStoreKey::new(
1469+
scope.clone(),
1470+
template_store::TemplateKind::IpfixData,
1471+
*id,
1472+
));
1473+
}
1474+
for (id, _) in self.ipfix_parser.ipfix_options_templates.iter() {
1475+
let _ = store.remove(&template_store::TemplateStoreKey::new(
1476+
scope.clone(),
1477+
template_store::TemplateKind::IpfixOptions,
1478+
*id,
1479+
));
1480+
}
1481+
for (id, _) in self.ipfix_parser.v9_templates.iter() {
1482+
let _ = store.remove(&template_store::TemplateStoreKey::new(
1483+
scope.clone(),
1484+
template_store::TemplateKind::IpfixV9Data,
1485+
*id,
1486+
));
1487+
}
1488+
for (id, _) in self.ipfix_parser.v9_options_templates.iter() {
1489+
let _ = store.remove(&template_store::TemplateStoreKey::new(
1490+
scope.clone(),
1491+
template_store::TemplateKind::IpfixV9Options,
1492+
*id,
1493+
));
1494+
}
1495+
}
13611496
self.ipfix_parser.templates.clear();
13621497
self.ipfix_parser.v9_templates.clear();
13631498
self.ipfix_parser.ipfix_options_templates.clear();
@@ -1581,6 +1716,22 @@ impl NetflowParser {
15811716
if let ParsedNetflow::Success { ref packet, .. } = result {
15821717
self.fire_template_events(packet);
15831718
}
1719+
// Fire Restored events for templates pulled in from the secondary
1720+
// store during this parse. Drained from each parser whether the
1721+
// outer result was Success or Error — restoration that happened
1722+
// before a later flowset failed should still be reported.
1723+
for (protocol, template_id) in self.v9_parser.drain_restored_templates() {
1724+
self.template_hooks.trigger(&TemplateEvent::Restored {
1725+
template_id: Some(template_id),
1726+
protocol,
1727+
});
1728+
}
1729+
for (protocol, template_id) in self.ipfix_parser.drain_restored_templates() {
1730+
self.template_hooks.trigger(&TemplateEvent::Restored {
1731+
template_id: Some(template_id),
1732+
protocol,
1733+
});
1734+
}
15841735
// Fire metric-based events for collisions, evictions, and expirations.
15851736
// Copy the after-metrics to avoid borrowing self immutably while
15861737
// fire_metric_delta_events borrows self mutably (for hook_errors).
@@ -1592,6 +1743,11 @@ impl NetflowParser {
15921743
let after = self.ipfix_parser.metrics;
15931744
self.fire_metric_delta_events(&before, &after, TemplateProtocol::Ipfix);
15941745
}
1746+
} else {
1747+
// Even when no hooks are registered, drain the buffers so they
1748+
// do not accumulate across parse_bytes calls.
1749+
let _ = self.v9_parser.drain_restored_templates();
1750+
let _ = self.ipfix_parser.drain_restored_templates();
15951751
}
15961752

15971753
result

0 commit comments

Comments
 (0)