diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 05549c84603..9c77cd8229a 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -36,6 +36,10 @@ jobs:
       - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         name: Checkout code
 
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update && sudo apt-get install -y libcurl4-openssl-dev
+
       - uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
         with:
           enable-cache: false
@@ -63,6 +67,10 @@ jobs:
           app_id: ${{ vars.SENTRY_INTERNAL_APP_ID }}
           private_key: ${{ secrets.SENTRY_INTERNAL_APP_PRIVATE_KEY }}
 
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update && sudo apt-get install -y libcurl4-openssl-dev
+
       - uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
         with:
           # we just cache the venv-dir directly in action-setup-venv
@@ -122,6 +130,10 @@ jobs:
       - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         name: Checkout code
 
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update && sudo apt-get install -y libcurl4-openssl-dev
+
       - name: Install protoc
         uses: arduino/setup-protoc@c65c819552d16ad3c9b72d9dfd5ba5237b9c906b # v3
         with:
@@ -144,6 +156,10 @@ jobs:
       - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         name: Checkout code
 
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update && sudo apt-get install -y libcurl4-openssl-dev
+
       - uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
         with:
           # we just cache the venv-dir directly in action-setup-venv
diff --git a/.github/workflows/ddl-changes.yml b/.github/workflows/ddl-changes.yml
index 739ea7c6d0a..b2ba16c7ae4 100644
--- a/.github/workflows/ddl-changes.yml
+++ b/.github/workflows/ddl-changes.yml
@@ -24,6 +24,10 @@ jobs:
           clean: false
           fetch-depth: 200
 
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update && sudo apt-get install -y libcurl4-openssl-dev
+
       - uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
         with:
           # we just cache the venv-dir directly in action-setup-venv
diff --git a/.github/workflows/docs.yml b/.github/workflows/docs.yml
index 9b8a36f7786..db28035a7b3 100644
--- a/.github/workflows/docs.yml
+++ b/.github/workflows/docs.yml
@@ -17,6 +17,10 @@ jobs:
           # we just cache the venv-dir directly in action-setup-venv
           enable-cache: false
 
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update && sudo apt-get install -y libcurl4-openssl-dev
+
       - uses: getsentry/action-setup-venv@5a80476d175edf56cb205b08bc58986fa99d1725 # v3.2.0
         with:
 
diff --git a/CODEOWNERS b/CODEOWNERS
index 923e42b1b51..7623065df5b 100644
--- a/CODEOWNERS
+++ b/CODEOWNERS
@@ -15,7 +15,7 @@
 
 # rust_snuba
 /rust_snuba @getsentry/owners-snuba
-/rust_snuba/src/processors/replays.rs @getsentry/replay-backend
+/rust_snuba/src/processors/replays.rs @getsentry/replay-backend @getsentry/owners-snuba
 
 # EAP-related
 /snuba/web/rpc @getsentry/events-analytics-platform @getsentry/owners-snuba
diff --git a/Dockerfile b/Dockerfile
index bd89e01f234..8209730070f 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -32,6 +32,7 @@ RUN set -ex; \
         g++ \
         gnupg \
         protobuf-compiler \
+        libcurl4-openssl-dev \
     '; \
     runtimeDeps=' \
         curl \
diff --git a/rust_snuba/Cargo.lock b/rust_snuba/Cargo.lock
index 59fe5b25fb7..a1ec439c8ef 100644
--- a/rust_snuba/Cargo.lock
+++ b/rust_snuba/Cargo.lock
@@ -73,54 +73,12 @@ version = "0.1.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299"
 
-[[package]]
-name = "anstream"
-version = "0.6.13"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d96bd03f33fe50a863e394ee9718a706f988b9079b20c3784fb726e7678b62fb"
-dependencies = [
- "anstyle",
- "anstyle-parse",
- "anstyle-query",
- "anstyle-wincon",
- "colorchoice",
- "utf8parse",
-]
-
 [[package]]
 name = "anstyle"
 version = "1.0.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8901269c6307e8d93993578286ac0edf7f195079ffff5ebdeea6a59ffb7e36bc"
 
-[[package]]
-name = "anstyle-parse"
-version = "0.2.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c75ac65da39e5fe5ab759307499ddad880d724eed2f6ce5b5e8a26f4f387928c"
-dependencies = [
- "utf8parse",
-]
-
-[[package]]
-name = "anstyle-query"
-version = "1.0.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e28923312444cdd728e4738b3f9c9cac739500909bb3d3c94b43551b16517648"
-dependencies = [
- "windows-sys 0.52.0",
-]
-
-[[package]]
-name = "anstyle-wincon"
-version = "3.0.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1cd54b81ec8d6180e24654d0b371ad22fc3dd083b6ff8ba325b72e00c87660a7"
-dependencies = [
- "anstyle",
- "windows-sys 0.52.0",
-]
-
 [[package]]
 name = "anyhow"
 version = "1.0.102"
@@ -574,15 +532,6 @@ version = "1.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33"
 
-[[package]]
-name = "cadence"
-version = "1.6.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3075f133bee430b7644c54fb629b9b4420346ffa275a45c81a6babe8b09b4f51"
-dependencies = [
- "crossbeam-channel",
-]
-
 [[package]]
 name = "cast"
 version = "0.3.0"
@@ -668,7 +617,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "90bc066a67923782aa8515dbaea16946c5bcc5addbd668bb80af688e53e548a0"
 dependencies = [
  "clap_builder",
- "clap_derive",
 ]
 
 [[package]]
@@ -677,22 +625,8 @@ version = "4.5.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ae129e2e766ae0ec03484e609954119f123cc1fe650337e155d03b022f24f7b4"
 dependencies = [
- "anstream",
  "anstyle",
  "clap_lex",
- "strsim",
-]
-
-[[package]]
-name = "clap_derive"
-version = "4.5.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "528131438037fd55894f62d6e9f068b8f45ac57ffa77517819645d10aed04f64"
-dependencies = [
- "heck 0.5.0",
- "proc-macro2",
- "quote",
- "syn 2.0.117",
 ]
 
 [[package]]
@@ -721,12 +655,6 @@ dependencies = [
  "wasm-bindgen",
 ]
 
-[[package]]
-name = "colorchoice"
-version = "1.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "acbf1af155f9b9ef647e42cdc158db4b64a1b61f743629225fde6f3e0be2a7c7"
-
 [[package]]
 name = "concurrent-queue"
 version = "2.4.0"
@@ -773,15 +701,6 @@ dependencies = [
  "libc",
 ]
 
-[[package]]
-name = "crc32fast"
-version = "1.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b3855a8a784b474f333699ef2bbca9db2c4a1f6d9088a90a2d25b1eb53111eaa"
-dependencies = [
- "cfg-if",
-]
-
 [[package]]
 name = "criterion"
 version = "0.5.1"
@@ -1049,27 +968,10 @@ dependencies = [
 ]
 
 [[package]]
-name = "env_filter"
-version = "0.1.3"
+name = "endian-type"
+version = "0.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "186e05a59d4c50738528153b83b0b0194d3a29507dfec16eccd4b342903397d0"
-dependencies = [
- "log",
- "regex",
-]
-
-[[package]]
-name = "env_logger"
-version = "0.11.8"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "13c863f0904021b108aa8b2f55046443e6b1ebde8fd4a15c399893aae4fa069f"
-dependencies = [
- "anstream",
- "anstyle",
- "env_filter",
- "jiff",
- "log",
-]
+checksum = "c34f04666d835ff5d62e058c3995147c06f42fe86ff053337632bca83e42702d"
 
 [[package]]
 name = "equivalent"
@@ -2107,30 +2009,6 @@ version = "1.0.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "49f1f14873335454500d59611f1cf4a4b0f786f9ac11f4312a78e4cf2566695b"
 
-[[package]]
-name = "jiff"
-version = "0.2.15"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "be1f93b8b1eb69c77f24bbb0afdf66f54b632ee39af40ca21c4365a1d7347e49"
-dependencies = [
- "jiff-static",
- "log",
- "portable-atomic",
- "portable-atomic-util",
- "serde",
-]
-
-[[package]]
-name = "jiff-static"
-version = "0.2.15"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "03343451ff899767262ec32146f6d559dd759fdadf42ff0e227c7c48f72594b4"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.117",
-]
-
 [[package]]
 name = "jobserver"
 version = "0.1.31"
@@ -2398,6 +2276,52 @@ dependencies = [
  "autocfg",
 ]
 
+[[package]]
+name = "metrics"
+version = "0.24.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "89550ee9f79e88fef3119de263694973a8adb26c21d75322164fb8c493039fe2"
+dependencies = [
+ "portable-atomic",
+ "rapidhash",
+]
+
+[[package]]
+name = "metrics-exporter-dogstatsd"
+version = "0.9.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b4f39e912d07239992146dc617951ef78a909daec25f0b31c629767cf59f34bb"
+dependencies = [
+ "bytes",
+ "itoa",
+ "metrics",
+ "metrics-util",
+ "ryu",
+ "thiserror 2.0.18",
+ "tracing",
+]
+
+[[package]]
+name = "metrics-util"
+version = "0.20.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "96f8722f8562635f92f8ed992f26df0532266eb03d5202607c20c0d7e9745e13"
+dependencies = [
+ "aho-corasick",
+ "crossbeam-epoch",
+ "crossbeam-utils",
+ "hashbrown 0.16.1",
+ "indexmap 2.13.0",
+ "metrics",
+ "ordered-float",
+ "quanta",
+ "radix_trie",
+ "rand 0.9.3",
+ "rand_xoshiro",
+ "rapidhash",
+ "sketches-ddsketch",
+]
+
 [[package]]
 name = "mime"
 version = "0.3.17"
@@ -2454,6 +2378,15 @@ version = "1.0.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "650eef8c711430f1a879fdd01d4745a7deea475becfb90269c06775983bbf086"
 
+[[package]]
+name = "nibble_vec"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "77a5d83df9f36fe23f0c3648c6bbb8b0298bb5f1939c8f2704431371f4b84d43"
+dependencies = [
+ "smallvec",
+]
+
 [[package]]
 name = "nix"
 version = "0.28.0"
@@ -2655,6 +2588,15 @@ dependencies = [
  "vcpkg",
 ]
 
+[[package]]
+name = "ordered-float"
+version = "5.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b7d950ca161dc355eaf28f82b11345ed76c6e1f6eb1f4f4479e0323b9e2fbd0e"
+dependencies = [
+ "num-traits",
+]
+
 [[package]]
 name = "os_info"
 version = "3.8.2"
@@ -2915,15 +2857,6 @@ version = "1.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f84267b20a16ea918e43c6a88433c2d54fa145c92a811b5b047ccbe153674483"
 
-[[package]]
-name = "portable-atomic-util"
-version = "0.2.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d8a2f0d8d040d7848a709caf78912debcc3f33ee4b3cac47d73d1e1069e83507"
-dependencies = [
- "portable-atomic",
-]
-
 [[package]]
 name = "powerfmt"
 version = "0.2.0"
@@ -3083,6 +3016,21 @@ dependencies = [
  "syn 2.0.117",
 ]
 
+[[package]]
+name = "quanta"
+version = "0.12.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f3ab5a9d756f0d97bdc89019bd2e4ea098cf9cde50ee7564dde6b81ccc8f06c7"
+dependencies = [
+ "crossbeam-utils",
+ "libc",
+ "once_cell",
+ "raw-cpuid",
+ "wasi",
+ "web-sys",
+ "winapi",
+]
+
 [[package]]
 name = "quote"
 version = "1.0.40"
@@ -3104,6 +3052,16 @@ version = "6.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
 
+[[package]]
+name = "radix_trie"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c069c179fcdc6a2fe24d8d18305cf085fdbd4f922c041943e203685d6a1c58fd"
+dependencies = [
+ "endian-type",
+ "nibble_vec",
+]
+
 [[package]]
 name = "rand"
 version = "0.8.6"
@@ -3163,6 +3121,33 @@ dependencies = [
  "getrandom 0.3.4",
 ]
 
+[[package]]
+name = "rand_xoshiro"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f703f4665700daf5512dcca5f43afa6af89f09db47fb56be587f80636bda2d41"
+dependencies = [
+ "rand_core 0.9.3",
+]
+
+[[package]]
+name = "rapidhash"
+version = "4.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5da7e78a036ce858e8d55b7e7dc8ba3a88b78350fd2155d3591bbd966b58589e"
+dependencies = [
+ "rustversion",
+]
+
+[[package]]
+name = "raw-cpuid"
+version = "11.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "498cd0dc59d73224351ee52a95fee0f1a617a2eae0e7d9d720cc622c73a54186"
+dependencies = [
+ "bitflags 2.11.0",
+]
+
 [[package]]
 name = "rayon"
 version = "1.10.0"
@@ -3397,7 +3382,6 @@ dependencies = [
  "adler",
  "anyhow",
  "bytes",
- "cadence",
  "chrono",
  "cityhash-rs",
  "criterion",
@@ -3410,6 +3394,8 @@ dependencies = [
  "json-schema-diff",
  "lz4_flex",
  "md5",
+ "metrics",
+ "metrics-exporter-dogstatsd",
  "once_cell",
  "parking_lot",
  "procspawn",
@@ -3432,7 +3418,6 @@ dependencies = [
  "serde_json",
  "serde_path_to_error",
  "serde_with",
- "statsdproxy",
  "thiserror 1.0.58",
  "tikv-jemallocator",
  "tokio",
@@ -4030,16 +4015,6 @@ version = "1.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64"
 
-[[package]]
-name = "signal-hook"
-version = "0.3.17"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8621587d4798caf8eb44879d42e56b9a93ea5dcd315a6487c357130095b62801"
-dependencies = [
- "libc",
- "signal-hook-registry",
-]
-
 [[package]]
 name = "signal-hook-registry"
 version = "1.4.1"
@@ -4061,6 +4036,12 @@ version = "0.3.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "38b58827f4464d87d377d175e90bf58eb00fd8716ff0a62f80356b5e61555d0d"
 
+[[package]]
+name = "sketches-ddsketch"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0c6f73aeb92d671e0cc4dca167e59b2deb6387c375391bc99ee743f326994a2b"
+
 [[package]]
 name = "slab"
 version = "0.4.9"
@@ -4118,25 +4099,6 @@ version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3"
 
-[[package]]
-name = "statsdproxy"
-version = "0.4.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8ad88f372ad4e60c8235f74feb2fcc047377c55b1872d0b9247b8a6aed8dcba9"
-dependencies = [
- "anyhow",
- "cadence",
- "clap",
- "crc32fast",
- "env_logger",
- "log",
- "rand 0.8.6",
- "serde",
- "serde_yaml",
- "signal-hook",
- "thread_local",
-]
-
 [[package]]
 name = "string_cache"
 version = "0.8.7"
@@ -4796,12 +4758,6 @@ version = "1.0.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be"
 
-[[package]]
-name = "utf8parse"
-version = "0.2.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "711b9620af191e0cdc7468a8d14e709c3dcdb115b36f838e601583af800a370a"
-
 [[package]]
 name = "uuid"
 version = "1.8.0"
diff --git a/rust_snuba/Cargo.toml b/rust_snuba/Cargo.toml
index d3baabbdad0..bb3ee4999cb 100644
--- a/rust_snuba/Cargo.toml
+++ b/rust_snuba/Cargo.toml
@@ -24,7 +24,6 @@ crate-type = ["cdylib", "rlib"]
 [dependencies]
 adler = "1.0.2"
 anyhow = { version = "1.0.69", features = ["backtrace"] }
-cadence = "1.0.0"
 chrono = { version = "0.4.26", features = ["serde"] }
 cityhash-rs = "1.0.1"
 ctrlc = { version = "3.2.5", features = ["termination"] }
@@ -34,6 +33,8 @@ hyper = "1.2.0"
 json-schema-diff = "0.1.7"
 lz4_flex = "0.11"
 md5 = "0.7.0"
+metrics = "0.24.3"
+metrics-exporter-dogstatsd = "0.9.6"
 parking_lot = "0.12.1"
 procspawn = { version = "1.0.0", features = ["json"] }
 prost = "0.14"
@@ -64,7 +65,6 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0", features = ["raw_value"] }
 serde_path_to_error = "0.1.15"
 serde_with = "3.8.1"
-statsdproxy = { version = "0.4.1", features = ["cadence"] }
 thiserror = "1.0"
 tokio = { version = "1.38.2", features = ["full"] }
 tokio-stream = "0.1.15"
diff --git a/rust_snuba/benches/processors.rs b/rust_snuba/benches/processors.rs
index 8f6793224b3..b7c3ebd49d5 100644
--- a/rust_snuba/benches/processors.rs
+++ b/rust_snuba/benches/processors.rs
@@ -7,9 +7,9 @@ use criterion::{black_box, BenchmarkGroup, BenchmarkId, Criterion, Throughput};
 use once_cell::sync::Lazy;
 use parking_lot::Mutex;
 use rust_snuba::{
-    BatchSizeCalculation, BrokerConfig, ClickhouseConfig, ConsumerStrategyFactoryV2, EnvConfig,
-    KafkaMessageMetadata, MessageProcessorConfig, ProcessingFunction, ProcessingFunctionType,
-    ProcessorConfig, StatsDBackend, StorageConfig, TopicConfig, PROCESSORS,
+    BatchSizeCalculation, BrokerConfig, ClickhouseConfig, ConsumerStrategyFactoryV2,
+    DogStatsDBackend, EnvConfig, KafkaMessageMetadata, MessageProcessorConfig, ProcessingFunction,
+    ProcessingFunctionType, ProcessorConfig, StorageConfig, TopicConfig, PROCESSORS,
 };
 use sentry_arroyo::backends::kafka::types::KafkaPayload;
 use sentry_arroyo::backends::local::broker::LocalBroker;
@@ -209,7 +209,13 @@ fn run_processor_bench(
 
 fn main() {
     // this sends to nowhere, but because it's UDP we won't error.
-    metrics::init(StatsDBackend::new("127.0.0.1", 8081, "snuba.consumer")).unwrap();
+    metrics::init(DogStatsDBackend::new_udp(
+        "127.0.0.1",
+        8081,
+        "snuba.consumer",
+        &[],
+    ))
+    .unwrap();
 
     let mut c = Criterion::default().configure_from_args();
 
diff --git a/rust_snuba/src/accepted_outcomes_consumer.rs b/rust_snuba/src/accepted_outcomes_consumer.rs
index 756835e0f83..55d3084a994 100644
--- a/rust_snuba/src/accepted_outcomes_consumer.rs
+++ b/rust_snuba/src/accepted_outcomes_consumer.rs
@@ -19,7 +19,7 @@ use pyo3::prelude::*;
 use crate::config;
 use crate::logging::{setup_logging, setup_sentry};
 use crate::metrics::global_tags::set_global_tag;
-use crate::metrics::statsd::StatsDBackend;
+use crate::metrics::statsd::create_dogstatsd_backend;
 use crate::strategies::accepted_outcomes::aggregator::OutcomesAggregator;
 use crate::strategies::accepted_outcomes::commit_outcomes::CommitOutcomes;
 use crate::strategies::accepted_outcomes::produce_outcome::ProduceAcceptedOutcome;
@@ -127,6 +127,8 @@ pub fn accepted_outcomes_consumer_impl(
 
     assert_eq!(consumer_config.storages.len(), 1);
 
+    let env_config = consumer_config.env.clone();
+
     let mut _sentry_guard = None;
 
     // setup sentry
@@ -137,20 +139,28 @@ pub fn accepted_outcomes_consumer_impl(
     }
 
     // setup arroyo metrics
-    if let (Some(host), Some(port)) = (
-        consumer_config.env.dogstatsd_host,
-        consumer_config.env.dogstatsd_port,
-    ) {
+    {
         let storage_name = consumer_config
             .storages
             .iter()
             .map(|s| s.name.clone())
             .collect::<Vec<_>>()
             .join(",");
-        set_global_tag("storage".to_owned(), storage_name);
-        set_global_tag("consumer_group".to_owned(), consumer_group.to_owned());
 
-        metrics::init(StatsDBackend::new(&host, port, "snuba.consumer")).unwrap();
+        // Set tags on Sentry scope for error observability
+        sentry::configure_scope(|scope| {
+            scope.set_tag("storage", &storage_name);
+            scope.set_tag("consumer_group", consumer_group);
+        });
+
+        let tags = [
+            ("storage", storage_name.clone()),
+            ("consumer_group", consumer_group.to_owned()),
+        ];
+
+        if let Some(backend) = create_dogstatsd_backend(&env_config, "snuba.consumer", &tags) {
+            metrics::init(backend).unwrap();
+        }
     }
 
     // DLQ setup
diff --git a/rust_snuba/src/config.rs b/rust_snuba/src/config.rs
index 8754d412723..a75104aca1e 100644
--- a/rust_snuba/src/config.rs
+++ b/rust_snuba/src/config.rs
@@ -116,6 +116,7 @@ pub struct EnvConfig {
     pub sentry_dsn: Option<String>,
     pub dogstatsd_host: Option<String>,
     pub dogstatsd_port: Option<u16>,
+    pub dogstatsd_socket_path: Option<String>,
     pub default_retention_days: u16,
     pub lower_retention_days: u16,
     pub valid_retention_days: HashSet<u16>,
@@ -129,6 +130,7 @@ impl Default for EnvConfig {
             sentry_dsn: None,
             dogstatsd_host: None,
             dogstatsd_port: None,
+            dogstatsd_socket_path: None,
             default_retention_days: 90,
             lower_retention_days: 30,
             valid_retention_days: [30, 60, 90].iter().cloned().collect(),
diff --git a/rust_snuba/src/consumer.rs b/rust_snuba/src/consumer.rs
index ff3e33b52d0..e95eb92bb07 100644
--- a/rust_snuba/src/consumer.rs
+++ b/rust_snuba/src/consumer.rs
@@ -21,8 +21,7 @@ use sentry_options::init_with_schemas;
 use crate::config;
 use crate::factory_v2::ConsumerStrategyFactoryV2;
 use crate::logging::{setup_logging, setup_sentry};
-use crate::metrics::global_tags::set_global_tag;
-use crate::metrics::statsd::StatsDBackend;
+use crate::metrics::statsd::create_dogstatsd_backend;
 use crate::processors;
 use crate::rebalancing;
 use crate::types::{InsertOrReplacement, KafkaMessageMetadata};
@@ -144,20 +143,28 @@ pub fn consumer_impl(
     }
 
     // setup arroyo metrics
-    if let (Some(host), Some(port)) = (
-        consumer_config.env.dogstatsd_host,
-        consumer_config.env.dogstatsd_port,
-    ) {
+    {
         let storage_name = consumer_config
             .storages
             .iter()
             .map(|s| s.name.clone())
             .collect::<Vec<_>>()
             .join(",");
-        set_global_tag("storage".to_owned(), storage_name);
-        set_global_tag("consumer_group".to_owned(), consumer_group.to_owned());
 
-        metrics::init(StatsDBackend::new(&host, port, "snuba.consumer")).unwrap();
+        // Set tags on Sentry scope for error observability
+        sentry::configure_scope(|scope| {
+            scope.set_tag("storage", &storage_name);
+            scope.set_tag("consumer_group", consumer_group);
+        });
+
+        let tags = [
+            ("storage", storage_name.clone()),
+            ("consumer_group", consumer_group.to_owned()),
+        ];
+
+        if let Some(backend) = create_dogstatsd_backend(&env_config, "snuba.consumer", &tags) {
+            metrics::init(backend).unwrap();
+        }
     }
 
     if !use_rust_processor {
diff --git a/rust_snuba/src/factory_v2.rs b/rust_snuba/src/factory_v2.rs
index 83acfeb9503..bf4abbd4ed9 100644
--- a/rust_snuba/src/factory_v2.rs
+++ b/rust_snuba/src/factory_v2.rs
@@ -162,6 +162,7 @@ impl ProcessingStrategyFactory<KafkaPayload> for ConsumerStrategyFactoryV2 {
             insert_columns,
         );
 
+        #[allow(clippy::result_large_err)]
         let accumulator = Arc::new(
             |batch: BytesInsertBatch<RowData>, small_batch: Message<BytesInsertBatch<RowData>>| {
                 Ok(batch.merge(small_batch.into_payload()))
diff --git a/rust_snuba/src/lib.rs b/rust_snuba/src/lib.rs
index 7fd2bedde6f..2a03900a7dc 100644
--- a/rust_snuba/src/lib.rs
+++ b/rust_snuba/src/lib.rs
@@ -35,7 +35,7 @@ pub use config::{
     ProcessorConfig, StorageConfig, TopicConfig,
 };
 pub use factory_v2::ConsumerStrategyFactoryV2;
-pub use metrics::statsd::StatsDBackend;
+pub use metrics::statsd::DogStatsDBackend;
 pub use processors::{ProcessingFunction, ProcessingFunctionType, PROCESSORS};
 pub use strategies::noop::Noop;
 pub use strategies::python::PythonTransformStep;
diff --git a/rust_snuba/src/metrics/global_tags.rs b/rust_snuba/src/metrics/global_tags.rs
index 8d5177ae568..88e320a96fc 100644
--- a/rust_snuba/src/metrics/global_tags.rs
+++ b/rust_snuba/src/metrics/global_tags.rs
@@ -1,125 +1,76 @@
 use std::collections::BTreeMap;
+use std::sync::{LazyLock, RwLock};
 
-use parking_lot::RwLock;
-use statsdproxy::middleware::Middleware;
-use statsdproxy::types::Metric;
-
-static GLOBAL_TAGS: RwLock<BTreeMap<String, String>> = RwLock::new(BTreeMap::new());
+static GLOBAL_TAGS: LazyLock<RwLock<BTreeMap<String, String>>> =
+    LazyLock::new(|| RwLock::new(BTreeMap::new()));
 
+/// Sets a tag on the current Sentry scope and stores it for DogStatsD metrics.
 pub fn set_global_tag(key: String, value: String) {
     sentry::configure_scope(|scope| {
         scope.set_tag(&key, &value);
     });
-    GLOBAL_TAGS.write().insert(key, value);
-}
-
-pub struct AddGlobalTags<'a, M> {
-    next: M,
-    global_tags: &'a RwLock<BTreeMap<String, String>>,
-}
-
-impl<M> AddGlobalTags<'static, M>
-where
-    M: Middleware,
-{
-    pub fn new(next: M) -> Self {
-        Self::new_with_tagmap(next, &GLOBAL_TAGS)
-    }
+    GLOBAL_TAGS.write().unwrap().insert(key, value);
 }
 
-impl<'a, M> AddGlobalTags<'a, M>
-where
-    M: Middleware,
-{
-    fn new_with_tagmap(next: M, global_tags: &'a RwLock<BTreeMap<String, String>>) -> Self {
-        AddGlobalTags { next, global_tags }
-    }
-}
-
-impl<M> Middleware for AddGlobalTags<'_, M>
-where
-    M: Middleware,
-{
-    fn poll(&mut self) {
-        self.next.poll()
-    }
-
-    fn submit(&mut self, metric: &mut Metric) {
-        let global_tags = self.global_tags.read();
-
-        if global_tags.is_empty() {
-            return self.next.submit(metric);
-        }
-
-        let mut tag_buffer: Vec<u8> = Vec::new();
-        let mut add_comma = false;
-        match metric.tags() {
-            Some(tags) if !tags.is_empty() => {
-                tag_buffer.extend(tags);
-                add_comma = true;
-            }
-            _ => (),
-        }
-        for (k, v) in global_tags.iter() {
-            if add_comma {
-                tag_buffer.push(b',');
-            }
-            tag_buffer.extend(k.as_bytes());
-            tag_buffer.push(b':');
-            tag_buffer.extend(v.as_bytes());
-            add_comma = true;
-        }
-
-        metric.set_tags(&tag_buffer);
-
-        self.next.submit(metric)
-    }
+pub fn get_global_tags() -> Vec<(String, String)> {
+    GLOBAL_TAGS
+        .read()
+        .unwrap()
+        .iter()
+        .map(|(k, v)| (k.clone(), v.clone()))
+        .collect()
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
 
-    use std::cell::RefCell;
-    use std::collections::BTreeMap;
-
-    use statsdproxy::{middleware::Middleware, types::Metric};
+    // GLOBAL_TAGS is a process-wide static and Rust runs tests in parallel, so each
+    // test uses its own key namespace to avoid racing on shared keys.
 
-    struct FnStep<F>(pub F);
+    #[test]
+    fn test_set_and_get_global_tags() {
+        set_global_tag("set_get.env".to_owned(), "production".to_owned());
+        set_global_tag("set_get.region".to_owned(), "us-east".to_owned());
 
-    impl<F> Middleware for FnStep<F>
-    where
-        F: FnMut(&mut Metric),
-    {
-        fn submit(&mut self, metric: &mut Metric) {
-            (self.0)(metric)
-        }
+        let tags = get_global_tags();
+        assert!(tags.contains(&("set_get.env".to_owned(), "production".to_owned())));
+        assert!(tags.contains(&("set_get.region".to_owned(), "us-east".to_owned())));
     }
 
     #[test]
-    fn test_basic() {
-        let test_cases = [
-            // Without tags
-            ("users.online:1|c", "users.online:1|c|#env:prod"),
-            // With tags
-            (
-                "users.online:1|c|#tag1:a",
-                "users.online:1|c|#tag1:a,env:prod",
-            ),
-        ];
-
-        for test_case in test_cases {
-            let results = RefCell::new(vec![]);
-            let global_tags = RwLock::new(BTreeMap::from([("env".to_owned(), "prod".to_owned())]));
+    fn test_overwrite_global_tag() {
+        set_global_tag("overwrite.env".to_owned(), "staging".to_owned());
+        set_global_tag("overwrite.env".to_owned(), "production".to_owned());
+
+        let tags = get_global_tags();
+        let env_tags: Vec<_> = tags.iter().filter(|(k, _)| k == "overwrite.env").collect();
+        assert_eq!(env_tags.len(), 1);
+        assert_eq!(env_tags[0].1, "production");
+    }
 
-            let step = FnStep(|metric: &mut Metric| results.borrow_mut().push(metric.clone()));
-            let mut middleware = AddGlobalTags::new_with_tagmap(step, &global_tags);
+    #[test]
+    fn test_tags_sorted_by_key() {
+        set_global_tag("sorted.z_key".to_owned(), "z_val".to_owned());
+        set_global_tag("sorted.a_key".to_owned(), "a_val".to_owned());
+
+        // get_global_tags() returns entries from a BTreeMap, so keys come back sorted
+        // regardless of which other tests have written to the shared static.
+        let keys: Vec<_> = tags_for_prefix("sorted.");
+        let mut sorted_keys = keys.clone();
+        sorted_keys.sort();
+        assert_eq!(keys, sorted_keys);
+        assert_eq!(
+            keys,
+            vec!["sorted.a_key".to_owned(), "sorted.z_key".to_owned()]
+        );
+    }
 
-            let mut metric = Metric::new(test_case.0.as_bytes().to_vec());
-            middleware.submit(&mut metric);
-            assert_eq!(results.borrow().len(), 1);
-            let updated_metric = Metric::new(results.borrow_mut()[0].raw.clone());
-            assert_eq!(updated_metric.raw, test_case.1.as_bytes());
-        }
+    fn tags_for_prefix(prefix: &str) -> Vec<String> {
+        get_global_tags()
+            .into_iter()
+            .map(|(k, _)| k)
+            .filter(|k| k.starts_with(prefix))
+            .collect()
     }
 }
diff --git a/rust_snuba/src/metrics/statsd.rs b/rust_snuba/src/metrics/statsd.rs
index de7d91be2aa..1f4ada20df9 100644
--- a/rust_snuba/src/metrics/statsd.rs
+++ b/rust_snuba/src/metrics/statsd.rs
@@ -1,56 +1,192 @@
-use std::time::Duration;
+use metrics::Label;
+use metrics_exporter_dogstatsd::DogStatsDBuilder;
+use sentry_arroyo::metrics::{Metric, MetricType, MetricValue, Recorder};
 
-use sentry_arroyo::metrics::{Metric, MetricSink, Recorder, StatsdRecorder};
-use statsdproxy::cadence::StatsdProxyMetricSink;
-use statsdproxy::config::AggregateMetricsConfig;
-use statsdproxy::middleware::aggregate::AggregateMetrics;
-use statsdproxy::middleware::upstream::Upstream;
-
-use crate::metrics::global_tags::AddGlobalTags;
+use crate::config::EnvConfig;
+use crate::metrics::global_tags::get_global_tags;
 
+/// A metrics backend that uses `metrics-exporter-dogstatsd` to send metrics
+/// to DogStatsD over UDP or Unix domain sockets. Adapts arroyo's [`Recorder`]
+/// trait to the `metrics` crate facade installed by the exporter.
 #[derive(Debug)]
-pub struct StatsDBackend {
-    recorder: StatsdRecorder<Wrapper>,
+pub struct DogStatsDBackend;
+
+impl DogStatsDBackend {
+    pub fn new_udp(host: &str, port: u16, prefix: &str, tags: &[(&str, String)]) -> Self {
+        let addr = format!("{host}:{port}");
+        Self::build(&addr, prefix, tags)
+    }
+
+    /// `socket_path` is passed to the exporter verbatim, so it must be a full DogStatsD
+    /// remote address including the transport scheme, e.g. `unixgram:///run/dogstatsd.sock`
+    /// (datagram) or `unix:///run/dogstatsd.sock` (stream). The scheme is supplied via the
+    /// `SNUBA_DOGSTATSD_SOCKET_PATH` env var rather than hardcoded here, so the same value
+    /// also works for the Python datadog client (which strips the scheme itself).
+    pub fn new_uds(socket_path: &str, prefix: &str, tags: &[(&str, String)]) -> Self {
+        Self::build(socket_path, prefix, tags)
+    }
+
+    fn build(addr: &str, prefix: &str, tags: &[(&str, String)]) -> Self {
+        let global_labels: Vec<Label> = tags
+            .iter()
+            .map(|(k, v)| Label::new(k.to_string(), v.clone()))
+            .collect();
+
+        DogStatsDBuilder::default()
+            .with_remote_address(addr)
+            .expect("invalid DogStatsD address")
+            .set_global_prefix(prefix)
+            .with_global_labels(global_labels)
+            .send_histograms_as_distributions(false)
+            // Disable the exporter's client telemetry so we don't start emitting new
+            // `datadog.dogstatsd.client.*` metrics that the previous statsdproxy pipeline
+            // never sent. This keeps the set of emitted metrics unchanged by the migration.
+            .with_telemetry(false)
+            .install()
+            .expect("failed to install DogStatsD exporter");
+
+        Self
+    }
 }
 
-impl Recorder for StatsDBackend {
-    fn record_metric(&self, metric: Metric<'_>) {
-        self.recorder.record_metric(metric)
+/// The DogStatsD transport selected for the current runtime configuration.
+#[derive(Debug, PartialEq, Eq)]
+enum DogStatsDTransport<'a> {
+    /// Unix domain socket at the given path.
+    Uds(&'a str),
+    /// UDP to the given host and port.
+    Udp(&'a str, u16),
+    /// No transport configured; metrics are disabled.
+    Disabled,
+}
+
+/// Decide which DogStatsD transport to use.
+///
+/// The `use_uds` flag selects the *preferred* transport: when it is set and a socket path
+/// is configured, metrics go over UDS; otherwise they go over UDP (host/port). If only one
+/// transport is configured it is used regardless of the flag — a socket-only deployment
+/// sends over UDS, and a host/port-only deployment sends over UDP. The flag is therefore
+/// authoritative only when both are configured (a configured socket never overrides an
+/// available UDP target while the flag is off), which keeps host/port as the UDP rollback
+/// target. With neither configured, metrics are disabled. Matches the Python
+/// `create_metrics()` gating. Kept pure (no global recorder install) so it is unit-testable.
+fn select_transport(env: &EnvConfig, use_uds: bool) -> DogStatsDTransport<'_> {
+    let socket = env.dogstatsd_socket_path.as_deref();
+    let udp = match (env.dogstatsd_host.as_deref(), env.dogstatsd_port) {
+        (Some(host), Some(port)) => Some((host, port)),
+        _ => None,
+    };
+    match socket {
+        // UDS when a socket is configured and either the flag selects it or there is no
+        // UDP (host/port) target to fall back to.
+        Some(socket_path) if use_uds || udp.is_none() => DogStatsDTransport::Uds(socket_path),
+        // Otherwise use UDP when host/port are configured; with neither, metrics are off.
+        _ => match udp {
+            Some((host, port)) => DogStatsDTransport::Udp(host, port),
+            None => DogStatsDTransport::Disabled,
+        },
     }
 }
 
-struct Wrapper(Box<dyn cadence::MetricSink + Send + Sync + 'static>);
+/// Read the `use_dogstatsd_uds` rollout flag from the `snuba` sentry-options namespace.
+///
+/// Reads the same store as the rest of snuba's Rust runtime config (see
+/// `runtime_config.rs`) and as the Python `create_metrics()`
+/// (`snuba.state.sentry_options.get_option`). The option is declared as a boolean in
+/// `sentry-options/schemas/snuba/schema.json`. Any failure (options client
+/// uninitialized, key missing, wrong type) defaults to `false`, i.e. the stable UDP path.
+/// `sentry_options::init_with_schemas` is called at consumer startup before the metrics
+/// backend is built, so the option is available by the time this runs.
+fn use_dogstatsd_uds_enabled() -> bool {
+    sentry_options::options("snuba")
+        .ok()
+        .and_then(|o| o.get("use_dogstatsd_uds").ok())
+        .and_then(|v| v.as_bool())
+        .unwrap_or(false)
+}
+
+/// Build the DogStatsD metrics backend, choosing the transport at runtime.
+///
+/// The `use_dogstatsd_uds` sentry-option selects the preferred transport: with it `true`
+/// and a socket path configured, metrics go over UDS; otherwise UDP (host/port). A
+/// deployment configured with only one of the two uses that one regardless of the flag.
+/// This mirrors the gating in the Python `create_metrics()` so the transport can be
+/// switched by flipping the sentry-option (followed by a restart) without a redeploy. If
+/// the flag cannot be read (e.g. options unavailable) we default to `false` (prefer UDP).
+///
+/// Returns `None` when neither transport is configured, leaving metrics disabled.
+pub fn create_dogstatsd_backend(
+    env: &EnvConfig,
+    prefix: &str,
+    tags: &[(&str, String)],
+) -> Option<DogStatsDBackend> {
+    let use_uds = use_dogstatsd_uds_enabled();
 
-impl MetricSink for Wrapper {
-    fn emit(&self, metric: &str) {
-        let _ = self.0.emit(metric);
+    match select_transport(env, use_uds) {
+        DogStatsDTransport::Uds(socket_path) => {
+            Some(DogStatsDBackend::new_uds(socket_path, prefix, tags))
+        }
+        DogStatsDTransport::Udp(host, port) => {
+            Some(DogStatsDBackend::new_udp(host, port, prefix, tags))
+        }
+        DogStatsDTransport::Disabled => None,
     }
 }
 
-impl StatsDBackend {
-    pub fn new(host: &str, port: u16, prefix: &str) -> Self {
-        let upstream_addr = format!("{host}:{port}");
-        let aggregator_sink = StatsdProxyMetricSink::new(move || {
-            let upstream = Upstream::new(upstream_addr.clone()).unwrap();
-
-            let config = AggregateMetricsConfig {
-                aggregate_counters: true,
-                flush_offset: 0,
-                flush_interval: Duration::from_secs(1),
-                aggregate_gauges: true,
-                max_map_size: None,
-            };
-            let aggregate = AggregateMetrics::new(config, upstream);
-
-            // adding global tags *after* aggregation is more performant than trying to do the same
-            // in cadence, as it means more bytes and more memory to deal with in
-            // AggregateMetricsConfig
-            AddGlobalTags::new(aggregate)
-        });
-
-        let recorder = StatsdRecorder::new(prefix, Wrapper(Box::new(aggregator_sink)));
-
-        Self { recorder }
+impl Recorder for DogStatsDBackend {
+    fn record_metric(&self, metric: Metric<'_>) {
+        let key: metrics::SharedString = metric.key.to_string().into();
+        let mut labels: Vec<Label> = metric
+            .tags
+            .iter()
+            .map(|(k, v)| Label::new(k.to_string(), v.to_string()))
+            .collect();
+
+        for (k, v) in get_global_tags() {
+            labels.push(Label::new(k, v));
+        }
+        let metadata = metrics::Metadata::new("snuba", metrics::Level::INFO, None);
+        let key = metrics::Key::from_parts(key, labels);
+
+        match metric.ty {
+            MetricType::Counter => {
+                let value = match metric.value {
+                    MetricValue::I64(v) => v as u64,
+                    MetricValue::U64(v) => v,
+                    MetricValue::F64(v) => v as u64,
+                    MetricValue::Duration(d) => d.as_millis() as u64,
+                    _ => return,
+                };
+                metrics::with_recorder(|rec| {
+                    rec.register_counter(&key, &metadata).increment(value);
+                });
+            }
+            MetricType::Gauge => {
+                let value = match metric.value {
+                    MetricValue::I64(v) => v as f64,
+                    MetricValue::U64(v) => v as f64,
+                    MetricValue::F64(v) => v,
+                    MetricValue::Duration(d) => d.as_millis() as f64,
+                    _ => return,
+                };
+                metrics::with_recorder(|rec| {
+                    rec.register_gauge(&key, &metadata).set(value);
+                });
+            }
+            MetricType::Timer => {
+                let value = match metric.value {
+                    MetricValue::I64(v) => v as f64,
+                    MetricValue::U64(v) => v as f64,
+                    MetricValue::F64(v) => v,
+                    MetricValue::Duration(d) => d.as_millis() as f64,
+                    _ => return,
+                };
+                metrics::with_recorder(|rec| {
+                    rec.register_histogram(&key, &metadata).record(value);
+                });
+            }
+            _ => {}
+        }
     }
 }
 
@@ -61,11 +197,78 @@ mod tests {
     use super::*;
 
     #[test]
-    fn statsd_metric_backend() {
-        let backend = StatsDBackend::new("0.0.0.0", 8125, "test");
+    fn dogstatsd_metric_backend() {
+        let backend = DogStatsDBackend::new_udp("0.0.0.0", 8125, "test", &[]);
 
         backend.record_metric(metric!(Counter: "a", 1, "tag1" => "value1"));
         backend.record_metric(metric!(Gauge: "b", 20, "tag2" => "value2"));
         backend.record_metric(metric!(Timer: "c", 30, "tag3" => "value3"));
     }
+
+    fn env_with(host: Option<&str>, port: Option<u16>, socket: Option<&str>) -> EnvConfig {
+        EnvConfig {
+            dogstatsd_host: host.map(str::to_string),
+            dogstatsd_port: port,
+            dogstatsd_socket_path: socket.map(str::to_string),
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn uds_only_when_flag_on_and_socket_set() {
+        let env = env_with(
+            Some("localhost"),
+            Some(8125),
+            Some("unixgram:///var/run/dd.sock"),
+        );
+        // Flag on + socket present -> UDS.
+        assert_eq!(
+            select_transport(&env, true),
+            DogStatsDTransport::Uds("unixgram:///var/run/dd.sock")
+        );
+        // Flag off + socket still present -> stays on UDP. This is the key property:
+        // a deployed socket path does not force UDS, so flipping the flag back to "0"
+        // rolls back to UDP without a redeploy.
+        assert_eq!(
+            select_transport(&env, false),
+            DogStatsDTransport::Udp("localhost", 8125)
+        );
+    }
+
+    #[test]
+    fn falls_back_to_udp_without_socket() {
+        let env = env_with(Some("localhost"), Some(8125), None);
+        // Even with the flag on, no socket path means UDP.
+        assert_eq!(
+            select_transport(&env, true),
+            DogStatsDTransport::Udp("localhost", 8125)
+        );
+        assert_eq!(
+            select_transport(&env, false),
+            DogStatsDTransport::Udp("localhost", 8125)
+        );
+    }
+
+    #[test]
+    fn disabled_when_nothing_configured() {
+        // Neither UDP (host/port) nor UDS (socket) configured -> metrics disabled.
+        let env = env_with(None, None, None);
+        assert_eq!(select_transport(&env, true), DogStatsDTransport::Disabled);
+        assert_eq!(select_transport(&env, false), DogStatsDTransport::Disabled);
+    }
+
+    #[test]
+    fn uds_when_socket_only() {
+        // A socket alone (no host/port) is a usable transport: metrics go over UDS
+        // regardless of the flag, since there is no UDP target to prefer or fall back to.
+        let env = env_with(None, None, Some("unixgram:///var/run/dd.sock"));
+        assert_eq!(
+            select_transport(&env, true),
+            DogStatsDTransport::Uds("unixgram:///var/run/dd.sock")
+        );
+        assert_eq!(
+            select_transport(&env, false),
+            DogStatsDTransport::Uds("unixgram:///var/run/dd.sock")
+        );
+    }
 }
diff --git a/rust_snuba/src/processors/generic_metrics.rs b/rust_snuba/src/processors/generic_metrics.rs
index 37768f3e8fe..3839f1826ab 100644
--- a/rust_snuba/src/processors/generic_metrics.rs
+++ b/rust_snuba/src/processors/generic_metrics.rs
@@ -187,7 +187,7 @@ fn decode_encoded_into_numeric_array<T, const SIZE: usize>(
 where
     T: Decodable<SIZE>,
 {
-    if data.len() % T::SIZE == 0 {
+    if data.len().is_multiple_of(T::SIZE) {
         Ok(data
             .chunks_exact(T::SIZE)
             .map(TryInto::try_into)
diff --git a/rust_snuba/src/processors/replays.rs b/rust_snuba/src/processors/replays.rs
index 21189f91908..fc640d4f6e4 100644
--- a/rust_snuba/src/processors/replays.rs
+++ b/rust_snuba/src/processors/replays.rs
@@ -911,7 +911,7 @@ mod tests {
         assert_eq!(&replay_row.user_geo_country_code, "");
         assert_eq!(&replay_row.user_geo_region, "");
         assert_eq!(&replay_row.user_geo_subdivision, "");
-        assert_eq!(replay_row.error_ids, vec![]);
+        assert_eq!(replay_row.error_ids, Vec::<Uuid>::new());
         assert_eq!(replay_row.error_sample_rate, -1.0);
         assert_eq!(replay_row.ip_address_v4, None);
         assert_eq!(replay_row.ip_address_v6, None);
@@ -927,7 +927,7 @@ mod tests {
         assert_eq!(replay_row.segment_id, None);
         assert_eq!(replay_row.session_sample_rate, -1.0);
         assert_eq!(replay_row.title, None);
-        assert_eq!(replay_row.trace_ids, vec![]);
+        assert_eq!(replay_row.trace_ids, Vec::<Uuid>::new());
         assert_eq!(replay_row.urls, Vec::<String>::new());
         assert_eq!(replay_row.segment_names, Vec::<String>::new());
 
@@ -1040,7 +1040,7 @@ mod tests {
         assert_eq!(&replay_row.user, "");
         assert_eq!(replay_row.debug_id, Uuid::nil());
         assert_eq!(replay_row.error_id, Uuid::nil());
-        assert_eq!(replay_row.error_ids, vec![]);
+        assert_eq!(replay_row.error_ids, Vec::<Uuid>::new());
         assert_eq!(replay_row.error_sample_rate, -1.0);
         assert_eq!(replay_row.fatal_id, Uuid::nil());
         assert_eq!(replay_row.info_id, Uuid::nil());
@@ -1051,7 +1051,7 @@ mod tests {
         assert_eq!(replay_row.replay_start_timestamp, None);
         assert_eq!(replay_row.session_sample_rate, -1.0);
         assert_eq!(replay_row.title, None);
-        assert_eq!(replay_row.trace_ids, vec![]);
+        assert_eq!(replay_row.trace_ids, Vec::<Uuid>::new());
         assert_eq!(replay_row.urls, Vec::<String>::new());
         assert_eq!(replay_row.segment_names, Vec::<String>::new());
         assert_eq!(replay_row.viewed_by_id, 0);
@@ -1126,7 +1126,7 @@ mod tests {
         assert_eq!(&replay_row.user, "");
         assert_eq!(replay_row.debug_id, Uuid::nil());
         assert_eq!(replay_row.error_id, Uuid::nil());
-        assert_eq!(replay_row.error_ids, vec![]);
+        assert_eq!(replay_row.error_ids, Vec::<Uuid>::new());
         assert_eq!(replay_row.error_sample_rate, -1.0);
         assert_eq!(replay_row.fatal_id, Uuid::nil());
         assert_eq!(replay_row.info_id, Uuid::nil());
@@ -1137,7 +1137,7 @@ mod tests {
         assert_eq!(replay_row.replay_start_timestamp, None);
         assert_eq!(replay_row.session_sample_rate, -1.0);
         assert_eq!(replay_row.title, None);
-        assert_eq!(replay_row.trace_ids, vec![]);
+        assert_eq!(replay_row.trace_ids, Vec::<Uuid>::new());
         assert_eq!(replay_row.urls, Vec::<String>::new());
         assert_eq!(replay_row.segment_names, Vec::<String>::new());
         assert_eq!(replay_row.viewed_by_id, 0);
@@ -1224,7 +1224,7 @@ mod tests {
         assert_eq!(replay_row.click_is_dead, 0);
         assert_eq!(replay_row.click_is_rage, 0);
         assert_eq!(replay_row.click_node_id, 0);
-        assert_eq!(replay_row.error_ids, vec![]);
+        assert_eq!(replay_row.error_ids, Vec::<Uuid>::new());
         assert_eq!(replay_row.error_sample_rate, -1.0);
         assert_eq!(replay_row.ip_address_v4, None);
         assert_eq!(replay_row.ip_address_v6, None);
@@ -1233,7 +1233,7 @@ mod tests {
         assert_eq!(replay_row.replay_start_timestamp, None);
         assert_eq!(replay_row.session_sample_rate, -1.0);
         assert_eq!(replay_row.title, None);
-        assert_eq!(replay_row.trace_ids, vec![]);
+        assert_eq!(replay_row.trace_ids, Vec::<Uuid>::new());
         assert_eq!(replay_row.urls, Vec::<String>::new());
         assert_eq!(replay_row.segment_names, Vec::<String>::new());
         assert_eq!(replay_row.viewed_by_id, 0);
@@ -1307,7 +1307,7 @@ mod tests {
         assert_eq!(replay_row.click_node_id, 0);
         assert_eq!(replay_row.debug_id, Uuid::nil());
         assert_eq!(replay_row.error_id, Uuid::nil());
-        assert_eq!(replay_row.error_ids, vec![]);
+        assert_eq!(replay_row.error_ids, Vec::<Uuid>::new());
         assert_eq!(replay_row.error_sample_rate, -1.0);
         assert_eq!(replay_row.fatal_id, Uuid::nil());
         assert_eq!(replay_row.info_id, Uuid::nil());
@@ -1317,7 +1317,7 @@ mod tests {
         assert_eq!(replay_row.replay_start_timestamp, None);
         assert_eq!(replay_row.session_sample_rate, -1.0);
         assert_eq!(replay_row.title, None);
-        assert_eq!(replay_row.trace_ids, vec![]);
+        assert_eq!(replay_row.trace_ids, Vec::<Uuid>::new());
         assert_eq!(replay_row.urls, Vec::<String>::new());
         assert_eq!(replay_row.segment_names, Vec::<String>::new());
         assert_eq!(replay_row.viewed_by_id, 0);
@@ -1379,7 +1379,7 @@ mod tests {
         assert_eq!(&replay_row.dist, "");
         assert_eq!(&replay_row.environment, "");
         assert_eq!(replay_row.error_id, Uuid::nil());
-        assert_eq!(replay_row.error_ids, vec![]);
+        assert_eq!(replay_row.error_ids, Vec::<Uuid>::new());
         assert_eq!(replay_row.error_sample_rate, -1.0);
         assert_eq!(replay_row.fatal_id, Uuid::nil());
         assert_eq!(replay_row.info_id, Uuid::nil());
@@ -1402,7 +1402,7 @@ mod tests {
         assert_eq!(replay_row.tags_key, Vec::<String>::new());
         assert_eq!(replay_row.tags_value, Vec::<String>::new());
         assert_eq!(replay_row.title, None);
-        assert_eq!(replay_row.trace_ids, vec![]);
+        assert_eq!(replay_row.trace_ids, Vec::<Uuid>::new());
         assert_eq!(replay_row.urls, Vec::<String>::new());
         assert_eq!(replay_row.segment_names, Vec::<String>::new());
         assert_eq!(&replay_row.user_email, "");
diff --git a/sentry-options/schemas/snuba/schema.json b/sentry-options/schemas/snuba/schema.json
index 28ed7468fd3..c76eb6c60bc 100644
--- a/sentry-options/schemas/snuba/schema.json
+++ b/sentry-options/schemas/snuba/schema.json
@@ -565,6 +565,11 @@
       "default": false,
       "description": "When true, the dual-write metrics backend also emits to the Sentry metrics backend (sampled by settings.DDM_METRICS_SAMPLE_RATE)."
     },
+    "use_dogstatsd_uds": {
+      "type": "boolean",
+      "default": false,
+      "description": "Selects the preferred DogStatsD transport for snuba's Python and Rust metrics backends. When true and a socket path (DOGSTATSD_SOCKET_PATH) is configured, metrics are sent over the Unix domain socket; otherwise they use UDP (host/port). A deployment configured with only one transport uses that one regardless of the flag (socket-only sends over UDS, host/port-only over UDP), so the flag is authoritative only when both are configured, keeping host/port as the UDP rollback target. Read at metrics-backend creation, so flipping it takes effect on the next process restart -- no redeploy required."
+    },
     "ondemand_profiler_hostnames": {
       "type": "string",
       "default": "",
diff --git a/snuba/consumers/consumer_config.py b/snuba/consumers/consumer_config.py
index 34aa2bdf8f6..52b001cdbfa 100644
--- a/snuba/consumers/consumer_config.py
+++ b/snuba/consumers/consumer_config.py
@@ -50,6 +50,7 @@ class EnvConfig:
     sentry_dsn: str | None
     dogstatsd_host: str | None
     dogstatsd_port: int | None
+    dogstatsd_socket_path: str | None
     default_retention_days: int
     lower_retention_days: int
     valid_retention_days: list[int]
@@ -134,6 +135,7 @@ def _resolve_env_config() -> EnvConfig:
         sentry_dsn=sentry_dsn,
         dogstatsd_host=dogstatsd_host,
         dogstatsd_port=dogstatsd_port,
+        dogstatsd_socket_path=settings.DOGSTATSD_SOCKET_PATH,
         default_retention_days=default_retention_days,
         lower_retention_days=lower_retention_days,
         valid_retention_days=valid_retention_days,
diff --git a/snuba/settings/__init__.py b/snuba/settings/__init__.py
index 1f90bfb36dd..8518fa8a34e 100644
--- a/snuba/settings/__init__.py
+++ b/snuba/settings/__init__.py
@@ -151,6 +151,7 @@
 
 NEW_DOGSTATSD_HOST: str | None = os.environ.get("SNUBA_NEW_STATSD_HOST") or None
 NEW_DOGSTATSD_PORT: int | None = int(os.environ.get("SNUBA_NEW_STATSD_PORT") or 0) or None
+DOGSTATSD_SOCKET_PATH: str | None = os.environ.get("SNUBA_DOGSTATSD_SOCKET_PATH") or None
 
 CLICKHOUSE_READONLY_USER = os.environ.get("CLICKHOUSE_READONLY_USER", "default")
 CLICKHOUSE_READONLY_PASSWORD = os.environ.get("CLICKHOUSE_READONLY_PASSWORD", "")
diff --git a/snuba/utils/metrics/util.py b/snuba/utils/metrics/util.py
index a756eabb86b..feec1ca12f6 100644
--- a/snuba/utils/metrics/util.py
+++ b/snuba/utils/metrics/util.py
@@ -1,6 +1,7 @@
 import _strptime  # NOQA fixes _strptime deferred import issue
 import inspect
-from functools import partial, wraps
+import threading
+from functools import wraps
 from typing import Any, TypeVar, cast
 from collections.abc import Callable, Mapping
 
@@ -16,45 +17,105 @@ def create_metrics(
     tags: Tags | None = None,
     sample_rates: Mapping[str, float] | None = None,
 ) -> MetricsBackend:
-    """Create a DogStatsd object if DOGSTATSD_HOST and DOGSTATSD_PORT are defined,
-    with the specified prefix and tags. Return a DummyMetricsBackend otherwise.
+    """Create a DogStatsd object if a DogStatsD transport (UDP or UDS) is configured.
+
+    The ``use_dogstatsd_uds`` sentry-option selects the preferred transport: with it
+    ``True`` and ``DOGSTATSD_SOCKET_PATH`` configured, metrics are sent over the Unix domain
+    socket; otherwise they use UDP (host/port). A deployment configured with only one of the
+    two uses that one regardless of the flag -- a socket alone sends over UDS, host/port
+    alone over UDP -- so the flag is authoritative only when both are configured (a socket
+    never overrides an available UDP target while the flag is off), keeping host/port as the
+    UDP rollback target. Return a DummyMetricsBackend when neither transport is configured.
+    ``DOGSTATSD_SOCKET_PATH`` is a full address including the transport scheme, e.g.
+    ``unixgram:///run/dogstatsd.sock``; it is passed to the datadog client verbatim (the
+    client strips the scheme), the same value the Rust exporter consumes.
     Prefixes must start with `snuba.<category>`, for example: `snuba.processor`.
     """
     host: str | None = settings.DOGSTATSD_HOST
     port: int | None = settings.DOGSTATSD_PORT
+    socket_path: str | None = settings.DOGSTATSD_SOCKET_PATH
 
     if settings.TESTING:
         from snuba.utils.metrics.backends.testing import TestingMetricsBackend
 
         return TestingMetricsBackend()
-    if host is None and port is None:
-        from snuba.utils.metrics.backends.dummy import DummyMetricsBackend
 
-        return DummyMetricsBackend()
-    if host is None or port is None:
+    # host/port must be both set or both unset; a partial UDP config is a misconfiguration.
+    if (host is None) != (port is None):
         raise ValueError(
             f"DOGSTATSD_HOST and DOGSTATSD_PORT should both be None or not None. Found DOGSTATSD_HOST: {host}, DOGSTATSD_PORT: {port} instead."
         )
 
+    # No transport configured at all -> no metrics. A socket alone is enough (it enables
+    # UDS); host/port alone enable UDP.
+    if host is None and port is None and socket_path is None:
+        from snuba.utils.metrics.backends.dummy import DummyMetricsBackend
+
+        return DummyMetricsBackend()
+
     from datadog import DogStatsd  # type: ignore[attr-defined]  # datadog lacks explicit re-export
 
     from snuba.utils.metrics.backends.datadog import DatadogMetricsBackend
     from snuba.utils.metrics.backends.dualwrite import SentryDatadogMetricsBackend
     from snuba.utils.metrics.backends.sentry import SentryMetricsBackend
 
-    return SentryDatadogMetricsBackend(
-        DatadogMetricsBackend(
-            partial(
-                DogStatsd,
-                host=host,
-                port=port,
+    constant_tags = [f"{key}:{value}" for key, value in tags.items()] if tags is not None else None
+
+    # Resolve UDS-vs-UDP once for the whole process. DatadogMetricsBackend builds a
+    # DogStatsd client per thread, so the factory below runs on each thread's first
+    # metric emission; caching the decision keeps every thread on the same transport
+    # instead of letting a mid-process option flip put some threads on UDP and others on
+    # UDS. The value is fixed until restart, matching the documented "flip the option,
+    # then restart" contract. Double-checked locking: the lock is only contended on the
+    # first emission per thread until the decision is cached.
+    resolved_use_uds: bool | None = None
+    resolve_lock = threading.Lock()
+
+    def make_client() -> DogStatsd:
+        nonlocal resolved_use_uds
+        if resolved_use_uds is None:
+            # The use_dogstatsd_uds sentry-option is read lazily here -- on the first
+            # metric emission -- not at create_metrics() time. create_metrics() runs while
+            # snuba.environment is being imported, and importing snuba.state.sentry_options
+            # pulls in the snuba.state package, whose __init__ binds
+            # MetricsWrapper(environment.metrics, ...) at import time, so importing it any
+            # earlier would be a circular import. By first-emit time sentry-options has been
+            # initialized (snuba.environment.setup_sentry -> init_options); if it hasn't,
+            # get_option returns the False default and we prefer UDP.
+            from snuba.state.sentry_options import get_option
+
+            with resolve_lock:
+                if resolved_use_uds is None:
+                    # Use UDS when a socket is configured and either the flag selects it or
+                    # there is no UDP (host/port) target to fall back to (socket-only).
+                    resolved_use_uds = socket_path is not None and (
+                        bool(get_option("use_dogstatsd_uds", False)) or port is None
+                    )
+
+        if resolved_use_uds:
+            # socket_path is passed to the datadog client verbatim. It is expected to be a
+            # full address including the transport scheme (e.g.
+            # "unixgram:///run/dogstatsd.sock"); the datadog client strips the scheme and
+            # selects the socket kind itself. The same env var (SNUBA_DOGSTATSD_SOCKET_PATH)
+            # is passed verbatim to the Rust exporter, which parses the scheme too, so no
+            # scheme is hardcoded on either side.
+            return DogStatsd(
+                socket_path=socket_path,
                 namespace=prefix,
-                constant_tags=(
-                    [f"{key}:{value}" for key, value in tags.items()] if tags is not None else None
-                ),
-            ),
-            sample_rates,
-        ),
+                constant_tags=constant_tags,
+            )
+        # UDP branch: reached only when a host/port target is configured -- socket-only
+        # deployments always resolve to UDS above -- so host/port are set here.
+        assert host is not None and port is not None
+        return DogStatsd(
+            host=host,
+            port=port,
+            namespace=prefix,
+            constant_tags=constant_tags,
+        )
+
+    return SentryDatadogMetricsBackend(
+        DatadogMetricsBackend(make_client, sample_rates),
         SentryMetricsBackend(),
     )
 
diff --git a/tests/utils/metrics/test_create_metrics.py b/tests/utils/metrics/test_create_metrics.py
new file mode 100644
index 00000000000..28ae150600d
--- /dev/null
+++ b/tests/utils/metrics/test_create_metrics.py
@@ -0,0 +1,174 @@
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+from snuba.utils.metrics.backends.dualwrite import SentryDatadogMetricsBackend
+from snuba.utils.metrics.util import create_metrics
+
+
+def _get_option(use_uds: bool) -> Any:
+    """Return a ``sentry_options.get_option`` replacement that only overrides the
+    ``use_dogstatsd_uds`` flag and falls back to the provided default otherwise."""
+
+    def get_option(key: str, default: Any) -> Any:
+        if key == "use_dogstatsd_uds":
+            return use_uds
+        return default
+
+    return get_option
+
+
+@patch("datadog.DogStatsd")
+def test_create_metrics_uses_uds_when_flag_enabled(dogstatsd: MagicMock) -> None:
+    with (
+        patch.multiple(
+            "snuba.settings",
+            TESTING=False,
+            DOGSTATSD_HOST="localhost",
+            DOGSTATSD_PORT=8125,
+            DOGSTATSD_SOCKET_PATH="unixgram:///var/run/dogstatsd.sock",
+        ),
+        patch("snuba.state.sentry_options.get_option", side_effect=_get_option(True)),
+    ):
+        backend = create_metrics("snuba.test")
+        assert isinstance(backend, SentryDatadogMetricsBackend)
+        # Force the lazily-created DogStatsd client to be built.
+        backend.increment("snuba.test.metric")
+
+    dogstatsd.assert_called_once_with(
+        socket_path="unixgram:///var/run/dogstatsd.sock",
+        namespace="snuba.test",
+        constant_tags=None,
+    )
+
+
+@patch("datadog.DogStatsd")
+def test_create_metrics_uses_udp_when_flag_disabled(dogstatsd: MagicMock) -> None:
+    with (
+        patch.multiple(
+            "snuba.settings",
+            TESTING=False,
+            DOGSTATSD_HOST="localhost",
+            DOGSTATSD_PORT=8125,
+            # A socket path alone must not force UDS: the sentry-option gates it.
+            DOGSTATSD_SOCKET_PATH="unixgram:///var/run/dogstatsd.sock",
+        ),
+        patch("snuba.state.sentry_options.get_option", side_effect=_get_option(False)),
+    ):
+        backend = create_metrics("snuba.test")
+        assert isinstance(backend, SentryDatadogMetricsBackend)
+        backend.increment("snuba.test.metric")
+
+    dogstatsd.assert_called_once_with(
+        host="localhost",
+        port=8125,
+        namespace="snuba.test",
+        constant_tags=None,
+    )
+
+
+@patch("datadog.DogStatsd")
+def test_create_metrics_transport_decision_is_process_wide(dogstatsd: MagicMock) -> None:
+    # DatadogMetricsBackend builds a DogStatsd client per thread, so the client factory
+    # runs on each thread's first emission. The use_dogstatsd_uds decision must be
+    # resolved once per process so a mid-flight flip cannot put one thread on UDP and
+    # another on UDS. Emit once with the flag off (-> UDP), flip it on, then emit from a
+    # fresh thread and assert that thread still builds a UDP client (never a socket one).
+    import threading
+
+    with patch.multiple(
+        "snuba.settings",
+        TESTING=False,
+        DOGSTATSD_HOST="localhost",
+        DOGSTATSD_PORT=8125,
+        DOGSTATSD_SOCKET_PATH="unixgram:///var/run/dogstatsd.sock",
+    ):
+        with patch("snuba.state.sentry_options.get_option", side_effect=_get_option(False)):
+            backend = create_metrics("snuba.test")
+            backend.increment("snuba.test.metric")  # main thread resolves -> UDP
+
+        with patch("snuba.state.sentry_options.get_option", side_effect=_get_option(True)):
+
+            def emit_from_thread() -> None:
+                backend.increment("snuba.test.metric")
+
+            worker = threading.Thread(target=emit_from_thread)
+            worker.start()
+            worker.join()
+
+    # One client built per thread, and every one uses UDP (host/port), not the socket,
+    # even though the flag flipped to True before the worker thread emitted.
+    assert dogstatsd.call_count == 2
+    for call in dogstatsd.call_args_list:
+        assert "socket_path" not in call.kwargs
+        assert call.kwargs["host"] == "localhost"
+        assert call.kwargs["port"] == 8125
+
+
+@patch("datadog.DogStatsd")
+def test_create_metrics_socket_only_uses_uds(dogstatsd: MagicMock) -> None:
+    # A socket alone (no host/port) is a usable transport: metrics go over UDS regardless
+    # of the flag, since there is no UDP target to prefer or fall back to. Matches the Rust
+    # select_transport behavior.
+    for flag in (True, False):
+        dogstatsd.reset_mock()
+        with (
+            patch.multiple(
+                "snuba.settings",
+                TESTING=False,
+                DOGSTATSD_HOST=None,
+                DOGSTATSD_PORT=None,
+                DOGSTATSD_SOCKET_PATH="unixgram:///var/run/dogstatsd.sock",
+            ),
+            patch("snuba.state.sentry_options.get_option", side_effect=_get_option(flag)),
+        ):
+            backend = create_metrics("snuba.test")
+            assert isinstance(backend, SentryDatadogMetricsBackend)
+            backend.increment("snuba.test.metric")
+
+        dogstatsd.assert_called_once_with(
+            socket_path="unixgram:///var/run/dogstatsd.sock",
+            namespace="snuba.test",
+            constant_tags=None,
+        )
+
+
+def test_create_metrics_partial_host_port_raises() -> None:
+    # host/port must be both set or both unset; a partial UDP config is a misconfiguration.
+    import pytest
+
+    with (
+        patch.multiple(
+            "snuba.settings",
+            TESTING=False,
+            DOGSTATSD_HOST="localhost",
+            DOGSTATSD_PORT=None,
+            DOGSTATSD_SOCKET_PATH=None,
+        ),
+        pytest.raises(ValueError),
+    ):
+        create_metrics("snuba.test")
+
+
+def test_create_metrics_dummy_does_not_read_options() -> None:
+    # With nothing configured, create_metrics() must return the dummy backend without
+    # consulting sentry-options. create_metrics() runs at snuba.environment import time,
+    # and importing snuba.state.sentry_options there is a circular import, so the
+    # no-metrics path must not reach the option lookup.
+    with (
+        patch.multiple(
+            "snuba.settings",
+            TESTING=False,
+            DOGSTATSD_HOST=None,
+            DOGSTATSD_PORT=None,
+            DOGSTATSD_SOCKET_PATH=None,
+        ),
+        patch("snuba.state.sentry_options.get_option") as get_option,
+    ):
+        backend = create_metrics("snuba.test")
+        get_option.assert_not_called()
+
+    from snuba.utils.metrics.backends.dummy import DummyMetricsBackend
+
+    assert isinstance(backend, DummyMetricsBackend)