1use std::{num::NonZeroUsize, sync::Arc, time::Duration};
2
3use saluki_common::{collections::PrehashedHashSet, hash::NoopU64BuildHasher, sync::shutdown::ShutdownHandle};
4use saluki_error::{generic_error, GenericError};
5use saluki_metrics::{static_metrics, Counter, Gauge};
6use stringtheory::{
7 interning::{GenericMapInterner, Interner as _},
8 CheapMetaString, MetaString,
9};
10use tokio::{pin, select, time::sleep};
11use tracing::debug;
12
13use super::{
14 hash::{hash_context_with_host_and_seen, ContextKey, TagSetKey},
15 Context, ContextInner,
16};
17use crate::{
18 cache::{weight::ItemCountWeighter, Cache, CacheBuilder},
19 data_model::{
20 origin::{OriginTagsResolver, RawOrigin},
21 tags::{SharedTagSet, TagSet},
22 },
23 runtime::{self, ScopeGuard},
24};
25
26const DEFAULT_CONTEXT_RESOLVER_CACHED_CONTEXTS_LIMIT: NonZeroUsize = NonZeroUsize::new(500_000).unwrap();
28
29const DEFAULT_CONTEXT_RESOLVER_INTERNER_CAPACITY_BYTES: NonZeroUsize = NonZeroUsize::new(2 * 1024 * 1024).unwrap();
31
32const SEEN_HASHSET_INITIAL_CAPACITY: usize = 128;
33
34type ContextCache = Cache<ContextKey, Context, ItemCountWeighter, NoopU64BuildHasher>;
35type TagSetCache = Cache<TagSetKey, SharedTagSet, ItemCountWeighter, NoopU64BuildHasher>;
36
37#[static_metrics(prefix = context_resolver, labels(resolver_id))]
38#[derive(Clone)]
39struct Telemetry {
40 interner_capacity_bytes: Gauge,
41 interner_len_bytes: Gauge,
42 interner_entries: Gauge,
43 #[metric(level = debug)]
44 intern_fallback_total: Counter,
45 #[metric(level = debug)]
46 resolved_existing_context_total: Counter,
47 #[metric(level = debug)]
48 resolved_new_context_total: Counter,
49 active_contexts: Gauge,
50 #[metric(level = debug)]
51 resolved_existing_tagset_total: Counter,
52 #[metric(level = debug)]
53 resolved_new_tagset_total: Counter,
54}
55
56pub struct ContextResolverBuilder {
62 name: String,
63 caching_enabled: bool,
64 cached_contexts_limit: Option<NonZeroUsize>,
65 idle_context_expiration: Option<Duration>,
66 interner_capacity_bytes: Option<NonZeroUsize>,
67 allow_heap_allocations: Option<bool>,
68 tags_resolver: Option<TagsResolver>,
69 interner: Option<GenericMapInterner>,
70 origin_tags_resolver: Option<Arc<dyn OriginTagsResolver>>,
71 telemetry_enabled: bool,
72}
73
74impl ContextResolverBuilder {
75 pub fn from_name<S: Into<String>>(name: S) -> Result<Self, GenericError> {
85 let name = name.into();
86 if name.is_empty() {
87 return Err(generic_error!("resolver name must not be empty"));
88 }
89
90 Ok(Self {
91 name,
92 caching_enabled: true,
93 cached_contexts_limit: None,
94 idle_context_expiration: None,
95 interner_capacity_bytes: None,
96 allow_heap_allocations: None,
97 tags_resolver: None,
98 interner: None,
99 origin_tags_resolver: None,
100 telemetry_enabled: true,
101 })
102 }
103
104 pub fn without_caching(mut self) -> Self {
119 self.caching_enabled = false;
120 self.idle_context_expiration = None;
121 self
122 }
123
124 pub fn with_cached_contexts_limit(mut self, limit: usize) -> Self {
139 match NonZeroUsize::new(limit) {
140 Some(limit) => {
141 self.cached_contexts_limit = Some(limit);
142 self
143 }
144 None => self.without_caching(),
145 }
146 }
147
148 pub fn with_idle_context_expiration(mut self, time_to_idle: Duration) -> Self {
157 self.idle_context_expiration = Some(time_to_idle);
158 self
159 }
160
161 pub fn with_interner_capacity_bytes(mut self, capacity: NonZeroUsize) -> Self {
177 self.interner_capacity_bytes = Some(capacity);
178 self
179 }
180
181 pub fn with_heap_allocations(mut self, allow: bool) -> Self {
190 self.allow_heap_allocations = Some(allow);
191 self
192 }
193
194 pub fn with_tags_resolver(mut self, resolver: Option<TagsResolver>) -> Self {
198 self.tags_resolver = resolver;
199 self
200 }
201
202 pub fn without_telemetry(mut self) -> Self {
211 self.telemetry_enabled = false;
212 self
213 }
214
215 pub fn with_interner(mut self, interner: GenericMapInterner) -> Self {
219 self.interner = Some(interner);
220 self
221 }
222
223 pub fn for_tests() -> Self {
236 ContextResolverBuilder::from_name("noop")
237 .expect("resolver name not empty")
238 .with_cached_contexts_limit(usize::MAX)
239 .with_interner_capacity_bytes(NonZeroUsize::new(1).expect("not zero"))
240 .with_heap_allocations(true)
241 .with_tags_resolver(Some(TagsResolverBuilder::for_tests().build()))
242 .without_telemetry()
243 }
244
245 pub fn build(self) -> ContextResolver {
247 let interner_capacity_bytes = self
248 .interner_capacity_bytes
249 .unwrap_or(DEFAULT_CONTEXT_RESOLVER_INTERNER_CAPACITY_BYTES);
250
251 let interner = match self.interner {
252 Some(interner) => interner,
253 None => GenericMapInterner::new(interner_capacity_bytes),
254 };
255
256 let cached_context_limit = self
257 .cached_contexts_limit
258 .unwrap_or(DEFAULT_CONTEXT_RESOLVER_CACHED_CONTEXTS_LIMIT);
259
260 let allow_heap_allocations = self.allow_heap_allocations.unwrap_or(true);
261
262 let telemetry = Telemetry::new(&self.name);
263 telemetry
264 .interner_capacity_bytes()
265 .set(interner.capacity_bytes() as f64);
266
267 let needs_scope =
277 self.telemetry_enabled || self.idle_context_expiration.is_some() || self.tags_resolver.is_none();
278 let mut drivers =
279 needs_scope.then(|| runtime::scope::nested_or_detached(format!("context_resolver_{}", self.name)));
280
281 let build_caches = || {
282 let context_cache = CacheBuilder::from_identifier(format!("{}/contexts", self.name))
285 .expect("cache identifier cannot possibly be empty")
286 .with_capacity(cached_context_limit)
287 .with_time_to_idle(self.idle_context_expiration)
288 .with_hasher::<NoopU64BuildHasher>()
289 .with_telemetry(self.telemetry_enabled)
290 .build();
291
292 let tags_resolver = match self.tags_resolver {
295 Some(tags_resolver) => tags_resolver,
296 None => TagsResolverBuilder::new(format!("{}/tags", self.name), interner.clone())
297 .expect("tags resolver name not empty")
298 .with_cached_tagsets_limit(cached_context_limit.get())
299 .with_idle_tagsets_expiration(self.idle_context_expiration.unwrap_or_default())
300 .with_heap_allocations(allow_heap_allocations)
301 .with_origin_tags_resolver(self.origin_tags_resolver.clone())
302 .build(),
303 };
304
305 (context_cache, tags_resolver)
306 };
307 let (context_cache, tags_resolver) = match &drivers {
308 Some(drivers) => drivers.enter_sync(build_caches),
309 None => build_caches(),
310 };
311
312 if let (Some(drivers), true) = (drivers.as_mut(), self.telemetry_enabled) {
313 let (interner, telemetry) = (interner.clone(), telemetry.clone());
314 drivers.spawn(
315 runtime::worker_with_shutdown("telemetry", move |shutdown| {
316 drive_telemetry(interner, telemetry, shutdown)
317 })
318 .build(),
319 );
320 }
321
322 ContextResolver {
323 telemetry,
324 interner,
325 caching_enabled: self.caching_enabled,
326 context_cache,
327 hash_seen_buffer: PrehashedHashSet::with_capacity_and_hasher(
328 SEEN_HASHSET_INITIAL_CAPACITY,
329 NoopU64BuildHasher,
330 ),
331 allow_heap_allocations,
332 tags_resolver,
333 _drivers: drivers.map(Arc::new),
334 }
335 }
336}
337
338pub struct ContextResolver {
362 telemetry: Telemetry,
363 interner: GenericMapInterner,
364 caching_enabled: bool,
365 context_cache: ContextCache,
366 hash_seen_buffer: PrehashedHashSet<u64>,
367 allow_heap_allocations: bool,
368 tags_resolver: TagsResolver,
369
370 _drivers: Option<Arc<ScopeGuard>>,
375}
376
377impl ContextResolver {
378 fn intern<S>(&self, s: S) -> Option<MetaString>
379 where
380 S: AsRef<str> + CheapMetaString,
381 {
382 s.try_cheap_clone()
385 .or_else(|| self.interner.try_intern(s.as_ref()).map(MetaString::from))
386 .or_else(|| {
387 self.allow_heap_allocations.then(|| {
388 saluki_antithesis::sometimes!(
392 true,
393 "context string interner spilled to the heap (unbounded under default config)"
394 );
395 self.telemetry.intern_fallback_total().increment(1);
396 MetaString::from(s.as_ref())
397 })
398 })
399 }
400
401 fn create_context_key_with_host<N, H, I, I2, T, T2>(
402 &mut self, name: N, host: Option<&H>, tags: I, origin_tags: I2,
403 ) -> (ContextKey, TagSetKey)
404 where
405 N: AsRef<str>,
406 H: AsRef<str>,
407 I: IntoIterator<Item = T>,
408 T: AsRef<str>,
409 I2: IntoIterator<Item = T2>,
410 T2: AsRef<str>,
411 {
412 hash_context_with_host_and_seen(
413 name.as_ref(),
414 host.map(AsRef::as_ref),
415 tags,
416 origin_tags,
417 &mut self.hash_seen_buffer,
418 )
419 }
420
421 fn create_context<N, H>(
422 &self, key: ContextKey, name: N, host: Option<H>, context_tags: SharedTagSet, origin_tags: SharedTagSet,
423 ) -> Option<Context>
424 where
425 N: AsRef<str> + CheapMetaString,
426 H: AsRef<str> + CheapMetaString,
427 {
428 let context_name = self.intern(name)?;
430 let context_host = match host {
431 Some(host) => Some(self.intern(host)?),
432 None => None,
433 };
434
435 self.telemetry.resolved_new_context_total().increment(1);
436 self.telemetry.active_contexts().increment(1);
437
438 Some(Context::from_inner(ContextInner::from_parts(
439 key,
440 context_name,
441 context_host,
442 context_tags.into(),
443 origin_tags.into(),
444 self.telemetry.active_contexts().clone(),
445 )))
446 }
447
448 pub fn resolve<N, I, T>(&mut self, name: N, tags: I, maybe_origin: Option<RawOrigin<'_>>) -> Option<Context>
457 where
458 N: AsRef<str> + CheapMetaString,
459 I: IntoIterator<Item = T> + Clone,
460 T: AsRef<str> + CheapMetaString,
461 {
462 let origin_tags = self.tags_resolver.resolve_origin_tags(maybe_origin);
464
465 self.resolve_inner(name, tags, origin_tags)
466 }
467
468 pub fn resolve_with_origin_tags<N, I, T>(
487 &mut self, name: N, tags: I, origin_tags: impl Into<SharedTagSet>,
488 ) -> Option<Context>
489 where
490 N: AsRef<str> + CheapMetaString,
491 I: IntoIterator<Item = T> + Clone,
492 T: AsRef<str> + CheapMetaString,
493 {
494 self.resolve_inner(name, tags, origin_tags.into())
495 }
496
497 pub fn resolve_with_host_and_origin_tags<N, H, I, T>(
499 &mut self, name: N, host: H, tags: I, origin_tags: impl Into<SharedTagSet>,
500 ) -> Option<Context>
501 where
502 N: AsRef<str> + CheapMetaString,
503 H: AsRef<str> + CheapMetaString,
504 I: IntoIterator<Item = T> + Clone,
505 T: AsRef<str> + CheapMetaString,
506 {
507 self.resolve_inner_with_host(name, Some(host), tags, origin_tags.into())
508 }
509
510 pub fn resolve_with_optional_host_and_origin_tags<N, H, I, T>(
512 &mut self, name: N, host: Option<H>, tags: I, origin_tags: impl Into<SharedTagSet>,
513 ) -> Option<Context>
514 where
515 N: AsRef<str> + CheapMetaString,
516 H: AsRef<str> + CheapMetaString,
517 I: IntoIterator<Item = T> + Clone,
518 T: AsRef<str> + CheapMetaString,
519 {
520 self.resolve_inner_with_host(name, host, tags, origin_tags.into())
521 }
522
523 pub fn resolve_with_host<N, H, I, T>(
527 &mut self, name: N, host: H, tags: I, maybe_origin: Option<RawOrigin<'_>>,
528 ) -> Option<Context>
529 where
530 N: AsRef<str> + CheapMetaString,
531 H: AsRef<str> + CheapMetaString,
532 I: IntoIterator<Item = T> + Clone,
533 T: AsRef<str> + CheapMetaString,
534 {
535 self.resolve_with_optional_host(name, Some(host), tags, maybe_origin)
536 }
537
538 pub fn resolve_with_optional_host<N, H, I, T>(
540 &mut self, name: N, host: Option<H>, tags: I, maybe_origin: Option<RawOrigin<'_>>,
541 ) -> Option<Context>
542 where
543 N: AsRef<str> + CheapMetaString,
544 H: AsRef<str> + CheapMetaString,
545 I: IntoIterator<Item = T> + Clone,
546 T: AsRef<str> + CheapMetaString,
547 {
548 let origin_tags = self.tags_resolver.resolve_origin_tags(maybe_origin);
549
550 self.resolve_inner_with_host(name, host, tags, origin_tags)
551 }
552
553 fn resolve_inner<N, I, T>(&mut self, name: N, tags: I, origin_tags: SharedTagSet) -> Option<Context>
554 where
555 N: AsRef<str> + CheapMetaString,
556 I: IntoIterator<Item = T> + Clone,
557 T: AsRef<str> + CheapMetaString,
558 {
559 self.resolve_inner_with_host(name, None::<&str>, tags, origin_tags)
560 }
561
562 fn resolve_inner_with_host<N, H, I, T>(
563 &mut self, name: N, host: Option<H>, tags: I, origin_tags: SharedTagSet,
564 ) -> Option<Context>
565 where
566 N: AsRef<str> + CheapMetaString,
567 H: AsRef<str> + CheapMetaString,
568 I: IntoIterator<Item = T> + Clone,
569 T: AsRef<str> + CheapMetaString,
570 {
571 let (context_key, tagset_key) =
572 self.create_context_key_with_host(&name, host.as_ref(), tags.clone(), &origin_tags);
573
574 if !self.caching_enabled {
576 let tag_set = self.tags_resolver.create_tag_set(tags).unwrap_or_default();
577
578 let context = self.create_context(context_key, name, host, tag_set, origin_tags)?;
579
580 debug!(?context_key, ?context, "Resolved new non-cached context.");
581 return Some(context);
582 }
583
584 match self.context_cache.get(&context_key) {
585 Some(context) => {
586 self.telemetry.resolved_existing_context_total().increment(1);
587 Some(context)
588 }
589 None => {
590 let tag_set = match self.tags_resolver.get_tag_set(tagset_key) {
592 Some(tag_set) => {
593 self.telemetry.resolved_existing_tagset_total().increment(1);
594 tag_set
595 }
596 None => {
597 let tag_set = self.tags_resolver.create_tag_set(tags.clone()).unwrap_or_default();
599
600 self.tags_resolver.insert_tag_set(tagset_key, tag_set.clone());
601
602 tag_set
603 }
604 };
605
606 let context = self.create_context(context_key, name, host, tag_set, origin_tags)?;
607 self.context_cache.insert(context_key, context.clone());
608
609 debug!(?context_key, ?context, "Resolved new context.");
610 Some(context)
611 }
612 }
613 }
614}
615
616impl Clone for ContextResolver {
617 fn clone(&self) -> Self {
618 Self {
619 telemetry: self.telemetry.clone(),
620 interner: self.interner.clone(),
621 caching_enabled: self.caching_enabled,
622 context_cache: self.context_cache.clone(),
623 hash_seen_buffer: PrehashedHashSet::with_capacity_and_hasher(
624 SEEN_HASHSET_INITIAL_CAPACITY,
625 NoopU64BuildHasher,
626 ),
627 allow_heap_allocations: self.allow_heap_allocations,
628 tags_resolver: self.tags_resolver.clone(),
629 _drivers: self._drivers.clone(),
630 }
631 }
632}
633
634async fn drive_telemetry(interner: GenericMapInterner, telemetry: Telemetry, shutdown: ShutdownHandle) {
635 pin!(shutdown);
636
637 loop {
638 select! {
639 _ = &mut shutdown => break,
640 _ = sleep(Duration::from_secs(1)) => {}
641 }
642
643 telemetry.interner_entries().set(interner.len() as f64);
644 telemetry
645 .interner_capacity_bytes()
646 .set(interner.capacity_bytes() as f64);
647 telemetry.interner_len_bytes().set(interner.len_bytes() as f64);
648 }
649}
650
651pub struct TagsResolverBuilder {
653 name: String,
654 caching_enabled: bool,
655 cached_tagset_limit: Option<NonZeroUsize>,
656 idle_tagset_expiration: Option<Duration>,
657 allow_heap_allocations: Option<bool>,
658 origin_tags_resolver: Option<Arc<dyn OriginTagsResolver>>,
659 telemetry_enabled: bool,
660 interner: GenericMapInterner,
661}
662
663impl TagsResolverBuilder {
664 pub fn new<S: Into<String>>(name: S, interner: GenericMapInterner) -> Result<Self, GenericError> {
666 let name = name.into();
667 if name.is_empty() {
668 return Err(generic_error!("resolver name must not be empty"));
669 }
670
671 Ok(Self {
672 name,
673 caching_enabled: true,
674 cached_tagset_limit: None,
675 idle_tagset_expiration: None,
676 allow_heap_allocations: None,
677 origin_tags_resolver: None,
678 telemetry_enabled: true,
679 interner,
680 })
681 }
682
683 pub fn with_interner(mut self, interner: GenericMapInterner) -> Self {
689 self.interner = interner;
690 self
691 }
692
693 pub fn without_caching(mut self) -> Self {
708 self.caching_enabled = false;
709 self.idle_tagset_expiration = None;
710 self
711 }
712
713 pub fn with_cached_tagsets_limit(mut self, limit: usize) -> Self {
728 match NonZeroUsize::new(limit) {
729 Some(limit) => {
730 self.cached_tagset_limit = Some(limit);
731 self
732 }
733 None => self.without_caching(),
734 }
735 }
736
737 pub fn with_idle_tagsets_expiration(mut self, time_to_idle: Duration) -> Self {
746 self.idle_tagset_expiration = Some(time_to_idle);
747 self
748 }
749
750 pub fn with_heap_allocations(mut self, allow: bool) -> Self {
759 self.allow_heap_allocations = Some(allow);
760 self
761 }
762
763 pub fn with_origin_tags_resolver(mut self, resolver: Option<Arc<dyn OriginTagsResolver>>) -> Self {
776 self.origin_tags_resolver = resolver;
777 self
778 }
779
780 pub fn without_telemetry(mut self) -> Self {
789 self.telemetry_enabled = false;
790 self
791 }
792
793 pub fn build(self) -> TagsResolver {
795 let cached_tagsets_limit = self
796 .cached_tagset_limit
797 .unwrap_or(DEFAULT_CONTEXT_RESOLVER_CACHED_CONTEXTS_LIMIT);
798
799 let allow_heap_allocations = self.allow_heap_allocations.unwrap_or(true);
800
801 let telemetry = Telemetry::new(self.name.clone());
802 telemetry
803 .interner_capacity_bytes()
804 .set(self.interner.capacity_bytes() as f64);
805
806 let tagset_cache = CacheBuilder::from_identifier(format!("{}/tagsets", self.name))
807 .expect("cache identifier cannot possibly be empty")
808 .with_capacity(cached_tagsets_limit)
809 .with_time_to_idle(self.idle_tagset_expiration)
810 .with_hasher::<NoopU64BuildHasher>()
811 .with_telemetry(self.telemetry_enabled)
812 .build();
813
814 TagsResolver {
815 telemetry,
816 interner: self.interner,
817 caching_enabled: self.caching_enabled,
818 tagset_cache,
819 origin_tags_resolver: self.origin_tags_resolver,
820 allow_heap_allocations,
821 }
822 }
823
824 pub fn for_tests() -> Self {
837 TagsResolverBuilder::new("noop", GenericMapInterner::new(NonZeroUsize::new(1).expect("not zero")))
838 .expect("resolver name not empty")
839 .with_cached_tagsets_limit(usize::MAX)
840 .with_heap_allocations(true)
841 .without_telemetry()
842 }
843}
844
845pub struct TagsResolver {
847 telemetry: Telemetry,
848 interner: GenericMapInterner,
849 caching_enabled: bool,
850 tagset_cache: TagSetCache,
851 origin_tags_resolver: Option<Arc<dyn OriginTagsResolver>>,
852 allow_heap_allocations: bool,
853}
854
855impl TagsResolver {
856 fn intern<S>(&self, s: S) -> Option<MetaString>
857 where
858 S: AsRef<str> + CheapMetaString,
859 {
860 s.try_cheap_clone()
863 .or_else(|| self.interner.try_intern(s.as_ref()).map(MetaString::from))
864 .or_else(|| {
865 self.allow_heap_allocations.then(|| {
866 saluki_antithesis::sometimes!(
870 true,
871 "tag string interner spilled to the heap (unbounded under default config)"
872 );
873 self.telemetry.intern_fallback_total().increment(1);
874 MetaString::from(s.as_ref())
875 })
876 })
877 }
878
879 pub fn create_tag_set<I, T>(&mut self, tags: I) -> Option<SharedTagSet>
886 where
887 I: IntoIterator<Item = T>,
888 T: AsRef<str> + CheapMetaString,
889 {
890 let mut tag_set = TagSet::default();
891 for tag in tags {
892 let tag = self.intern(tag)?;
893 tag_set.insert_tag(tag);
894 }
895
896 self.telemetry.resolved_new_tagset_total().increment(1);
897
898 Some(tag_set.into_shared())
899 }
900
901 pub fn resolve_origin_tags(&self, maybe_origin: Option<RawOrigin<'_>>) -> SharedTagSet {
905 self.origin_tags_resolver
906 .as_ref()
907 .and_then(|resolver| maybe_origin.map(|origin| resolver.resolve_origin_tags(origin)))
908 .unwrap_or_default()
909 }
910
911 fn get_tag_set(&self, key: TagSetKey) -> Option<SharedTagSet> {
912 self.tagset_cache.get(&key)
913 }
914
915 fn insert_tag_set(&self, key: TagSetKey, tag_set: SharedTagSet) {
916 self.tagset_cache.insert(key, tag_set);
917 }
918}
919
920impl Clone for TagsResolver {
921 fn clone(&self) -> Self {
922 Self {
923 telemetry: self.telemetry.clone(),
924 interner: self.interner.clone(),
925 caching_enabled: self.caching_enabled,
926 tagset_cache: self.tagset_cache.clone(),
927 origin_tags_resolver: self.origin_tags_resolver.clone(),
928 allow_heap_allocations: self.allow_heap_allocations,
929 }
930 }
931}
932
933#[cfg(test)]
934mod tests {
935 use metrics::{SharedString, Unit};
936 use metrics_util::{
937 debugging::{DebugValue, DebuggingRecorder},
938 CompositeKey,
939 };
940 use saluki_common::hash::hash_single_fast;
941
942 use super::*;
943 use crate::data_model::tags::Tag;
944
945 fn get_gauge_value(metrics: &[(CompositeKey, Option<Unit>, Option<SharedString>, DebugValue)], key: &str) -> f64 {
946 metrics
947 .iter()
948 .find(|(k, _, _, _)| k.key().name() == key)
949 .map(|(_, _, _, value)| match value {
950 DebugValue::Gauge(value) => value.into_inner(),
951 other => panic!("expected a gauge, got: {:?}", other),
952 })
953 .unwrap_or_else(|| panic!("no metric found with key: {}", key))
954 }
955
956 fn find<'a>(node: &'a runtime::NodeSnapshot, name: &str) -> Option<&'a runtime::NodeSnapshot> {
958 if node.name == name {
959 return Some(node);
960 }
961 node.children.iter().find_map(|child| find(child, name))
962 }
963
964 fn child<'a>(node: &'a runtime::NodeSnapshot, name: &str) -> Option<&'a runtime::NodeSnapshot> {
966 node.children.iter().find(|child| child.name == name)
967 }
968
969 fn run_owner(
973 owner: runtime::ChildSpecification<runtime::WorkerSpec>,
974 ) -> (
975 runtime::SupervisionTreeHandle,
976 tokio::sync::oneshot::Sender<()>,
977 tokio::task::JoinHandle<Result<(), runtime::SupervisorError>>,
978 ) {
979 let mut supervisor = runtime::Supervisor::new("resolver-test").expect("valid name");
980 let tree = supervisor.tree_handle();
981 supervisor.add_worker(owner);
982
983 let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
984 let run = tokio::spawn(async move { supervisor.run_with_shutdown(shutdown_rx).await });
985 (tree, shutdown_tx, run)
986 }
987
988 async fn stop(
989 shutdown_tx: tokio::sync::oneshot::Sender<()>,
990 run: tokio::task::JoinHandle<Result<(), runtime::SupervisorError>>,
991 ) {
992 shutdown_tx.send(()).expect("supervisor should be running");
993 let result = tokio::time::timeout(Duration::from_secs(5), run)
994 .await
995 .expect("supervisor should stop")
996 .expect("supervisor task should not panic");
997 assert!(result.is_ok(), "every task should have stopped on its own: {result:?}");
998 }
999
1000 #[tokio::test]
1001 async fn background_tasks_run_beneath_the_resolver_and_stop_with_its_last_copy() {
1002 use crate::test_support::wait_until;
1003
1004 let (drop_first_tx, drop_first_rx) = tokio::sync::oneshot::channel::<()>();
1005 let (drop_last_tx, drop_last_rx) = tokio::sync::oneshot::channel::<()>();
1006
1007 let (tree, shutdown_tx, run) = run_owner(
1008 runtime::worker_with_shutdown("owner", move |shutdown| async move {
1009 let resolver = ContextResolverBuilder::from_name("test")
1011 .expect("valid name")
1012 .with_idle_context_expiration(Duration::from_secs(60))
1013 .build();
1014 let copy = resolver.clone();
1015
1016 let _ = drop_first_rx.await;
1017 drop(resolver);
1018 let _ = drop_last_rx.await;
1019 drop(copy);
1020 shutdown.await;
1021 })
1022 .build(),
1023 );
1024
1025 wait_until("the tasks of the resolver run beneath it", || {
1028 let root = tree.snapshot().root;
1029 find(&root, "context_resolver_test").is_some_and(|resolver| {
1030 child(resolver, "telemetry").is_some()
1031 && child(resolver, "cache_test/contexts").is_some_and(|cache| child(cache, "expiration").is_some())
1032 && child(resolver, "cache_test/tags/tagsets").is_some()
1033 })
1034 })
1035 .await;
1036
1037 drop_first_tx.send(()).expect("owner should be waiting");
1039 tokio::time::sleep(Duration::from_millis(100)).await;
1040 assert!(
1041 find(&tree.snapshot().root, "context_resolver_test").is_some(),
1042 "the scope of the resolver must stay while a copy of the resolver exists"
1043 );
1044
1045 drop_last_tx.send(()).expect("owner should be waiting");
1046 wait_until("dropping the last copy stops the tasks of the resolver", || {
1047 find(&tree.snapshot().root, "context_resolver_test").is_none()
1048 })
1049 .await;
1050
1051 stop(shutdown_tx, run).await;
1052 }
1053
1054 #[tokio::test]
1055 async fn a_supplied_tags_resolver_keeps_its_tasks_where_it_was_built() {
1056 use crate::test_support::wait_until;
1057
1058 let (tree, shutdown_tx, run) = run_owner(
1059 runtime::worker_with_shutdown("owner", |shutdown| async move {
1060 let interner = GenericMapInterner::new(NonZeroUsize::new(1024).unwrap());
1061 let tags_resolver = TagsResolverBuilder::new("shared", interner)
1062 .expect("valid name")
1063 .build();
1064 let _resolver = ContextResolverBuilder::from_name("test")
1065 .expect("valid name")
1066 .with_tags_resolver(Some(tags_resolver))
1067 .build();
1068 shutdown.await;
1069 })
1070 .build(),
1071 );
1072
1073 wait_until("each cache runs below the value that built it", || {
1076 let root = tree.snapshot().root;
1077 let Some(owner) = find(&root, "owner") else {
1078 return false;
1079 };
1080 child(owner, "cache_shared/tagsets").is_some()
1081 && child(owner, "context_resolver_test").is_some_and(|resolver| {
1082 child(resolver, "cache_test/contexts").is_some() && find(resolver, "cache_shared/tagsets").is_none()
1083 })
1084 })
1085 .await;
1086
1087 stop(shutdown_tx, run).await;
1088 }
1089
1090 struct DummyOriginTagsResolver;
1091
1092 impl OriginTagsResolver for DummyOriginTagsResolver {
1093 fn resolve_origin_tags(&self, origin: RawOrigin<'_>) -> SharedTagSet {
1094 let origin_key = hash_single_fast(origin);
1095
1096 let mut tags = TagSet::default();
1097 tags.insert_tag(format!("origin_key:{}", origin_key));
1098 tags.into_shared()
1099 }
1100 }
1101
1102 #[test]
1103 fn basic() {
1104 let mut resolver = ContextResolverBuilder::for_tests().build();
1105
1106 let name = "metric_name";
1108 let tags1: [&str; 0] = [];
1109 let tags2 = ["tag1"];
1110
1111 assert_ne!(&tags1[..], &tags2[..]);
1112
1113 let context1 = resolver
1114 .resolve(name, &tags1[..], None)
1115 .expect("should not fail to resolve");
1116 let context2 = resolver
1117 .resolve(name, &tags2[..], None)
1118 .expect("should not fail to resolve");
1119
1120 assert_ne!(context1, context2);
1123 assert!(!context1.ptr_eq(&context2));
1124
1125 let context1_redo = resolver
1127 .resolve(name, &tags1[..], None)
1128 .expect("should not fail to resolve");
1129 let context2_redo = resolver
1130 .resolve(name, &tags2[..], None)
1131 .expect("should not fail to resolve");
1132
1133 assert_ne!(context1_redo, context2_redo);
1134 assert_eq!(context1, context1_redo);
1135 assert_eq!(context2, context2_redo);
1136 assert!(context1.ptr_eq(&context1_redo));
1137 assert!(context2.ptr_eq(&context2_redo));
1138 }
1139
1140 #[test]
1141 fn tag_order() {
1142 let mut resolver = ContextResolverBuilder::for_tests().build();
1143
1144 let name = "metric_name";
1146 let tags1 = ["tag1", "tag2"];
1147 let tags2 = ["tag2", "tag1"];
1148
1149 assert_ne!(&tags1[..], &tags2[..]);
1150
1151 let context1 = resolver
1152 .resolve(name, &tags1[..], None)
1153 .expect("should not fail to resolve");
1154 let context2 = resolver
1155 .resolve(name, &tags2[..], None)
1156 .expect("should not fail to resolve");
1157
1158 assert_eq!(context1, context2);
1161 assert!(context1.ptr_eq(&context2));
1162 }
1163
1164 #[test]
1165 fn host_affects_identity_but_not_visible_tags() {
1166 let mut resolver = ContextResolverBuilder::for_tests().build();
1167
1168 let context1 = resolver
1169 .resolve_with_host("metric_name", "host-a", &[] as &[&str], None)
1170 .expect("should not fail to resolve");
1171 let context2 = resolver
1172 .resolve_with_host("metric_name", "host-b", &[] as &[&str], None)
1173 .expect("should not fail to resolve");
1174 let context1_redo = resolver
1175 .resolve_with_host("metric_name", "host-a", &[] as &[&str], None)
1176 .expect("should not fail to resolve");
1177
1178 assert_ne!(context1, context2);
1179 assert_eq!(context1, context1_redo);
1180 assert!(context1.ptr_eq(&context1_redo));
1181 assert_eq!(context1.host(), Some("host-a"));
1182 assert_eq!(context2.host(), Some("host-b"));
1183 assert!(context1.tags().is_empty());
1184 assert!(context2.tags().is_empty());
1185
1186 let mut uncached_resolver = ContextResolverBuilder::for_tests().without_caching().build();
1187 let uncached1 = uncached_resolver
1188 .resolve_with_host("metric_name", "host-a", &[] as &[&str], None)
1189 .expect("should not fail to resolve");
1190 let uncached2 = uncached_resolver
1191 .resolve_with_host("metric_name", "host-b", &[] as &[&str], None)
1192 .expect("should not fail to resolve");
1193
1194 assert_ne!(uncached1, uncached2);
1195 assert!(!uncached1.ptr_eq(&uncached2));
1196 }
1197
1198 #[test]
1199 fn host_survives_rewrites() {
1200 let mut resolver = ContextResolverBuilder::for_tests().build();
1201
1202 let context1 = resolver
1203 .resolve_with_host("metric_name", "host-a", &["env:prod"][..], None)
1204 .expect("should not fail to resolve");
1205 let context2 = resolver
1206 .resolve_with_host("metric_name", "host-b", &["env:prod"][..], None)
1207 .expect("should not fail to resolve");
1208
1209 assert_ne!(context1, context2);
1210 let service_tag_set = TagSet::from_iter([Tag::from("service:api")]);
1211 assert_ne!(context1.with_name("renamed"), context2.with_name("renamed"));
1212 assert_ne!(
1213 context1.with_tags(service_tag_set.clone()),
1214 context2.with_tags(service_tag_set)
1215 );
1216
1217 let mut context1_filtered = context1.clone();
1218 let mut context2_filtered = context2.clone();
1219 let mut state1 = crate::data_model::event::metric::context::TagSetMutViewState::new();
1220 let mut state2 = crate::data_model::event::metric::context::TagSetMutViewState::new();
1221 {
1222 let mut view = context1_filtered.tags_mut_view(&mut state1);
1223 view.retain_tags(|_| false);
1224 view.finish();
1225 }
1226 {
1227 let mut view = context2_filtered.tags_mut_view(&mut state2);
1228 view.retain_tags(|_| false);
1229 view.finish();
1230 }
1231
1232 assert_ne!(context1_filtered, context2_filtered);
1233 assert!(context1_filtered.tags().is_empty());
1234 assert!(context2_filtered.tags().is_empty());
1235 assert_eq!(context1_filtered.host(), Some("host-a"));
1236 assert_eq!(context2_filtered.host(), Some("host-b"));
1237 }
1238
1239 #[test]
1240 fn active_contexts() {
1241 let recorder = DebuggingRecorder::new();
1242 let snapshotter = recorder.snapshotter();
1243
1244 let context = metrics::with_local_recorder(&recorder, || {
1246 let mut resolver = ContextResolverBuilder::for_tests().build();
1247 resolver
1248 .resolve("name", &["tag"][..], None)
1249 .expect("should not fail to resolve")
1250 });
1251
1252 let metrics_before = snapshotter.snapshot().into_vec();
1254 let active_contexts = get_gauge_value(&metrics_before, Telemetry::active_contexts_name());
1255 assert_eq!(active_contexts, 1.0);
1256
1257 drop(context);
1259 let metrics_after = snapshotter.snapshot().into_vec();
1260 let active_contexts = get_gauge_value(&metrics_after, Telemetry::active_contexts_name());
1261 assert_eq!(active_contexts, -1.0);
1262 }
1263
1264 #[test]
1265 fn duplicate_tags() {
1266 let mut resolver = ContextResolverBuilder::for_tests().build();
1267
1268 let name = "metric_name";
1270 let tags1 = ["tag1"];
1271 let tags1_duplicated = ["tag1", "tag1"];
1272 let tags2 = ["tag2"];
1273 let tags2_duplicated = ["tag2", "tag2"];
1274
1275 let context1 = resolver
1276 .resolve(name, &tags1[..], None)
1277 .expect("should not fail to resolve");
1278 let context1_duplicated = resolver
1279 .resolve(name, &tags1_duplicated[..], None)
1280 .expect("should not fail to resolve");
1281 let context2 = resolver
1282 .resolve(name, &tags2[..], None)
1283 .expect("should not fail to resolve");
1284 let context2_duplicated = resolver
1285 .resolve(name, &tags2_duplicated[..], None)
1286 .expect("should not fail to resolve");
1287
1288 assert_eq!(context1, context1_duplicated);
1290 assert_eq!(context2, context2_duplicated);
1291
1292 assert_ne!(context1, context2);
1301 assert_ne!(context1_duplicated, context2_duplicated);
1302 assert_ne!(context1, context2_duplicated);
1303 assert_ne!(context2, context1_duplicated);
1304 }
1305
1306 #[test]
1307 fn differing_origins_with_without_resolver() {
1308 let mut resolver = ContextResolverBuilder::for_tests().build();
1311
1312 let name = "metric_name";
1313 let tags = ["tag1"];
1314 let mut origin1 = RawOrigin::default();
1315 origin1.set_local_data("container1");
1316 let mut origin2 = RawOrigin::default();
1317 origin2.set_local_data("container2");
1318
1319 let context1 = resolver
1320 .resolve(name, &tags[..], Some(origin1.clone()))
1321 .expect("should not fail to resolve");
1322 let context2 = resolver
1323 .resolve(name, &tags[..], Some(origin2.clone()))
1324 .expect("should not fail to resolve");
1325
1326 assert_eq!(context1, context2);
1327
1328 let tags_resolver = TagsResolverBuilder::for_tests()
1329 .with_origin_tags_resolver(Some(Arc::new(DummyOriginTagsResolver)))
1330 .build();
1331 let mut resolver = ContextResolverBuilder::for_tests()
1335 .with_tags_resolver(Some(tags_resolver))
1336 .build();
1337
1338 let context1 = resolver
1339 .resolve(name, &tags[..], Some(origin1))
1340 .expect("should not fail to resolve");
1341 let context2 = resolver
1342 .resolve(name, &tags[..], Some(origin2))
1343 .expect("should not fail to resolve");
1344
1345 assert_ne!(context1, context2);
1346 }
1347
1348 #[test]
1349 fn caching_disabled() {
1350 let tags_resolver = TagsResolverBuilder::for_tests()
1351 .with_origin_tags_resolver(Some(Arc::new(DummyOriginTagsResolver)))
1352 .build();
1353 let mut resolver = ContextResolverBuilder::for_tests()
1354 .without_caching()
1355 .with_tags_resolver(Some(tags_resolver))
1356 .build();
1357
1358 let name = "metric_name";
1359 let tags = ["tag1"];
1360 let mut origin1 = RawOrigin::default();
1361 origin1.set_local_data("container1");
1362
1363 let context1 = resolver
1365 .resolve(name, &tags[..], Some(origin1.clone()))
1366 .expect("should not fail to resolve");
1367 assert_eq!(resolver.context_cache.len(), 0);
1368
1369 let context2 = resolver
1371 .resolve(name, &tags[..], Some(origin1))
1372 .expect("should not fail to resolve");
1373 assert_eq!(resolver.context_cache.len(), 0);
1374
1375 assert_eq!(context1, context2);
1378 assert!(!context1.ptr_eq(&context2));
1379 }
1380
1381 #[test]
1382 fn cheaply_cloneable_name_and_tags() {
1383 const BIG_TAG_ONE: &str = "long-tag-that-cannot-be-inlined-just-to-be-doubly-sure-on-top-of-being-static";
1384 const BIG_TAG_TWO: &str = "another-long-boye-that-we-are-also-sure-wont-be-inlined-and-we-stand-on-that";
1385
1386 let mut resolver = ContextResolverBuilder::for_tests()
1388 .with_interner_capacity_bytes(NonZeroUsize::new(1024).expect("not zero"))
1389 .build();
1390
1391 let name = MetaString::from_static("long-metric-name-that-shouldnt-be-inlined-and-should-end-up-interned");
1393 let tags = [
1394 MetaString::from_static(BIG_TAG_ONE),
1395 MetaString::from_static(BIG_TAG_TWO),
1396 ];
1397 assert!(tags[0].is_cheaply_cloneable());
1398 assert!(tags[1].is_cheaply_cloneable());
1399
1400 assert_eq!(resolver.interner.len(), 0);
1403 assert_eq!(resolver.interner.len_bytes(), 0);
1404
1405 let context = resolver
1406 .resolve(&name, &tags[..], None)
1407 .expect("should not fail to resolve");
1408 assert_eq!(resolver.interner.len(), 0);
1409 assert_eq!(resolver.interner.len_bytes(), 0);
1410
1411 assert_eq!(context.name(), &name);
1413
1414 let context_tags = context.tags();
1415 assert_eq!(context_tags.len(), 2);
1416 assert!(context_tags.has_tag(&tags[0]));
1417 assert!(context_tags.has_tag(&tags[1]));
1418 }
1419}