saluki_components/transforms/dogstatsd_mapper/

mod.rs

use std::collections::HashMap;
use std::num::NonZeroUsize;
use std::str::FromStr;
use std::sync::LazyLock;
use std::time::Duration;

use async_trait::async_trait;
use bytesize::ByteSize;
use regex::Regex;
use resource_accounting::{MemoryBounds, MemoryBoundsBuilder};
use saluki_common::cache::{Cache, CacheBuilder};
use saluki_config::GenericConfiguration;
use saluki_context::tags::SharedTagSet;
use saluki_context::tags::TagSet;
use saluki_context::{Context, ContextResolver, ContextResolverBuilder};
use saluki_core::{
    components::{
        transforms::{SynchronousTransform, SynchronousTransformBuilder},
        ComponentContext,
    },
    topology::EventsBuffer,
};
use saluki_error::{generic_error, ErrorContext, GenericError};
use serde::{Deserialize, Serialize};
use serde_with::{serde_as, DisplayFromStr, PickFirst};
use stringtheory::MetaString;

const MATCH_TYPE_WILDCARD: &str = "wildcard";
const MATCH_TYPE_REGEX: &str = "regex";

static ALLOWED_WILDCARD_MATCH_PATTERN: LazyLock<Regex> =
    LazyLock::new(|| Regex::new(r"^[a-zA-Z0-9\-_*.]+$").expect("Invalid regex in ALLOWED_WILDCARD_MATCH_PATTERN"));

const fn default_context_string_interner_size() -> ByteSize {
    ByteSize::kib(64)
}

const fn default_dogstatsd_mapper_cache_size() -> usize {
    1000
}
/// DogStatsD mapper transform.
#[serde_as]
#[derive(Deserialize)]
#[cfg_attr(test, derive(Debug, PartialEq, serde::Serialize))]
pub struct DogStatsDMapperConfiguration {
    /// Total size of the string interner used for contexts, in bytes.
    ///
    /// This controls the amount of memory that will be pre-allocated for the purpose
    /// of interning mapped metric names and tags, which can help to avoid unnecessary
    /// allocations and allocator fragmentation.
    #[serde(
        rename = "dogstatsd_mapper_string_interner_size",
        default = "default_context_string_interner_size"
    )]
    context_string_interner_bytes: ByteSize,

    /// Maximum number of mapped results to cache.
    ///
    /// When enabled, mapped metrics will be cached by name to avoid repeat evaluation of the configured mapper rules.
    ///
    /// When set to `0`, the cache is disabled.
    ///
    /// Defaults to `1000`.
    #[serde(
        rename = "dogstatsd_mapper_cache_size",
        default = "default_dogstatsd_mapper_cache_size"
    )]
    cache_size: usize,

    /// Configuration related to metric mapping.
    #[serde_as(as = "PickFirst<(DisplayFromStr, _)>")]
    #[serde(default)]
    dogstatsd_mapper_profiles: MapperProfileConfigs,
}

#[derive(Clone, Debug, Default, Deserialize, Serialize, PartialEq)]
struct MappingProfileConfig {
    name: String,
    prefix: String,
    mappings: Vec<MetricMappingConfig>,
}
#[derive(Clone, Debug, Default, Deserialize, Serialize, PartialEq)]
struct MapperProfileConfigs(pub Vec<MappingProfileConfig>);

impl FromStr for MapperProfileConfigs {
    type Err = serde_json::Error;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        let profiles: Vec<MappingProfileConfig> = serde_json::from_str(s)?;
        Ok(MapperProfileConfigs(profiles))
    }
}

#[cfg(test)]
impl std::fmt::Display for MapperProfileConfigs {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", serde_json::to_string(&self.0).unwrap_or_default())
    }
}

impl MapperProfileConfigs {
    fn build(
        &self, context: ComponentContext, context_string_interner_bytes: ByteSize, cache_size: usize,
    ) -> Result<MetricMapper, GenericError> {
        let mut profiles = Vec::with_capacity(self.0.len());
        for (i, config_profile) in self.0.iter().enumerate() {
            if config_profile.name.is_empty() {
                return Err(generic_error!("missing profile name"));
            }
            if config_profile.prefix.is_empty() {
                return Err(generic_error!("missing prefix for profile: {}", config_profile.name));
            }

            let mut profile = MappingProfile {
                prefix: config_profile.prefix.clone(),
                mappings: Vec::with_capacity(config_profile.mappings.len()),
            };

            for mapping in &config_profile.mappings {
                let match_type = match mapping.match_type.as_str() {
                    // Default to wildcard when not set.
                    "" => MATCH_TYPE_WILDCARD,
                    MATCH_TYPE_WILDCARD => MATCH_TYPE_WILDCARD,
                    MATCH_TYPE_REGEX => MATCH_TYPE_REGEX,
                    unknown => {
                        return Err(generic_error!(
                            "profile: {}, mapping num {}: invalid match type `{}`, expected `wildcard` or `regex`",
                            config_profile.name,
                            i,
                            unknown,
                        ))
                    }
                };
                if mapping.name.is_empty() {
                    return Err(generic_error!(
                        "profile: {}, mapping num {}: name is required",
                        config_profile.name,
                        i
                    ));
                }
                if mapping.metric_match.is_empty() {
                    return Err(generic_error!(
                        "profile: {}, mapping num {}: match is required",
                        config_profile.name,
                        i
                    ));
                }
                let regex = build_regex(&mapping.metric_match, match_type)?;
                profile.mappings.push(MetricMapping {
                    name: mapping.name.clone(),
                    tags: mapping.tags.clone(),
                    regex,
                });
            }
            profiles.push(profile);
        }

        let context_string_interner_size = NonZeroUsize::new(context_string_interner_bytes.as_u64() as usize)
            .ok_or_else(|| generic_error!("context_string_interner_size must be greater than 0"))
            .unwrap();

        let context_resolver =
            ContextResolverBuilder::from_name(format!("{}/dsd_mapper/primary", context.component_id()))
                .expect("resolver name is not empty")
                .with_interner_capacity_bytes(context_string_interner_size)
                .with_idle_context_expiration(Duration::from_secs(30))
                .build();

        let cache = match NonZeroUsize::new(cache_size) {
            Some(capacity) => Some(
                CacheBuilder::from_identifier(format!("{}/dsd_mapper/result_cache", context.component_id()))?
                    .with_capacity(capacity)
                    .build(),
            ),
            None => None,
        };

        Ok(MetricMapper {
            context_resolver,
            profiles,
            cache,
        })
    }
}

fn build_regex(match_re: &str, match_type: &str) -> Result<Regex, GenericError> {
    let mut pattern = match_re.to_owned();
    if match_type == MATCH_TYPE_WILDCARD {
        // Check it against the allowed wildcard pattern
        if !ALLOWED_WILDCARD_MATCH_PATTERN.is_match(&pattern) {
            return Err(generic_error!(
                "invalid wildcard match pattern `{}`, it does not match allowed match regex `{}`",
                pattern,
                ALLOWED_WILDCARD_MATCH_PATTERN.as_str()
            ));
        }
        if pattern.contains("**") {
            return Err(generic_error!(
                "invalid wildcard match pattern `{}`, it should not contain consecutive `*`",
                pattern
            ));
        }
        pattern = pattern.replace(".", "\\.");
        pattern = pattern.replace("*", "([^.]*)");
    }

    let final_pattern = format!("^{}$", pattern);

    Regex::new(&final_pattern).with_error_context(|| {
        format!(
            "Failed to compile regular expression `{}` for `{}` match type",
            final_pattern, match_type
        )
    })
}

#[derive(Clone, Debug, Default, Deserialize, Serialize, PartialEq)]
struct MetricMappingConfig {
    // The metric name to extract groups from with the Wildcard or Regex match logic.
    #[serde(rename = "match")]
    metric_match: String,

    // The type of match to apply to the `metric_match`. Either wildcard or regex.
    #[serde(default)]
    match_type: String,

    // The new metric name to send to Datadog with the tags defined in the same group.
    name: String,

    // Map with the tag key and tag values collected from the `match_type` to inline.
    #[serde(default)]
    tags: HashMap<String, String>,
}

struct MappingProfile {
    prefix: String,
    mappings: Vec<MetricMapping>,
}

struct MetricMapping {
    name: String,
    tags: HashMap<String, String>,
    regex: Regex,
}

#[derive(Clone)]
struct CachedMapResult {
    name: MetaString,
    extra_tags: SharedTagSet,
}

struct MetricMapper {
    profiles: Vec<MappingProfile>,
    context_resolver: ContextResolver,
    cache: Option<Cache<MetaString, Option<CachedMapResult>>>,
}

impl MetricMapper {
    fn try_map(&mut self, context: &Context) -> Option<Context> {
        // TODO: We should really be able to immutably borrow both the incoming tag set and the cached extra tags and
        // chain them together for our call into `resolve_with_origin_tags`, avoiding any allocations... but we need
        // some supporting work on the `TagSet` side to make it possible.

        let metric_name = context.name();
        let tags = context.tags();
        let origin_tags = context.origin_tags();

        // See if we have a cached result for this metric name.
        if let Some(cache) = &self.cache {
            if let Some(cached) = cache.get(metric_name) {
                return match cached {
                    None => None,
                    Some(result) => {
                        let mut merged_tags = tags.clone();
                        merged_tags.merge_shared(&result.extra_tags);

                        self.context_resolver.resolve_with_origin_tags(
                            result.name.clone(),
                            merged_tags,
                            origin_tags.clone(),
                        )
                    }
                };
            }
        }

        // Slow path: iterate profiles and run regexes.
        let mut new_name = String::new();
        let mut expanded_tag_value = String::new();

        for profile in &self.profiles {
            if !metric_name.starts_with(&profile.prefix) && profile.prefix != "*" {
                continue;
            }

            for mapping in &profile.mappings {
                if let Some(captures) = mapping.regex.captures(metric_name) {
                    new_name.clear();
                    captures.expand(&mapping.name, &mut new_name);

                    let mut extra_tags = TagSet::with_capacity(mapping.tags.len());
                    for (tag_key, tag_value_expr) in &mapping.tags {
                        expanded_tag_value.clear();
                        expanded_tag_value.push_str(tag_key);
                        expanded_tag_value.push(':');
                        captures.expand(tag_value_expr, &mut expanded_tag_value);

                        extra_tags.insert_tag(expanded_tag_value.as_str());
                    }

                    // Freeze the tags here so they can be shared / cached.
                    let extra_tags = extra_tags.into_shared();

                    let mut merged_tags = tags.clone();
                    merged_tags.merge_shared(&extra_tags);

                    let resolved = self.context_resolver.resolve_with_origin_tags(
                        new_name.as_str(),
                        merged_tags,
                        origin_tags.clone(),
                    )?;

                    if let Some(cache) = &self.cache {
                        cache.insert(
                            metric_name.clone(),
                            Some(CachedMapResult {
                                name: resolved.name().clone(),
                                extra_tags,
                            }),
                        );
                    }
                    return Some(resolved);
                }
            }
        }

        // We also cache "negative" results -- no match for this metric in the configured profiles -- to save ourselves some work.
        if let Some(cache) = &self.cache {
            cache.insert(metric_name.clone(), None);
        }
        None
    }

    #[cfg(test)]
    fn cache_len(&self) -> Option<usize> {
        self.cache.as_ref().map(|c| c.len())
    }
}

impl DogStatsDMapperConfiguration {
    /// Creates a new `DogstatsDMapperConfiguration` from the given configuration.
    pub fn from_configuration(config: &GenericConfiguration) -> Result<Self, GenericError> {
        Ok(config.as_typed()?)
    }
}

#[async_trait]
impl SynchronousTransformBuilder for DogStatsDMapperConfiguration {
    async fn build(&self, context: ComponentContext) -> Result<Box<dyn SynchronousTransform + Send>, GenericError> {
        let metric_mapper =
            self.dogstatsd_mapper_profiles
                .build(context, self.context_string_interner_bytes, self.cache_size)?;
        Ok(Box::new(DogStatsDMapper { metric_mapper }))
    }
}

impl MemoryBounds for DogStatsDMapperConfiguration {
    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
        let mut min = builder.minimum();
        min
            // Capture the size of the heap allocation when the component is built.
            .with_single_value::<DogStatsDMapper>("component struct")
            // We also allocate the backing storage for the string interner up front, which is used by our context
            // resolver.
            .with_fixed_amount("string interner", self.context_string_interner_bytes.as_u64() as usize);

        // Account for the per-name result cache when enabled.
        if self.cache_size > 0 {
            min.with_array::<(MetaString, Option<CachedMapResult>)>("mapper result cache", self.cache_size);
        }
    }
}

pub struct DogStatsDMapper {
    metric_mapper: MetricMapper,
}

impl SynchronousTransform for DogStatsDMapper {
    fn transform_buffer(&mut self, event_buffer: &mut EventsBuffer) {
        for event in event_buffer {
            if let Some(metric) = event.try_as_metric_mut() {
                if let Some(new_context) = self.metric_mapper.try_map(metric.context()) {
                    *metric.context_mut() = new_context;
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {

    use bytesize::ByteSize;
    use saluki_context::Context;
    use saluki_core::{components::ComponentContext, data_model::event::metric::Metric, topology::ComponentId};
    use saluki_error::GenericError;
    use serde_json::{json, Value};

    use super::{MapperProfileConfigs, MetricMapper};

    fn counter_metric(name: &'static str, tags: &[&'static str]) -> Metric {
        let context = Context::from_static_parts(name, tags);
        Metric::counter(context, 1.0)
    }

    fn mapper(json_data: Value) -> Result<MetricMapper, GenericError> {
        mapper_with_cache(json_data, 1000)
    }

    fn mapper_with_cache(json_data: Value, cache_size: usize) -> Result<MetricMapper, GenericError> {
        let context = ComponentContext::transform(ComponentId::try_from("test_mapper").unwrap());
        let mpc: MapperProfileConfigs = serde_json::from_value(json_data)?;
        let context_string_interner_bytes = ByteSize::kib(64);
        mpc.build(context, context_string_interner_bytes, cache_size)
    }

    fn assert_tags(context: &Context, expected_tags: &[&str]) {
        for tag in expected_tags {
            assert!(context.tags().has_tag(tag), "missing tag: {}", tag);
        }
        assert_eq!(context.tags().len(), expected_tags.len(), "unexpected number of tags");
    }

    #[tokio::test]
    async fn test_mapper_wildcard_simple() {
        let json_data = json!([{
          "name": "test",
          "prefix": "test.",
          "mappings": [
            {
              "match": "test.job.duration.*.*",
              "name": "test.job.duration",
              "tags": {
                "job_type": "$1",
                "job_name": "$2"
              }
            },
            {
              "match": "test.job.size.*.*",
              "name": "test.job.size",
              "tags": {
                "foo": "$1",
                "bar": "$2"
              }
            }
          ]
        }]);

        let mut mapper = mapper(json_data).expect("should have parsed mapping config");
        let metric = counter_metric("test.job.duration.my_job_type.my_job_name", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.job.duration");
        assert_tags(&context, &["job_type:my_job_type", "job_name:my_job_name"]);

        let metric = counter_metric("test.job.size.my_job_type.my_job_name", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.job.size");
        assert_tags(&context, &["foo:my_job_type", "bar:my_job_name"]);

        let metric = counter_metric("test.job.size.not_match", &[]);
        assert!(mapper.try_map(metric.context()).is_none(), "should not have remapped");
    }

    #[tokio::test]
    async fn test_partial_match() {
        let json_data = json!([{
          "name": "test",
          "prefix": "test.",
          "mappings": [
            {
              "match": "test.job.duration.*.*",
              "name": "test.job.duration",
              "tags": {
                "job_type": "$1"
              }
            },
            {
              "match": "test.task.duration.*.*",
              "name": "test.task.duration",
            }
          ]
        }]);
        let mut mapper = mapper(json_data).expect("should have parsed mapping config");
        let metric = counter_metric("test.job.duration.my_job_type.my_job_name", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.job.duration");
        assert!(context.tags().has_tag("job_type:my_job_type"));

        let metric = counter_metric("test.task.duration.my_job_type.my_job_name", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.task.duration");
    }

    #[tokio::test]
    async fn test_use_regex_expansion_alternative_syntax() {
        let json_data = json!([{
            "name": "test",
            "prefix": "test.",
            "mappings": [
                {
                    "match": "test.job.duration.*.*",
                    "name": "test.job.duration",
                    "tags": {
                        "job_type": "${1}_x",
                        "job_name": "${2}_y"
                    }
                }
            ]
        }]);

        let mut mapper = mapper(json_data).expect("should have parsed mapping config");

        let metric = counter_metric("test.job.duration.my_job_type.my_job_name", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.job.duration");
        assert_tags(&context, &["job_type:my_job_type_x", "job_name:my_job_name_y"]);
    }

    #[tokio::test]
    async fn test_expand_name() {
        let json_data = json!([{
            "name": "test",
            "prefix": "test.",
            "mappings": [
                {
                    "match": "test.job.duration.*.*",
                    "name": "test.hello.$2.$1",
                    "tags": {
                        "job_type": "$1",
                        "job_name": "$2"
                    }
                }
            ]
        }]);

        let mut mapper = mapper(json_data).expect("should have parsed mapping config");

        let metric = counter_metric("test.job.duration.my_job_type.my_job_name", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.hello.my_job_name.my_job_type");
        assert_tags(&context, &["job_type:my_job_type", "job_name:my_job_name"]);
    }

    #[tokio::test]
    async fn test_match_before_underscore() {
        let json_data = json!([{
            "name": "test",
            "prefix": "test.",
            "mappings": [
                {
                    "match": "test.*_start",
                    "name": "test.start",
                    "tags": {
                        "job": "$1"
                    }
                }
            ]
        }]);

        let mut mapper = mapper(json_data).expect("should have parsed mapping config");

        let metric = counter_metric("test.my_job_start", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.start");
        assert!(context.tags().has_tag("job:my_job"));
    }

    #[tokio::test]
    async fn test_no_tags() {
        let json_data = json!([{
            "name": "test",
            "prefix": "test.",
            "mappings": [
                {
                    "match": "test.my-worker.start",
                    "name": "test.worker.start"
                },
                {
                    "match": "test.my-worker.stop.*",
                    "name": "test.worker.stop"
                }
            ]
        }]);

        let mut mapper = mapper(json_data).expect("should have parsed mapping config");

        let metric = counter_metric("test.my-worker.start", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.worker.start");
        assert!(context.tags().is_empty(), "Expected no tags");

        let metric = counter_metric("test.my-worker.stop.worker-name", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.worker.stop");
        assert!(context.tags().is_empty(), "Expected no tags");
    }

    #[tokio::test]
    async fn test_all_allowed_characters() {
        let json_data = json!([{
            "name": "test",
            "prefix": "test.",
            "mappings": [
                {
                    "match": "test.abcdefghijklmnopqrstuvwxyz_ABCDEFGHIJKLMNOPQRSTUVWXYZ-01234567.*",
                    "name": "test.alphabet"
                }
            ]
        }]);

        let mut mapper = mapper(json_data).expect("should have parsed mapping config");

        let metric = counter_metric(
            "test.abcdefghijklmnopqrstuvwxyz_ABCDEFGHIJKLMNOPQRSTUVWXYZ-01234567.123",
            &[],
        );
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.alphabet");
        assert!(context.tags().is_empty(), "Expected no tags");
    }

    #[tokio::test]
    async fn test_regex_match_type() {
        let json_data = json!([{
            "name": "test",
            "prefix": "test.",
            "mappings": [
                {
                    "match": "test\\.job\\.duration\\.(.*)",
                    "match_type": "regex",
                    "name": "test.job.duration",
                    "tags": {
                        "job_name": "$1"
                    }
                },
                {
                    "match": "test\\.task\\.duration\\.(.*)",
                    "match_type": "regex",
                    "name": "test.task.duration",
                    "tags": {
                        "task_name": "$1"
                    }
                }
            ]
        }]);

        let mut mapper = mapper(json_data).expect("should have parsed mapping config");
        let metric = counter_metric("test.job.duration.my.funky.job$name-abc/123", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.job.duration");
        assert!(context.tags().has_tag("job_name:my.funky.job$name-abc/123"));

        let metric = counter_metric("test.task.duration.MY_task_name", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.task.duration");
        assert!(context.tags().has_tag("task_name:MY_task_name"));
    }

    #[tokio::test]
    async fn test_complex_regex_match_type() {
        let json_data = json!([{
            "name": "test",
            "prefix": "test.",
            "mappings": [
                {
                    "match": "test\\.job\\.([a-z][0-9]-\\w+)\\.(.*)",
                    "match_type": "regex",
                    "name": "test.job",
                    "tags": {
                        "job_type": "$1",
                        "job_name": "$2"
                    }
                }
            ]
        }]);

        let mut mapper = mapper(json_data).expect("should have parsed mapping config");

        let metric = counter_metric("test.job.a5-foo.bar", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.job");
        assert_tags(&context, &["job_type:a5-foo", "job_name:bar"]);

        let metric = counter_metric("test.job.foo.bar-not-match", &[]);
        assert!(mapper.try_map(metric.context()).is_none(), "should not have remapped");
    }

    #[tokio::test]
    async fn test_profile_and_prefix() {
        let json_data = json!([{
            "name": "test",
            "prefix": "foo.",
            "mappings": [
                {
                    "match": "foo.duration.*",
                    "name": "foo.duration",
                    "tags": {
                        "name": "$1"
                    }
                }
            ]
        },
        {
            "name": "test",
            "prefix": "bar.",
            "mappings": [
                {
                    "match": "bar.count.*",
                    "name": "bar.count",
                    "tags": {
                        "name": "$1"
                    }
                },
                {
                    "match": "foo.duration2.*",
                    "name": "foo.duration2",
                    "tags": {
                        "name": "$1"
                    }
                }
            ]
        }]);

        let mut mapper = mapper(json_data).expect("should have parsed mapping config");

        let metric = counter_metric("foo.duration.foo_name1", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "foo.duration");
        assert!(context.tags().has_tag("name:foo_name1"));

        let metric = counter_metric("foo.duration2.foo_name1", &[]);
        assert!(
            mapper.try_map(metric.context()).is_none(),
            "should not have remapped due to wrong group"
        );

        let metric = counter_metric("bar.count.bar_name1", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "bar.count");
        assert!(context.tags().has_tag("name:bar_name1"));

        let metric = counter_metric("z.not.mapped", &[]);
        assert!(mapper.try_map(metric.context()).is_none(), "should not have remapped");
    }

    #[tokio::test]
    async fn test_wildcard_prefix() {
        let json_data = json!([{
            "name": "test",
            "prefix": "*",
            "mappings": [
                {
                    "match": "foo.duration.*",
                    "name": "foo.duration",
                    "tags": {
                        "name": "$1"
                    }
                }
            ]
        }]);

        let mut mapper = mapper(json_data).expect("should have parsed mapping config");

        let metric = counter_metric("foo.duration.foo_name1", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "foo.duration");
        assert!(context.tags().has_tag("name:foo_name1"));
    }

    #[tokio::test]
    async fn test_wildcard_prefix_order() {
        let json_data = json!([{
            "name": "test",
            "prefix": "*",
            "mappings": [
                {
                    "match": "foo.duration.*",
                    "name": "foo.duration",
                    "tags": {
                        "name1": "$1"
                    }
                }
            ]
        },
        {
            "name": "test",
            "prefix": "*",
            "mappings": [
                {
                    "match": "foo.duration.*",
                    "name": "foo.duration",
                    "tags": {
                        "name2": "$1"
                    }
                }
            ]
        }]);

        let mut mapper = mapper(json_data).expect("should have parsed mapping config");
        let metric = counter_metric("foo.duration.foo_name", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "foo.duration");
        assert!(context.tags().has_tag("name1:foo_name"));
        assert!(
            !context.tags().has_tag("name2:foo_name"),
            "Only the first matching profile should apply"
        );
    }

    #[tokio::test]
    async fn test_multiple_profiles_order() {
        let json_data = json!([{
            "name": "test",
            "prefix": "foo.",
            "mappings": [
                {
                    "match": "foo.*.duration.*",
                    "name": "foo.bar1.duration",
                    "tags": {
                        "bar": "$1",
                        "foo": "$2"
                    }
                }
            ]
        },
        {
            "name": "test",
            "prefix": "foo.bar.",
            "mappings": [
                {
                    "match": "foo.bar.duration.*",
                    "name": "foo.bar2.duration",
                    "tags": {
                        "foo_bar": "$1"
                    }
                }
            ]
        }]);

        let mut mapper = mapper(json_data).expect("should have parsed mapping config");

        let metric = counter_metric("foo.bar.duration.foo_name", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "foo.bar1.duration");
        assert_tags(&context, &["bar:bar", "foo:foo_name"]);
        assert!(
            !context.tags().has_tag("foo_bar:foo_name"),
            "Only the first matching profile should apply"
        );
    }

    #[tokio::test]
    async fn test_different_regex_expansion_syntax() {
        let json_data = json!([{
            "name": "test",
            "prefix": "test.",
            "mappings": [
                {
                    "match": "test.user.(\\w+).action.(\\w+)",
                    "match_type": "regex",
                    "name": "test.user.action",
                    "tags": {
                        "user": "$1",
                        "action": "$2"
                    }
                }
            ]
        }]);

        let mut mapper = mapper(json_data).expect("should have parsed mapping config");
        let metric = counter_metric("test.user.john_doe.action.login", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.user.action");
        assert_tags(&context, &["user:john_doe", "action:login"]);
    }

    #[tokio::test]
    async fn test_retain_existing_tags() {
        let json_data = json!([{
          "name": "test",
          "prefix": "test.",
          "mappings": [
            {
              "match": "test.job.duration.*.*",
              "name": "test.job.duration.$2",
              "tags": {
                "job_type": "$1",
                "job_name": "$2"
              }
            },
          ]
        }]);
        let mut mapper = mapper(json_data).expect("should have parsed mapping config");
        let metric = counter_metric("test.job.duration.abc.def", &["foo:bar", "baz"]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.job.duration.def");
        assert!(context.tags().has_tag("foo:bar"));
        assert!(context.tags().has_tag("baz"));
    }

    #[test]
    fn test_empty_name() {
        let json_data = json!([{
            "name": "test",
            "prefix": "test.",
            "mappings": [
                {
                    "match": "test.job.duration.*.*",
                    "name": "",
                    "tags": {
                        "job_type": "$1"
                    }
                }
            ]
        }]);
        assert!(mapper(json_data).is_err())
    }

    #[test]
    fn test_missing_name() {
        let json_data = json!([{
            "name": "test",
            "prefix": "test.",
            "mappings": [
                {
                    "match": "test.job.duration.*.*",
                    "tags": {
                        "job_type": "$1",
                        "job_name": "$2"
                    }
                }
            ]
        }]);
        assert!(mapper(json_data).is_err());
    }

    #[test]
    fn test_invalid_match_regex_brackets() {
        let json_data = json!([{
            "name": "test",
            "prefix": "test.",
            "mappings": [
                {
                    "match": "test.[]duration.*.*", // Invalid regex
                    "name": "test.job.duration"
                }
            ]
        }]);
        assert!(mapper(json_data).is_err());
    }

    #[test]
    fn test_invalid_match_regex_caret() {
        let json_data = json!([{
            "name": "test",
            "prefix": "test.",
            "mappings": [
                {
                    "match": "^test.invalid.duration.*.*", // Invalid regex
                    "name": "test.job.duration"
                }
            ]
        }]);
        assert!(mapper(json_data).is_err());
    }

    #[test]
    fn test_consecutive_wildcards() {
        let json_data = json!([{
            "name": "test",
            "prefix": "test.",
            "mappings": [
                {
                    "match": "test.invalid.duration.**", // Consecutive *
                    "name": "test.job.duration"
                }
            ]
        }]);
        assert!(mapper(json_data).is_err());
    }

    #[test]
    fn test_invalid_match_type() {
        let json_data = json!([{
            "name": "test",
            "prefix": "test.",
            "mappings": [
                {
                    "match": "test.invalid.duration",
                    "match_type": "invalid", // Invalid match_type
                    "name": "test.job.duration"
                }
            ]
        }]);
        assert!(mapper(json_data).is_err());
    }

    #[test]
    fn test_missing_profile_name() {
        let json_data = json!([{
            // "name" is missing here
            "prefix": "test.",
            "mappings": [
                {
                    "match": "test.invalid.duration",
                    "match_type": "invalid",
                    "name": "test.job.duration"
                }
            ]
        }]);
        assert!(mapper(json_data).is_err());
    }

    #[test]
    fn test_missing_profile_prefix() {
        let json_data = json!([{
            "name": "test",
            // "prefix" is missing here
            "mappings": [
                {
                    "match": "test.invalid.duration",
                    "match_type": "invalid",
                    "name": "test.job.duration"
                }
            ]
        }]);
        assert!(mapper(json_data).is_err());
    }

    fn simple_mapping_profile() -> Value {
        json!([{
            "name": "test",
            "prefix": "test.",
            "mappings": [
                {
                    "match": "test.job.duration.*.*",
                    "name": "test.job.duration",
                    "tags": {
                        "job_type": "$1",
                        "job_name": "$2"
                    }
                }
            ]
        }])
    }

    #[tokio::test]
    async fn test_cache_hit_returns_same_result_as_miss() {
        let mut mapper = mapper_with_cache(simple_mapping_profile(), 1000).expect("should have parsed mapping config");
        assert_eq!(mapper.cache_len(), Some(0));

        let metric = counter_metric("test.job.duration.my_type.my_name", &[]);
        let first = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(mapper.cache_len(), Some(1));

        let metric = counter_metric("test.job.duration.my_type.my_name", &[]);
        let second = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(mapper.cache_len(), Some(1));

        assert_eq!(first.name(), second.name());
        assert_eq!(first.name(), "test.job.duration");
        assert_tags(&first, &["job_type:my_type", "job_name:my_name"]);
        assert_tags(&second, &["job_type:my_type", "job_name:my_name"]);
    }

    #[tokio::test]
    async fn test_negative_cache() {
        let mut mapper = mapper_with_cache(simple_mapping_profile(), 1000).expect("should have parsed mapping config");

        let metric = counter_metric("unrelated.metric.name", &[]);
        assert!(mapper.try_map(metric.context()).is_none());
        assert_eq!(mapper.cache_len(), Some(1));

        let metric = counter_metric("unrelated.metric.name", &[]);
        assert!(mapper.try_map(metric.context()).is_none());
        assert_eq!(mapper.cache_len(), Some(1));
    }

    #[tokio::test]
    async fn test_cache_disabled_when_zero() {
        let mut mapper = mapper_with_cache(simple_mapping_profile(), 0).expect("should have parsed mapping config");
        assert_eq!(mapper.cache_len(), None);

        let metric = counter_metric("test.job.duration.my_type.my_name", &[]);
        let context = mapper.try_map(metric.context()).expect("should have remapped");
        assert_eq!(context.name(), "test.job.duration");
        assert_tags(&context, &["job_type:my_type", "job_name:my_name"]);

        assert!(mapper
            .try_map(counter_metric("unrelated.metric", &[]).context())
            .is_none());
        assert_eq!(mapper.cache_len(), None);
    }

    #[tokio::test]
    async fn test_cache_eviction() {
        let mut mapper = mapper_with_cache(simple_mapping_profile(), 2).expect("should have parsed mapping config");

        for suffix in ["a", "b", "c"] {
            let name = format!("test.job.duration.t.{}", suffix);
            let metric = counter_metric(Box::leak(name.into_boxed_str()), &[]);
            mapper.try_map(metric.context()).expect("should have remapped");
        }

        assert!(
            mapper.cache_len().unwrap() <= 2,
            "cache should not exceed configured capacity (got {})",
            mapper.cache_len().unwrap()
        );
    }

    #[tokio::test]
    async fn test_flood_does_not_grow_cache() {
        // Many profiles, only the last one matches the test metric. A flood of identical
        // names should be served from the cache after the first call.
        let mut profiles: Vec<Value> = (0..50)
            .map(|i| {
                json!({
                    "name": format!("noise-{}", i),
                    "prefix": format!("noise{}.", i),
                    "mappings": [{
                        "match": format!("noise{}.*", i),
                        "name": "noise.mapped"
                    }]
                })
            })
            .collect();
        profiles.push(json!({
            "name": "real",
            "prefix": "real.",
            "mappings": [{
                "match": "real.metric.*",
                "name": "real.mapped",
                "tags": { "x": "$1" }
            }]
        }));
        let json_data = Value::Array(profiles);

        let mut mapper = mapper_with_cache(json_data, 16).expect("should have parsed mapping config");

        for _ in 0..10_000 {
            let metric = counter_metric("real.metric.flood", &[]);
            let context = mapper.try_map(metric.context()).expect("should have remapped");
            assert_eq!(context.name(), "real.mapped");
        }

        assert_eq!(
            mapper.cache_len(),
            Some(1),
            "flood of identical names should populate exactly one cache entry"
        );
    }
}

#[cfg(test)]
mod config_smoke {
    use serde_json::json;

    use super::DogStatsDMapperConfiguration;
    use crate::config_registry::structs;
    use crate::config_registry::test_support::run_config_smoke_tests;

    #[tokio::test]
    async fn smoke_test() {
        run_config_smoke_tests(structs::DOGSTATSD_MAPPER_CONFIGURATION, &[], json!({}), |cfg| {
            cfg.as_typed::<DogStatsDMapperConfiguration>()
                .expect("DogStatsDMapperConfiguration should deserialize")
        })
        .await
    }
}