Skip to main content

saluki_core/data_model/event/metric/context/
resolver.rs

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
26// SAFETY: We know, unquestionably, that this value is not zero.
27const DEFAULT_CONTEXT_RESOLVER_CACHED_CONTEXTS_LIMIT: NonZeroUsize = NonZeroUsize::new(500_000).unwrap();
28
29// SAFETY: We know, unquestionably, that this value is not zero.
30const 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
56/// Builder for creating a [`ContextResolver`].
57///
58/// # Missing
59///
60/// - Support for configuring the size limit of cached contexts.
61pub 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    /// Creates a new `ContextResolverBuilder` with the given resolver name.
76    ///
77    /// The resolver name _should_ be unique, but it isn't required to be. Metrics for the resolver will be
78    /// emitted using the given name, so in cases where the name isn't unique, those metrics will be aggregated
79    /// together and it won't be possible to distinguish between the different resolvers.
80    ///
81    /// # Errors
82    ///
83    /// If the given resolver name is empty, an error is returned.
84    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    /// Sets whether or not to enable caching of resolved contexts.
105    ///
106    /// [`ContextResolver`] provides two main benefits: consistent behavior for resolving contexts (interning, origin
107    /// tags, etc), and the caching of those resolved contexts to speed up future resolutions. However, caching contexts
108    /// means that we pay a memory cost for the cache itself, even if the contexts aren't ever reused or are seen
109    /// infrequently. While expiration can help free up cache capacity, it can't help recover the memory used by the
110    /// underlying cache data structure once they have expanded to hold the contexts.
111    ///
112    /// Disabling caching allows normal resolving to take place without the overhead of caching the contexts. This can
113    /// lead to lower average memory usage, as contexts will only live as long as they're needed, but it will reduce
114    /// memory determinism as memory will be allocated for every resolved context (minus interned strings), which means
115    /// that resolving the same context ten times in a row will result in ten separate allocations, and so on.
116    ///
117    /// Defaults to caching enabled.
118    pub fn without_caching(mut self) -> Self {
119        self.caching_enabled = false;
120        self.idle_context_expiration = None;
121        self
122    }
123
124    /// Sets the limit on the number of cached contexts.
125    ///
126    /// This is the maximum number of resolved contexts that can be cached at any given time. This limit doesn't affect
127    /// the total number of contexts that can be _alive_ at any given time, which is dependent on the interner capacity
128    /// and whether or not heap allocations are allowed.
129    ///
130    /// Caching contexts is beneficial when the same context is resolved frequently, and it's generally worth
131    /// allowing for higher limits on cached contexts when heap allocations are allowed, as this can better amortize the
132    /// cost of those heap allocations.
133    ///
134    /// If value is zero, caching will be disabled, and no contexts will be cached. This is equivalent to calling
135    /// `without_caching`.
136    ///
137    /// Defaults to 500,000.
138    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    /// Sets the time before contexts are considered "idle" and eligible for expiration.
149    ///
150    /// This controls how long a context will be kept in the cache after its last access or creation time. This value is
151    /// a lower bound, as contexts eligible for expiration may not be expired immediately. Contexts may still be removed
152    /// prior to their natural expiration time if the cache is full and evictions are required to make room for a new
153    /// context.
154    ///
155    /// Defaults to no expiration.
156    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    /// Sets the capacity of the string interner, in bytes.
162    ///
163    /// This is the maximum number of bytes that the interner will use for interning strings that are present in
164    /// contexts being resolved. This capacity may or may not be allocated entirely when the resolver is built, but the
165    /// interner won't exceed the configured capacity when allocating any backing storage.
166    ///
167    /// This value directly impacts the number of contexts that can be resolved when heap allocations are disabled, as
168    /// all resolved contexts must either have values (name or tags) that can be inlined or interned. Once the interner
169    /// is full, contexts may fail to be resolved if heap allocations are disabled.
170    ///
171    /// The optimal value will almost always be workload-dependent, but a good starting point can be to estimate around
172    /// 150 - 200 bytes per context based on empirical measurements around common metric name and tag lengths. This
173    /// translate to around 5000 unique contexts per 1 MB of interner size.
174    ///
175    /// Defaults to 2 MB.
176    pub fn with_interner_capacity_bytes(mut self, capacity: NonZeroUsize) -> Self {
177        self.interner_capacity_bytes = Some(capacity);
178        self
179    }
180
181    /// Sets whether or not to allow heap allocations when interning strings.
182    ///
183    /// In cases where the interner is full, this setting determines whether or not we refuse to resolve a context, or
184    /// if we allow it be resolved by allocating strings on the heap. When heap allocations are enabled, the amount of
185    /// memory that can be used by the interner is effectively unlimited, as contexts that can't be interned will be
186    /// simply spill to the heap instead of being limited in any way.
187    ///
188    /// Defaults to `true`.
189    pub fn with_heap_allocations(mut self, allow: bool) -> Self {
190        self.allow_heap_allocations = Some(allow);
191        self
192    }
193
194    /// Sets the tags resolver.
195    ///
196    /// Defaults to unset.
197    pub fn with_tags_resolver(mut self, resolver: Option<TagsResolver>) -> Self {
198        self.tags_resolver = resolver;
199        self
200    }
201
202    /// Sets whether or not to enable telemetry for this resolver.
203    ///
204    /// Reporting the telemetry of the resolver requires running an asynchronous task to override adding additional
205    /// overhead in the hot path of resolving contexts. In some cases, it may be cumbersome to always create the
206    /// resolver in an asynchronous context so that the telemetry task can be spawned. This method allows disabling
207    /// telemetry reporting in those cases.
208    ///
209    /// Defaults to telemetry enabled.
210    pub fn without_telemetry(mut self) -> Self {
211        self.telemetry_enabled = false;
212        self
213    }
214
215    /// Sets the interner to use for this resolver.
216    ///
217    /// If an interner isn't provided, an interner will be created in [`ContextResolverBuilder::build`]
218    pub fn with_interner(mut self, interner: GenericMapInterner) -> Self {
219        self.interner = Some(interner);
220        self
221    }
222
223    /// Configures a [`ContextResolverBuilder`] that's suitable for tests.
224    ///
225    /// This configures the builder with the following defaults:
226    ///
227    /// - resolver name of "noop"
228    /// - unlimited cache capacity
229    /// - no-op interner (all strings are heap-allocated)
230    /// - heap allocations allowed
231    /// - telemetry disabled
232    ///
233    /// This is generally only useful for testing purposes, and is exposed publicly in order to be used in cross-crate
234    /// testing scenarios.
235    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    /// Builds a [`ContextResolver`] from the current configuration.
246    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        // The background tasks of this resolver run in a scope that the resolver owns. The caches that the resolver
268        // builds start their own tasks inside that scope. Thus, all of these tasks appear below the resolver, and they
269        // stop when the last copy of the resolver is dropped. A supplied tags resolver can be shared with other
270        // resolvers, so it keeps its tasks where it was built.
271        //
272        // The resolver needs the scope only if something runs in it: the telemetry task of the resolver, or a driver of
273        // a cache that it builds. The context cache has drivers if it expires idle contexts or reports telemetry. A tags
274        // resolver that this resolver builds always expires idle tagsets. If the resolver is built outside supervision,
275        // its tasks run detached.
276        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            // NOTE: We should switch to using a size-based weighter so that we can do more firm bounding of what we
283            // cache.
284            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            // If no tags resolver is provided, we need to create one using the same interner used for the context
293            // resolver.
294            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
338/// A centralized store for resolved contexts.
339///
340/// Contexts are the combination of a name and a set of tags. They're used to identify a specific metric series. As
341/// contexts are constructed entirely of strings, they're expensive to construct in a way that allows sending between
342/// tasks, as this usually requires allocations. Additionally, some context are "hotter" than others, used frequently by
343/// the applications/services sending us metrics.
344///
345/// In order to optimize this, the context resolver is responsible for both interning the strings involved where
346/// possible, as well as keeping a map of contexts that can be referred to with a cheap handle. We can cheaply search
347/// for an existing context without needing to allocate an entirely new one, and get a clone of the handle to use going
348/// forward.
349///
350/// # Design
351///
352/// `ContextResolver` specifically manages interning and mapping of contexts. It can be cheaply cloned itself.
353///
354/// In order to resolve a context, `resolve` must be called which requires taking a lock to check for an existing
355/// context. A read/write lock is used in order to prioritize lookups over inserts, as lookups are expected to be more
356/// common given how often a given context is used and resolved.
357///
358/// Once a context is resolved, a cheap handle -- `Context` -- is returned. This handle, like `ContextResolver`, can be
359/// cheaply cloned. It points directly to the underlying context data (name and tags) and provides access to these
360/// components.
361pub 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    /// The background tasks of the resolver and of the caches that it built, if it has any.
371    ///
372    /// Every copy of the resolver shares this guard. Thus, the tasks stop when the last copy is dropped. They stop
373    /// earlier if the process that owns them stops first.
374    _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        // Try to cheaply clone the string, and if that fails, try to intern it. If that fails, then we fall back to
383        // allocating it on the heap if we allow it.
384        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                    // Heap spill: with `allow_context_heap_allocations` true (the default), a full interner silently
389                    // falls back to the heap, so the bounded-memory guarantee no longer holds. Anchor that this path is
390                    // reached so the unbounded-growth behavior is observable.
391                    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        // Intern the name, host, and tags of the context.
429        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    /// Resolves the given context.
449    ///
450    /// If the context hasn't yet been resolved, the name and tags are interned and a new context is created and
451    /// stored. Otherwise, the existing context is returned. If an origin tags resolver is configured, and origin info
452    /// is available, any enriched tags will be added to the context.
453    ///
454    /// `None` may be returned if the interner is full and outside allocations are disallowed. See
455    /// `allow_heap_allocations` for more information.
456    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        // Try and resolve our origin tags from the provided origin information, if any.
463        let origin_tags = self.tags_resolver.resolve_origin_tags(maybe_origin);
464
465        self.resolve_inner(name, tags, origin_tags)
466    }
467
468    /// Resolves the given context using the provided origin tags.
469    ///
470    /// If the context hasn't yet been resolved, the name and tags are interned and a new context is created and
471    /// stored. Otherwise, the existing context is returned. The provided origin tags are used to enrich the context.
472    ///
473    /// `None` may be returned if the interner is full and outside allocations are disallowed. See
474    /// `allow_heap_allocations` for more information.
475    ///
476    /// ## Origin tags resolver mismatch
477    ///
478    /// When passing in origin tags, they will be inherently tied to a specific `OriginTagsResolver`, which may
479    /// differ from the configured origin tags resolver in this context resolver. This means that the context that's
480    /// generated and cached may not be reused in the future if an attempt is made to resolve it using the raw origin
481    /// information instead.
482    ///
483    /// This method is intended primarily to allow for resolving contexts in a consistent way while _reusing_ the origin
484    /// tags from another context, such as when remapping the name and/or instrumented tags of a given metric, while
485    /// maintaining its origin association.
486    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    /// Resolves the given context using the provided host and origin tags.
498    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    /// Resolves the given context using the provided optional host and origin tags.
511    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    /// Resolves the given context with an explicit host dimension.
524    ///
525    /// The host participates in context identity but is not part of the visible tag set.
526    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    /// Resolves the given context with an optional host dimension.
539    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        // Fast path to avoid looking up the context in the cache if caching is disabled.
575        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                // Try seeing if we have the tagset cached already, and create it if not.
591                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                        // If the tagset is not cached, we need to create it.
598                        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
651/// A builder for a tag resolver.
652pub 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    /// Creates a new [`TagsResolverBuilder`] with the given name and interner.
665    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    /// Sets the interner to use for this resolver.
684    ///
685    /// This is used when we want to use a separate internet for tagsets, different from the one used for contexts.
686    ///
687    /// Defaults to using the interner passed to the builder.
688    pub fn with_interner(mut self, interner: GenericMapInterner) -> Self {
689        self.interner = interner;
690        self
691    }
692
693    /// Sets whether or not to enable caching of resolved tag sets.
694    ///
695    /// [`TagsResolver`] provides two main benefits: consistent behavior for resolving tag sets (interning, origin
696    /// tags, etc), and the caching of those resolved tag sets to speed up future resolutions. However, caching tag
697    /// sets means that we pay a memory cost for the cache itself, even if the tag sets aren't ever reused or are seen
698    /// infrequently. While expiration can help free up cache capacity, it can't help recover the memory used by the
699    /// underlying cache data structure once they have expanded to hold the tag sets.
700    ///
701    /// Disabling caching allows normal resolving to take place without the overhead of caching the tag sets. This can
702    /// lead to lower average memory usage, as tag sets will only live as long as they're needed, but it will reduce
703    /// memory determinism as memory will be allocated for every resolved tag set (minus interned strings), which means
704    /// that resolving the same tag set ten times in a row will result in ten separate allocations, and so on.
705    ///
706    /// Defaults to caching enabled.
707    pub fn without_caching(mut self) -> Self {
708        self.caching_enabled = false;
709        self.idle_tagset_expiration = None;
710        self
711    }
712
713    /// Sets the limit on the number of cached tagsets.
714    ///
715    /// This is the maximum number of resolved tag sets that can be cached at any given time. This limit doesn't affect
716    /// the total number of tag sets that can be _alive_ at any given time, which is dependent on the interner capacity
717    /// and whether or not heap allocations are allowed.
718    ///
719    /// Caching tag sets is beneficial when the same tag set is resolved frequently, and it's generally worth
720    /// allowing for higher limits on cached tag sets when heap allocations are allowed, as this can better amortize the
721    /// cost of those heap allocations.
722    ///
723    /// If value is zero, caching will be disabled, and no tag sets will be cached. This is equivalent to calling
724    /// `without_caching`.
725    ///
726    /// Defaults to 500,000.
727    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    /// Sets the time before tag sets are considered "idle" and eligible for expiration.
738    ///
739    /// This controls how long a tag set will be kept in the cache after its last access or creation time. This value is
740    /// a lower bound, as tag sets eligible for expiration may not be expired immediately. Tag sets may still be removed
741    /// prior to their natural expiration time if the cache is full and evictions are required to make room for a new
742    /// context.
743    ///
744    /// Defaults to no expiration.
745    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    /// Sets whether or not to allow heap allocations when interning strings.
751    ///
752    /// In cases where the interner is full, this setting determines whether or not we refuse to resolve a context, or
753    /// if we allow it be resolved by allocating strings on the heap. When heap allocations are enabled, the amount of
754    /// memory that can be used by the interner is effectively unlimited, as contexts that can't be interned will be
755    /// simply spill to the heap instead of being limited in any way.
756    ///
757    /// Defaults to `true`.
758    pub fn with_heap_allocations(mut self, allow: bool) -> Self {
759        self.allow_heap_allocations = Some(allow);
760        self
761    }
762
763    /// Sets the origin tags resolver to use when building a context.
764    ///
765    /// In some cases, metrics, events, and service checks may have enriched tags based on their origin -- the
766    /// application/host/container/etc that emitted the metric -- which has to be considered when building the context
767    /// itself. As this can be expensive, it's useful to split the logic of actually grabbing the enriched tags based
768    /// on the available origin info into a separate phase, and implementation, that can run separately from the
769    /// initial hash-based approach of checking if a context has already been resolved.
770    ///
771    /// When set, any origin information provided will be considered during hashing when looking up a context, and any
772    /// enriched tags attached to the detected origin will be accessible from the context.
773    ///
774    /// Defaults to unset.
775    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    /// Sets whether or not to enable telemetry for this resolver.
781    ///
782    /// Reporting the telemetry of the resolver requires running an asynchronous task to override adding additional
783    /// overhead in the hot path of resolving contexts. In some cases, it may be cumbersome to always create the
784    /// resolver in an asynchronous context so that the telemetry task can be spawned. This method allows disabling
785    /// telemetry reporting in those cases.
786    ///
787    /// Defaults to telemetry enabled.
788    pub fn without_telemetry(mut self) -> Self {
789        self.telemetry_enabled = false;
790        self
791    }
792
793    /// Builds a [`TagsResolver`] from the current configuration.
794    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    /// Configures a [`TagsResolverBuilder`] that's suitable for tests.
825    ///
826    /// This configures the builder with the following defaults:
827    ///
828    /// - resolver name of "noop"
829    /// - unlimited cache capacity
830    /// - no-op interner (all strings are heap-allocated)
831    /// - heap allocations allowed
832    /// - telemetry disabled
833    ///
834    /// This is generally only useful for testing purposes, and is exposed publicly in order to be used in cross-crate
835    /// testing scenarios.
836    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
845/// A resolver for tags.
846pub 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        // Try to cheaply clone the string, and if that fails, try to intern it. If that fails, then we fall back to
861        // allocating it on the heap if we allow it.
862        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                    // Heap spill: with `allow_context_heap_allocations` true (the default), a full interner silently
867                    // falls back to the heap, so the bounded-memory guarantee no longer holds. Anchor that this path is
868                    // reached so the unbounded-growth behavior is observable.
869                    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    /// Creates a new tag set from the given tags.
880    ///
881    /// This will intern the tags, and then return a shared tag set. If the interner is full, and heap allocations are
882    /// not allowed, then this will return `None`.
883    ///
884    /// If heap allocations are allowed, then this will return a shared tag set, and the tag set will be cached.
885    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    /// Resolves the origin tags for the given origin.
902    ///
903    /// This will return the origin tags for the given origin, or an empty tag set if no origin tags resolver is set.
904    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    /// Finds the first node named `name` with a depth-first search.
957    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    /// Returns the direct child of `node` named `name`.
965    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    /// Runs `owner` as the only worker of a test supervisor.
970    ///
971    /// Returns a handle to the supervision tree, the sender that stops the supervisor, and the task of the supervisor.
972    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                // The resolver builds its own tags resolver, because none is supplied.
1010                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        // The telemetry task of the resolver runs in the scope of the resolver, and so do the scopes of the two caches
1026        // that the resolver built.
1027        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        // Every copy of the resolver shares its scope. Thus, the scope stays while one copy remains.
1038        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        // The owner built the tags resolver, and other resolvers can share it. Thus, its cache stays below the owner, and
1074        // the context resolver holds only its own cache.
1075        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        // Create two distinct contexts with the same name but different tags:
1107        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        // The contexts should not be equal to each other, and should have distinct underlying pointers to the shared
1121        // context state:
1122        assert_ne!(context1, context2);
1123        assert!(!context1.ptr_eq(&context2));
1124
1125        // If we create the context references again, we _should_ get back the same contexts as before:
1126        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        // Create two distinct contexts with the same name and tags, but with the tags in a different order:
1145        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        // The contexts should be equal to each other, and should have the same underlying pointer to the shared context
1159        // state:
1160        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        // Create our resolver and then create a context, which will have its metrics attached to our local recorder:
1245        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        // We should be able to see that the active context count is one, representing the context we created:
1253        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        // Now drop the context, and observe the active context count is negative one, representing the context we dropped:
1258        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        // Two contexts with the same name, but each with a different set of duplicate tags:
1269        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        // Each non-duplicated/duplicated context pair should be equal to one another:
1289        assert_eq!(context1, context1_duplicated);
1290        assert_eq!(context2, context2_duplicated);
1291
1292        // Each pair should not be equal to the other pair, however.
1293        //
1294        // What we're asserting here is that, if we didn't handle duplicate tags correctly, the XOR hashing of [tag1,
1295        // tag1] and [tag2, tag2] would result in the same hash value, since the second duplicate hash of tag1/tag2
1296        // would cancel out the first... and thus all that would be left is the hash of the name itself, which is the
1297        // same in this test. This would lead to the contexts being equal, which is obviously wrong.
1298        //
1299        // If we're handling duplicates properly, then the resulting context hashes _shouldn't_ be equal.
1300        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        // Create a regular context resolver, without any origin tags resolver, which should result in contexts being
1309        // the same so long as the name and tags are the same, disregarding any difference in origin information:
1310        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        // Now build a context resolver with an origin tags resolver that trivially returns the hash of the origin info
1332        // as a tag, which should result in differeing sets of origin tags between the two origins, thus no longer
1333        // comparing as equal:
1334        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        // Create a context with caching disabled, and verify that the context is not cached:
1364        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        // Create a second context with the same name and tags, and verify that it is not cached:
1370        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        // The contexts should be equal to each other, but the underlying `Arc` pointers should be different since
1376        // they're two distinct contexts in terms of not being cached:
1377        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        // Create a context resolver with a proper string interner configured:
1387        let mut resolver = ContextResolverBuilder::for_tests()
1388            .with_interner_capacity_bytes(NonZeroUsize::new(1024).expect("not zero"))
1389            .build();
1390
1391        // Create our context with cheaply cloneable tags, aka static strings:
1392        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        // Make sure the interner is empty before we resolve the context, and that it's empty afterwards, since we
1401        // should be able to cheaply clone both the metric name and both tags:
1402        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        // And just a sanity check that we have the expected name and tags in the context:
1412        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}