1use crate::registry::SliceSet;
22use zenkey::grammar::{self, BlobTier, Class, ClassOrPlane, Origin, Plane, StructuralKey};
23
24#[derive(Debug, Clone, PartialEq, Eq)]
26pub struct KeyFacts {
27 pub shape: KeyShape,
28 pub registration: Registration,
29}
30
31#[derive(Debug, Clone, PartialEq, Eq)]
33pub enum KeyShape {
34 V1(Box<V1Facts>),
36 NotUnderBase,
46 Unparsed { reason: String },
49}
50
51#[derive(Debug, Clone, PartialEq, Eq)]
53pub struct V1Facts {
54 pub origin: String,
56 pub origin_kind: OriginKind,
57 pub class: String,
59 pub class_kind: ClassKind,
60 pub producer: Option<String>,
63 pub instance: Option<u32>,
64 pub blob_tier: Option<String>,
66 pub subject: Vec<String>,
68}
69
70#[derive(Debug, Clone, Copy, PartialEq, Eq)]
74pub enum OriginKind {
75 Host,
76 Service,
77}
78
79#[derive(Debug, Clone, Copy, PartialEq, Eq)]
80pub enum ClassKind {
81 Telemetry,
82 State,
83 Events,
84 Rpc,
85 Media,
86 Blob,
87}
88
89#[derive(Debug, Clone, PartialEq, Eq)]
97pub enum Registration {
98 Unknown,
100 NoSliceForProducer,
102 Unregistered,
106 Registered(Box<SubjectFacts>),
107 NotApplicable,
110}
111
112#[derive(Debug, Clone, PartialEq, Eq)]
114pub struct SubjectFacts {
115 pub path: String,
117 pub type_name: String,
118 pub vars: Vec<(String, String)>,
120 pub unit: Option<String>,
121 pub qos: Option<String>,
122 pub encoding: Option<String>,
123 pub ttl_s: Option<i64>,
124}
125
126impl KeyFacts {
127 pub fn project(base: &str, wire_key: &str) -> KeyFacts {
134 let Some(relative) = grammar::strip_base(base, wire_key) else {
135 return KeyFacts {
136 shape: KeyShape::NotUnderBase,
137 registration: Registration::NotApplicable,
138 };
139 };
140 match grammar::parse(relative) {
141 Ok(parsed) => {
142 let facts = V1Facts::from_parsed(&parsed);
143 let registration = if facts.class_kind.is_data_class() {
144 Registration::Unknown
145 } else {
146 Registration::NotApplicable
148 };
149 KeyFacts {
150 shape: KeyShape::V1(Box::new(facts)),
151 registration,
152 }
153 }
154 Err(e) => KeyFacts {
155 shape: KeyShape::Unparsed {
156 reason: e.to_string(),
157 },
158 registration: Registration::NotApplicable,
159 },
160 }
161 }
162
163 pub fn resolve(&mut self, slices: &SliceSet) {
170 let KeyShape::V1(facts) = &self.shape else {
171 return;
172 };
173 if !facts.class_kind.is_data_class() {
174 return;
175 }
176 let producer = match facts.origin_kind {
179 OriginKind::Host => facts.producer.clone(),
180 OriginKind::Service => slices
181 .by_service_origin(&facts.origin)
182 .map(|s| s.name.clone()),
183 };
184 let Some(producer) = producer else {
185 self.registration = Registration::NoSliceForProducer;
186 return;
187 };
188 if slices.get(&producer).is_none() {
189 self.registration = Registration::NoSliceForProducer;
190 return;
191 }
192 let tail: Vec<&str> = facts.subject.iter().map(String::as_str).collect();
193 self.registration = match slices.refine(&producer, &facts.class, &tail) {
194 Some((decl, vars)) => Registration::Registered(Box::new(SubjectFacts {
195 path: decl.path.clone(),
196 type_name: decl.type_name.clone(),
197 vars,
198 unit: decl.unit.clone(),
199 qos: decl.qos.clone(),
200 encoding: decl.encoding.clone(),
201 ttl_s: decl.ttl_s,
202 })),
203 None => Registration::Unregistered,
204 };
205 }
206
207 pub fn type_name(&self) -> Option<&str> {
210 match &self.registration {
211 Registration::Registered(s) => Some(&s.type_name),
212 _ => None,
213 }
214 }
215}
216
217const EVICT_FRACTION: usize = 16;
221
222struct Entry {
223 facts: KeyFacts,
224 seen: u64,
228}
229
230impl std::fmt::Debug for Entry {
231 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
232 f.debug_struct("Entry").field("seen", &self.seen).finish()
233 }
234}
235
236#[derive(Debug)]
264pub struct FactsCache {
265 entries: std::collections::HashMap<String, Entry>,
266 max_keys: usize,
267 inserted: u64,
268 evicted: u64,
269 seq: u64,
270}
271
272impl Default for FactsCache {
273 fn default() -> Self {
274 FactsCache::with_capacity(crate::stats::DEFAULT_MAX_KEYS)
275 }
276}
277
278impl FactsCache {
279 pub fn with_capacity(max_keys: usize) -> FactsCache {
283 FactsCache {
284 entries: std::collections::HashMap::new(),
285 max_keys: max_keys.max(1),
286 inserted: 0,
287 evicted: 0,
288 seq: 0,
289 }
290 }
291
292 pub fn ensure(&mut self, base: &str, key: &str, slices: Option<&SliceSet>) {
296 self.seq += 1;
297 let seq = self.seq;
298 if let Some(entry) = self.entries.get_mut(key) {
299 entry.seen = seq;
300 return;
301 }
302 if self.entries.len() >= self.max_keys {
303 self.evict();
304 }
305 let mut facts = KeyFacts::project(base, key);
306 if let Some(slices) = slices {
307 facts.resolve(slices);
308 }
309 self.entries
310 .insert(key.to_string(), Entry { facts, seen: seq });
311 self.inserted += 1;
312 }
313
314 pub fn get(&self, key: &str) -> Option<&KeyFacts> {
317 self.entries.get(key).map(|e| &e.facts)
318 }
319
320 pub fn keys(&self) -> impl Iterator<Item = &str> {
321 self.entries.keys().map(String::as_str)
322 }
323
324 pub fn len(&self) -> usize {
325 self.entries.len()
326 }
327
328 pub fn is_empty(&self) -> bool {
329 self.entries.is_empty()
330 }
331
332 pub fn max_keys(&self) -> usize {
333 self.max_keys
334 }
335
336 pub fn evicted(&self) -> u64 {
341 self.evicted
342 }
343
344 pub fn inserted(&self) -> u64 {
353 self.inserted
354 }
355
356 pub fn resolve_all(&mut self, slices: &SliceSet) {
359 for entry in self.entries.values_mut() {
360 entry.facts.resolve(slices);
361 }
362 }
363
364 pub fn clear(&mut self) {
367 self.entries.clear();
368 self.inserted = 0;
369 self.evicted = 0;
370 self.seq = 0;
371 }
372
373 fn evict(&mut self) {
375 let target = self.max_keys - (self.max_keys / EVICT_FRACTION).max(1);
376 let mut seen: Vec<(u64, String)> = self
377 .entries
378 .iter()
379 .map(|(k, e)| (e.seen, k.clone()))
380 .collect();
381 seen.sort_unstable_by_key(|(seen, _)| *seen);
382 for (_, key) in seen.into_iter().take(self.entries.len() - target) {
383 self.entries.remove(&key);
384 self.evicted += 1;
385 }
386 }
387}
388
389impl V1Facts {
390 fn from_parsed(parsed: &StructuralKey<'_>) -> V1Facts {
391 let (origin, origin_kind) = match &parsed.origin {
392 Origin::Host(id) => (id.as_str().to_string(), OriginKind::Host),
393 Origin::Service(s) => (s.clone(), OriginKind::Service),
394 };
395 let (class, class_kind) = match parsed.class {
396 ClassOrPlane::Class(c) => (c.chunk().to_string(), ClassKind::from_class(c)),
397 ClassOrPlane::Plane(p) => (p.chunk().to_string(), ClassKind::from_plane(p)),
398 };
399 V1Facts {
400 origin,
401 origin_kind,
402 class,
403 class_kind,
404 producer: parsed.producer.as_ref().map(|p| p.name().to_string()),
405 instance: parsed.producer.as_ref().and_then(|p| p.instance()),
406 blob_tier: parsed.blob_tier.map(|t| tier_chunk(t).to_string()),
407 subject: parsed.subject.iter().map(|s| (*s).to_string()).collect(),
408 }
409 }
410}
411
412fn tier_chunk(tier: BlobTier) -> &'static str {
413 tier.chunk()
414}
415
416impl ClassKind {
417 fn from_class(c: Class) -> ClassKind {
418 match c {
419 Class::Telemetry => ClassKind::Telemetry,
420 Class::State => ClassKind::State,
421 Class::Events => ClassKind::Events,
422 }
423 }
424
425 fn from_plane(p: Plane) -> ClassKind {
426 match p {
427 Plane::Rpc => ClassKind::Rpc,
428 Plane::Media => ClassKind::Media,
429 Plane::Blob => ClassKind::Blob,
430 }
431 }
432
433 pub fn is_data_class(self) -> bool {
436 matches!(
437 self,
438 ClassKind::Telemetry | ClassKind::State | ClassKind::Events
439 )
440 }
441}
442
443#[derive(Debug, Clone, PartialEq, Eq)]
451pub struct KeyDescription {
452 pub key: String,
454 pub facts: KeyFacts,
455}
456
457pub fn describe_key(base: &str, key: &str, slices: Option<&SliceSet>) -> KeyDescription {
463 let mut facts = KeyFacts::project(base, key);
464 if let Some(slices) = slices {
465 facts.resolve(slices);
466 }
467 KeyDescription {
468 key: key.to_string(),
469 facts,
470 }
471}
472
473#[cfg(test)]
474mod tests {
475 use super::*;
476
477 fn v1(facts: &KeyFacts) -> &V1Facts {
478 match &facts.shape {
479 KeyShape::V1(f) => f,
480 other => panic!("expected a v1 key, got {other:?}"),
481 }
482 }
483
484 #[test]
485 fn projects_a_host_telemetry_key() {
486 let f = KeyFacts::project(
487 "zensight",
488 "zensight/v1/h-3fa9c2d41b7e/telemetry/sysinfo/cpu/usage",
489 );
490 let v = v1(&f);
491 assert_eq!(v.origin, "h-3fa9c2d41b7e");
492 assert_eq!(v.origin_kind, OriginKind::Host);
493 assert_eq!(v.class, "telemetry");
494 assert_eq!(v.producer.as_deref(), Some("sysinfo"));
495 assert_eq!(v.instance, None);
496 assert_eq!(v.subject, ["cpu", "usage"]);
497 assert_eq!(f.registration, Registration::Unknown);
499 }
500
501 #[test]
505 fn positions_are_base_relative_never_absolute() {
506 let subject = "v1/h-3fa9c2d41b7e/telemetry/sysinfo/cpu/usage";
507 let cases = [
508 ("", subject.to_string()),
509 ("zensight", format!("zensight/{subject}")),
510 ("acme/fleet-a", format!("acme/fleet-a/{subject}")),
511 ];
512 let projected: Vec<V1Facts> = cases
513 .iter()
514 .map(|(base, key)| v1(&KeyFacts::project(base, key)).clone())
515 .collect();
516 assert_eq!(projected[0], projected[1]);
517 assert_eq!(projected[1], projected[2]);
518 assert_eq!(projected[0].producer.as_deref(), Some("sysinfo"));
519 }
520
521 #[test]
524 fn origin_chunk_alone_decides_whether_chunk_five_is_a_producer() {
525 let host = KeyFacts::project("", "v1/h-3fa9c2d41b7e/state/sysinfo/health");
526 assert_eq!(v1(&host).producer.as_deref(), Some("sysinfo"));
527 assert_eq!(v1(&host).subject, ["health"]);
528
529 let service = KeyFacts::project("", "v1/@catalog/state/entity/x");
530 assert_eq!(v1(&service).origin_kind, OriginKind::Service);
531 assert_eq!(v1(&service).origin, "@catalog");
532 assert_eq!(v1(&service).producer, None);
533 assert_eq!(v1(&service).subject, ["entity", "x"]);
535 }
536
537 #[test]
538 fn parses_a_producer_instance_suffix() {
539 let f = KeyFacts::project("", "v1/h-3fa9c2d41b7e/telemetry/snmp-2/if/eth0/in");
540 assert_eq!(v1(&f).producer.as_deref(), Some("snmp"));
541 assert_eq!(v1(&f).instance, Some(2));
542 }
543
544 #[test]
546 fn blob_tier_occupies_the_producer_position() {
547 let f = KeyFacts::project("", "v1/h-3fa9c2d41b7e/@blob/store/sha256/abcdef01");
548 let v = v1(&f);
549 assert_eq!(v.class_kind, ClassKind::Blob);
550 assert_eq!(v.producer, None);
551 assert_eq!(v.blob_tier.as_deref(), Some("store"));
552 assert_eq!(f.registration, Registration::NotApplicable);
554 }
555
556 #[test]
557 fn a_key_under_another_base_is_a_fact_not_an_error() {
558 let f = KeyFacts::project("zensight", "other/v1/h-3fa9c2d41b7e/state/sysinfo/health");
559 assert_eq!(f.shape, KeyShape::NotUnderBase);
560 }
563
564 #[test]
567 fn empty_base_makes_not_under_base_unreachable() {
568 for key in [
569 "v1/h-3fa9c2d41b7e/state/sysinfo/health",
570 "zensight/v1/h-3fa9c2d41b7e/state/sysinfo/health",
571 "demo/example/foo",
572 "",
573 ] {
574 assert_ne!(
575 KeyFacts::project("", key).shape,
576 KeyShape::NotUnderBase,
577 "{key}"
578 );
579 }
580 }
581
582 #[test]
585 fn arbitrary_keys_degrade_to_a_stated_reason() {
586 for key in ["demo/example/foo", "v2/h-3fa9c2d41b7e/state/x/y", "a", ""] {
587 let f = KeyFacts::project("", key);
588 match f.shape {
589 KeyShape::Unparsed { reason } => assert!(!reason.is_empty(), "{key}"),
590 other => panic!("{key} should be unparsed, got {other:?}"),
591 }
592 assert_eq!(f.registration, Registration::NotApplicable);
593 }
594 }
595
596 #[test]
599 fn foreign_keys_with_verbatim_chunks_are_merely_unparsed() {
600 let f = KeyFacts::project("", "demo/@thing/foo");
601 assert!(matches!(f.shape, KeyShape::Unparsed { .. }));
602 }
603
604 #[test]
605 fn unknown_registration_is_not_unregistered() {
606 assert_ne!(Registration::Unknown, Registration::Unregistered);
608 }
609
610 #[test]
615 fn describe_key_prefers_the_literal_over_the_variable() {
616 use zenkey::slice::{RegistrySlice, SubjectDecl};
617 let subject = |path: &str| SubjectDecl {
618 path: path.to_string(),
619 class: "telemetry".to_string(),
620 type_name: if path.contains('{') {
621 "VarPoint"
622 } else {
623 "SpecialPoint"
624 }
625 .to_string(),
626 common: None,
627 since: None,
628 description: None,
629 qos: None,
630 ttl_s: None,
631 unit: None,
632 rate: None,
633 cardinality: None,
634 encoding: None,
635 };
636 let slice = RegistrySlice {
637 version: "1.0".into(),
638 app: "test".into(),
639 convention: 1,
640 name: "flowd".into(),
641 service_origin: None,
642 description: None,
643 subjects: vec![subject("flow/{q}"), subject("flow/special")],
645 procedures: vec![],
646 blob: vec![],
647 media: vec![],
648 deprecated: vec![],
649 };
650 let slices = SliceSet::from_slices(vec![slice]);
651 let d = describe_key(
652 "",
653 "v1/h-3fa9c2d41b7e/telemetry/flowd/flow/special",
654 Some(&slices),
655 );
656 match &d.facts.registration {
657 Registration::Registered(s) => {
658 assert_eq!(s.path, "flow/special", "literal must beat {{var}}");
659 assert_eq!(s.type_name, "SpecialPoint");
660 }
661 other => panic!("expected Registered, got {other:?}"),
662 }
663 let d = describe_key(
665 "",
666 "v1/h-3fa9c2d41b7e/telemetry/flowd/flow/p95",
667 Some(&slices),
668 );
669 match &d.facts.registration {
670 Registration::Registered(s) => assert_eq!(s.path, "flow/{q}"),
671 other => panic!("expected Registered, got {other:?}"),
672 }
673 }
674
675 #[test]
677 fn describe_key_never_fails() {
678 for key in ["demo/example/foo", "", "v2/x", "@weird/key"] {
679 let d = describe_key("", key, None);
680 assert_eq!(d.key, key);
681 assert!(matches!(d.facts.shape, KeyShape::Unparsed { .. }), "{key}");
682 }
683 let d = describe_key("zensight", "other/v1/h-3fa9c2d41b7e/state/x/y", None);
684 assert_eq!(d.facts.shape, KeyShape::NotUnderBase);
685 }
686}
687
688#[cfg(test)]
691mod cache_tests {
692 use super::*;
693
694 fn key(i: usize) -> String {
695 format!("v1/h-3fa9c2d41b7e/telemetry/sysinfo/k{i}")
696 }
697
698 #[test]
699 fn the_bound_holds_and_every_drop_is_counted() {
700 let mut cache = FactsCache::with_capacity(100);
701 for i in 0..1_000 {
702 cache.ensure("", &key(i), None);
703 }
704 assert!(cache.len() <= 100, "held {}", cache.len());
705 assert!(cache.evicted() > 0, "the fixture must trip the bound");
706 assert_eq!(cache.inserted(), 1_000, "every key here was distinct");
708 assert_eq!(cache.len() as u64 + cache.evicted(), cache.inserted());
709 }
710
711 #[test]
715 fn a_re_observed_eviction_is_projected_again() {
716 let mut cache = FactsCache::with_capacity(2);
717 for i in 0..10 {
718 cache.ensure("", &key(i), None);
719 }
720 let after_first_pass = cache.inserted();
721 for i in 0..10 {
722 cache.ensure("", &key(i), None);
723 }
724 assert!(
725 cache.inserted() > after_first_pass,
726 "a second pass over evicted keys re-projects them"
727 );
728 assert_eq!(cache.len() as u64 + cache.evicted(), cache.inserted());
729 }
730
731 #[test]
732 fn the_least_recently_observed_is_the_one_that_goes() {
733 let mut cache = FactsCache::with_capacity(4);
734 for i in 0..4 {
735 cache.ensure("", &key(i), None);
736 }
737 cache.ensure("", &key(0), None);
740 cache.ensure("", &key(99), None);
741 assert!(cache.get(&key(0)).is_some(), "the re-observed key survives");
742 assert!(cache.get(&key(1)).is_none(), "the oldest went instead");
743 }
744
745 #[test]
746 fn ensure_is_idempotent_and_does_not_reproject() {
747 let slices = SliceSet::default();
748 let mut cache = FactsCache::with_capacity(10);
749 cache.ensure("", &key(0), None);
750 let before = cache.get(&key(0)).cloned();
751 cache.ensure("", &key(0), Some(&slices));
752 assert_eq!(
753 cache.get(&key(0)).cloned(),
754 before,
755 "a second ensure must not re-resolve behind the caller's back"
756 );
757 assert_eq!(cache.len(), 1);
758 }
759
760 #[test]
761 fn resolve_all_reaches_entries_projected_before_the_registry_arrived() {
762 let mut cache = FactsCache::with_capacity(10);
764 cache.ensure("", &key(0), None);
765 assert_eq!(
766 cache.get(&key(0)).map(|f| f.registration.clone()),
767 Some(Registration::Unknown)
768 );
769 cache.resolve_all(&SliceSet::default());
770 assert_ne!(
771 cache.get(&key(0)).map(|f| f.registration.clone()),
772 Some(Registration::Unknown),
773 "a registry that arrives late still reaches what was already cached"
774 );
775 }
776
777 #[test]
778 fn clearing_keeps_the_bound_and_forgets_the_count() {
779 let mut cache = FactsCache::with_capacity(4);
780 for i in 0..40 {
781 cache.ensure("", &key(i), None);
782 }
783 assert!(cache.evicted() > 0);
784 cache.clear();
785 assert!(cache.is_empty());
786 assert_eq!(cache.max_keys(), 4, "the bound is a setting, not a state");
787 assert_eq!(
788 cache.evicted(),
789 0,
790 "retirements under another deployment are not this one's"
791 );
792 assert_eq!(cache.inserted(), 0);
793 }
794
795 #[test]
796 fn a_degenerate_bound_is_still_a_bound() {
797 let mut cache = FactsCache::with_capacity(0);
798 for i in 0..10 {
799 cache.ensure("", &key(i), None);
800 }
801 assert_eq!(cache.max_keys(), 1);
802 assert!(cache.len() <= 1);
803 }
804}