diff --git a/changelog.d/21939_log_to_metric_tags_expansion.feature.md b/changelog.d/21939_log_to_metric_tags_expansion.feature.md
new file mode 100644
index 0000000000000..d6223975ae7e9
--- /dev/null
+++ b/changelog.d/21939_log_to_metric_tags_expansion.feature.md
@@ -0,0 +1,3 @@
+The `log_to_metric` transformer tag key are now template-able which enables tags expansion.
+
+authors: titaneric
diff --git a/src/common/expansion.rs b/src/common/expansion.rs
new file mode 100644
index 0000000000000..11d6940183de7
--- /dev/null
+++ b/src/common/expansion.rs
@@ -0,0 +1,48 @@
+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: &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`.
+pub(crate) fn pair_expansion(
+    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: 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 {
+            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.to_string(), value_s.to_string());
+        expanded_pairs.insert(key_s.to_string(), value_s.to_string());
+    }
+    Ok(expanded_pairs)
+}
diff --git a/src/common/mod.rs b/src/common/mod.rs
index dad3e63de35e4..6e29cbeebeadf 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", feature = "sinks-loki"))]
+pub(crate) mod expansion;
diff --git a/src/sinks/loki/sink.rs b/src/sinks/loki/sink.rs
index f12dbc3b09e4a..37be3a9ee7478 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,41 +179,13 @@ 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 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
+                );
             }
         }
 
@@ -284,52 +256,25 @@ 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 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
                 );
@@ -620,13 +565,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};
diff --git a/src/transforms/log_to_metric.rs b/src/transforms/log_to_metric.rs
index 1e3d966d2c43a..48cfbecfeb501 100644
--- a/src/transforms/log_to_metric.rs
+++ b/src/transforms/log_to_metric.rs
@@ -19,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,
@@ -109,8 +110,11 @@ pub struct MetricConfig {
     pub namespace: Option<Template>,
 
     /// Tags to apply to the metric.
+    ///
+    /// 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<String, TagConfig>>,
+    pub tags: Option<IndexMap<Template, TagConfig>>,
 
     #[configurable(derived)]
     #[serde(flatten)]
@@ -262,6 +266,7 @@ enum TransformError {
         error: ParseFloatError,
     },
     TemplateRenderingError(TemplateRenderingError),
+    PairExpansionError,
 }
 
 fn render_template(template: &Template, event: &Event) -> Result<String, TransformError> {
@@ -271,9 +276,11 @@ 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_tags: HashMap<String, String> = HashMap::new();
+    let mut dynamic_tags: HashMap<String, String> = HashMap::new();
     Ok(match tags {
         None => None,
         Some(tags) => {
@@ -281,15 +288,38 @@ fn render_tags(
             for (name, config) in tags {
                 match config {
                     TagConfig::Plain(template) => {
-                        render_tag_into(event, name, template, &mut result)?
+                        render_tag_into(
+                            event,
+                            name,
+                            template.as_ref(),
+                            &mut result,
+                            &mut static_tags,
+                            &mut dynamic_tags,
+                        )?;
                     }
                     TagConfig::Multi(vec) => {
                         for template in vec {
-                            render_tag_into(event, name, template, &mut result)?;
+                            render_tag_into(
+                                event,
+                                name,
+                                template.as_ref(),
+                                &mut result,
+                                &mut static_tags,
+                                &mut dynamic_tags,
+                            )?;
                         }
                     }
                 }
             }
+            for (k, v) in static_tags {
+                if let Some(discarded_v) = dynamic_tags.insert(k.clone(), v.clone()) {
+                    warn!(
+                        "Static tags overrides dynamic tags. \
+                key: {}, value: {:?}, discarded value: {:?}",
+                        k, v, discarded_v
+                    );
+                };
+            }
             result.as_option()
         }
     })
@@ -297,26 +327,45 @@ fn render_tags(
 
 fn render_tag_into(
     event: &Event,
-    name: &str,
-    template: &Option<Template>,
+    key_template: &Template,
+    value_template: Option<&Template>,
     result: &mut MetricTags,
+    static_tags: &mut HashMap<String, String>,
+    dynamic_tags: &mut HashMap<String, String>,
 ) -> Result<(), TransformError> {
-    let value = match template {
-        None => TagValue::Bare,
+    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_s, TagValue::Bare);
+        }
         Some(template) => match render_template(template, event) {
-            Ok(result) => TagValue::Value(result),
-            Err(TransformError::TemplateRenderingError(error)) => {
+            Ok(value_s) => {
+                let expanded_pairs = pair_expansion(&key_s, &value_s, static_tags, dynamic_tags)
+                    .map_err(|_| TransformError::PairExpansionError)?;
+                result.extend(expanded_pairs);
+            }
+            Err(TransformError::TemplateRenderingError(value_error)) => {
                 emit!(crate::internal_events::TemplateRenderingError {
-                    error,
+                    error: value_error,
                     drop_event: false,
-                    field: Some(name),
+                    field: Some(template.get_ref()),
                 });
                 return Ok(());
             }
             Err(other) => return Err(other),
         },
     };
-    result.insert(name.to_string(), value);
     Ok(())
 }
 
@@ -910,7 +959,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]
@@ -1073,6 +1122,111 @@ mod tests {
         );
     }
 
+    #[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 = {"*" = "{{ 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("foo"));
+        test_dict.insert("two".into(), Value::from("baz"));
+        log.insert("dict", Value::from(test_dict));
+
+        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!(
+                "one" => "foo",
+                "two" => "baz",
+            )))
+            .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
@@ -1101,6 +1255,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() {
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..6275706f44869 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 can be templated, allowing you to attach dynamic tags to events.
+					"""
+				required: false
 				type: object: options: "*": {
 					description: "A metric tag."
 					required:    true
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