From 1fc4d4db54c82a86d529667c22ed469ee22037b0 Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Sun, 1 Dec 2024 01:50:47 +0800
Subject: [PATCH 01/26] Try tag expansion

---
 src/transforms/log_to_metric.rs | 103 ++++++++++++++++++++++++++++----
 1 file changed, 93 insertions(+), 10 deletions(-)

diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index 1e3d966d2c43a..9518863a5f637 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -1,8 +1,9 @@
 use std::sync::Arc;
-use std::{collections::HashMap, num::ParseFloatError};
+use std::{collections::HashMap, num::ParseFloatError, sync::LazyLock};
 
 use chrono::Utc;
 use indexmap::IndexMap;
+use regex::Regex;
 use vector_lib::configurable::configurable_component;
 use vector_lib::event::LogEvent;
 use vector_lib::{
@@ -110,7 +111,7 @@ pub struct MetricConfig {
 
     /// Tags to apply to the metric.
     #[configurable(metadata(docs::additional_props_description = "A metric tag."))]
-    pub tags: Option<IndexMap<String, TagConfig>>,
+    pub tags: Option<IndexMap<Template, TagConfig>>,
 
     #[configurable(derived)]
     #[serde(flatten)]
@@ -271,25 +272,48 @@ fn render_template(template: &Template, event: &Event) -> Result<String, Transfo
 }
 
 fn render_tags(
-    tags: &Option<IndexMap<String, TagConfig>>,
+    tags: &Option<IndexMap<Template, TagConfig>>,
     event: &Event,
 ) -> Result<Option<MetricTags>, TransformError> {
+    let mut static_labels: HashMap<String, String> = HashMap::new();
+    let mut dynamic_labels: HashMap<String, String> = HashMap::new();
     Ok(match tags {
         None => None,
         Some(tags) => {
             let mut result = MetricTags::default();
             for (name, config) in tags {
                 match config {
-                    TagConfig::Plain(template) => {
-                        render_tag_into(event, name, template, &mut result)?
-                    }
+                    TagConfig::Plain(template) => render_tag_into(
+                        event,
+                        name,
+                        template,
+                        &mut result,
+                        &mut static_labels,
+                        &mut dynamic_labels,
+                    )?,
                     TagConfig::Multi(vec) => {
                         for template in vec {
-                            render_tag_into(event, name, template, &mut result)?;
+                            render_tag_into(
+                                event,
+                                name,
+                                template,
+                                &mut result,
+                                &mut static_labels,
+                                &mut dynamic_labels,
+                            )?;
                         }
                     }
                 }
             }
+            for (k, v) in static_labels {
+                if let Some(discarded_v) = dynamic_labels.insert(k.clone(), v.clone()) {
+                    warn!(
+                        "Static label overrides dynamic label. \
+                key: {}, value: {}, discarded value: {}",
+                        k, v, discarded_v
+                    );
+                };
+            }
             result.as_option()
         }
     })
@@ -297,19 +321,24 @@ fn render_tags(
 
 fn render_tag_into(
     event: &Event,
-    name: &str,
+    name: &Template,
     template: &Option<Template>,
     result: &mut MetricTags,
+    static_labels: &mut HashMap<String, String>,
+    dynamic_labels: &mut HashMap<String, String>,
 ) -> Result<(), TransformError> {
     let value = match template {
         None => TagValue::Bare,
         Some(template) => match render_template(template, event) {
-            Ok(result) => TagValue::Value(result),
+            Ok(result) => {
+                tag_expansion(event, name, template, static_labels, dynamic_labels);
+                TagValue::Value(result)
+            }
             Err(TransformError::TemplateRenderingError(error)) => {
                 emit!(crate::internal_events::TemplateRenderingError {
                     error,
                     drop_event: false,
-                    field: Some(name),
+                    field: Some(name.get_ref()),
                 });
                 return Ok(());
             }
@@ -320,6 +349,60 @@ fn render_tag_into(
     Ok(())
 }
 
+static RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"[^0-9A-Za-z_]").unwrap());
+fn slugify_text(input: String) -> String {
+    let result = RE.replace_all(&input, "_");
+    result.to_lowercase()
+}
+
+fn tag_expansion(
+    event: &Event,
+    key_template: &Template,
+    value_template: &Template,
+    static_labels: &mut HashMap<String, String>,
+    dynamic_labels: &mut HashMap<String, String>,
+) {
+    let key = key_template.render_string(event);
+    let value = value_template.render_string(event);
+
+    let key_s = key.unwrap();
+    let value_s = value.unwrap();
+
+    if let Some(opening_prefix) = key_s.strip_suffix('*') {
+        let output: Result<serde_json::map::Map<String, serde_json::Value>, serde_json::Error> =
+            serde_json::from_str(value_s.clone().as_str());
+
+        if output.is_err() {
+            warn!(
+                "Failed to expand dynamic label. value: {}, err: {}",
+                value_s,
+                output.err().unwrap()
+            );
+            return;
+        }
+
+        // key_* -> key_one, key_two, key_three
+        // * -> one, two, three
+        for (k, v) in output.unwrap() {
+            let key = slugify_text(format!("{}{}", opening_prefix, k));
+            let val = Value::from(v).to_string_lossy().into_owned();
+            if val == "<null>" {
+                warn!("Encountered \"null\" value for dynamic label. key: {}", key);
+                continue;
+            }
+            if let Some(prev) = dynamic_labels.insert(key.clone(), val.clone()) {
+                warn!(
+                    "Encountered duplicated dynamic label. \
+                                key: {}, value: {}, discarded value: {}",
+                    key, val, prev
+                );
+            };
+        }
+    } else {
+        static_labels.insert(key_s, value_s);
+    }
+}
+
 fn to_metric_with_config(config: &MetricConfig, event: &Event) -> Result<Metric, TransformError> {
     let log = event.as_log();
 

From f49b068b3d4bf7e80ae10fa87acbb4e656ae66f3 Mon Sep 17 00:00:00 2001
From: titaneric <chenyihuang001@gmail.com>
Date: Mon, 2 Dec 2024 22:56:13 +0800
Subject: [PATCH 02/26] Add tags expansion test

---
 src/transforms/log_to_metric.rs | 51 +++++++++++++++++++++++++++++++++
 1 file changed, 51 insertions(+)

diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index 9518863a5f637..e3525b420ca5b 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -306,6 +306,7 @@ fn render_tags(
                 }
             }
             for (k, v) in static_labels {
+                println!("{} {}", k, v);
                 if let Some(discarded_v) = dynamic_labels.insert(k.clone(), v.clone()) {
                     warn!(
                         "Static label overrides dynamic label. \
@@ -1155,6 +1156,56 @@ mod tests {
             .with_timestamp(Some(ts()))
         );
     }
+    
+    #[tokio::test]
+    async fn count_http_requests_with_tags_expansion() {
+        let config = parse_config(
+            r#"
+            [[metrics]]
+            type = "counter"
+            field = "message"
+            name = "http_requests_total"
+            namespace = "app"
+            tags = {"*" = "{{.}}"}
+            "#,
+        );
+
+        let mut event = create_event("message", "i am log");
+        event.as_mut_log().insert("method", "post");
+        event.as_mut_log().insert("code", "200");
+        let mut metadata =
+            event
+                .metadata()
+                .clone()
+                .with_origin_metadata(DatadogMetricOriginMetadata::new(
+                    None,
+                    None,
+                    Some(ORIGIN_SERVICE_VALUE),
+                ));
+        // definitions aren't valid for metrics yet, it's just set to the default (anything).
+        metadata.set_schema_definition(&Arc::new(Definition::any()));
+        metadata.set_upstream_id(Arc::new(OutputId::from("transform")));
+        metadata.set_source_id(Arc::new(ComponentKey::from("in")));
+
+        let metric = do_transform(config, event).await.unwrap();
+
+        assert_eq!(
+            metric.into_metric(),
+            Metric::new_with_metadata(
+                "http_requests_total",
+                MetricKind::Incremental,
+                MetricValue::Counter { value: 1.0 },
+                metadata,
+            )
+            .with_namespace(Some("app"))
+            .with_tags(Some(metric_tags!(
+                "method" => "post",
+                "code" => "200",
+                "host" => "localhost",
+            )))
+            .with_timestamp(Some(ts()))
+        );
+    }
 
     #[tokio::test]
     async fn multi_value_tags_yaml() {

From 7a94bd1b14b708b797b33ce346af37818363daee Mon Sep 17 00:00:00 2001
From: titaneric <chenyihuang001@gmail.com>
Date: Tue, 3 Dec 2024 01:41:55 +0900
Subject: [PATCH 03/26] Extract dynamic labels into tags result

---
 src/transforms/log_to_metric.rs | 18 +++++++++++++-----
 1 file changed, 13 insertions(+), 5 deletions(-)

diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index e3525b420ca5b..92adb08a005f7 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -306,7 +306,6 @@ fn render_tags(
                 }
             }
             for (k, v) in static_labels {
-                println!("{} {}", k, v);
                 if let Some(discarded_v) = dynamic_labels.insert(k.clone(), v.clone()) {
                     warn!(
                         "Static label overrides dynamic label. \
@@ -315,6 +314,16 @@ fn render_tags(
                     );
                 };
             }
+            println!("{:?}", dynamic_labels);
+            for (name, value) in dynamic_labels {
+                if name == "timestamp".to_string() {
+                    continue;
+                }
+                if name == "message".to_string() {
+                    continue;
+                }
+                result.insert(name.to_string(), value);
+            }
             result.as_option()
         }
     })
@@ -324,11 +333,11 @@ fn render_tag_into(
     event: &Event,
     name: &Template,
     template: &Option<Template>,
-    result: &mut MetricTags,
+    _result: &mut MetricTags,
     static_labels: &mut HashMap<String, String>,
     dynamic_labels: &mut HashMap<String, String>,
 ) -> Result<(), TransformError> {
-    let value = match template {
+    let _value = match template {
         None => TagValue::Bare,
         Some(template) => match render_template(template, event) {
             Ok(result) => {
@@ -346,7 +355,7 @@ fn render_tag_into(
             Err(other) => return Err(other),
         },
     };
-    result.insert(name.to_string(), value);
+    // result.insert(name.to_string(), value);
     Ok(())
 }
 
@@ -1201,7 +1210,6 @@ mod tests {
             .with_tags(Some(metric_tags!(
                 "method" => "post",
                 "code" => "200",
-                "host" => "localhost",
             )))
             .with_timestamp(Some(ts()))
         );

From e85cb44deb775f8dbd5943db6ad807c65e2b2378 Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Tue, 3 Dec 2024 22:01:30 +0800
Subject: [PATCH 04/26] Update tags expansion testcase

---
 src/transforms/log_to_metric.rs | 25 ++++++++++++-------------
 1 file changed, 12 insertions(+), 13 deletions(-)

diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index 92adb08a005f7..4e4275143a2fe 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -316,12 +316,6 @@ fn render_tags(
             }
             println!("{:?}", dynamic_labels);
             for (name, value) in dynamic_labels {
-                if name == "timestamp".to_string() {
-                    continue;
-                }
-                if name == "message".to_string() {
-                    continue;
-                }
                 result.insert(name.to_string(), value);
             }
             result.as_option()
@@ -1003,7 +997,7 @@ mod tests {
     use tokio::sync::mpsc;
     use tokio_stream::wrappers::ReceiverStream;
     use vector_lib::config::ComponentKey;
-    use vector_lib::event::EventMetadata;
+    use vector_lib::event::{EventMetadata, ObjectMap};
     use vector_lib::metric_tags;
 
     #[test]
@@ -1165,7 +1159,7 @@ mod tests {
             .with_timestamp(Some(ts()))
         );
     }
-    
+
     #[tokio::test]
     async fn count_http_requests_with_tags_expansion() {
         let config = parse_config(
@@ -1175,13 +1169,18 @@ mod tests {
             field = "message"
             name = "http_requests_total"
             namespace = "app"
-            tags = {"*" = "{{.}}"}
+            tags = {"*" = "{{dict}}"}
             "#,
         );
 
         let mut event = create_event("message", "i am log");
-        event.as_mut_log().insert("method", "post");
-        event.as_mut_log().insert("code", "200");
+        let log = event.as_mut_log();
+
+        let mut test_dict = ObjectMap::default();
+        test_dict.insert("one".into(), Value::from("foo"));
+        test_dict.insert("two".into(), Value::from("baz"));
+        log.insert("dict", Value::from(test_dict));
+
         let mut metadata =
             event
                 .metadata()
@@ -1208,8 +1207,8 @@ mod tests {
             )
             .with_namespace(Some("app"))
             .with_tags(Some(metric_tags!(
-                "method" => "post",
-                "code" => "200",
+                "one" => "foo",
+                "two" => "baz",
             )))
             .with_timestamp(Some(ts()))
         );

From 7c39d9886538461ac3996493c8f8485301ae2446 Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Tue, 3 Dec 2024 23:33:24 +0800
Subject: [PATCH 05/26] Refactor render_tag_into

---
 src/transforms/log_to_metric.rs | 52 +++++++++++++++++----------------
 1 file changed, 27 insertions(+), 25 deletions(-)

diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index 4e4275143a2fe..c0a5b8504640b 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -314,9 +314,8 @@ fn render_tags(
                     );
                 };
             }
-            println!("{:?}", dynamic_labels);
             for (name, value) in dynamic_labels {
-                result.insert(name.to_string(), value);
+                result.insert(name.to_string(), TagValue::Value(value));
             }
             result.as_option()
         }
@@ -325,31 +324,41 @@ fn render_tags(
 
 fn render_tag_into(
     event: &Event,
-    name: &Template,
-    template: &Option<Template>,
-    _result: &mut MetricTags,
+    key_template: &Template,
+    value_template: &Option<Template>,
+    result: &mut MetricTags,
     static_labels: &mut HashMap<String, String>,
     dynamic_labels: &mut HashMap<String, String>,
 ) -> Result<(), TransformError> {
-    let _value = match template {
-        None => TagValue::Bare,
-        Some(template) => match render_template(template, event) {
-            Ok(result) => {
-                tag_expansion(event, name, template, static_labels, dynamic_labels);
-                TagValue::Value(result)
+    let key = render_template(key_template, event);
+    match value_template {
+        None => {
+            result.insert(key_template.to_string(), TagValue::Bare);
+        }
+        Some(template) => match (key, render_template(template, event)) {
+            (Ok(key_s), Ok(value_s)) => {
+                tag_expansion(key_s, value_s, static_labels, dynamic_labels);
             }
-            Err(TransformError::TemplateRenderingError(error)) => {
+            (Err(TransformError::TemplateRenderingError(key_error)), _) => {
                 emit!(crate::internal_events::TemplateRenderingError {
-                    error,
+                    error: key_error,
                     drop_event: false,
-                    field: Some(name.get_ref()),
+                    field: Some(key_template.get_ref()),
                 });
                 return Ok(());
             }
-            Err(other) => return Err(other),
+            (_, Err(TransformError::TemplateRenderingError(value_error))) => {
+                emit!(crate::internal_events::TemplateRenderingError {
+                    error: value_error,
+                    drop_event: false,
+                    field: Some(template.get_ref()),
+                });
+                return Ok(());
+            }
+            (Err(key_error), _) => return Err(key_error),
+            (_, Err(value_error)) => return Err(value_error),
         },
     };
-    // result.insert(name.to_string(), value);
     Ok(())
 }
 
@@ -360,18 +369,11 @@ fn slugify_text(input: String) -> String {
 }
 
 fn tag_expansion(
-    event: &Event,
-    key_template: &Template,
-    value_template: &Template,
+    key_s: String,
+    value_s: String,
     static_labels: &mut HashMap<String, String>,
     dynamic_labels: &mut HashMap<String, String>,
 ) {
-    let key = key_template.render_string(event);
-    let value = value_template.render_string(event);
-
-    let key_s = key.unwrap();
-    let value_s = value.unwrap();
-
     if let Some(opening_prefix) = key_s.strip_suffix('*') {
         let output: Result<serde_json::map::Map<String, serde_json::Value>, serde_json::Error> =
             serde_json::from_str(value_s.clone().as_str());

From 788a25757aaf25f7d6613aab4f54788591d59b26 Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Wed, 4 Dec 2024 00:56:02 +0800
Subject: [PATCH 06/26] Add multi_value_tags_expansion_yaml test

---
 src/transforms/log_to_metric.rs | 34 +++++++++++++++++++++++++++++++++
 1 file changed, 34 insertions(+)

diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index c0a5b8504640b..5675b2498d32e 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -1244,6 +1244,40 @@ mod tests {
             assert!(value.is_none() || value == Some("one") || value == Some("two"));
         }
     }
+    #[tokio::test]
+    async fn multi_value_tags_expansion_yaml() {
+        // Have to use YAML to represent bare tags
+        let config = parse_yaml_config(
+            r#"
+            metrics:
+            - field: "message"
+              type: "counter"
+              tags:
+                "*": "{{dict}}"
+            "#,
+        );
+
+        let mut event = create_event("message", "I am log");
+        let log = event.as_mut_log();
+
+        let mut test_dict = ObjectMap::default();
+        test_dict.insert("one".into(), Value::from(vec!["foo", "baz"]));
+        log.insert("dict", Value::from(test_dict));
+
+        let metric = do_transform(config, event).await.unwrap().into_metric();
+        let tags = metric.tags().expect("Metric should have tags");
+
+        assert_eq!(
+            tags.iter_single().collect::<Vec<_>>(),
+            vec![("one", "[\"foo\",\"baz\"]")]
+        );
+
+        assert_eq!(tags.iter_all().count(), 1);
+        for (name, value) in tags.iter_all() {
+            assert_eq!(name, "one");
+            assert_eq!(value, Some("[\"foo\",\"baz\"]"));
+        }
+    }
 
     #[tokio::test]
     async fn multi_value_tags_toml() {

From 293b15a71c2e420d18a6869112fc30fd5b464575 Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Wed, 4 Dec 2024 01:04:51 +0800
Subject: [PATCH 07/26] Add basic docs for tags

---
 src/transforms/log_to_metric.rs | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index 5675b2498d32e..032cc4d248f16 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -110,6 +110,9 @@ pub struct MetricConfig {
     pub namespace: Option<Template>,
 
     /// Tags to apply to the metric.
+    ///
+    /// Both keys and values are templateable, which enables you to attach dynamic labels to events.
+    ///
     #[configurable(metadata(docs::additional_props_description = "A metric tag."))]
     pub tags: Option<IndexMap<Template, TagConfig>>,
 

From ba98fb2b3448ec118ab8ca31986e00ba67e27525 Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Wed, 4 Dec 2024 01:13:25 +0800
Subject: [PATCH 08/26] Add change log

---
 changelog.d/21939_log_to_metric_tags_expansion.features.md | 3 +++
 1 file changed, 3 insertions(+)
 create mode 100644 changelog.d/21939_log_to_metric_tags_expansion.features.md

diff --git a/changelog.d/21939_log_to_metric_tags_expansion.features.md b/changelog.d/21939_log_to_metric_tags_expansion.features.md
new file mode 100644
index 0000000000000..bca187fa1bf09
--- /dev/null
+++ b/changelog.d/21939_log_to_metric_tags_expansion.features.md
@@ -0,0 +1,3 @@
+The `log_to_metric` transformer now support tags expansion similar to Loki sink's labels expansion.
+
+authors: titaneric

From c4bba1be5a81cafb5499ca1a64657ae831d9f388 Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Wed, 4 Dec 2024 01:29:15 +0800
Subject: [PATCH 09/26] Rename changelog file

---
 ....features.md => 21939_log_to_metric_tags_expansion.feature.md} | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 rename changelog.d/{21939_log_to_metric_tags_expansion.features.md => 21939_log_to_metric_tags_expansion.feature.md} (100%)

diff --git a/changelog.d/21939_log_to_metric_tags_expansion.features.md b/changelog.d/21939_log_to_metric_tags_expansion.feature.md
similarity index 100%
rename from changelog.d/21939_log_to_metric_tags_expansion.features.md
rename to changelog.d/21939_log_to_metric_tags_expansion.feature.md

From 8ced9a8332c1ee717ea494183c17ec602c20b8ed Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Fri, 6 Dec 2024 00:51:16 +0800
Subject: [PATCH 10/26] rename labels to tags and inset the tag value in
 render_tag_into

---
 src/transforms/log_to_metric.rs | 97 +++++++++++++++++++--------------
 1 file changed, 55 insertions(+), 42 deletions(-)

diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index 032cc4d248f16..aed611ab5c93c 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -278,22 +278,24 @@ fn render_tags(
     tags: &Option<IndexMap<Template, TagConfig>>,
     event: &Event,
 ) -> Result<Option<MetricTags>, TransformError> {
-    let mut static_labels: HashMap<String, String> = HashMap::new();
-    let mut dynamic_labels: HashMap<String, String> = HashMap::new();
+    let mut static_tags: HashMap<String, TagValue> = HashMap::new();
+    let mut dynamic_tags: HashMap<String, TagValue> = HashMap::new();
     Ok(match tags {
         None => None,
         Some(tags) => {
             let mut result = MetricTags::default();
             for (name, config) in tags {
                 match config {
-                    TagConfig::Plain(template) => render_tag_into(
-                        event,
-                        name,
-                        template,
-                        &mut result,
-                        &mut static_labels,
-                        &mut dynamic_labels,
-                    )?,
+                    TagConfig::Plain(template) => {
+                        render_tag_into(
+                            event,
+                            name,
+                            template,
+                            &mut result,
+                            &mut static_tags,
+                            &mut dynamic_tags,
+                        )?;
+                    }
                     TagConfig::Multi(vec) => {
                         for template in vec {
                             render_tag_into(
@@ -301,25 +303,24 @@ fn render_tags(
                                 name,
                                 template,
                                 &mut result,
-                                &mut static_labels,
-                                &mut dynamic_labels,
+                                &mut static_tags,
+                                &mut dynamic_tags,
                             )?;
                         }
                     }
                 }
             }
-            for (k, v) in static_labels {
-                if let Some(discarded_v) = dynamic_labels.insert(k.clone(), v.clone()) {
+            // println!("{:?}", static_tags);
+            for (k, v) in static_tags {
+                if let Some(discarded_v) = dynamic_tags.insert(k.clone(), v.clone()) {
                     warn!(
-                        "Static label overrides dynamic label. \
-                key: {}, value: {}, discarded value: {}",
+                        "Static static overrides dynamic tag. \
+                key: {}, value: {:?}, discarded value: {:?}",
                         k, v, discarded_v
                     );
                 };
             }
-            for (name, value) in dynamic_labels {
-                result.insert(name.to_string(), TagValue::Value(value));
-            }
+            // println!("result is {:?}", result);
             result.as_option()
         }
     })
@@ -330,27 +331,34 @@ fn render_tag_into(
     key_template: &Template,
     value_template: &Option<Template>,
     result: &mut MetricTags,
-    static_labels: &mut HashMap<String, String>,
-    dynamic_labels: &mut HashMap<String, String>,
+    static_tags: &mut HashMap<String, TagValue>,
+    dynamic_tags: &mut HashMap<String, TagValue>,
 ) -> Result<(), TransformError> {
-    let key = render_template(key_template, event);
+    // println!(
+    //     "key_template: {:?}, value_template: {:?}",
+    //     key_template, value_template
+    // );
+    let key_s = match render_template(key_template, event) {
+        Ok(key_s) => key_s,
+        Err(TransformError::TemplateRenderingError(err)) => {
+            emit!(crate::internal_events::TemplateRenderingError {
+                error: err,
+                drop_event: false,
+                field: Some(key_template.get_ref()),
+            });
+            return Ok(());
+        }
+        Err(err) => return Err(err),
+    };
     match value_template {
         None => {
-            result.insert(key_template.to_string(), TagValue::Bare);
+            result.insert(key_s, TagValue::Bare);
         }
-        Some(template) => match (key, render_template(template, event)) {
-            (Ok(key_s), Ok(value_s)) => {
-                tag_expansion(key_s, value_s, static_labels, dynamic_labels);
-            }
-            (Err(TransformError::TemplateRenderingError(key_error)), _) => {
-                emit!(crate::internal_events::TemplateRenderingError {
-                    error: key_error,
-                    drop_event: false,
-                    field: Some(key_template.get_ref()),
-                });
-                return Ok(());
+        Some(template) => match render_template(template, event) {
+            Ok(value_s) => {
+                tag_expansion(key_s.clone(), value_s, result, static_tags, dynamic_tags);
             }
-            (_, Err(TransformError::TemplateRenderingError(value_error))) => {
+            Err(TransformError::TemplateRenderingError(value_error)) => {
                 emit!(crate::internal_events::TemplateRenderingError {
                     error: value_error,
                     drop_event: false,
@@ -358,8 +366,7 @@ fn render_tag_into(
                 });
                 return Ok(());
             }
-            (Err(key_error), _) => return Err(key_error),
-            (_, Err(value_error)) => return Err(value_error),
+            Err(err) => return Err(err),
         },
     };
     Ok(())
@@ -374,9 +381,11 @@ fn slugify_text(input: String) -> String {
 fn tag_expansion(
     key_s: String,
     value_s: String,
-    static_labels: &mut HashMap<String, String>,
-    dynamic_labels: &mut HashMap<String, String>,
+    result: &mut MetricTags,
+    static_tags: &mut HashMap<String, TagValue>,
+    dynamic_tags: &mut HashMap<String, TagValue>,
 ) {
+    // println!("key_s: {:?}, value_s: {:?}", key_s, value_s);
     if let Some(opening_prefix) = key_s.strip_suffix('*') {
         let output: Result<serde_json::map::Map<String, serde_json::Value>, serde_json::Error> =
             serde_json::from_str(value_s.clone().as_str());
@@ -399,16 +408,20 @@ fn tag_expansion(
                 warn!("Encountered \"null\" value for dynamic label. key: {}", key);
                 continue;
             }
-            if let Some(prev) = dynamic_labels.insert(key.clone(), val.clone()) {
+            let val = TagValue::Value(val);
+            if let Some(prev) = dynamic_tags.insert(key.clone(), val.clone()) {
                 warn!(
                     "Encountered duplicated dynamic label. \
-                                key: {}, value: {}, discarded value: {}",
+                                key: {}, value: {:?}, discarded value: {:?}",
                     key, val, prev
                 );
             };
+            result.insert(key, val);
         }
     } else {
-        static_labels.insert(key_s, value_s);
+        let val = TagValue::Value(value_s);
+        result.insert(key_s.clone(), val.clone());
+        static_tags.insert(key_s, val);
     }
 }
 

From 3884f83145cfb4ab7aa92cca7534faea27147e7d Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Fri, 6 Dec 2024 01:10:16 +0800
Subject: [PATCH 11/26] Remove debug msg

---
 src/transforms/log_to_metric.rs | 7 -------
 1 file changed, 7 deletions(-)

diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index aed611ab5c93c..7b04ce427ad7b 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -310,7 +310,6 @@ fn render_tags(
                     }
                 }
             }
-            // println!("{:?}", static_tags);
             for (k, v) in static_tags {
                 if let Some(discarded_v) = dynamic_tags.insert(k.clone(), v.clone()) {
                     warn!(
@@ -320,7 +319,6 @@ fn render_tags(
                     );
                 };
             }
-            // println!("result is {:?}", result);
             result.as_option()
         }
     })
@@ -334,10 +332,6 @@ fn render_tag_into(
     static_tags: &mut HashMap<String, TagValue>,
     dynamic_tags: &mut HashMap<String, TagValue>,
 ) -> Result<(), TransformError> {
-    // println!(
-    //     "key_template: {:?}, value_template: {:?}",
-    //     key_template, value_template
-    // );
     let key_s = match render_template(key_template, event) {
         Ok(key_s) => key_s,
         Err(TransformError::TemplateRenderingError(err)) => {
@@ -385,7 +379,6 @@ fn tag_expansion(
     static_tags: &mut HashMap<String, TagValue>,
     dynamic_tags: &mut HashMap<String, TagValue>,
 ) {
-    // println!("key_s: {:?}, value_s: {:?}", key_s, value_s);
     if let Some(opening_prefix) = key_s.strip_suffix('*') {
         let output: Result<serde_json::map::Map<String, serde_json::Value>, serde_json::Error> =
             serde_json::from_str(value_s.clone().as_str());

From 21a532c072c5444d5b4e7950001624b2c36551ab Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Fri, 6 Dec 2024 01:52:12 +0800
Subject: [PATCH 12/26] Add colliding tags test

---
 src/transforms/log_to_metric.rs | 53 ++++++++++++++++++++++++++++++++-
 1 file changed, 52 insertions(+), 1 deletion(-)

diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index 7b04ce427ad7b..d16aa1ea76960 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -1180,7 +1180,7 @@ mod tests {
             field = "message"
             name = "http_requests_total"
             namespace = "app"
-            tags = {"*" = "{{dict}}"}
+            tags = {"*" = "{{ dict }}"}
             "#,
         );
 
@@ -1224,7 +1224,58 @@ mod tests {
             .with_timestamp(Some(ts()))
         );
     }
+    #[tokio::test]
+    async fn count_http_requests_with_colliding_dynamic_tags() {
+        let config = parse_config(
+            r#"
+            [[metrics]]
+            type = "counter"
+            field = "message"
+            name = "http_requests_total"
+            namespace = "app"
+            tags = {"l1_*" = "{{ map1 }}", "*" = "{{ map2 }}"}
+            "#,
+        );
+
+        let mut event = create_event("message", "i am log");
+        let log = event.as_mut_log();
+
+        let mut map1 = ObjectMap::default();
+        map1.insert("key1".into(), Value::from("val1"));
+        log.insert("map1", Value::from(map1));
+
+        let mut map2 = ObjectMap::default();
+        map2.insert("l1_key1".into(), Value::from("val2"));
+        log.insert("map2", Value::from(map2));
+
+        let mut metadata =
+            event
+                .metadata()
+                .clone()
+                .with_origin_metadata(DatadogMetricOriginMetadata::new(
+                    None,
+                    None,
+                    Some(ORIGIN_SERVICE_VALUE),
+                ));
+        // definitions aren't valid for metrics yet, it's just set to the default (anything).
+        metadata.set_schema_definition(&Arc::new(Definition::any()));
+        metadata.set_upstream_id(Arc::new(OutputId::from("transform")));
+        metadata.set_source_id(Arc::new(ComponentKey::from("in")));
+
+        let metric = do_transform(config, event).await.unwrap().into_metric();
+        let tags = metric.tags().expect("Metric should have tags");
+
+        assert_eq!(
+            tags.iter_single().collect::<Vec<_>>(),
+            vec![("l1_key1", "val2")]
+        );
 
+        assert_eq!(tags.iter_all().count(), 2);
+        for (name, value) in tags.iter_all() {
+            assert_eq!(name, "l1_key1");
+            assert!(value == Some("val1") || value == Some("val2"));
+        }
+    }
     #[tokio::test]
     async fn multi_value_tags_yaml() {
         // Have to use YAML to represent bare tags

From 9bf3219e6e649d837a181f7cce13e90c83609685 Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Fri, 6 Dec 2024 10:38:59 +0800
Subject: [PATCH 13/26] Add source reference

---
 src/transforms/log_to_metric.rs | 1 +
 1 file changed, 1 insertion(+)

diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index d16aa1ea76960..8f8fb8973ddb5 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -372,6 +372,7 @@ fn slugify_text(input: String) -> String {
     result.to_lowercase()
 }
 
+// borrow from src/sinks/loki/sink.rs
 fn tag_expansion(
     key_s: String,
     value_s: String,

From 5957703f0dc78631c40eaa5152970e152e1f37fd Mon Sep 17 00:00:00 2001
From: titaneric <chenyihuang001@gmail.com>
Date: Sun, 22 Dec 2024 14:40:18 +0900
Subject: [PATCH 14/26] fix comment and changelog

---
 changelog.d/21939_log_to_metric_tags_expansion.feature.md | 2 +-
 src/transforms/log_to_metric.rs                           | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/changelog.d/21939_log_to_metric_tags_expansion.feature.md b/changelog.d/21939_log_to_metric_tags_expansion.feature.md
index bca187fa1bf09..d6223975ae7e9 100644
--- a/changelog.d/21939_log_to_metric_tags_expansion.feature.md
+++ b/changelog.d/21939_log_to_metric_tags_expansion.feature.md
@@ -1,3 +1,3 @@
-The `log_to_metric` transformer now support tags expansion similar to Loki sink's labels expansion.
+The `log_to_metric` transformer tag key are now template-able which enables tags expansion.
 
 authors: titaneric
diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index 8f8fb8973ddb5..db6cedefa7dc4 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -313,7 +313,7 @@ fn render_tags(
             for (k, v) in static_tags {
                 if let Some(discarded_v) = dynamic_tags.insert(k.clone(), v.clone()) {
                     warn!(
-                        "Static static overrides dynamic tag. \
+                        "Static tags overrides dynamic tags. \
                 key: {}, value: {:?}, discarded value: {:?}",
                         k, v, discarded_v
                     );

From c23fb3177fcbf61758cbf6fb10eb72dd163edf73 Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Mon, 23 Dec 2024 01:58:36 +0800
Subject: [PATCH 15/26] Extract pair expansion to common mod

---
 src/common/expansion.rs         | 54 ++++++++++++++++++++++
 src/common/mod.rs               |  5 ++-
 src/transforms/log_to_metric.rs | 80 ++++++++-------------------------
 3 files changed, 77 insertions(+), 62 deletions(-)
 create mode 100644 src/common/expansion.rs

diff --git a/src/common/expansion.rs b/src/common/expansion.rs
new file mode 100644
index 0000000000000..46f3acf1be517
--- /dev/null
+++ b/src/common/expansion.rs
@@ -0,0 +1,54 @@
+use regex::Regex;
+use std::{collections::HashMap, sync::LazyLock};
+
+use crate::event::Value;
+
+static RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"[^0-9A-Za-z_]").unwrap());
+fn slugify_text(input: String) -> String {
+    let result = RE.replace_all(&input, "_");
+    result.to_lowercase()
+}
+
+pub(crate) fn pair_expansion(
+    key_s: String,
+    value_s: String,
+    static_pairs: &mut HashMap<String, String>,
+    dynamic_pairs: &mut HashMap<String, String>,
+) -> Result<HashMap<String, String>, serde_json::Error> {
+    let mut expanded_pairs = HashMap::new();
+    if let Some(opening_prefix) = key_s.strip_suffix('*') {
+        let output: Result<serde_json::map::Map<String, serde_json::Value>, serde_json::Error> =
+            serde_json::from_str(value_s.clone().as_str());
+
+        if let Err(err) = output {
+            warn!(
+                "Failed to expand dynamic pair. value: {}, err: {}",
+                value_s, err
+            );
+            return Err(err);
+        }
+
+        // key_* -> key_one, key_two, key_three
+        // * -> one, two, three
+        for (k, v) in output.unwrap() {
+            let key = slugify_text(format!("{}{}", opening_prefix, k));
+            let val = Value::from(v).to_string_lossy().into_owned();
+            if val == "<null>" {
+                warn!("Encountered \"null\" value for dynamic pair. key: {}", key);
+                continue;
+            }
+            if let Some(prev) = dynamic_pairs.insert(key.clone(), val.clone()) {
+                warn!(
+                    "Encountered duplicated dynamic pair. \
+                                key: {}, value: {:?}, discarded value: {:?}",
+                    key, val, prev
+                );
+            };
+            expanded_pairs.insert(key, val);
+        }
+    } else {
+        static_pairs.insert(key_s.clone(), value_s.clone());
+        expanded_pairs.insert(key_s, value_s);
+    }
+    Ok(expanded_pairs)
+}
diff --git a/src/common/mod.rs b/src/common/mod.rs
index dad3e63de35e4..7b66ca1c2bec3 100644
--- a/src/common/mod.rs
+++ b/src/common/mod.rs
@@ -1,4 +1,4 @@
-//! Modules that are common between sources and sinks.
+//! Modules that are common between sources, transforms, and sinks.
 #[cfg(any(
     feature = "sources-datadog_agent",
     feature = "sinks-datadog_events",
@@ -17,3 +17,6 @@ pub(crate) mod sqs;
 
 #[cfg(any(feature = "sources-aws_s3", feature = "sinks-aws_s3"))]
 pub(crate) mod s3;
+
+#[cfg(any(feature = "transforms-log_to_metric"))]
+pub(crate) mod expansion;
diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index db6cedefa7dc4..d03c93f5bb222 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -1,9 +1,8 @@
 use std::sync::Arc;
-use std::{collections::HashMap, num::ParseFloatError, sync::LazyLock};
+use std::{collections::HashMap, num::ParseFloatError};
 
 use chrono::Utc;
 use indexmap::IndexMap;
-use regex::Regex;
 use vector_lib::configurable::configurable_component;
 use vector_lib::event::LogEvent;
 use vector_lib::{
@@ -20,6 +19,7 @@ use vrl::{event_path, path};
 use crate::config::schema::Definition;
 use crate::transforms::log_to_metric::TransformError::PathNotFound;
 use crate::{
+    common::expansion::pair_expansion,
     config::{
         DataType, GenerateConfig, Input, OutputId, TransformConfig, TransformContext,
         TransformOutput,
@@ -111,7 +111,7 @@ pub struct MetricConfig {
 
     /// Tags to apply to the metric.
     ///
-    /// Both keys and values are templateable, which enables you to attach dynamic labels to events.
+    /// Both keys and values are templateable, which enables you to attach dynamic tags to events.
     ///
     #[configurable(metadata(docs::additional_props_description = "A metric tag."))]
     pub tags: Option<IndexMap<Template, TagConfig>>,
@@ -278,8 +278,8 @@ fn render_tags(
     tags: &Option<IndexMap<Template, TagConfig>>,
     event: &Event,
 ) -> Result<Option<MetricTags>, TransformError> {
-    let mut static_tags: HashMap<String, TagValue> = HashMap::new();
-    let mut dynamic_tags: HashMap<String, TagValue> = HashMap::new();
+    let mut static_tags: HashMap<String, String> = HashMap::new();
+    let mut dynamic_tags: HashMap<String, String> = HashMap::new();
     Ok(match tags {
         None => None,
         Some(tags) => {
@@ -329,8 +329,8 @@ fn render_tag_into(
     key_template: &Template,
     value_template: &Option<Template>,
     result: &mut MetricTags,
-    static_tags: &mut HashMap<String, TagValue>,
-    dynamic_tags: &mut HashMap<String, TagValue>,
+    static_tags: &mut HashMap<String, String>,
+    dynamic_tags: &mut HashMap<String, String>,
 ) -> Result<(), TransformError> {
     let key_s = match render_template(key_template, event) {
         Ok(key_s) => key_s,
@@ -350,7 +350,18 @@ fn render_tag_into(
         }
         Some(template) => match render_template(template, event) {
             Ok(value_s) => {
-                tag_expansion(key_s.clone(), value_s, result, static_tags, dynamic_tags);
+                let expanded_pairs = pair_expansion(key_s, value_s, static_tags, dynamic_tags)
+                    .map_err(|_| {
+                        TransformError::TemplateRenderingError(
+                            TemplateRenderingError::MissingKeys {
+                                missing_keys: vec![],
+                            },
+                        )
+                    })?;
+
+                for (key, value) in expanded_pairs {
+                    result.insert(key, TagValue::Value(value));
+                }
             }
             Err(TransformError::TemplateRenderingError(value_error)) => {
                 emit!(crate::internal_events::TemplateRenderingError {
@@ -366,59 +377,6 @@ fn render_tag_into(
     Ok(())
 }
 
-static RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"[^0-9A-Za-z_]").unwrap());
-fn slugify_text(input: String) -> String {
-    let result = RE.replace_all(&input, "_");
-    result.to_lowercase()
-}
-
-// borrow from src/sinks/loki/sink.rs
-fn tag_expansion(
-    key_s: String,
-    value_s: String,
-    result: &mut MetricTags,
-    static_tags: &mut HashMap<String, TagValue>,
-    dynamic_tags: &mut HashMap<String, TagValue>,
-) {
-    if let Some(opening_prefix) = key_s.strip_suffix('*') {
-        let output: Result<serde_json::map::Map<String, serde_json::Value>, serde_json::Error> =
-            serde_json::from_str(value_s.clone().as_str());
-
-        if output.is_err() {
-            warn!(
-                "Failed to expand dynamic label. value: {}, err: {}",
-                value_s,
-                output.err().unwrap()
-            );
-            return;
-        }
-
-        // key_* -> key_one, key_two, key_three
-        // * -> one, two, three
-        for (k, v) in output.unwrap() {
-            let key = slugify_text(format!("{}{}", opening_prefix, k));
-            let val = Value::from(v).to_string_lossy().into_owned();
-            if val == "<null>" {
-                warn!("Encountered \"null\" value for dynamic label. key: {}", key);
-                continue;
-            }
-            let val = TagValue::Value(val);
-            if let Some(prev) = dynamic_tags.insert(key.clone(), val.clone()) {
-                warn!(
-                    "Encountered duplicated dynamic label. \
-                                key: {}, value: {:?}, discarded value: {:?}",
-                    key, val, prev
-                );
-            };
-            result.insert(key, val);
-        }
-    } else {
-        let val = TagValue::Value(value_s);
-        result.insert(key_s.clone(), val.clone());
-        static_tags.insert(key_s, val);
-    }
-}
-
 fn to_metric_with_config(config: &MetricConfig, event: &Event) -> Result<Metric, TransformError> {
     let log = event.as_log();
 

From 3a10f42fbaceb1be633c90a26a0090eee534e2d8 Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Mon, 23 Dec 2024 02:21:04 +0800
Subject: [PATCH 16/26] Adopt pair expansion in loki sink

---
 src/common/mod.rs      |  2 +-
 src/sinks/loki/sink.rs | 94 ++++--------------------------------------
 2 files changed, 10 insertions(+), 86 deletions(-)

diff --git a/src/common/mod.rs b/src/common/mod.rs
index 7b66ca1c2bec3..6e29cbeebeadf 100644
--- a/src/common/mod.rs
+++ b/src/common/mod.rs
@@ -18,5 +18,5 @@ pub(crate) mod sqs;
 #[cfg(any(feature = "sources-aws_s3", feature = "sinks-aws_s3"))]
 pub(crate) mod s3;
 
-#[cfg(any(feature = "transforms-log_to_metric"))]
+#[cfg(any(feature = "transforms-log_to_metric", feature = "sinks-loki"))]
 pub(crate) mod expansion;
diff --git a/src/sinks/loki/sink.rs b/src/sinks/loki/sink.rs
index f12dbc3b09e4a..64376701bee81 100644
--- a/src/sinks/loki/sink.rs
+++ b/src/sinks/loki/sink.rs
@@ -1,7 +1,6 @@
-use std::{collections::HashMap, num::NonZeroUsize, sync::LazyLock};
+use std::{collections::HashMap, num::NonZeroUsize};
 
 use bytes::{Bytes, BytesMut};
-use regex::Regex;
 use snafu::Snafu;
 use tokio_util::codec::Encoder as _;
 use vrl::path::parse_target_path;
@@ -13,6 +12,7 @@ use super::{
 };
 use crate::sinks::loki::event::LokiBatchEncoding;
 use crate::{
+    common::expansion::pair_expansion,
     http::{get_http_scheme_from_uri, HttpClient},
     internal_events::{
         LokiEventUnlabeledError, LokiOutOfOrderEventDroppedError, LokiOutOfOrderEventRewritten,
@@ -179,42 +179,7 @@ impl EventEncoder {
 
             let key_s = key.unwrap();
             let value_s = value.unwrap();
-
-            if let Some(opening_prefix) = key_s.strip_suffix('*') {
-                let output: Result<
-                    serde_json::map::Map<String, serde_json::Value>,
-                    serde_json::Error,
-                > = serde_json::from_str(value_s.clone().as_str());
-
-                if output.is_err() {
-                    warn!(
-                        "Failed to expand dynamic label. value: {}, err: {}",
-                        value_s,
-                        output.err().unwrap()
-                    );
-                    continue;
-                }
-
-                // key_* -> key_one, key_two, key_three
-                // * -> one, two, three
-                for (k, v) in output.unwrap() {
-                    let key = slugify_text(format!("{}{}", opening_prefix, k));
-                    let val = Value::from(v).to_string_lossy().into_owned();
-                    if val == "<null>" {
-                        warn!("Encountered \"null\" value for dynamic label. key: {}", key);
-                        continue;
-                    }
-                    if let Some(prev) = dynamic_labels.insert(key.clone(), val.clone()) {
-                        warn!(
-                            "Encountered duplicated dynamic label. \
-                                key: {}, value: {}, discarded value: {}",
-                            key, val, prev
-                        );
-                    };
-                }
-            } else {
-                static_labels.insert(key_s, value_s);
-            }
+            let _ = pair_expansion(key_s, value_s, &mut static_labels, &mut dynamic_labels);
         }
 
         for (k, v) in static_labels {
@@ -284,46 +249,12 @@ impl EventEncoder {
 
             let key_s = key.unwrap();
             let value_s = value.unwrap();
-
-            if let Some(opening_prefix) = key_s.strip_suffix('*') {
-                let output: Result<
-                    serde_json::map::Map<String, serde_json::Value>,
-                    serde_json::Error,
-                > = serde_json::from_str(value_s.clone().as_str());
-
-                if output.is_err() {
-                    warn!(
-                        "Failed to expand dynamic structured metadata. value: {}, err: {}",
-                        value_s,
-                        output.err().unwrap()
-                    );
-                    continue;
-                }
-
-                // key_* -> key_one, key_two, key_three
-                // * -> one, two, three
-                for (k, v) in output.unwrap() {
-                    let key = slugify_text(format!("{}{}", opening_prefix, k));
-                    let val = Value::from(v).to_string_lossy().into_owned();
-                    if val == "<null>" {
-                        warn!(
-                            "Encountered \"null\" value for dynamic structured_metadata. key: {}",
-                            key
-                        );
-                        continue;
-                    }
-                    if let Some(prev) = dynamic_structured_metadata.insert(key.clone(), val.clone())
-                    {
-                        warn!(
-                            "Encountered duplicated dynamic structured_metadata. \
-                        key: {}, value: {}, discarded value: {}",
-                            key, val, prev
-                        );
-                    };
-                }
-            } else {
-                static_structured_metadata.insert(key_s, value_s);
-            }
+            let _ = pair_expansion(
+                key_s,
+                value_s,
+                &mut static_structured_metadata,
+                &mut dynamic_structured_metadata,
+            );
         }
 
         for (k, v) in static_structured_metadata {
@@ -620,13 +551,6 @@ impl StreamSink<Event> for LokiSink {
     }
 }
 
-static RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"[^0-9A-Za-z_]").unwrap());
-
-fn slugify_text(input: String) -> String {
-    let result = RE.replace_all(&input, "_");
-    result.to_lowercase()
-}
-
 #[cfg(test)]
 mod tests {
     use std::{collections::HashMap, convert::TryFrom};

From f12e9e20a41b4b92e34fc6f9dd5390b3f7bf396e Mon Sep 17 00:00:00 2001
From: titaneric <chenyihuang001@gmail.com>
Date: Mon, 23 Dec 2024 11:36:45 +0900
Subject: [PATCH 17/26] Generate component docs

---
 .../components/transforms/base/log_to_metric.cue          | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/website/cue/reference/components/transforms/base/log_to_metric.cue b/website/cue/reference/components/transforms/base/log_to_metric.cue
index 24ff4a3bc4bc1..4b2dfaccb1323 100644
--- a/website/cue/reference/components/transforms/base/log_to_metric.cue
+++ b/website/cue/reference/components/transforms/base/log_to_metric.cue
@@ -85,8 +85,12 @@ base: components: transforms: log_to_metric: configuration: {
 				type: string: syntax: "template"
 			}
 			tags: {
-				description: "Tags to apply to the metric."
-				required:    false
+				description: """
+					Tags to apply to the metric.
+
+					Both keys and values are templateable, which enables you to attach dynamic tags to events.
+					"""
+				required: false
 				type: object: options: "*": {
 					description: "A metric tag."
 					required:    true

From 057456d8e4eededd5d0c92413b533bac972fceb0 Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Mon, 23 Dec 2024 12:24:39 +0800
Subject: [PATCH 18/26] Add PairExpansion Error

---
 src/transforms/log_to_metric.rs | 9 ++-------
 1 file changed, 2 insertions(+), 7 deletions(-)

diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index d03c93f5bb222..44ece4ad550eb 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -266,6 +266,7 @@ enum TransformError {
         error: ParseFloatError,
     },
     TemplateRenderingError(TemplateRenderingError),
+    PairExpansionError,
 }
 
 fn render_template(template: &Template, event: &Event) -> Result<String, TransformError> {
@@ -351,13 +352,7 @@ fn render_tag_into(
         Some(template) => match render_template(template, event) {
             Ok(value_s) => {
                 let expanded_pairs = pair_expansion(key_s, value_s, static_tags, dynamic_tags)
-                    .map_err(|_| {
-                        TransformError::TemplateRenderingError(
-                            TemplateRenderingError::MissingKeys {
-                                missing_keys: vec![],
-                            },
-                        )
-                    })?;
+                    .map_err(|_| TransformError::PairExpansionError)?;
 
                 for (key, value) in expanded_pairs {
                     result.insert(key, TagValue::Value(value));

From 6af30a48d7ffaf0865b6dd91731b561ca9f16f98 Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Mon, 23 Dec 2024 12:39:36 +0800
Subject: [PATCH 19/26] Extend the result with extended pairs

---
 src/transforms/log_to_metric.rs | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index 44ece4ad550eb..7bd74c82415ca 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -353,10 +353,7 @@ fn render_tag_into(
             Ok(value_s) => {
                 let expanded_pairs = pair_expansion(key_s, value_s, static_tags, dynamic_tags)
                     .map_err(|_| TransformError::PairExpansionError)?;
-
-                for (key, value) in expanded_pairs {
-                    result.insert(key, TagValue::Value(value));
-                }
+                result.extend(expanded_pairs);
             }
             Err(TransformError::TemplateRenderingError(value_error)) => {
                 emit!(crate::internal_events::TemplateRenderingError {

From c337c041b576d4ab472fd9874849b0b82a68e23e Mon Sep 17 00:00:00 2001
From: "eric.cy.huang" <eric.cy.huang@linecorp.com>
Date: Tue, 24 Dec 2024 01:46:25 +0800
Subject: [PATCH 20/26] Update docs

---
 src/transforms/log_to_metric.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index 7bd74c82415ca..74af687824103 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -111,7 +111,7 @@ pub struct MetricConfig {
 
     /// Tags to apply to the metric.
     ///
-    /// Both keys and values are templateable, which enables you to attach dynamic tags to events.
+    /// Both keys and values can be templated, allowing you to attach dynamic tags to events.
     ///
     #[configurable(metadata(docs::additional_props_description = "A metric tag."))]
     pub tags: Option<IndexMap<Template, TagConfig>>,

From fbdebbb6b7a3ca309a7e28636fdd5886a4926f89 Mon Sep 17 00:00:00 2001
From: titaneric <chenyihuang001@gmail.com>
Date: Tue, 24 Dec 2024 02:56:11 +0900
Subject: [PATCH 21/26] Re-generate docs

---
 .../cue/reference/components/transforms/base/log_to_metric.cue  | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/website/cue/reference/components/transforms/base/log_to_metric.cue b/website/cue/reference/components/transforms/base/log_to_metric.cue
index 4b2dfaccb1323..6275706f44869 100644
--- a/website/cue/reference/components/transforms/base/log_to_metric.cue
+++ b/website/cue/reference/components/transforms/base/log_to_metric.cue
@@ -88,7 +88,7 @@ base: components: transforms: log_to_metric: configuration: {
 				description: """
 					Tags to apply to the metric.
 
-					Both keys and values are templateable, which enables you to attach dynamic tags to events.
+					Both keys and values can be templated, allowing you to attach dynamic tags to events.
 					"""
 				required: false
 				type: object: options: "*": {

From f08e7ffaed8eaf4fa7394e42aefa3f4a8e010877 Mon Sep 17 00:00:00 2001
From: "chenyee.cy.huang" <eric.cy.huang@linecorp.com>
Date: Sat, 4 Jan 2025 12:37:54 +0800
Subject: [PATCH 22/26] Add docs for pair_expansion and change signature to
 string literal

---
 src/common/expansion.rs         | 17 ++++++++++-------
 src/sinks/loki/sink.rs          |  6 +++---
 src/transforms/log_to_metric.rs |  2 +-
 3 files changed, 14 insertions(+), 11 deletions(-)

diff --git a/src/common/expansion.rs b/src/common/expansion.rs
index 46f3acf1be517..07e241b0a1405 100644
--- a/src/common/expansion.rs
+++ b/src/common/expansion.rs
@@ -4,21 +4,24 @@ use std::{collections::HashMap, sync::LazyLock};
 use crate::event::Value;
 
 static RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"[^0-9A-Za-z_]").unwrap());
-fn slugify_text(input: String) -> String {
+fn slugify_text(input: &str) -> String {
     let result = RE.replace_all(&input, "_");
     result.to_lowercase()
 }
 
+/// Expands the given possibly template-able `key_s` and `value_s`, and return the expanded owned pairs
+/// it would also insert the pairs into either `static_pairs` or `dynamic_pairs` depending on the template-ability of `key_s`.
+/// Refer to loki sink and log_to_metric transform for further information.
 pub(crate) fn pair_expansion(
-    key_s: String,
-    value_s: String,
+    key_s: &str,
+    value_s: &str,
     static_pairs: &mut HashMap<String, String>,
     dynamic_pairs: &mut HashMap<String, String>,
 ) -> Result<HashMap<String, String>, serde_json::Error> {
     let mut expanded_pairs = HashMap::new();
     if let Some(opening_prefix) = key_s.strip_suffix('*') {
         let output: Result<serde_json::map::Map<String, serde_json::Value>, serde_json::Error> =
-            serde_json::from_str(value_s.clone().as_str());
+            serde_json::from_str(value_s);
 
         if let Err(err) = output {
             warn!(
@@ -31,7 +34,7 @@ pub(crate) fn pair_expansion(
         // key_* -> key_one, key_two, key_three
         // * -> one, two, three
         for (k, v) in output.unwrap() {
-            let key = slugify_text(format!("{}{}", opening_prefix, k));
+            let key = slugify_text(&format!("{}{}", opening_prefix, k));
             let val = Value::from(v).to_string_lossy().into_owned();
             if val == "<null>" {
                 warn!("Encountered \"null\" value for dynamic pair. key: {}", key);
@@ -47,8 +50,8 @@ pub(crate) fn pair_expansion(
             expanded_pairs.insert(key, val);
         }
     } else {
-        static_pairs.insert(key_s.clone(), value_s.clone());
-        expanded_pairs.insert(key_s, value_s);
+        static_pairs.insert(key_s.to_string(), value_s.to_string());
+        expanded_pairs.insert(key_s.to_string(), value_s.to_string());
     }
     Ok(expanded_pairs)
 }
diff --git a/src/sinks/loki/sink.rs b/src/sinks/loki/sink.rs
index 64376701bee81..a3765b6f5723f 100644
--- a/src/sinks/loki/sink.rs
+++ b/src/sinks/loki/sink.rs
@@ -179,7 +179,7 @@ impl EventEncoder {
 
             let key_s = key.unwrap();
             let value_s = value.unwrap();
-            let _ = pair_expansion(key_s, value_s, &mut static_labels, &mut dynamic_labels);
+            let _ = pair_expansion(&key_s, &value_s, &mut static_labels, &mut dynamic_labels);
         }
 
         for (k, v) in static_labels {
@@ -250,8 +250,8 @@ impl EventEncoder {
             let key_s = key.unwrap();
             let value_s = value.unwrap();
             let _ = pair_expansion(
-                key_s,
-                value_s,
+                &key_s,
+                &value_s,
                 &mut static_structured_metadata,
                 &mut dynamic_structured_metadata,
             );
diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index 74af687824103..8608b7530d5b1 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -351,7 +351,7 @@ fn render_tag_into(
         }
         Some(template) => match render_template(template, event) {
             Ok(value_s) => {
-                let expanded_pairs = pair_expansion(key_s, value_s, static_tags, dynamic_tags)
+                let expanded_pairs = pair_expansion(&key_s, &value_s, static_tags, dynamic_tags)
                     .map_err(|_| TransformError::PairExpansionError)?;
                 result.extend(expanded_pairs);
             }

From 97a87568a5afe777fdc84ecf27f05a98e8fc4dd1 Mon Sep 17 00:00:00 2001
From: "chenyee.cy.huang" <eric.cy.huang@linecorp.com>
Date: Sat, 4 Jan 2025 12:48:18 +0800
Subject: [PATCH 23/26] Update signature for render_tag_into

---
 src/transforms/log_to_metric.rs | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index 8608b7530d5b1..48cfbecfeb501 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -291,7 +291,7 @@ fn render_tags(
                         render_tag_into(
                             event,
                             name,
-                            template,
+                            template.as_ref(),
                             &mut result,
                             &mut static_tags,
                             &mut dynamic_tags,
@@ -302,7 +302,7 @@ fn render_tags(
                             render_tag_into(
                                 event,
                                 name,
-                                template,
+                                template.as_ref(),
                                 &mut result,
                                 &mut static_tags,
                                 &mut dynamic_tags,
@@ -328,7 +328,7 @@ fn render_tags(
 fn render_tag_into(
     event: &Event,
     key_template: &Template,
-    value_template: &Option<Template>,
+    value_template: Option<&Template>,
     result: &mut MetricTags,
     static_tags: &mut HashMap<String, String>,
     dynamic_tags: &mut HashMap<String, String>,
@@ -363,7 +363,7 @@ fn render_tag_into(
                 });
                 return Ok(());
             }
-            Err(err) => return Err(err),
+            Err(other) => return Err(other),
         },
     };
     Ok(())

From a4ecc09da121e61883fb6371200c6cc4417a1644 Mon Sep 17 00:00:00 2001
From: "chenyee.cy.huang" <eric.cy.huang@linecorp.com>
Date: Tue, 7 Jan 2025 00:55:03 +0800
Subject: [PATCH 24/26] Handle the `pair_expansion` error in caller side

---
 src/common/expansion.rs |  7 +------
 src/sinks/loki/sink.rs  | 20 +++++++++++++++++---
 2 files changed, 18 insertions(+), 9 deletions(-)

diff --git a/src/common/expansion.rs b/src/common/expansion.rs
index 07e241b0a1405..305eed7f9ac2f 100644
--- a/src/common/expansion.rs
+++ b/src/common/expansion.rs
@@ -5,13 +5,12 @@ use crate::event::Value;
 
 static RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"[^0-9A-Za-z_]").unwrap());
 fn slugify_text(input: &str) -> String {
-    let result = RE.replace_all(&input, "_");
+    let result = RE.replace_all(input, "_");
     result.to_lowercase()
 }
 
 /// Expands the given possibly template-able `key_s` and `value_s`, and return the expanded owned pairs
 /// it would also insert the pairs into either `static_pairs` or `dynamic_pairs` depending on the template-ability of `key_s`.
-/// Refer to loki sink and log_to_metric transform for further information.
 pub(crate) fn pair_expansion(
     key_s: &str,
     value_s: &str,
@@ -24,10 +23,6 @@ pub(crate) fn pair_expansion(
             serde_json::from_str(value_s);
 
         if let Err(err) = output {
-            warn!(
-                "Failed to expand dynamic pair. value: {}, err: {}",
-                value_s, err
-            );
             return Err(err);
         }
 
diff --git a/src/sinks/loki/sink.rs b/src/sinks/loki/sink.rs
index a3765b6f5723f..37be3a9ee7478 100644
--- a/src/sinks/loki/sink.rs
+++ b/src/sinks/loki/sink.rs
@@ -179,7 +179,14 @@ impl EventEncoder {
 
             let key_s = key.unwrap();
             let value_s = value.unwrap();
-            let _ = pair_expansion(&key_s, &value_s, &mut static_labels, &mut dynamic_labels);
+            let result = pair_expansion(&key_s, &value_s, &mut static_labels, &mut dynamic_labels);
+            // we just need to check the error since the result have been inserted in the static_pairs or dynamic_pairs
+            if let Err(err) = result {
+                warn!(
+                    "Failed to expand dynamic label. value: {}, err: {}",
+                    value_s, err
+                );
+            }
         }
 
         for (k, v) in static_labels {
@@ -249,18 +256,25 @@ impl EventEncoder {
 
             let key_s = key.unwrap();
             let value_s = value.unwrap();
-            let _ = pair_expansion(
+            let result = pair_expansion(
                 &key_s,
                 &value_s,
                 &mut static_structured_metadata,
                 &mut dynamic_structured_metadata,
             );
+            // we just need to check the error since the result have been inserted in the static_pairs or dynamic_pairs
+            if let Err(err) = result {
+                warn!(
+                    "Failed to expand dynamic structured metadata. value: {}, err: {}",
+                    value_s, err
+                );
+            }
         }
 
         for (k, v) in static_structured_metadata {
             if let Some(discarded_v) = dynamic_structured_metadata.insert(k.clone(), v.clone()) {
                 warn!(
-                    "Static label overrides dynamic label. \
+                    "Static structured_metadata overrides dynamic structured_metadata. \
         key: {}, value: {}, discarded value: {}",
                     k, v, discarded_v
                 );

From 32ed91dd5545c78730ea54dce8c7cd1af328cfbc Mon Sep 17 00:00:00 2001
From: "chenyee.cy.huang" <eric.cy.huang@linecorp.com>
Date: Tue, 7 Jan 2025 01:13:17 +0800
Subject: [PATCH 25/26] Better error handling in `pair_expansion`

---
 src/common/expansion.rs | 10 +++-------
 1 file changed, 3 insertions(+), 7 deletions(-)

diff --git a/src/common/expansion.rs b/src/common/expansion.rs
index 305eed7f9ac2f..11d6940183de7 100644
--- a/src/common/expansion.rs
+++ b/src/common/expansion.rs
@@ -19,16 +19,12 @@ pub(crate) fn pair_expansion(
 ) -> Result<HashMap<String, String>, serde_json::Error> {
     let mut expanded_pairs = HashMap::new();
     if let Some(opening_prefix) = key_s.strip_suffix('*') {
-        let output: Result<serde_json::map::Map<String, serde_json::Value>, serde_json::Error> =
-            serde_json::from_str(value_s);
-
-        if let Err(err) = output {
-            return Err(err);
-        }
+        let output: serde_json::map::Map<String, serde_json::Value> =
+            serde_json::from_str(value_s)?;
 
         // key_* -> key_one, key_two, key_three
         // * -> one, two, three
-        for (k, v) in output.unwrap() {
+        for (k, v) in output {
             let key = slugify_text(&format!("{}{}", opening_prefix, k));
             let val = Value::from(v).to_string_lossy().into_owned();
             if val == "<null>" {

From 5d4b3865c11ca17bced51aa061cc6bceb344e6f0 Mon Sep 17 00:00:00 2001
From: "chenyee.cy.huang" <eric.cy.huang@linecorp.com>
Date: Tue, 7 Jan 2025 23:20:08 +0800
Subject: [PATCH 26/26] Update example

---
 .../components/transforms/log_to_metric.cue        | 14 +++++++++-----
 1 file changed, 9 insertions(+), 5 deletions(-)

diff --git a/website/cue/reference/components/transforms/log_to_metric.cue b/website/cue/reference/components/transforms/log_to_metric.cue
index 1bdc6a333924c..80a49c2fa54b7 100644
--- a/website/cue/reference/components/transforms/log_to_metric.cue
+++ b/website/cue/reference/components/transforms/log_to_metric.cue
@@ -79,7 +79,7 @@ components: transforms: log_to_metric: {
 		},
 		{
 			title: "Sum"
-			notes: "In this example we'll demonstrate computing a sum by computing the total of orders placed."
+			notes: "In this example we'll demonstrate computing a sum by computing the total of orders placed, and we are trying the tags expansion feature."
 
 			configuration: {
 				metrics: [
@@ -89,22 +89,26 @@ components: transforms: log_to_metric: {
 						name:               "order_total"
 						increment_by_value: true
 						tags: {
-							host: "{{host}}"
+							"*": "{{labels}}"
 						}
 					},
 				]
 			}
 
 			input: log: {
-				host:    "10.22.11.222"
 				message: "Order placed for $122.20"
-				total:   122.2
+				labels: {
+					host:     "10.22.11.222"
+					hostname: "vector-dev"
+				}
+				total: 122.2
 			}
 			output: [{metric: {
 				kind: "incremental"
 				name: "order_total"
 				tags: {
-					host: "10.22.11.222"
+					host:     "10.22.11.222"
+					hostname: "vector-dev"
 				}
 				counter: {
 					value: 122.2