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 aa16a04 [SPARK-59074] Add scg-pool-k8s crate
aa16a04 is described below
commit aa16a0406cd5dd81759dabd45d361d6893e911f4
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Sat Aug 29 00:44:23 2026 -0700
[SPARK-59074] Add scg-pool-k8s 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-pool-k8s` crate — a Kubernetes service-watch backend
`Pool`.
It watches a `Service`'s `Endpoints` object in a given namespace and
exposes the
ready endpoints as the routing backend set, so the gateway's backend
membership
tracks the Spark Connect pods dynamically. This is the dynamic counterpart
to
the fixed-membership `scg-pool-static`.
`scg-pool-k8s` depends on the `scg-routing` crate whose `Pool` trait it
implements, and pulls in the kube / k8s-openapi client stack. It is
imported as
its own PR (rather than grouped) because of that larger dependency surface.
It
compiles and tests on its own (unit tests plus a router integration test)
and is
appended to the workspace `members` list. It is consumed later by the
`gateway`
binary.
### 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.
### Was this patch authored or co-authored using generative AI tooling?
Yes, co-authored with Claude Code.
Closes #13 from viirya/import-10-pool-k8s.
Authored-by: Liang-Chi Hsieh <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
Cargo.lock | 382 +++++++++++++++++++++++++++-
Cargo.toml | 1 +
crates/pool-k8s/Cargo.toml | 40 +++
crates/pool-k8s/src/lib.rs | 380 +++++++++++++++++++++++++++
crates/pool-k8s/tests/router_integration.rs | 110 ++++++++
5 files changed, 912 insertions(+), 1 deletion(-)
diff --git a/Cargo.lock b/Cargo.lock
index 856f3d0..9f0b511 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2,6 +2,19 @@
# It is not intended for manual editing.
version = 4
+[[package]]
+name = "ahash"
+version = "0.8.12"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75"
+dependencies = [
+ "cfg-if",
+ "getrandom 0.3.4",
+ "once_cell",
+ "version_check",
+ "zerocopy",
+]
+
[[package]]
name = "aho-corasick"
version = "1.1.5"
@@ -11,6 +24,12 @@ dependencies = [
"memchr",
]
+[[package]]
+name = "allocator-api2"
+version = "0.2.21"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
+
[[package]]
name = "android_system_properties"
version = "0.1.6"
@@ -67,6 +86,18 @@ dependencies = [
"xattr",
]
+[[package]]
+name = "async-broadcast"
+version = "0.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "435a87a52755b8f27fcf321ac4f04b2802e337c8c4872923137471ec39c37532"
+dependencies = [
+ "event-listener",
+ "event-listener-strategy",
+ "futures-core",
+ "pin-project-lite",
+]
+
[[package]]
name = "async-lock"
version = "3.4.2"
@@ -173,6 +204,8 @@ source =
"registry+https://github.com/rust-lang/crates.io-index";
checksum = "cffb0e931875b666fc4fcb20fee52e9bbd1ef836fd9e9e04ec21555f9f85f7ef"
dependencies = [
"fastrand",
+ "gloo-timers",
+ "tokio",
]
[[package]]
@@ -503,6 +536,27 @@ dependencies = [
"serde_core",
]
+[[package]]
+name = "derive_more"
+version = "2.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "d751e9e49156b02b44f9c1815bcb94b984cdcc4396ecc32521c739452808b134"
+dependencies = [
+ "derive_more-impl",
+]
+
+[[package]]
+name = "derive_more-impl"
+version = "2.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "799a97264921d8623a957f6c3b9011f3b5492f557bbb7a5a19b7fa6d06ba8dcb"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "rustc_version",
+ "syn 2.0.119",
+]
+
[[package]]
name = "digest"
version = "0.10.7"
@@ -541,12 +595,44 @@ version = "1.0.20"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"
+[[package]]
+name = "educe"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "1d7bc049e1bd8cdeb31b68bbd586a9464ecf9f3944af3958a7a9d0f8b9799417"
+dependencies = [
+ "enum-ordinalize",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
[[package]]
name = "either"
version = "1.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "9e5e8f6c15a24b9a3ee5efec809ccd006d3b30e8b3bb63c39af737c7f87daa1d"
+[[package]]
+name = "enum-ordinalize"
+version = "4.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "89dd01549b09589510cf0647475075d12071456586d70f5c75c98ae2a5537677"
+dependencies = [
+ "enum-ordinalize-derive",
+]
+
+[[package]]
+name = "enum-ordinalize-derive"
+version = "4.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "a65863d15a4ce2888bd2f0f543cc963d3879c3a022c8ee43f6141d479a3ac815"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.3",
+]
+
[[package]]
name = "equivalent"
version = "1.0.2"
@@ -634,6 +720,12 @@ version = "0.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2"
+[[package]]
+name = "foldhash"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb"
+
[[package]]
name = "form_urlencoded"
version = "1.2.2"
@@ -780,6 +872,18 @@ dependencies = [
"wasm-bindgen",
]
+[[package]]
+name = "gloo-timers"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "bbb143cf96099802033e0d4f4963b19fd2e0b728bcf076cd9cf7f6634f092994"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "js-sys",
+ "wasm-bindgen",
+]
+
[[package]]
name = "h2"
version = "0.4.15"
@@ -811,7 +915,18 @@ version = "0.15.5"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1"
dependencies = [
- "foldhash",
+ "foldhash 0.1.5",
+]
+
+[[package]]
+name = "hashbrown"
+version = "0.16.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100"
+dependencies = [
+ "allocator-api2",
+ "equivalent",
+ "foldhash 0.2.0",
]
[[package]]
@@ -847,6 +962,17 @@ dependencies = [
"windows-sys 0.61.2",
]
+[[package]]
+name = "hostname"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "617aaa3557aef3810a6369d0a99fac8a080891b68bd9f9812a1eeda0c0730cbd"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "windows-link",
+]
+
[[package]]
name = "http"
version = "1.5.0"
@@ -937,7 +1063,9 @@ dependencies = [
"http",
"hyper",
"hyper-util",
+ "log",
"rustls",
+ "rustls-native-certs",
"tokio",
"tokio-rustls",
"tower-service",
@@ -1236,6 +1364,41 @@ dependencies = [
"wasm-bindgen",
]
+[[package]]
+name = "json-patch"
+version = "4.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "7421438de105a0827e44fadd05377727847d717c80ce29a229f85fd04c427b72"
+dependencies = [
+ "jsonptr",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.20",
+]
+
+[[package]]
+name = "jsonpath-rust"
+version = "1.0.8"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "2e07021eefc09f897611b067ba7897b5e9266edfc902768418968772e5bb06a7"
+dependencies = [
+ "pest",
+ "pest_derive",
+ "regex",
+ "serde_json",
+ "thiserror 2.0.20",
+]
+
+[[package]]
+name = "jsonptr"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "a5a3cc660ba5d72bce0b3bb295bf20847ccbb40fd423f3f05b61273672e561fe"
+dependencies = [
+ "serde",
+ "serde_json",
+]
+
[[package]]
name = "jsonwebtoken"
version = "9.3.1"
@@ -1251,6 +1414,110 @@ dependencies = [
"simple_asn1",
]
+[[package]]
+name = "k8s-openapi"
+version = "0.27.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "51b326f5219dd55872a72c1b6ddd1b830b8334996c667449c29391d657d78d5e"
+dependencies = [
+ "base64 0.22.1",
+ "jiff",
+ "serde",
+ "serde_json",
+]
+
+[[package]]
+name = "kube"
+version = "3.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "acc5a6a69da2975ed9925d56b5dcfc9cc739b66f37add06785b7c9f6d1e88741"
+dependencies = [
+ "k8s-openapi",
+ "kube-client",
+ "kube-core",
+ "kube-runtime",
+]
+
+[[package]]
+name = "kube-client"
+version = "3.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "0fcaf2d1f1a91e1805d4cd82e8333c022767ae8ffd65909bbef6802733a7dd40"
+dependencies = [
+ "base64 0.22.1",
+ "bytes",
+ "either",
+ "futures",
+ "http",
+ "http-body",
+ "http-body-util",
+ "hyper",
+ "hyper-rustls",
+ "hyper-timeout",
+ "hyper-util",
+ "jiff",
+ "jsonpath-rust",
+ "k8s-openapi",
+ "kube-core",
+ "pem",
+ "rustls",
+ "secrecy",
+ "serde",
+ "serde_json",
+ "serde_yaml",
+ "thiserror 2.0.20",
+ "tokio",
+ "tokio-util",
+ "tower",
+ "tower-http",
+ "tracing",
+]
+
+[[package]]
+name = "kube-core"
+version = "3.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "f126d2db7a8b532ec1d839ece2a71e2485dc3bbca6cc3c3f929becaa810e719e"
+dependencies = [
+ "derive_more",
+ "form_urlencoded",
+ "http",
+ "jiff",
+ "json-patch",
+ "k8s-openapi",
+ "serde",
+ "serde-value",
+ "serde_json",
+ "thiserror 2.0.20",
+]
+
+[[package]]
+name = "kube-runtime"
+version = "3.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "c072737075826ee74d3e615e80334e41e617ca3d14fb46ef7cdfda822d6f15f2"
+dependencies = [
+ "ahash",
+ "async-broadcast",
+ "async-stream",
+ "backon",
+ "educe",
+ "futures",
+ "hashbrown 0.16.1",
+ "hostname",
+ "json-patch",
+ "k8s-openapi",
+ "kube-client",
+ "parking_lot",
+ "pin-project",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.20",
+ "tokio",
+ "tokio-util",
+ "tracing",
+]
+
[[package]]
name = "lazy_static"
version = "1.5.0"
@@ -1552,6 +1819,15 @@ dependencies = [
"tokio-stream",
]
+[[package]]
+name = "ordered-float"
+version = "2.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "68f19d67e5a2795c94e73e0bb1cc1a7edeb2e28efd39e2e1c9b7a40c1108b11c"
+dependencies = [
+ "num-traits",
+]
+
[[package]]
name = "parking"
version = "2.2.1"
@@ -1631,6 +1907,48 @@ version = "2.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
+[[package]]
+name = "pest"
+version = "2.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "5a07a60cc7a4d00c91f95c685609d1d2f79050e6804b70ebedd7650f0b839bcf"
+dependencies = [
+ "memchr",
+ "ucd-trie",
+]
+
+[[package]]
+name = "pest_derive"
+version = "2.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "b3a83744a5c8455b8b3e0dc5031362780a347c878bdd11584d1a8984228cc88d"
+dependencies = [
+ "pest",
+ "pest_generator",
+]
+
+[[package]]
+name = "pest_generator"
+version = "2.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "e0cd3451aa3de60d4b9a1e736885e4dea6b31617598026f12256ad566d63304a"
+dependencies = [
+ "pest",
+ "pest_meta",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
+[[package]]
+name = "pest_meta"
+version = "2.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "e04d3a0849e241d7dfce834c83b1c5edc8622009e8dd51a12ba1927c32f05496"
+dependencies = [
+ "pest",
+]
+
[[package]]
name = "petgraph"
version = "0.8.3"
@@ -2180,6 +2498,15 @@ version = "2.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d"
+[[package]]
+name = "rustc_version"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92"
+dependencies = [
+ "semver",
+]
+
[[package]]
name = "rustix"
version = "1.1.4"
@@ -2343,6 +2670,23 @@ dependencies = [
"uuid",
]
+[[package]]
+name = "scg-pool-k8s"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "futures",
+ "k8s-openapi",
+ "kube",
+ "parking_lot",
+ "scg-routing",
+ "serde",
+ "thiserror 2.0.20",
+ "tokio",
+ "tokio-stream",
+ "tracing",
+]
+
[[package]]
name = "scg-pool-static"
version = "0.1.0"
@@ -2447,6 +2791,15 @@ version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
+[[package]]
+name = "secrecy"
+version = "0.10.3"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "e891af845473308773346dc847b2c23ee78fe442e0472ac50e22a18a93d3ae5a"
+dependencies = [
+ "zeroize",
+]
+
[[package]]
name = "security-framework"
version = "3.7.0"
@@ -2470,6 +2823,12 @@ dependencies = [
"libc",
]
+[[package]]
+name = "semver"
+version = "1.0.28"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd"
+
[[package]]
name = "serde"
version = "1.0.229"
@@ -2480,6 +2839,16 @@ dependencies = [
"serde_derive",
]
+[[package]]
+name = "serde-value"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "f3a1a3341211875ef120e117ea7fd5228530ae7e7036a779fdc9117be6b3282c"
+dependencies = [
+ "ordered-float",
+ "serde",
+]
+
[[package]]
name = "serde_core"
version = "1.0.229"
@@ -2974,6 +3343,7 @@ dependencies = [
"futures-sink",
"libc",
"pin-project-lite",
+ "slab",
"tokio",
]
@@ -3085,15 +3455,18 @@ version = "0.6.11"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840"
dependencies = [
+ "base64 0.22.1",
"bitflags 2.13.1",
"bytes",
"futures-util",
"http",
"http-body",
+ "mime",
"pin-project-lite",
"tower",
"tower-layer",
"tower-service",
+ "tracing",
"url",
]
@@ -3115,6 +3488,7 @@ version = "0.1.44"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
dependencies = [
+ "log",
"pin-project-lite",
"tracing-attributes",
"tracing-core",
@@ -3211,6 +3585,12 @@ version = "1.20.1"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20"
+[[package]]
+name = "ucd-trie"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971"
+
[[package]]
name = "unicase"
version = "2.9.0"
diff --git a/Cargo.toml b/Cargo.toml
index cc94747..0528ae0 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -28,6 +28,7 @@ members = [
"crates/pool-static",
"crates/store-redis",
"crates/healthcheck",
+ "crates/pool-k8s",
]
[workspace.package]
diff --git a/crates/pool-k8s/Cargo.toml b/crates/pool-k8s/Cargo.toml
new file mode 100644
index 0000000..cdf1319
--- /dev/null
+++ b/crates/pool-k8s/Cargo.toml
@@ -0,0 +1,40 @@
+# 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-pool-k8s"
+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 = "Kubernetes Endpoints-watching backend pool for the Spark
Connect Gateway."
+publish = false
+
+[dependencies]
+scg-routing = { path = "../routing" }
+kube = { workspace = true }
+k8s-openapi = { workspace = true }
+tokio = { workspace = true }
+tokio-stream = { workspace = true }
+futures = { workspace = true }
+parking_lot = { workspace = true }
+tracing = { workspace = true }
+thiserror = { workspace = true }
+serde = { workspace = true }
+async-trait = { workspace = true }
diff --git a/crates/pool-k8s/src/lib.rs b/crates/pool-k8s/src/lib.rs
new file mode 100644
index 0000000..6b83c86
--- /dev/null
+++ b/crates/pool-k8s/src/lib.rs
@@ -0,0 +1,380 @@
+// 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.
+
+//! Kubernetes service-watch backend pool.
+//!
+//! Watches a `Service`'s `Endpoints` object in a given namespace and exposes
+//! the ready-pod IPs (paired with the service's target port) as the live
+//! backend list. The pool is updated in-process whenever the K8s watcher
+//! emits an event, so a `kubectl scale`, a pod evict, or a rolling update
+//! flows through automatically — no gateway redeploy.
+//!
+//! Design notes:
+//!
+//! * We watch the legacy `core/v1/Endpoints` resource rather than the newer
+//! `discovery.k8s.io/v1/EndpointSlice`. Endpoints is universally
+//! supported (back to K8s 1.0) and gives us a single object per Service;
+//! EndpointSlice support can be added later as an alternative source if
+//! we need topology hints or 1000+-endpoint Services.
+//! * We only consider entries in `subset.addresses` (i.e. ready pods) and
+//! skip `subset.not_ready_addresses`.
+//! * Internal state is a `Vec<String>` of `host:port` strings. The
+//! watcher fully replaces the Vec on every event, so we don't need
+//! fine-grained add/remove handling. Placement over the members is
+//! the [`scg_routing::SelectionStrategy`]'s job, not the pool's.
+
+use std::sync::Arc;
+
+use futures::StreamExt;
+use k8s_openapi::api::core::v1::Endpoints;
+use kube::api::Api;
+use kube::runtime::watcher;
+use parking_lot::RwLock;
+use scg_routing::{BackendMember, Pool};
+use serde::Deserialize;
+use thiserror::Error;
+use tracing::{debug, info, warn};
+
+/// Configuration for the K8s pool.
+#[derive(Debug, Clone, Deserialize)]
+pub struct K8sPoolConfig {
+ /// Namespace the Service lives in.
+ pub namespace: String,
+ /// Name of the Service whose Endpoints we watch.
+ pub service_name: String,
+ /// Target port that pods listen on. K8s `Endpoints.subset.ports` may
+ /// have multiple named ports; we filter by the port *number* here.
+ pub port: u16,
+}
+
+#[derive(Debug, Error)]
+pub enum K8sPoolError {
+ #[error("kube client init: {0}")]
+ Client(#[from] kube::Error),
+}
+
+/// Callback invoked with the new backend count whenever the pool's
+/// membership changes. Used by the gateway binary to keep the
+/// `scg_backend_pool_size` Prometheus gauge in sync with the live
+/// pool — otherwise the gauge would freeze at its startup value (0
+/// for K8s discovery, since the watcher populates the pool
+/// asynchronously after `K8sPool::new`).
+///
+/// Defined as a `Fn(usize)` rather than a trait to keep this crate
+/// free of any dependency on `scg-observability`.
+pub type SizeObserver = Arc<dyn Fn(usize) + Send + Sync>;
+
+/// In-memory snapshot of the live backend list.
+#[derive(Default)]
+struct Inner {
+ backends: RwLock<Vec<String>>,
+ size_observer: RwLock<Option<SizeObserver>>,
+}
+
+/// K8s service-watch backend pool. Cheap to clone: an `Arc` wraps the
+/// shared state.
+#[derive(Clone)]
+pub struct K8sPool {
+ inner: Arc<Inner>,
+}
+
+impl K8sPool {
+ /// Build an empty pool. The watcher task must be spawned separately
+ /// via [`K8sPool::spawn_watcher`] for the pool to become populated.
+ pub fn new() -> Self {
+ Self {
+ inner: Arc::new(Inner::default()),
+ }
+ }
+
+ /// Register a callback that fires whenever the pool's backend list
+ /// changes (i.e. once per K8s Endpoints transition). The callback
+ /// runs synchronously inside `set_backends`, so it must be cheap
+ /// and non-blocking. Calling this replaces any previously registered
+ /// observer.
+ pub fn set_size_observer(&self, observer: SizeObserver) {
+ // Fire once on registration so the observer sees the current
+ // state — useful when a metric was already at a default before
+ // the pool was wired up, or when the watcher fired before the
+ // observer was attached.
+ let current_len = self.inner.backends.read().len();
+ observer(current_len);
+ *self.inner.size_observer.write() = Some(observer);
+ }
+
+ /// Replace the backend list. Used by the spawned watcher task
+ /// (in this crate) and by the in-crate integration tests.
+ /// Production callers never invoke this directly — it's an
+ /// internal seam exposed only because the integration-test
+ /// binary lives in a separate compilation unit.
+ #[doc(hidden)]
+ pub fn set_backends(&self, addrs: Vec<String>) {
+ let mut g = self.inner.backends.write();
+ if *g != addrs {
+ info!(count = addrs.len(), "k8s pool: backend list updated");
+ debug!(?addrs, "k8s pool: new backends");
+ let new_len = addrs.len();
+ *g = addrs;
+ // Drop the write lock before invoking the observer so a
+ // misbehaving observer can't deadlock the pool.
+ drop(g);
+ if let Some(obs) = self.inner.size_observer.read().as_ref() {
+ obs(new_len);
+ }
+ }
+ }
+
+ /// Spawn a Tokio task that drives the K8s `Endpoints` watcher and
+ /// updates the pool whenever the Endpoints object changes. Returns
+ /// the spawned `JoinHandle` so callers can `abort()` on shutdown.
+ pub async fn spawn_watcher(
+ &self,
+ cfg: K8sPoolConfig,
+ ) -> Result<tokio::task::JoinHandle<()>, K8sPoolError> {
+ let client = kube::Client::try_default().await?;
+ let api: Api<Endpoints> = Api::namespaced(client, &cfg.namespace);
+
+ // Filter to a single Endpoints object by name. The runtime watcher
+ // takes a `ListParams`-shaped config and we use field_selector to
+ // narrow to one resource.
+ let watcher_cfg =
+ watcher::Config::default().fields(&format!("metadata.name={}",
cfg.service_name));
+
+ let pool = self.clone();
+ let target_port = cfg.port;
+ let service_name = cfg.service_name.clone();
+
+ let handle = tokio::spawn(async move {
+ let mut stream = watcher(api, watcher_cfg).boxed();
+ while let Some(event) = stream.next().await {
+ match event {
+ Ok(watcher::Event::Apply(ep)) |
Ok(watcher::Event::InitApply(ep)) => {
+ let addrs = extract_addresses(&ep, target_port);
+ pool.set_backends(addrs);
+ }
+ Ok(watcher::Event::Delete(_)) => {
+ warn!(service = %service_name, "k8s pool: Endpoints
object deleted");
+ pool.set_backends(Vec::new());
+ }
+ Ok(watcher::Event::Init) | Ok(watcher::Event::InitDone) =>
{}
+ Err(e) => {
+ warn!(error = %e, "k8s pool: watcher error; will
reconnect");
+ }
+ }
+ }
+ warn!("k8s pool: watcher stream ended");
+ });
+ Ok(handle)
+ }
+}
+
+impl Default for K8sPool {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl Pool for K8sPool {
+ fn members(&self) -> Vec<BackendMember> {
+ // Addresses only for now — Endpoints objects don't carry pod
+ // labels, so surfacing metadata (e.g. spark-version) needs an
+ // additional Pod watch. Planned work; members get default
+ // labels/weight until then.
+ self.inner
+ .backends
+ .read()
+ .iter()
+ .map(BackendMember::new)
+ .collect()
+ }
+
+ fn mark_unhealthy(&self, addr: &str) {
+ // The K8s watcher is the source of truth for membership. A
+ // forward error against `addr` is usually transient (pod
+ // restarted) and the watcher will reconcile within seconds. We
+ // log the hint but do not optimistically evict — eviction would
+ // race with the watcher and risks oscillation.
+ debug!(%addr, "k8s pool: mark_unhealthy hint (no-op)");
+ }
+}
+
+/// Extract `host:port` strings from a single `Endpoints` object, keeping
+/// only the addresses that match `target_port`.
+///
+/// `Endpoints.subsets[i].addresses` are the ready pods;
+/// `subsets[i].not_ready_addresses` are excluded. Each subset has its own
+/// list of `ports`; we walk the cartesian product within each subset.
+fn extract_addresses(ep: &Endpoints, target_port: u16) -> Vec<String> {
+ let target_port_i32 = i32::from(target_port);
+ let Some(subsets) = ep.subsets.as_ref() else {
+ return Vec::new();
+ };
+ let mut out = Vec::new();
+ for subset in subsets {
+ let port_matches = subset
+ .ports
+ .as_ref()
+ .is_some_and(|ports| ports.iter().any(|p| p.port ==
target_port_i32));
+ if !port_matches {
+ continue;
+ }
+ let Some(addrs) = subset.addresses.as_ref() else {
+ continue;
+ };
+ for a in addrs {
+ out.push(format!("{}:{}", a.ip, target_port));
+ }
+ }
+ out
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use k8s_openapi::api::core::v1::{EndpointAddress, EndpointPort,
EndpointSubset, Endpoints};
+
+ fn ep(subsets: Vec<EndpointSubset>) -> Endpoints {
+ Endpoints {
+ metadata: Default::default(),
+ subsets: Some(subsets),
+ }
+ }
+
+ fn addr(ip: &str) -> EndpointAddress {
+ EndpointAddress {
+ ip: ip.into(),
+ ..Default::default()
+ }
+ }
+
+ fn port(p: i32) -> EndpointPort {
+ EndpointPort {
+ port: p,
+ ..Default::default()
+ }
+ }
+
+ #[test]
+ fn extracts_ready_addresses_only_for_matching_port() {
+ let e = ep(vec![EndpointSubset {
+ addresses: Some(vec![addr("10.0.0.1"), addr("10.0.0.2")]),
+ not_ready_addresses: Some(vec![addr("10.0.0.99")]),
+ ports: Some(vec![port(15002)]),
+ }]);
+ let got = extract_addresses(&e, 15002);
+ assert_eq!(got, vec!["10.0.0.1:15002", "10.0.0.2:15002"]);
+ }
+
+ #[test]
+ fn skips_subsets_with_wrong_port() {
+ let e = ep(vec![
+ EndpointSubset {
+ addresses: Some(vec![addr("10.0.0.1")]),
+ not_ready_addresses: None,
+ ports: Some(vec![port(80)]),
+ },
+ EndpointSubset {
+ addresses: Some(vec![addr("10.0.0.2")]),
+ not_ready_addresses: None,
+ ports: Some(vec![port(15002)]),
+ },
+ ]);
+ let got = extract_addresses(&e, 15002);
+ assert_eq!(got, vec!["10.0.0.2:15002"]);
+ }
+
+ #[test]
+ fn empty_subsets_returns_empty_list() {
+ let e = ep(Vec::new());
+ assert!(extract_addresses(&e, 15002).is_empty());
+
+ let e = Endpoints {
+ metadata: Default::default(),
+ subsets: None,
+ };
+ assert!(extract_addresses(&e, 15002).is_empty());
+ }
+
+ fn addrs(p: &K8sPool) -> Vec<String> {
+ p.members().into_iter().map(|m| m.addr).collect()
+ }
+
+ #[test]
+ fn members_reflect_set_backends() {
+ let p = K8sPool::new();
+ assert!(p.members().is_empty(), "fresh pool has no members");
+ p.set_backends(vec!["a".into(), "b".into(), "c".into()]);
+ assert_eq!(addrs(&p), vec!["a", "b", "c"]);
+ assert!(p
+ .members()
+ .iter()
+ .all(|m| m.weight == 1 && m.labels.is_empty()));
+ }
+
+ #[test]
+ fn pool_handles_membership_changes() {
+ let p = K8sPool::new();
+ p.set_backends(vec!["a".into(), "b".into()]);
+ assert_eq!(addrs(&p), vec!["a", "b"]);
+ p.set_backends(vec!["x".into(), "y".into(), "z".into()]);
+ assert_eq!(addrs(&p), vec!["x", "y", "z"]);
+ }
+
+ #[test]
+ fn pool_empties_after_drain() {
+ let p = K8sPool::new();
+ p.set_backends(vec!["a".into()]);
+ assert_eq!(addrs(&p), vec!["a"]);
+ p.set_backends(Vec::new());
+ assert!(p.members().is_empty());
+ }
+
+ #[test]
+ fn size_observer_fires_on_each_membership_change() {
+ use std::sync::Mutex;
+
+ let p = K8sPool::new();
+ let log: Arc<Mutex<Vec<usize>>> = Arc::new(Mutex::new(Vec::new()));
+ let sink = log.clone();
+ p.set_size_observer(Arc::new(move |n| sink.lock().unwrap().push(n)));
+
+ // Registration alone fires once with the current (empty) size.
+ assert_eq!(*log.lock().unwrap(), vec![0]);
+
+ p.set_backends(vec!["a".into(), "b".into()]);
+ p.set_backends(vec!["a".into(), "b".into(), "c".into()]);
+ p.set_backends(vec!["a".into(), "b".into(), "c".into()]); //
duplicate: no callback
+ p.set_backends(Vec::new());
+
+ assert_eq!(*log.lock().unwrap(), vec![0, 2, 3, 0]);
+ }
+
+ #[test]
+ fn size_observer_attached_after_first_update_sees_current_state() {
+ use std::sync::Mutex;
+
+ let p = K8sPool::new();
+ p.set_backends(vec!["a".into(), "b".into()]);
+
+ let log: Arc<Mutex<Vec<usize>>> = Arc::new(Mutex::new(Vec::new()));
+ let sink = log.clone();
+ p.set_size_observer(Arc::new(move |n| sink.lock().unwrap().push(n)));
+
+ // Late attach: observer is informed of the current size
+ // immediately so the downstream gauge isn't stuck at its prior
+ // value.
+ assert_eq!(*log.lock().unwrap(), vec![2]);
+ }
+}
diff --git a/crates/pool-k8s/tests/router_integration.rs
b/crates/pool-k8s/tests/router_integration.rs
new file mode 100644
index 0000000..26c3aa8
--- /dev/null
+++ b/crates/pool-k8s/tests/router_integration.rs
@@ -0,0 +1,110 @@
+// 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.
+
+//! Integration test: K8sPool wired into the Router exhibits the same
+//! stickiness invariants as the static pool, and gracefully reports
+//! "no backend" when the watcher hasn't populated yet.
+//!
+//! We don't talk to a real K8s API server here; the watcher is mocked
+//! by calling K8sPool::set_backends() directly. The point is to prove
+//! that Pool::pick() returning None propagates correctly through
+//! Router and that, once the pool is populated, sticky routing works.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use parking_lot::Mutex;
+use scg_pool_k8s::K8sPool;
+use scg_routing::{AffinityStore, Router, SessionKey};
+
+#[derive(Default)]
+struct StubStore {
+ sessions: Mutex<HashMap<SessionKey, String>>,
+ ops: Mutex<HashMap<String, String>>,
+}
+
+#[async_trait::async_trait]
+impl AffinityStore for StubStore {
+ async fn lookup_session(&self, k: &SessionKey) -> Option<String> {
+ self.sessions.lock().get(k).cloned()
+ }
+ async fn bind_session_if_absent(&self, k: SessionKey, v: String) -> String
{
+ self.sessions.lock().entry(k).or_insert(v).clone()
+ }
+ async fn forget_session(&self, k: &SessionKey) {
+ self.sessions.lock().remove(k);
+ }
+ async fn lookup_op(&self, o: &str) -> Option<String> {
+ self.ops.lock().get(o).cloned()
+ }
+ async fn bind_op(&self, o: String, v: String) {
+ self.ops.lock().insert(o, v);
+ }
+ async fn forget_op(&self, o: &str) {
+ self.ops.lock().remove(o);
+ }
+}
+
+#[tokio::test]
+async fn empty_pool_propagates_none_through_router() {
+ let pool = Arc::new(K8sPool::new());
+ let router = Router::single_pool(pool.clone(),
Arc::new(StubStore::default()));
+ let key = SessionKey::new("u", "s");
+ assert!(
+ router.resolve_session(&key).await.unwrap().is_none(),
+ "router should return None when pool is empty"
+ );
+ assert!(
+ router.resolve_op("op-1", &key).await.unwrap().is_none(),
+ "router should return None for op resolution too"
+ );
+}
+
+#[tokio::test]
+async fn stickiness_holds_across_membership_changes() {
+ let pool = Arc::new(K8sPool::new());
+ let router = Router::single_pool(pool.clone(),
Arc::new(StubStore::default()));
+
+ pool.set_backends(vec!["a:1".into(), "b:1".into()]);
+ let key = SessionKey::new("u", "s");
+ let first = router.resolve_session(&key).await.unwrap().unwrap();
+
+ // Even if the K8s watcher emits a totally different membership,
+ // the existing session must keep its binding.
+ pool.set_backends(vec!["x:1".into(), "y:1".into(), "z:1".into()]);
+ for _ in 0..5 {
+ assert_eq!(
+ router.resolve_session(&key).await.unwrap().unwrap(),
+ first,
+ "stickiness must outlive pool membership churn"
+ );
+ }
+}
+
+#[tokio::test]
+async fn new_session_after_repopulation_picks_from_new_set() {
+ let pool = Arc::new(K8sPool::new());
+ let router = Router::single_pool(pool.clone(),
Arc::new(StubStore::default()));
+
+ // Pool empty at first.
+ let early = SessionKey::new("u", "early");
+ assert!(router.resolve_session(&early).await.unwrap().is_none());
+
+ // Watcher fires → new sessions can now route.
+ pool.set_backends(vec!["a:1".into(), "b:1".into()]);
+ let late = SessionKey::new("u", "late");
+ let chosen = router.resolve_session(&late).await.unwrap().unwrap();
+ assert!(["a:1", "b:1"].contains(&chosen.as_str()));
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
