This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/spark-connect-gateway.git
The following commit(s) were added to refs/heads/main by this push:
new e6a8cba [SPARK-58930] Add scg-observability crate
e6a8cba is described below
commit e6a8cba030414c90e5734fb837cea4c0f1c14738
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Sun Aug 23 18:12:15 2026 -0700
[SPARK-58930] Add scg-observability crate
### What changes were proposed in this pull request?
Part of importing the Spark Connect Gateway reference implementation into
`apache/spark-connect-gateway` as a series of small, bottom-up PRs (tracked
in
the do-not-merge reference PR #1).
This adds the `scg-observability` crate — the metrics, tracing, and admin
surface for the gateway:
- Prometheus metrics for the gateway
- Distributed tracing — an OpenTelemetry exporter with W3C traceparent
propagation
- An admin HTTP server exposing `/metrics`, `/healthz`, and `/readyz`
`scg-observability` is a leaf crate (no internal dependencies), so it
compiles
and tests on its own (8 unit tests). It is appended to the workspace
`members`
list, and is consumed later by the `proxy` and `gateway` crates.
### Why are the changes needed?
Implements the reference gateway from the approved SPIP (SPARK-58455),
imported
one reviewable piece at a time.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
`cargo fmt --all --check`, `cargo clippy --workspace --all-targets -- -D
warnings`,
`cargo build --workspace`, and `cargo test --workspace` all pass locally
(scg-observability has 8 unit tests).
### Was this patch authored or co-authored using generative AI tooling?
Yes, co-authored with Claude Code.
Closes #7 from viirya/import-05-observability.
Authored-by: Liang-Chi Hsieh <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
Cargo.lock | 359 ++++++++++++++++++++++++++++++-
Cargo.toml | 1 +
crates/observability/Cargo.toml | 58 ++++++
crates/observability/src/admin.rs | 182 ++++++++++++++++
crates/observability/src/lib.rs | 51 +++++
crates/observability/src/metrics.rs | 406 ++++++++++++++++++++++++++++++++++++
crates/observability/src/tracing.rs | 340 ++++++++++++++++++++++++++++++
7 files changed, 1389 insertions(+), 8 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index b7227f0..49fc8fb 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -156,6 +156,15 @@ dependencies = [
"rand_core 0.10.1",
]
+[[package]]
+name = "chrono"
+version = "0.4.45"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327"
+dependencies = [
+ "num-traits",
+]
+
[[package]]
name = "const-oid"
version = "0.9.6"
@@ -409,6 +418,18 @@ dependencies = [
"wasm-bindgen",
]
+[[package]]
+name = "getrandom"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "r-efi 5.3.0",
+ "wasip2",
+]
+
[[package]]
name = "getrandom"
version = "0.4.3"
@@ -418,7 +439,7 @@ dependencies = [
"cfg-if",
"js-sys",
"libc",
- "r-efi",
+ "r-efi 6.0.0",
"rand_core 0.10.1",
"wasm-bindgen",
]
@@ -802,6 +823,15 @@ version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154"
+[[package]]
+name = "matchers"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9"
+dependencies = [
+ "regex-automata",
+]
+
[[package]]
name = "matchit"
version = "0.8.4"
@@ -837,6 +867,15 @@ version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084"
+[[package]]
+name = "nu-ansi-term"
+version = "0.50.3"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
+dependencies = [
+ "windows-sys 0.61.2",
+]
+
[[package]]
name = "num-bigint"
version = "0.4.8"
@@ -914,6 +953,83 @@ version = "1.21.4"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50"
+[[package]]
+name = "opentelemetry"
+version = "0.31.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "b84bcd6ae87133e903af7ef497404dda70c60d0ea14895fc8a5e6722754fc2a0"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+ "js-sys",
+ "pin-project-lite",
+ "thiserror 2.0.20",
+ "tracing",
+]
+
+[[package]]
+name = "opentelemetry-otlp"
+version = "0.31.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "1f69cd6acbb9af919df949cd1ec9e5e7fdc2ef15d234b6b795aaa525cc02f71f"
+dependencies = [
+ "http",
+ "opentelemetry",
+ "opentelemetry-proto",
+ "opentelemetry_sdk",
+ "prost",
+ "thiserror 2.0.20",
+ "tokio",
+ "tonic",
+]
+
+[[package]]
+name = "opentelemetry-proto"
+version = "0.31.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "a7175df06de5eaee9909d4805a3d07e28bb752c34cab57fa9cff549da596b30f"
+dependencies = [
+ "opentelemetry",
+ "opentelemetry_sdk",
+ "prost",
+ "tonic",
+ "tonic-prost",
+]
+
+[[package]]
+name = "opentelemetry-semantic-conventions"
+version = "0.31.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "e62e29dfe041afb8ed2a6c9737ab57db4907285d999ef8ad3a59092a36bdc846"
+
+[[package]]
+name = "opentelemetry-stdout"
+version = "0.31.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "bc8887887e169414f637b18751487cce4e095be787d23fad13c454e2fb1b3811"
+dependencies = [
+ "chrono",
+ "opentelemetry",
+ "opentelemetry_sdk",
+]
+
+[[package]]
+name = "opentelemetry_sdk"
+version = "0.31.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "e14ae4f5991976fd48df6d843de219ca6d31b01daaab2dad5af2badeded372bd"
+dependencies = [
+ "futures-channel",
+ "futures-executor",
+ "futures-util",
+ "opentelemetry",
+ "percent-encoding",
+ "rand 0.9.5",
+ "thiserror 2.0.20",
+ "tokio",
+ "tokio-stream",
+]
+
[[package]]
name = "parking_lot"
version = "0.12.5"
@@ -1063,6 +1179,21 @@ dependencies = [
"unicode-ident",
]
+[[package]]
+name = "prometheus"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "3ca5326d8d0b950a9acd87e6a3f94745394f62e4dae1b1ee22b2bc0c394af43a"
+dependencies = [
+ "cfg-if",
+ "fnv",
+ "lazy_static",
+ "memchr",
+ "parking_lot",
+ "protobuf",
+ "thiserror 2.0.20",
+]
+
[[package]]
name = "prost"
version = "0.14.4"
@@ -1116,6 +1247,26 @@ dependencies = [
"prost",
]
+[[package]]
+name = "protobuf"
+version = "3.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "d65a1d4ddae7d8b5de68153b48f6aa3bba8cb002b243dbdbc55a5afbc98f99f4"
+dependencies = [
+ "once_cell",
+ "protobuf-support",
+ "thiserror 1.0.69",
+]
+
+[[package]]
+name = "protobuf-support"
+version = "3.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "3e36c2f31e0a47f9280fb347ef5e461ffcd2c52dd520d8e216b52f93b0b0d7d6"
+dependencies = [
+ "thiserror 1.0.69",
+]
+
[[package]]
name = "pulldown-cmark"
version = "0.13.4"
@@ -1150,7 +1301,7 @@ dependencies = [
"rustc-hash",
"rustls",
"socket2",
- "thiserror",
+ "thiserror 2.0.20",
"tokio",
"tracing",
"web-time",
@@ -1172,7 +1323,7 @@ dependencies = [
"rustls",
"rustls-pki-types",
"slab",
- "thiserror",
+ "thiserror 2.0.20",
"tinyvec",
"tracing",
"web-time",
@@ -1201,6 +1352,12 @@ dependencies = [
"proc-macro2",
]
+[[package]]
+name = "r-efi"
+version = "5.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
+
[[package]]
name = "r-efi"
version = "6.0.0"
@@ -1214,10 +1371,20 @@ source =
"registry+https://github.com/rust-lang/crates.io-index";
checksum = "22f6172bdec972074665ed81ed53b71da00bfc44b65a753cfde883ec4c702a1a"
dependencies = [
"libc",
- "rand_chacha",
+ "rand_chacha 0.3.1",
"rand_core 0.6.4",
]
+[[package]]
+name = "rand"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41"
+dependencies = [
+ "rand_chacha 0.9.0",
+ "rand_core 0.9.5",
+]
+
[[package]]
name = "rand"
version = "0.10.2"
@@ -1239,6 +1406,16 @@ dependencies = [
"rand_core 0.6.4",
]
+[[package]]
+name = "rand_chacha"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb"
+dependencies = [
+ "ppv-lite86",
+ "rand_core 0.9.5",
+]
+
[[package]]
name = "rand_core"
version = "0.6.4"
@@ -1248,6 +1425,15 @@ dependencies = [
"getrandom 0.2.17",
]
+[[package]]
+name = "rand_core"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c"
+dependencies = [
+ "getrandom 0.3.4",
+]
+
[[package]]
name = "rand_core"
version = "0.10.1"
@@ -1398,6 +1584,7 @@ version = "0.23.43"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06"
dependencies = [
+ "log",
"once_cell",
"ring",
"rustls-pki-types",
@@ -1453,7 +1640,7 @@ dependencies = [
"serde",
"serde_json",
"subtle",
- "thiserror",
+ "thiserror 2.0.20",
"tokio",
"tonic",
"tracing",
@@ -1467,7 +1654,7 @@ dependencies = [
"serde",
"serde_yaml",
"tempfile",
- "thiserror",
+ "thiserror 2.0.20",
]
[[package]]
@@ -1481,6 +1668,31 @@ dependencies = [
"tonic-prost-build",
]
+[[package]]
+name = "scg-observability"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "bytes",
+ "futures",
+ "http-body-util",
+ "hyper",
+ "hyper-util",
+ "opentelemetry",
+ "opentelemetry-otlp",
+ "opentelemetry-semantic-conventions",
+ "opentelemetry-stdout",
+ "opentelemetry_sdk",
+ "prometheus",
+ "thiserror 2.0.20",
+ "tokio",
+ "tonic",
+ "tracing",
+ "tracing-opentelemetry",
+ "tracing-subscriber",
+ "uuid",
+]
+
[[package]]
name = "scg-routing"
version = "0.1.0"
@@ -1565,6 +1777,15 @@ dependencies = [
"unsafe-libyaml",
]
+[[package]]
+name = "sharded-slab"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6"
+dependencies = [
+ "lazy_static",
+]
+
[[package]]
name = "shlex"
version = "2.0.1"
@@ -1599,7 +1820,7 @@ checksum =
"0d585997b0ac10be3c5ee635f1bab02d512760d14b7c468801ac8a01d9ae5f1d"
dependencies = [
"num-bigint",
"num-traits",
- "thiserror",
+ "thiserror 2.0.20",
"time",
]
@@ -1708,13 +1929,33 @@ dependencies = [
"windows-sys 0.61.2",
]
+[[package]]
+name = "thiserror"
+version = "1.0.69"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52"
+dependencies = [
+ "thiserror-impl 1.0.69",
+]
+
[[package]]
name = "thiserror"
version = "2.0.20"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f"
dependencies = [
- "thiserror-impl",
+ "thiserror-impl 2.0.20",
+]
+
+[[package]]
+name = "thiserror-impl"
+version = "1.0.69"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
]
[[package]]
@@ -1728,6 +1969,15 @@ dependencies = [
"syn 3.0.3",
]
+[[package]]
+name = "thread_local"
+version = "1.1.10"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070"
+dependencies = [
+ "cfg-if",
+]
+
[[package]]
name = "time"
version = "0.3.55"
@@ -1868,11 +2118,13 @@ dependencies = [
"socket2",
"sync_wrapper",
"tokio",
+ "tokio-rustls",
"tokio-stream",
"tower",
"tower-layer",
"tower-service",
"tracing",
+ "webpki-roots",
]
[[package]]
@@ -1992,6 +2244,65 @@ source =
"registry+https://github.com/rust-lang/crates.io-index";
checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a"
dependencies = [
"once_cell",
+ "valuable",
+]
+
+[[package]]
+name = "tracing-log"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3"
+dependencies = [
+ "log",
+ "once_cell",
+ "tracing-core",
+]
+
+[[package]]
+name = "tracing-opentelemetry"
+version = "0.32.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "1ac28f2d093c6c477eaa76b23525478f38de514fa9aeb1285738d4b97a9552fc"
+dependencies = [
+ "js-sys",
+ "opentelemetry",
+ "smallvec",
+ "tracing",
+ "tracing-core",
+ "tracing-log",
+ "tracing-subscriber",
+ "web-time",
+]
+
+[[package]]
+name = "tracing-serde"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "704b1aeb7be0d0a84fc9828cae51dab5970fee5088f83d1dd7ee6f6246fc6ff1"
+dependencies = [
+ "serde",
+ "tracing-core",
+]
+
+[[package]]
+name = "tracing-subscriber"
+version = "0.3.23"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319"
+dependencies = [
+ "matchers",
+ "nu-ansi-term",
+ "once_cell",
+ "regex-automata",
+ "serde",
+ "serde_json",
+ "sharded-slab",
+ "smallvec",
+ "thread_local",
+ "tracing",
+ "tracing-core",
+ "tracing-log",
+ "tracing-serde",
]
[[package]]
@@ -2048,6 +2359,23 @@ version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be"
+[[package]]
+name = "uuid"
+version = "1.24.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239"
+dependencies = [
+ "getrandom 0.4.3",
+ "js-sys",
+ "wasm-bindgen",
+]
+
+[[package]]
+name = "valuable"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65"
+
[[package]]
name = "version_check"
version = "0.9.5"
@@ -2069,6 +2397,15 @@ version = "0.11.1+wasi-snapshot-preview1"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
+[[package]]
+name = "wasip2"
+version = "1.0.4+wasi-0.2.12"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487"
+dependencies = [
+ "wit-bindgen",
+]
+
[[package]]
name = "wasm-bindgen"
version = "0.2.127"
@@ -2264,6 +2601,12 @@ dependencies = [
"url",
]
+[[package]]
+name = "wit-bindgen"
+version = "0.57.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e"
+
[[package]]
name = "writeable"
version = "0.6.3"
diff --git a/Cargo.toml b/Cargo.toml
index c8b3f91..754343f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -20,6 +20,7 @@ members = [
"crates/routing",
"crates/config",
"crates/auth",
+ "crates/observability",
]
[workspace.package]
diff --git a/crates/observability/Cargo.toml b/crates/observability/Cargo.toml
new file mode 100644
index 0000000..4a1cf4d
--- /dev/null
+++ b/crates/observability/Cargo.toml
@@ -0,0 +1,58 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+[package]
+name = "scg-observability"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+authors.workspace = true
+repository.workspace = true
+readme.workspace = true
+keywords.workspace = true
+categories.workspace = true
+description = "Prometheus metrics, OpenTelemetry tracing, and admin HTTP
server for the Spark Connect Gateway."
+publish = false
+
+[dependencies]
+tokio = { workspace = true }
+tonic = { workspace = true }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
+prometheus = { workspace = true }
+uuid = { workspace = true }
+hyper = { workspace = true }
+hyper-util = { workspace = true }
+http-body-util = { workspace = true }
+bytes = { workspace = true }
+thiserror = { workspace = true }
+anyhow = { workspace = true }
+opentelemetry = { workspace = true }
+opentelemetry_sdk = { workspace = true }
+opentelemetry-otlp = { workspace = true }
+opentelemetry-semantic-conventions = { workspace = true }
+tracing-opentelemetry = { workspace = true }
+
+[features]
+# Enables `install_test_subscriber` — used by integration tests in
+# downstream crates so they can observe spans via
+# `opentelemetry_sdk::trace::InMemorySpanExporter` without each test
+# binary respelling the wiring.
+testing = ["opentelemetry_sdk/testing"]
+
+[dev-dependencies]
+opentelemetry-stdout = { workspace = true }
+opentelemetry_sdk = { workspace = true, features = ["rt-tokio", "testing"] }
+futures = { workspace = true }
diff --git a/crates/observability/src/admin.rs
b/crates/observability/src/admin.rs
new file mode 100644
index 0000000..362a02d
--- /dev/null
+++ b/crates/observability/src/admin.rs
@@ -0,0 +1,182 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Admin HTTP server: `/metrics`, `/healthz`, `/readyz`.
+//!
+//! Bound on a separate `admin_addr` from the gRPC server so that:
+//!
+//! * scrape traffic doesn't compete with hot-path gRPC for the
+//! listener's accept queue,
+//! * the gRPC port can be exposed externally while `/metrics` stays
+//! internal-only,
+//! * a Kubernetes Service can target `:9090` for liveness / readiness
+//! probes without leaking gRPC.
+//!
+//! Implementation uses a minimal Hyper 1.x server. We don't pull in
+//! `axum` or `warp` because the routing surface is three endpoints.
+
+use std::convert::Infallible;
+use std::future::Future;
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+use bytes::Bytes;
+use http_body_util::Full;
+use hyper::body::Incoming;
+use hyper::server::conn::http1;
+use hyper::service::service_fn;
+use hyper::{Method, Request, Response, StatusCode};
+use hyper_util::rt::TokioIo;
+use prometheus::Encoder;
+use tokio::net::TcpListener;
+use tracing::{debug, info, warn};
+
+use crate::metrics::Metrics;
+
+/// Configuration for the admin server.
+#[derive(Debug, Clone)]
+pub struct AdminConfig {
+ /// Address to bind, e.g. `0.0.0.0:9090`.
+ pub bind_addr: SocketAddr,
+}
+
+/// Read-only readiness signal driven by the gateway. Static-pool
+/// deployments mark themselves ready immediately; dynamic-pool
+/// deployments mark themselves ready once the watcher emits its first
+/// healthy backend list.
+#[derive(Clone)]
+pub struct ReadinessProbe {
+ inner: Arc<std::sync::atomic::AtomicBool>,
+}
+
+impl ReadinessProbe {
+ pub fn new(initial: bool) -> Self {
+ Self {
+ inner: Arc::new(std::sync::atomic::AtomicBool::new(initial)),
+ }
+ }
+ pub fn mark_ready(&self) {
+ self.inner.store(true, std::sync::atomic::Ordering::Release);
+ }
+ /// Flip readiness back to false. Used during drain on SIGTERM so
+ /// `/readyz` returns 503 and Kubernetes removes this pod from the
+ /// Service endpoints before any in-flight RPCs are torn down.
+ pub fn mark_not_ready(&self) {
+ self.inner
+ .store(false, std::sync::atomic::Ordering::Release);
+ }
+ pub fn is_ready(&self) -> bool {
+ self.inner.load(std::sync::atomic::Ordering::Acquire)
+ }
+}
+
+impl Default for ReadinessProbe {
+ fn default() -> Self {
+ Self::new(false)
+ }
+}
+
+/// Run the admin server until `shutdown` resolves. Returns when the
+/// listener has stopped accepting new connections.
+///
+/// Each accepted connection is handled on a fresh tokio task.
+pub async fn serve_admin<F>(
+ cfg: AdminConfig,
+ metrics: Metrics,
+ readiness: ReadinessProbe,
+ shutdown: F,
+) -> anyhow::Result<()>
+where
+ F: Future<Output = ()> + Send + 'static,
+{
+ let listener = TcpListener::bind(cfg.bind_addr).await?;
+ info!(addr = %cfg.bind_addr, "admin server listening");
+
+ tokio::pin!(shutdown);
+
+ loop {
+ tokio::select! {
+ _ = &mut shutdown => {
+ info!("admin server shutting down");
+ return Ok(());
+ }
+ accept = listener.accept() => {
+ let (stream, peer) = match accept {
+ Ok(x) => x,
+ Err(e) => {
+ warn!(error = %e, "admin accept error");
+ continue;
+ }
+ };
+ debug!(%peer, "admin: connection accepted");
+ let metrics = metrics.clone();
+ let readiness = readiness.clone();
+ tokio::spawn(async move {
+ let io = TokioIo::new(stream);
+ let svc = service_fn(move |req| {
+ let metrics = metrics.clone();
+ let readiness = readiness.clone();
+ async move { Ok::<_, Infallible>(handle(req, &metrics,
&readiness).await) }
+ });
+ if let Err(e) = http1::Builder::new().serve_connection(io,
svc).await {
+ debug!(error = %e, "admin connection ended");
+ }
+ });
+ }
+ }
+ }
+}
+
+async fn handle(
+ req: Request<Incoming>,
+ metrics: &Metrics,
+ readiness: &ReadinessProbe,
+) -> Response<Full<Bytes>> {
+ match (req.method(), req.uri().path()) {
+ (&Method::GET, "/metrics") => render_metrics(metrics),
+ (&Method::GET, "/healthz") => simple(StatusCode::OK, "ok"),
+ (&Method::GET, "/readyz") => {
+ if readiness.is_ready() {
+ simple(StatusCode::OK, "ready")
+ } else {
+ simple(StatusCode::SERVICE_UNAVAILABLE, "not ready")
+ }
+ }
+ _ => simple(StatusCode::NOT_FOUND, "not found"),
+ }
+}
+
+fn render_metrics(metrics: &Metrics) -> Response<Full<Bytes>> {
+ let encoder = prometheus::TextEncoder::new();
+ let mfs = metrics.registry().gather();
+ let mut buf = Vec::new();
+ if let Err(e) = encoder.encode(&mfs, &mut buf) {
+ warn!(error = %e, "failed to encode metrics");
+ return simple(StatusCode::INTERNAL_SERVER_ERROR, "encoding failed");
+ }
+ Response::builder()
+ .status(StatusCode::OK)
+ .header("content-type", encoder.format_type())
+ .body(Full::new(Bytes::from(buf)))
+ .expect("static response builder")
+}
+
+fn simple(status: StatusCode, body: &'static str) -> Response<Full<Bytes>> {
+ Response::builder()
+ .status(status)
+ .header("content-type", "text/plain; charset=utf-8")
+ .body(Full::new(Bytes::from(body)))
+ .expect("static response builder")
+}
diff --git a/crates/observability/src/lib.rs b/crates/observability/src/lib.rs
new file mode 100644
index 0000000..8261b59
--- /dev/null
+++ b/crates/observability/src/lib.rs
@@ -0,0 +1,51 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Metrics, structured spans, and the admin HTTP server.
+//!
+//! The crate exposes three things:
+//!
+//! * [`Metrics`] — a `Clone`-able handle owning a Prometheus `Registry`
+//! and the per-RPC counters / histograms. Hand one to anything that
+//! needs to instrument requests.
+//! * [`request_id`] — generates a per-RPC UUID v4 used as the
+//! correlation ID stamped into both the `tracing::Span` and the
+//! outbound gRPC `x-request-id` metadata.
+//! * [`admin`] — a small Hyper-based HTTP server that serves
+//! `/metrics`, `/healthz`, `/readyz` on a separate admin port.
+
+pub mod admin;
+pub mod metrics;
+pub mod tracing;
+
+pub use admin::{serve_admin, AdminConfig, ReadinessProbe};
+pub use metrics::{Metrics, MetricsError, RpcGuard, StreamGuard};
+#[cfg(feature = "testing")]
+pub use tracing::install_test_subscriber;
+pub use tracing::{
+ extract_parent, init_tracing, inject_context, TracingConfig, TracingError,
TracingHandle,
+ TRACEPARENT_HEADER, TRACER_NAME,
+};
+
+use uuid::Uuid;
+
+/// Generate a fresh correlation ID for the current RPC.
+pub fn request_id() -> String {
+ Uuid::new_v4().to_string()
+}
+
+/// gRPC metadata key under which the gateway forwards correlation IDs
+/// to backends. Public so the proxy crate can use the same constant.
+pub const REQUEST_ID_HEADER: &str = "x-request-id";
diff --git a/crates/observability/src/metrics.rs
b/crates/observability/src/metrics.rs
new file mode 100644
index 0000000..4957956
--- /dev/null
+++ b/crates/observability/src/metrics.rs
@@ -0,0 +1,406 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Prometheus metrics for the gateway.
+//!
+//! Naming follows the Prometheus best practices: `scg_*` prefix,
+//! `_total` suffix on counters, `_seconds` suffix on duration
+//! histograms, label cardinality kept small.
+//!
+//! Cardinality budget:
+//!
+//! * `rpc` — fixed set of 12 Spark Connect RPC names. Bounded.
+//! * `code` — gRPC status codes ("OK", "Unauthenticated", …). ~16 values.
+//! * `reason` / `scope` — small fixed enums.
+//! * `tenant` — appears **only** on the two rate-limit counters.
+//! Cardinality is bounded by the tenant set the resolver admits
+//! (operator-configured); deployments deriving tenants from token
+//! claims under a permissive unknown-tenant policy should count
+//! that toward their cardinality budget.
+//! * **Never** `user_id`, `session_id`, or `operation_id` labels —
+//! those are unbounded and would blow up Prometheus. Per-user and
+//! per-session observability belongs in logs, audit events, and
+//! trace attributes, not in metric labels.
+
+use std::sync::Arc;
+
+use prometheus::{
+ register_histogram_vec_with_registry,
register_int_counter_vec_with_registry,
+ register_int_gauge_with_registry, HistogramTimer, HistogramVec,
IntCounterVec, IntGauge,
+ Registry,
+};
+
+#[derive(Debug, thiserror::Error)]
+pub enum MetricsError {
+ #[error("registering metric: {0}")]
+ Register(#[from] prometheus::Error),
+}
+
+/// Cheap-to-clone handle to the gateway's Prometheus registry and
+/// per-RPC metric families. All metric families are eagerly registered
+/// at construction so scrapes always return a complete metric set, even
+/// before any traffic has flowed.
+#[derive(Clone)]
+pub struct Metrics {
+ inner: Arc<MetricsInner>,
+}
+
+struct MetricsInner {
+ registry: Registry,
+
+ // Per-RPC families.
+ rpcs_total: IntCounterVec,
+ rpc_duration_seconds: HistogramVec,
+
+ // Auth.
+ auth_failures_total: IntCounterVec,
+
+ // Pool / streams gauges.
+ backend_pool_size: IntGauge,
+ active_streams: IntGauge,
+
+ // Rate limit.
+ rate_limit_rejected_total: IntCounterVec,
+ /// Redis-side errors observed by the distributed rate limiter.
+ /// Counts errors, not rejects — a fail-open deployment
+ /// increments this without firing `rate_limit_rejected_total`.
+ /// `reason` is one of a small fixed set: `tenant_bucket`,
+ /// `user_bucket`.
+ rate_limit_redis_errors_total: IntCounterVec,
+}
+
+impl Metrics {
+ /// Build a fresh `Metrics` against a new `Registry`. Tests and the
+ /// gateway main both use this; every gateway process owns one
+ /// `Metrics` for the lifetime of the process.
+ pub fn new() -> Result<Self, MetricsError> {
+ let registry = Registry::new();
+ let rpcs_total = register_int_counter_vec_with_registry!(
+ "scg_rpcs_total",
+ "Total Spark Connect RPCs handled by the gateway, labelled by
method and final gRPC code.",
+ &["rpc", "code"],
+ registry,
+ )?;
+ let rpc_duration_seconds = register_histogram_vec_with_registry!(
+ "scg_rpc_duration_seconds",
+ "Per-RPC handler duration (gateway-side, end-to-end including
backend forward).",
+ &["rpc"],
+ // Buckets: 0.5 ms, 1 ms, 2.5 ms, 5 ms, 10 ms, 25 ms, 50 ms,
+ // 100 ms, 250 ms, 500 ms, 1 s, 2.5 s, 5 s, 10 s, 30 s, 60 s.
+ // Spans the range from in-memory unary RPCs to long-running
+ // ExecutePlan streams.
+ vec![
+ 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25,
0.5, 1.0, 2.5, 5.0,
+ 10.0, 30.0, 60.0,
+ ],
+ registry,
+ )?;
+ let auth_failures_total = register_int_counter_vec_with_registry!(
+ "scg_auth_failures_total",
+ "Authentication failures, labelled by reason (e.g. missing_token,
invalid_token, expired).",
+ &["reason"],
+ registry,
+ )?;
+ let backend_pool_size = register_int_gauge_with_registry!(
+ "scg_backend_pool_size",
+ "Current number of healthy backends the gateway can route to.",
+ registry,
+ )?;
+ let active_streams = register_int_gauge_with_registry!(
+ "scg_active_streams",
+ "Currently in-flight server-streaming or client-streaming RPCs.",
+ registry,
+ )?;
+ let rate_limit_rejected_total =
register_int_counter_vec_with_registry!(
+ "scg_rate_limit_rejected_total",
+ "RPCs rejected by per-tenant rate limiting, labelled by tenant and
scope (tenant|user).",
+ &["tenant", "scope"],
+ registry,
+ )?;
+ let rate_limit_redis_errors_total =
register_int_counter_vec_with_registry!(
+ "scg_rate_limit_redis_errors_total",
+ "Backend errors from the Redis-backed rate limiter. Counts
failures, not rejects.",
+ &["tenant", "reason"],
+ registry,
+ )?;
+ Ok(Self {
+ inner: Arc::new(MetricsInner {
+ registry,
+ rpcs_total,
+ rpc_duration_seconds,
+ auth_failures_total,
+ backend_pool_size,
+ active_streams,
+ rate_limit_rejected_total,
+ rate_limit_redis_errors_total,
+ }),
+ })
+ }
+
+ /// Bump `scg_rate_limit_redis_errors_total{tenant, reason}`.
+ /// `reason` is a small fixed-cardinality string from the limiter
+ /// (`tenant_bucket`, `user_bucket`).
+ pub fn record_rate_limit_redis_error(&self, tenant: &str, reason: &str) {
+ self.inner
+ .rate_limit_redis_errors_total
+ .with_label_values(&[tenant, reason])
+ .inc();
+ }
+
+ /// Bump `scg_rate_limit_rejected_total{tenant, scope}`. `scope`
+ /// is the fixed-cardinality string `"tenant"` or `"user"`. The
+ /// label cardinality is bounded by the configured-tenant set —
+ /// callers should not pass user-input tenants that aren't in
+ /// the resolver's allowlist.
+ pub fn record_rate_limit_reject(&self, tenant: &str, scope: &str) {
+ self.inner
+ .rate_limit_rejected_total
+ .with_label_values(&[tenant, scope])
+ .inc();
+ }
+
+ /// Underlying `Registry` — handed to the admin HTTP server so the
+ /// `/metrics` endpoint can render it in the Prometheus exposition
+ /// format.
+ pub fn registry(&self) -> &Registry {
+ &self.inner.registry
+ }
+
+ /// Start a per-RPC timer + observation guard. The guard records
+ /// completion (with the final gRPC code) on Drop, so callers don't
+ /// have to remember to call anything from error paths. See
+ /// [`RpcGuard::record`] to override the default code.
+ pub fn rpc_guard(&self, rpc: &'static str) -> RpcGuard {
+ let timer = self
+ .inner
+ .rpc_duration_seconds
+ .with_label_values(&[rpc])
+ .start_timer();
+ RpcGuard {
+ metrics: self.clone(),
+ rpc,
+ code: "Cancelled", // Replaced before drop unless caller bails
early.
+ recorded: false,
+ _timer: timer,
+ }
+ }
+
+ /// Bump `scg_rpcs_total{rpc, code}` directly. Called by the
+ /// `RpcGuard` Drop impl so the counter still reflects requests
+ /// where the handler returned early.
+ pub(crate) fn record_rpc(&self, rpc: &'static str, code: &str) {
+ self.inner.rpcs_total.with_label_values(&[rpc, code]).inc();
+ }
+
+ /// Bump the auth-failures counter. `reason` should be a small
+ /// closed set: "missing_token", "invalid_token", "expired",
+ /// "unknown_kid", "missing_tenant", "unknown".
+ pub fn record_auth_failure(&self, reason: &'static str) {
+ self.inner
+ .auth_failures_total
+ .with_label_values(&[reason])
+ .inc();
+ }
+
+ /// Set the live backend-pool size. Call from the K8s watcher / on
+ /// pool reconfiguration.
+ pub fn set_backend_pool_size(&self, n: i64) {
+ self.inner.backend_pool_size.set(n);
+ }
+
+ /// Increment the active-streams gauge for the lifetime of `_guard`.
+ /// Use the returned [`StreamGuard`] for RAII-style accounting.
+ pub fn stream_guard(&self) -> StreamGuard {
+ self.inner.active_streams.inc();
+ StreamGuard {
+ metrics: self.clone(),
+ }
+ }
+
+ /// Read the live `scg_active_streams` count. Used by the
+ /// graceful-shutdown loop to decide whether the server can stop.
+ pub fn active_streams_value(&self) -> i64 {
+ self.inner.active_streams.get()
+ }
+}
+
+impl std::fmt::Debug for Metrics {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("Metrics")
+ .field("registered_families", &self.inner.registry.gather().len())
+ .finish()
+ }
+}
+
+/// Per-RPC observation guard. Records the final gRPC code via
+/// [`RpcGuard::record`]; if the guard is dropped without `record`, the
+/// outcome is recorded as `Cancelled` so we don't lose the request from
+/// the totals.
+#[must_use = "RpcGuard records on Drop; capture it for the lifetime of the
handler"]
+pub struct RpcGuard {
+ metrics: Metrics,
+ rpc: &'static str,
+ code: &'static str,
+ recorded: bool,
+ _timer: HistogramTimer,
+}
+
+impl RpcGuard {
+ /// Override the final gRPC code reported by this guard. Common
+ /// values: "OK", "Unauthenticated", "Unavailable", "Internal".
+ /// Calling `record` more than once is a no-op after the first.
+ pub fn record(&mut self, code: &'static str) {
+ if self.recorded {
+ return;
+ }
+ self.recorded = true;
+ self.code = code;
+ }
+}
+
+impl Drop for RpcGuard {
+ fn drop(&mut self) {
+ // If the handler returned early without calling `record`, we
+ // still want a count so the user can see the totals match
+ // request volume. Default code "Cancelled" stays.
+ self.metrics.record_rpc(self.rpc, self.code);
+ // _timer also drops here, recording the histogram observation.
+ }
+}
+
+/// RAII guard that decrements `scg_active_streams` on drop.
+#[must_use = "drop the guard at the end of the streaming RPC"]
+pub struct StreamGuard {
+ metrics: Metrics,
+}
+
+impl Drop for StreamGuard {
+ fn drop(&mut self) {
+ self.metrics.inner.active_streams.dec();
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn read_counter(m: &Metrics, name: &str, labels: &[(&str, &str)]) -> u64 {
+ let mfs = m.registry().gather();
+ for mf in mfs {
+ if mf.name() != name {
+ continue;
+ }
+ for metric in mf.get_metric() {
+ let matches = labels.iter().all(|(k, v)| {
+ metric
+ .get_label()
+ .iter()
+ .any(|l| l.name() == *k && l.value() == *v)
+ });
+ if matches {
+ return metric.get_counter().value() as u64;
+ }
+ }
+ }
+ 0
+ }
+
+ fn read_gauge(m: &Metrics, name: &str) -> i64 {
+ let mfs = m.registry().gather();
+ for mf in mfs {
+ if mf.name() == name {
+ if let Some(metric) = mf.get_metric().first() {
+ return metric.get_gauge().value() as i64;
+ }
+ }
+ }
+ 0
+ }
+
+ #[test]
+ fn rpc_guard_records_ok_on_record() {
+ let m = Metrics::new().unwrap();
+ {
+ let mut g = m.rpc_guard("Config");
+ g.record("OK");
+ }
+ assert_eq!(
+ read_counter(&m, "scg_rpcs_total", &[("rpc", "Config"), ("code",
"OK")]),
+ 1
+ );
+ }
+
+ #[test]
+ fn rpc_guard_defaults_to_cancelled_on_early_drop() {
+ let m = Metrics::new().unwrap();
+ {
+ let _g = m.rpc_guard("ExecutePlan");
+ // simulate handler bailing without explicit record
+ }
+ assert_eq!(
+ read_counter(
+ &m,
+ "scg_rpcs_total",
+ &[("rpc", "ExecutePlan"), ("code", "Cancelled")]
+ ),
+ 1
+ );
+ }
+
+ #[test]
+ fn auth_failure_increments() {
+ let m = Metrics::new().unwrap();
+ m.record_auth_failure("invalid_token");
+ m.record_auth_failure("invalid_token");
+ m.record_auth_failure("missing_token");
+ assert_eq!(
+ read_counter(
+ &m,
+ "scg_auth_failures_total",
+ &[("reason", "invalid_token")]
+ ),
+ 2
+ );
+ assert_eq!(
+ read_counter(
+ &m,
+ "scg_auth_failures_total",
+ &[("reason", "missing_token")]
+ ),
+ 1
+ );
+ }
+
+ #[test]
+ fn pool_size_gauge_is_settable() {
+ let m = Metrics::new().unwrap();
+ m.set_backend_pool_size(5);
+ assert_eq!(read_gauge(&m, "scg_backend_pool_size"), 5);
+ m.set_backend_pool_size(0);
+ assert_eq!(read_gauge(&m, "scg_backend_pool_size"), 0);
+ }
+
+ #[test]
+ fn stream_guard_increments_then_decrements() {
+ let m = Metrics::new().unwrap();
+ let g1 = m.stream_guard();
+ let g2 = m.stream_guard();
+ assert_eq!(read_gauge(&m, "scg_active_streams"), 2);
+ drop(g1);
+ assert_eq!(read_gauge(&m, "scg_active_streams"), 1);
+ drop(g2);
+ assert_eq!(read_gauge(&m, "scg_active_streams"), 0);
+ }
+}
diff --git a/crates/observability/src/tracing.rs
b/crates/observability/src/tracing.rs
new file mode 100644
index 0000000..e94c0e2
--- /dev/null
+++ b/crates/observability/src/tracing.rs
@@ -0,0 +1,340 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Distributed tracing — OpenTelemetry exporter + W3C traceparent
+//! propagation across the gateway → backend hop.
+//!
+//! Two responsibilities:
+//!
+//! * `init_tracing(cfg)` — install the global `tracing_subscriber` with a
+//! `tracing-opentelemetry` layer that forwards spans to an OTLP/gRPC
+//! exporter. Returns a [`TracingHandle`] whose `shutdown()` flushes
+//! spans before the process exits.
+//! * `extract_parent` / `inject_traceparent` — translate W3C traceparent
+//! metadata between tonic `MetadataMap`s and OpenTelemetry `Context`s,
+//! so a request's parent trace from the inbound side becomes the
+//! parent of every outbound RPC the gateway makes.
+//!
+//! Tracing is disabled by default; passing a `TracingConfig` with
+//! `endpoint == None` returns a no-op handle and installs only the
+//! existing JSON formatter (so logs keep working unchanged).
+
+use std::time::Duration;
+
+use opentelemetry::global;
+use opentelemetry::propagation::{Extractor, Injector, TextMapPropagator};
+use opentelemetry::trace::TracerProvider as _;
+use opentelemetry::{Context, KeyValue};
+use opentelemetry_otlp::{SpanExporter, WithExportConfig};
+use opentelemetry_sdk::propagation::TraceContextPropagator;
+use opentelemetry_sdk::trace::{Sampler, SdkTracerProvider};
+use opentelemetry_sdk::Resource;
+use opentelemetry_semantic_conventions::resource as semres;
+use tonic::metadata::MetadataMap;
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
+use tracing_subscriber::EnvFilter;
+
+/// Tracer name reported on every span emitted by the gateway. The
+/// service name (an OTel `Resource` attribute) is set separately via
+/// `TracingConfig::service_name`.
+pub const TRACER_NAME: &str = "scg.gateway";
+
+/// W3C trace-context header. Lower-case because gRPC metadata keys are
+/// case-insensitive and stored lower-cased.
+pub const TRACEPARENT_HEADER: &str = "traceparent";
+pub const TRACESTATE_HEADER: &str = "tracestate";
+
+/// Runtime tracing configuration. Constructed by the gateway from its
+/// YAML config, or by tests directly.
+#[derive(Debug, Clone)]
+pub struct TracingConfig {
+ /// Logical service name reported as `service.name` on every span.
+ pub service_name: String,
+ /// Service version reported as `service.version`.
+ pub service_version: String,
+ /// OTLP/gRPC collector endpoint, e.g. `http://otel-collector:4317`.
+ /// `None` disables span export — only the JSON log formatter is
+ /// installed.
+ pub endpoint: Option<String>,
+ /// `TraceIdRatioBased` ratio in `[0.0, 1.0]`. Wrapped in
+ /// `ParentBased` so a sampled parent always wins.
+ pub sample_ratio: f64,
+ /// Per-batch export deadline for the OTLP exporter.
+ pub export_timeout: Duration,
+}
+
+impl Default for TracingConfig {
+ fn default() -> Self {
+ Self {
+ service_name: "spark-connect-gateway".into(),
+ service_version: env!("CARGO_PKG_VERSION").into(),
+ endpoint: None,
+ sample_ratio: 1.0,
+ export_timeout: Duration::from_secs(10),
+ }
+ }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum TracingError {
+ #[error("building OTLP exporter: {0}")]
+ Exporter(#[from] opentelemetry_sdk::error::OTelSdkError),
+ #[error("setting tracing subscriber: {0}")]
+ SetSubscriber(#[from] tracing_subscriber::util::TryInitError),
+ #[error("OTLP exporter init: {0}")]
+ OtlpInit(String),
+}
+
+/// RAII handle returned by [`init_tracing`]. When the gateway shuts
+/// down, call [`TracingHandle::shutdown`] to flush in-flight spans
+/// before the process exits.
+pub struct TracingHandle {
+ provider: Option<SdkTracerProvider>,
+}
+
+impl TracingHandle {
+ /// A handle that does nothing on shutdown. Used for the
+ /// `endpoint == None` case and in tests where the caller installs
+ /// its own provider.
+ pub fn noop() -> Self {
+ Self { provider: None }
+ }
+
+ /// Force-flush and shut down the OTLP exporter. Safe to call
+ /// multiple times; subsequent calls are no-ops.
+ pub fn shutdown(&mut self) {
+ if let Some(p) = self.provider.take() {
+ // `shutdown` returns Result; we log on error but never
+ // propagate — at this point the process is exiting.
+ if let Err(e) = p.shutdown() {
+ tracing::warn!(error = %e, "OTel tracer provider shutdown
failed");
+ }
+ }
+ }
+}
+
+impl Drop for TracingHandle {
+ fn drop(&mut self) {
+ self.shutdown();
+ }
+}
+
+/// Install the global tracing subscriber:
+///
+/// * Always: a `tracing_subscriber` JSON-formatter layer driven by
+/// `RUST_LOG` (defaults to `info`).
+/// * If `cfg.endpoint.is_some()`: also a `tracing-opentelemetry` layer
+/// that exports spans to the configured OTLP/gRPC endpoint, plus a
+/// global `TraceContextPropagator` so W3C traceparent extraction /
+/// injection works.
+///
+/// Returns a [`TracingHandle`] whose `shutdown` must be called before
+/// the process exits, otherwise the final batch of spans may be lost.
+pub fn init_tracing(cfg: TracingConfig) -> Result<TracingHandle, TracingError>
{
+ let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_|
EnvFilter::new("info"));
+ let fmt_layer = tracing_subscriber::fmt::layer().json().with_target(false);
+
+ let Some(endpoint) = cfg.endpoint.clone() else {
+ // Logs only — keep the existing JSON formatter behaviour.
+ tracing_subscriber::registry()
+ .with(env_filter)
+ .with(fmt_layer)
+ .try_init()?;
+ return Ok(TracingHandle::noop());
+ };
+
+ let exporter = SpanExporter::builder()
+ .with_tonic()
+ .with_endpoint(endpoint)
+ .with_timeout(cfg.export_timeout)
+ .build()
+ .map_err(|e| TracingError::OtlpInit(e.to_string()))?;
+
+ let resource = Resource::builder()
+ .with_attribute(KeyValue::new(
+ semres::SERVICE_NAME,
+ cfg.service_name.clone(),
+ ))
+ .with_attribute(KeyValue::new(
+ semres::SERVICE_VERSION,
+ cfg.service_version.clone(),
+ ))
+ .build();
+
+ // Wrap the configured ratio in ParentBased so a remote sampled
+ // decision always wins; fresh local roots fall through to the
+ // ratio sampler.
+ //
+ // KNOWN LIMITATION: when the inbound request carries a W3C
+ // `traceparent` header, the OTel SDK's interaction with
+ // `tracing-opentelemetry::set_parent` does not propagate the
+ // inbound trace_id onto the gateway's span (the parent_cx is set
+ // via `Context::with_remote_span_context`, which is invisible to
+ // `Context::has_active_span()` and to several internal codepaths
+ // in `tracer::build_with_context`). The resulting gateway span
+ // is dropped before reaching the OTLP exporter — fmt-side logs
+ // still record it (so JSON logs + correlation IDs work), but the
+ // distributed trace cannot link gateway → backend hops on those
+ // RPCs. Spans for RPCs without an inbound traceparent (i.e. the
+ // gateway is the trace root) export normally. Tracking upstream
+ // for resolution; switch back to a richer sampler when fixed.
+ let sampler = Sampler::ParentBased(Box::new(Sampler::TraceIdRatioBased(
+ cfg.sample_ratio.clamp(0.0, 1.0),
+ )));
+
+ let provider = SdkTracerProvider::builder()
+ .with_batch_exporter(exporter)
+ .with_resource(resource)
+ .with_sampler(sampler)
+ .build();
+
+ // Install the W3C traceparent propagator globally so extract /
+ // inject helpers below pick it up without re-creating an instance.
+ global::set_text_map_propagator(TraceContextPropagator::new());
+
+ let tracer = provider.tracer(TRACER_NAME);
+ let otel_layer = tracing_opentelemetry::layer().with_tracer(tracer);
+
+ tracing_subscriber::registry()
+ .with(env_filter)
+ .with(fmt_layer)
+ .with(otel_layer)
+ .try_init()?;
+
+ Ok(TracingHandle {
+ provider: Some(provider),
+ })
+}
+
+/// Tonic [`MetadataMap`] adapter for OpenTelemetry's `Extractor` /
+/// `Injector` traits. Both use lower-case keys.
+struct MetadataCarrier<'a>(&'a MetadataMap);
+
+impl<'a> Extractor for MetadataCarrier<'a> {
+ fn get(&self, key: &str) -> Option<&str> {
+ self.0.get(key).and_then(|v| v.to_str().ok())
+ }
+
+ fn keys(&self) -> Vec<&str> {
+ self.0
+ .keys()
+ .map(|k| match k {
+ tonic::metadata::KeyRef::Ascii(k) => k.as_str(),
+ tonic::metadata::KeyRef::Binary(k) => k.as_str(),
+ })
+ .collect()
+ }
+}
+
+struct MetadataInjector<'a>(&'a mut MetadataMap);
+
+impl<'a> Injector for MetadataInjector<'a> {
+ fn set(&mut self, key: &str, value: String) {
+ if let (Ok(k), Ok(v)) = (
+
tonic::metadata::MetadataKey::<tonic::metadata::Ascii>::from_bytes(key.as_bytes()),
+ tonic::metadata::MetadataValue::try_from(value),
+ ) {
+ self.0.insert(k, v);
+ }
+ }
+}
+
+/// Extract a W3C trace-context parent from inbound gRPC metadata.
+/// Returns the empty `Context` when no traceparent is present (the
+/// resulting span will be a fresh trace root).
+pub fn extract_parent(metadata: &MetadataMap) -> Context {
+ let propagator = TraceContextPropagator::new();
+ propagator.extract(&MetadataCarrier(metadata))
+}
+
+/// Inject the OpenTelemetry context held by `cx` into outbound gRPC
+/// metadata as a W3C `traceparent` (and `tracestate` when present).
+pub fn inject_context(cx: &Context, metadata: &mut MetadataMap) {
+ let propagator = TraceContextPropagator::new();
+ propagator.inject_context(cx, &mut MetadataInjector(metadata));
+}
+
+/// Test-only helper: install a global `tracing` subscriber that
+/// forwards spans to the supplied [`SdkTracerProvider`], plus the W3C
+/// trace-context propagator so [`extract_parent`] / [`inject_context`]
+/// work. Returns silently if a global subscriber is already installed
+/// (every test in a binary shares the same one).
+///
+/// Pair with `opentelemetry_sdk::trace::InMemorySpanExporter` for
+/// round-trip assertions; see `crates/proxy/tests/tracing_integration.rs`.
+#[cfg(feature = "testing")]
+pub fn install_test_subscriber(provider:
&opentelemetry_sdk::trace::SdkTracerProvider) {
+ use std::sync::Once;
+ static ONCE: Once = Once::new();
+ ONCE.call_once(|| {
+ global::set_text_map_propagator(TraceContextPropagator::new());
+ let tracer = provider.tracer(TRACER_NAME);
+ let layer = tracing_opentelemetry::layer().with_tracer(tracer);
+ let _ = tracing_subscriber::registry().with(layer).try_init();
+ });
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use opentelemetry::trace::TraceContextExt;
+
+ #[test]
+ fn extract_with_no_traceparent_yields_invalid_span_context() {
+ let md = MetadataMap::new();
+ let cx = extract_parent(&md);
+ assert!(!cx.span().span_context().is_valid());
+ }
+
+ #[test]
+ fn inject_then_extract_round_trips_traceparent() {
+ // Build a Context whose active span has a known SpanContext.
+ use opentelemetry::trace::{SpanContext, SpanId, TraceFlags, TraceId,
TraceState};
+ let span_ctx = SpanContext::new(
+ TraceId::from_hex("0af7651916cd43dd8448eb211c80319c").unwrap(),
+ SpanId::from_hex("b7ad6b7169203331").unwrap(),
+ TraceFlags::SAMPLED,
+ true, // remote
+ TraceState::default(),
+ );
+ let cx = Context::new().with_remote_span_context(span_ctx.clone());
+
+ let mut md = MetadataMap::new();
+ inject_context(&cx, &mut md);
+
+ let tp = md
+ .get(TRACEPARENT_HEADER)
+ .expect("traceparent injected")
+ .to_str()
+ .unwrap();
+ // W3C format: 00-<traceid>-<spanid>-<flags>
+ assert!(tp.starts_with("00-0af7651916cd43dd8448eb211c80319c-"));
+
+ let extracted = extract_parent(&md);
+ let extracted_span = extracted.span();
+ let extracted_sc = extracted_span.span_context();
+ assert!(extracted_sc.is_valid());
+ assert_eq!(extracted_sc.trace_id(), span_ctx.trace_id());
+ assert_eq!(extracted_sc.span_id(), span_ctx.span_id());
+ }
+
+ #[test]
+ fn noop_handle_shutdown_is_idempotent() {
+ let mut h = TracingHandle::noop();
+ h.shutdown();
+ h.shutdown(); // must not panic
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
