diff --git a/Cargo.lock b/Cargo.lock index a7b6d1d..c38a817 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -654,6 +654,22 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + +[[package]] +name = "scg-config" +version = "0.1.0" +dependencies = [ + "serde", + "serde_yaml", + "tempfile", + "thiserror", +] + [[package]] name = "scg-genproto" version = "0.1.0" @@ -681,6 +697,16 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + [[package]] name = "serde_core" version = "1.0.229" @@ -701,6 +727,19 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "serde_yaml" +version = "0.9.34+deprecated" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47" +dependencies = [ + "indexmap", + "itoa", + "ryu", + "serde", + "unsafe-libyaml", +] + [[package]] name = "signal-hook-registry" version = "1.4.8" @@ -774,6 +813,26 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "thiserror" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "tokio" version = "1.53.1" @@ -975,6 +1034,12 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "unsafe-libyaml" +version = "0.2.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861" + [[package]] name = "want" version = "0.3.1" diff --git a/Cargo.toml b/Cargo.toml index 8484085..d2c44c7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,7 @@ resolver = "2" members = [ "crates/genproto", "crates/routing", + "crates/config", ] [workspace.package] diff --git a/crates/config/Cargo.toml b/crates/config/Cargo.toml new file mode 100644 index 0000000..782ea30 --- /dev/null +++ b/crates/config/Cargo.toml @@ -0,0 +1,35 @@ +# 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-config" +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 = "YAML configuration schema for the Spark Connect Gateway." +publish = false + +[dependencies] +serde = { workspace = true } +serde_yaml = { workspace = true } +thiserror = { workspace = true } + +[dev-dependencies] +tempfile = { workspace = true } diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs new file mode 100644 index 0000000..2ecc98e --- /dev/null +++ b/crates/config/src/lib.rs @@ -0,0 +1,1631 @@ +// 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. + +//! YAML configuration for the gateway. +//! +//! Two equivalent forms are accepted for backend discovery: +//! +//! ```yaml +//! # Shorthand — equivalent to a static-list `backend_discovery`: +//! backends: +//! - "host1:15002" +//! - "host2:15002" +//! ``` +//! +//! ```yaml +//! # Tagged static form: +//! backend_discovery: +//! type: static +//! addresses: ["host1:15002", "host2:15002"] +//! ``` +//! +//! ```yaml +//! # Tagged K8s form — watches an Endpoints object: +//! backend_discovery: +//! type: k8s +//! namespace: spark-connect +//! service_name: spark-connect +//! port: 15002 +//! ``` + +use serde::Deserialize; +use std::path::Path; + +#[derive(Debug, thiserror::Error)] +pub enum ConfigError { + #[error("read config {path}: {source}")] + Io { + path: String, + #[source] + source: std::io::Error, + }, + #[error("parse config {path}: {source}")] + Parse { + path: String, + #[source] + source: serde_yaml::Error, + }, + #[error( + "config: must specify either `backends` or `backend_discovery` \ + (omitting both is allowed only when \ + `tenant_pools.on_unknown_tenant: reject` is set and at least \ + one tenant override is configured)" + )] + NoDiscoverySource, + #[error("config: cannot specify both `backends` and `backend_discovery`")] + ConflictingDiscovery, + #[error("config: static backend list must contain at least one address")] + EmptyStatic, + #[error("config: backend token env var {name} is not set")] + TokenEnvMissing { name: String }, + #[error("config: read backend token file {path}: {source}")] + TokenFile { + path: String, + #[source] + source: std::io::Error, + }, + #[error("config: backend token resolved to an empty string")] + TokenEmpty, +} + +/// One of the supported backend discovery sources. +#[derive(Debug, Clone, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum BackendDiscovery { + /// Fixed list of `host:port` addresses, configured at startup. + Static { addresses: Vec }, + /// Watch a Kubernetes Service's Endpoints object. + K8s { + namespace: String, + service_name: String, + port: u16, + }, +} + +/// Where a backend pre-shared token comes from. Backends started +/// with `spark.connect.authenticate.token` (Spark 4.0+) require a +/// matching `Bearer` credential on every request; the gateway +/// presents it on the gateway→backend hop so that operators can +/// hold the token *only* in the gateway and have the backend itself +/// reject clients that try to bypass it. +/// +/// `env` / `file` keep the token out of the YAML (and out of the +/// ConfigMap the Helm chart renders it into); `inline` is for +/// walkthroughs and tests. +#[derive(Debug, Clone, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum BackendTokenSource { + Inline { token: String }, + Env { name: String }, + File { path: String }, +} + +impl BackendTokenSource { + /// Resolve the token value. Reads the env var / file once, at + /// startup — the backend's own token is fixed for the server's + /// lifetime, so rotation implies restarting both sides anyway. + pub fn resolve(&self) -> Result { + let token = match self { + Self::Inline { token } => token.clone(), + Self::Env { name } => std::env::var(name) + .map_err(|_| ConfigError::TokenEnvMissing { name: name.clone() })?, + Self::File { path } => std::fs::read_to_string(path) + .map_err(|e| ConfigError::TokenFile { + path: path.clone(), + source: e, + })? + .trim_end_matches(['\r', '\n']) + .to_string(), + }; + if token.is_empty() { + return Err(ConfigError::TokenEmpty); + } + Ok(token) + } +} + +/// Authentication configuration. Defaults to `none` so an unset +/// `auth:` block lets every caller through as `user_id="anonymous"` +/// — fine for trusted in-cluster networks, not for external exposure. +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum AuthConfig { + /// No auth — anyone reaching the gateway is `user_id: anonymous`. + /// Acceptable on a trusted network; **not** for production. + #[default] + None, + /// Bearer-token allowlist. See [`scg-auth::token`]. + Static { tokens: Vec }, + /// Local-key JWT verification. See [`scg-auth::jwt`]. + Jwt(JwtSettings), + /// Remote JWKS / OIDC verification. See [`scg-auth::oidc`]. + Oidc(OidcSettings), +} + +/// One entry in the static-token table — kept here (rather than only in +/// `scg-auth`) so config files can describe auth without depending on +/// the auth crate's serde shape. +#[derive(Debug, Clone, Deserialize)] +pub struct TokenEntry { + pub token: String, + pub user_id: String, + #[serde(default)] + pub tenant: Option, + #[serde(default)] + pub groups: Vec, +} + +/// JWT verification settings; mirrors `scg_auth::jwt::JwtConfig`. +#[derive(Debug, Clone, Deserialize)] +pub struct JwtSettings { + pub key: KeySource, + pub algorithms: Vec, + #[serde(default)] + pub issuer: Option, + #[serde(default)] + pub audience: Option, + #[serde(default = "default_user_id_claim")] + pub user_id_claim: String, + #[serde(default)] + pub tenant_claim: Option, + #[serde(default)] + pub groups_claim: Option, +} + +#[derive(Debug, Clone, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum KeySource { + PemFile { path: String }, + PemInline { pem: String }, + HmacSecret { secret: String }, +} + +/// OIDC verification settings; mirrors `scg_auth::oidc::OidcConfig`. +#[derive(Debug, Clone, Deserialize)] +pub struct OidcSettings { + #[serde(default)] + pub jwks_url: Option, + #[serde(default)] + pub discovery_url: Option, + pub algorithms: Vec, + #[serde(default)] + pub issuer: Option, + #[serde(default)] + pub audience: Option, + #[serde(default = "default_user_id_claim")] + pub user_id_claim: String, + #[serde(default)] + pub tenant_claim: Option, + #[serde(default)] + pub groups_claim: Option, + #[serde(default = "default_refresh_floor_secs")] + pub refresh_floor_secs: u64, +} + +fn default_user_id_claim() -> String { + "sub".into() +} +fn default_refresh_floor_secs() -> u64 { + 60 +} + +/// Where the gateway keeps its `(user_id, session_id) -> backend` +/// affinity table. Default `memory` is single-replica only. +/// `redis` is required for HA across multiple gateway replicas — +/// without it, two replicas will pin the same session to different +/// backends and Spark Connect's per-driver session state stops +/// being consistent. +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum AffinityStoreConfig { + #[default] + Memory, + Redis(RedisStoreSettings), +} + +#[derive(Debug, Clone, Deserialize)] +pub struct RedisStoreSettings { + /// `redis://` URL. Supports password (`redis://:pw@host:6379`) + /// and database index (`redis://host:6379/2`). + pub url: String, + /// Key prefix; lets multiple gateway deployments share a Redis + /// without colliding. Default `scg`. + #[serde(default = "default_redis_prefix")] + pub key_prefix: String, + /// TTL for `(user_id, session_id) -> backend` bindings (seconds). + /// Refreshed on every read. + #[serde(default = "default_session_ttl_secs")] + pub session_ttl_secs: u64, + /// TTL for `op_id -> backend` bindings (seconds). + #[serde(default = "default_op_ttl_secs")] + pub op_ttl_secs: u64, +} + +fn default_redis_prefix() -> String { + "scg".into() +} +fn default_session_ttl_secs() -> u64 { + 60 * 60 +} +fn default_op_ttl_secs() -> u64 { + 15 * 60 +} + +/// Audit logging configuration. Records security- and +/// compliance-relevant events as structured `tracing` events with +/// `target = "scg::audit"`. On by default — the per-event cost is +/// negligible and the compliance value is high. +#[derive(Debug, Clone, Deserialize)] +pub struct AuditSettings { + /// Master switch. `true` (default) writes session lifecycle, + /// auth failures, and RPC errors to the audit stream. + #[serde(default = "default_audit_enabled")] + pub enabled: bool, + /// When `true`, every successful RPC also emits an `rpc.ok` + /// audit event. Off by default — successful RPCs are already + /// counted in `scg_rpcs_total{code="OK"}` and emitting one + /// audit event per successful RPC drowns out the events + /// operators actually care about. Switch on under strict + /// monitoring. + #[serde(default)] + pub log_successful_rpcs: bool, +} + +fn default_audit_enabled() -> bool { + true +} + +impl Default for AuditSettings { + fn default() -> Self { + Self { + enabled: true, + log_successful_rpcs: false, + } + } +} + +/// Per-tenant rate-limit configuration. Token bucket +/// per tenant with an optional per-user sub-bucket; both are +/// disabled (`rpcs_per_second: 0`) by default so the limiter is a +/// no-op until the operator opts in. +/// +/// Tenants not listed in `overrides` use `default`; an inbound RPC +/// is admitted only when *both* applicable buckets (tenant + +/// per-user, if enabled) have tokens. +#[derive(Debug, Clone, Default, Deserialize)] +pub struct RateLimitSettings { + /// Master switch. When `false` (default), no rate limiting is + /// performed even if `default` / `overrides` are populated. + #[serde(default)] + pub enabled: bool, + /// Backend store for the bucket state. `memory` (default) + /// enforces quotas per gateway replica; `redis` shares state + /// across all replicas via a Lua-driven token bucket. See + /// `crates/ratelimit/src/redis.rs` for the wire format and the + /// trade-off between the two stores. + #[serde(default)] + pub store: RateLimitStore, + /// Redis connection settings — only consulted when `store: redis`. + #[serde(default)] + pub redis: RateLimitRedisSettings, + #[serde(default)] + pub default: BucketSettings, + #[serde(default)] + pub overrides: std::collections::HashMap, +} + +/// Which backend stores the token-bucket state. +#[derive(Debug, Clone, Copy, Default, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum RateLimitStore { + /// Each gateway replica enforces its own bucket. Fine for + /// single-replica deployments or back-pressure-style limiting. + #[default] + Memory, + /// Atomic token bucket in Redis, shared across all replicas. + /// The effective cluster-wide quota matches the configured + /// rates exactly. + Redis, +} + +/// Redis settings for the distributed limiter. Defaults are +/// dev-friendly — production should override `url`. +#[derive(Debug, Clone, Deserialize)] +pub struct RateLimitRedisSettings { + /// Redis URL. e.g. `redis://redis.spark-connect.svc:6379`. + #[serde(default = "default_rate_limit_redis_url")] + pub url: String, + /// Key prefix. All limiter keys live under `{key_prefix}:t:*` + /// (tenant) and `{key_prefix}:u:*` (user). Default `scg-rl` — + /// distinct from the affinity store's prefix so flushing one + /// doesn't disturb the other. + #[serde(default = "default_rate_limit_redis_key_prefix")] + pub key_prefix: String, + /// TTL for an idle bucket key in seconds. Default 3600 (one + /// hour) — long enough that an idle tenant doesn't lose its + /// bucket, short enough that one-off keys GC themselves. + #[serde(default = "default_rate_limit_redis_key_ttl_secs")] + pub key_ttl_secs: u64, + /// Behaviour when Redis is unreachable. `open` (default) admits + /// the RPC and bumps `scg_rate_limit_redis_errors_total`; + /// `closed` rejects the RPC with `ResourceExhausted`. + #[serde(default)] + pub on_failure: RateLimitFailMode, +} + +impl Default for RateLimitRedisSettings { + fn default() -> Self { + Self { + url: default_rate_limit_redis_url(), + key_prefix: default_rate_limit_redis_key_prefix(), + key_ttl_secs: default_rate_limit_redis_key_ttl_secs(), + on_failure: RateLimitFailMode::default(), + } + } +} + +#[derive(Debug, Clone, Copy, Default, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum RateLimitFailMode { + /// Admit the RPC; bump the error metric. Recommended default — + /// availability over strict quotas. + #[default] + Open, + /// Reject the RPC with `ResourceExhausted`. Use for strict-SaaS + /// deployments where a Redis outage must not become a + /// quota-bypass vector; makes Redis a hard request-path + /// dependency. + Closed, +} + +fn default_rate_limit_redis_url() -> String { + "redis://localhost:6379".into() +} +fn default_rate_limit_redis_key_prefix() -> String { + "scg-rl".into() +} +fn default_rate_limit_redis_key_ttl_secs() -> u64 { + 3600 +} + +#[derive(Debug, Clone, Copy, Default, Deserialize)] +pub struct BucketSettings { + /// Per-tenant token-bucket refill rate (RPCs/second). `0` + /// disables the per-tenant bucket entirely. + #[serde(default)] + pub rpcs_per_second: f64, + /// Per-tenant token-bucket capacity (max consecutive RPCs + /// before the limiter kicks in). + #[serde(default)] + pub burst: u64, + /// Per-user token-bucket refill rate inside the tenant. `0` + /// (default) disables this dimension — only the per-tenant + /// bucket is consulted. + #[serde(default)] + pub per_user_rpcs_per_second: f64, + #[serde(default)] + pub per_user_burst: u64, +} + +/// Per-tenant backend pool overrides. A multi-tenant deployment +/// lists one entry per tenant that needs its own pool; +/// any tenant *not* listed here routes through the deployment's +/// default pool (the existing `backends:` / `backend_discovery:` +/// settings). +/// +/// The fallback `policy` decides what happens when an inbound RPC +/// carries a tenant that has neither an explicit override nor (in +/// the `Reject` case) any pool at all. Default `UseDefault` matches +/// the single-tenant baseline — everything routes to the default +/// pool. `Reject` is the right choice for SaaS-style deployments +/// where unconfigured tenants must not get any access. +#[derive(Debug, Clone, Default, Deserialize)] +pub struct TenantPoolsSettings { + /// Tenant name → its own pool's discovery configuration. The + /// `default` tenant is **not** special here; if you want the + /// default pool to be different from what `backends:` / + /// `backend_discovery:` provides, list it as an override too. + #[serde(default)] + pub overrides: std::collections::HashMap, + /// What to do when an inbound RPC has a tenant that's not in + /// `overrides`. `use_default` (default) routes through the + /// deployment's default pool; `reject` returns + /// `PermissionDenied` to the client. + #[serde(default = "default_unknown_tenant_policy")] + pub on_unknown_tenant: UnknownTenantPolicySetting, +} + +/// One tenant's pool configuration: a discovery source plus, +/// optionally, its own backend token. The discovery fields sit at +/// the same YAML level as `backend_token` (serde-flattened), so +/// existing override entries keep parsing unchanged: +/// +/// ```yaml +/// tenant_pools: +/// overrides: +/// team-a: +/// type: static +/// addresses: ["a-1:15002"] +/// backend_token: # optional +/// kind: env +/// name: TEAM_A_BACKEND_TOKEN +/// ``` +/// +/// A tenant override *without* `backend_token` inherits the +/// top-level `backend_token` (if any). +#[derive(Debug, Clone, Deserialize)] +pub struct TenantPoolConfig { + #[serde(flatten)] + pub discovery: BackendDiscovery, + #[serde(default)] + pub backend_token: Option, +} + +#[derive(Debug, Clone, Copy, Default, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum UnknownTenantPolicySetting { + #[default] + UseDefault, + Reject, +} + +fn default_unknown_tenant_policy() -> UnknownTenantPolicySetting { + UnknownTenantPolicySetting::UseDefault +} + +/// How the gateway figures out which tenant an inbound RPC belongs +/// to. The resolved tenant becomes the first segment of the routing +/// key, so two tenants with the same `session_id` get isolated +/// affinity buckets. +/// +/// Deployments without a `tenant_resolver:` block fall back to +/// `from_claim + use_default + "default"`, so every RPC ends up +/// in `tenant="default"` — the single-tenant baseline. +#[derive(Debug, Clone, Deserialize)] +#[serde(tag = "source", rename_all = "snake_case")] +pub enum TenantResolverSource { + /// Read from `Identity.tenant` produced by the auth interceptor + /// (JWT/OIDC `tenant` claim, static-token `tenant` field). + FromClaim, + /// Read from a gRPC metadata header. For deployments where auth + /// is disabled but clients still cooperate by declaring a + /// tenant. + FromMetadata { + #[serde(default = "default_tenant_header")] + header: String, + }, + /// Always use `default_name`. Single-tenant deployments that + /// don't want to bother with auth claims or headers. + AlwaysDefault, +} + +fn default_tenant_header() -> String { + "x-tenant".into() +} + +#[derive(Debug, Clone, Copy, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum TenantOnMissing { + /// Fall back to `default_name` when the source yields nothing. + /// The default — preserves single-tenant behaviour for + /// deployments that haven't opted into multi-tenant routing. + UseDefault, + /// Fail the RPC with `Unauthenticated`. Used by SaaS-style + /// deployments where a missing tenant claim almost always means + /// the IdP is misconfigured. + Reject, +} + +#[derive(Debug, Clone, Deserialize)] +pub struct TenantResolverSettings { + #[serde(flatten)] + pub source: TenantResolverSource, + #[serde(default = "default_tenant_on_missing")] + pub on_missing: TenantOnMissing, + #[serde(default = "default_tenant_name")] + pub default_name: String, +} + +fn default_tenant_on_missing() -> TenantOnMissing { + TenantOnMissing::UseDefault +} +fn default_tenant_name() -> String { + "default".into() +} + +impl Default for TenantResolverSettings { + fn default() -> Self { + Self { + source: TenantResolverSource::FromClaim, + on_missing: TenantOnMissing::UseDefault, + default_name: default_tenant_name(), + } + } +} + +/// Active gRPC health-check probing for backend pool members. Wraps +/// the configured pool with a probe loop that calls +/// `grpc.health.v1.Health/Check` on each backend and removes +/// repeatedly-failing ones from `pick()`. Off by default to avoid +/// breaking deployments where backends don't ship the standard +/// Health service. +#[derive(Debug, Clone, Deserialize)] +pub struct HealthCheckSettings { + /// Master switch. `false` (default) skips active probing — + /// routing fails through to the next session on a forward + /// error instead. + #[serde(default)] + pub enabled: bool, + #[serde(default = "default_hc_interval_secs")] + pub interval_secs: u64, + #[serde(default = "default_hc_timeout_secs")] + pub timeout_secs: u64, + #[serde(default = "default_hc_unhealthy_threshold")] + pub unhealthy_threshold: u32, + #[serde(default = "default_hc_healthy_threshold")] + pub healthy_threshold: u32, +} + +impl Default for HealthCheckSettings { + fn default() -> Self { + Self { + enabled: false, + interval_secs: default_hc_interval_secs(), + timeout_secs: default_hc_timeout_secs(), + unhealthy_threshold: default_hc_unhealthy_threshold(), + healthy_threshold: default_hc_healthy_threshold(), + } + } +} + +fn default_hc_interval_secs() -> u64 { + 5 +} +fn default_hc_timeout_secs() -> u64 { + 2 +} +fn default_hc_unhealthy_threshold() -> u32 { + 3 +} +fn default_hc_healthy_threshold() -> u32 { + 2 +} + +/// Graceful shutdown behaviour. On SIGINT/SIGTERM, the gateway +/// flips `/readyz` to 503 (so K8s drains it from the Service), then +/// waits for in-flight streaming RPCs (`ExecutePlan`, +/// `ReattachExecute`, `AddArtifacts`) to complete, up to +/// `deadline_secs`. +#[derive(Debug, Clone, Deserialize)] +pub struct ShutdownSettings { + /// Hard ceiling on the drain period. After this many seconds the + /// gateway forcibly shuts down regardless of in-flight streams. + /// Pick something compatible with your K8s + /// `terminationGracePeriodSeconds` (the chart defaults to 30). + #[serde(default = "default_shutdown_deadline_secs")] + pub deadline_secs: u64, +} + +impl Default for ShutdownSettings { + fn default() -> Self { + Self { + deadline_secs: default_shutdown_deadline_secs(), + } + } +} + +fn default_shutdown_deadline_secs() -> u64 { + 30 +} + +/// Distributed-tracing configuration. Off by default — configs +/// without a `tracing:` section keep working. +#[derive(Debug, Clone, Default, Deserialize)] +pub struct TracingSettings { + /// OTLP/gRPC collector endpoint (e.g. `http://otel-collector:4317`). + /// `None` disables span export — only the JSON log formatter runs. + #[serde(default)] + pub endpoint: Option, + /// `service.name` resource attribute reported on every span. + #[serde(default = "default_service_name")] + pub service_name: String, + /// `service.version` resource attribute. Defaults to the + /// gateway's compile-time CARGO_PKG_VERSION when omitted. + #[serde(default)] + pub service_version: Option, + /// `TraceIdRatioBased` sampling ratio in `[0.0, 1.0]`. Wrapped in + /// `ParentBased` at runtime so a sampled remote parent always wins. + #[serde(default = "default_sample_ratio")] + pub sample_ratio: f64, + /// Per-batch OTLP export deadline, in seconds. + #[serde(default = "default_export_timeout_secs")] + pub export_timeout_secs: u64, +} + +fn default_service_name() -> String { + "spark-connect-gateway".into() +} +fn default_sample_ratio() -> f64 { + 1.0 +} +fn default_export_timeout_secs() -> u64 { + 10 +} + +/// Raw YAML shape — accepts either the legacy `backends` shorthand or the +/// tagged `backend_discovery` form, never both. +#[derive(Debug, Deserialize)] +struct RawConfig { + #[serde(default = "default_bind_addr")] + bind_addr: String, + #[serde(default)] + backends: Option>, + #[serde(default)] + backend_discovery: Option, + #[serde(default)] + backend_token: Option, + #[serde(default)] + auth: Option, + /// Address for the admin / metrics HTTP server. `null` disables it. + /// Default `0.0.0.0:9090`. + #[serde(default = "default_admin_addr_opt")] + admin_addr: Option, + #[serde(default)] + tracing: Option, + #[serde(default)] + affinity_store: Option, + #[serde(default)] + health_check: Option, + #[serde(default)] + shutdown: Option, + #[serde(default)] + tenant_resolver: Option, + #[serde(default)] + tenant_pools: Option, + #[serde(default)] + rate_limit: Option, + #[serde(default)] + audit: Option, +} + +fn default_admin_addr_opt() -> Option { + Some(":9090".into()) +} + +#[derive(Debug, Clone)] +pub struct Config { + pub bind_addr: String, + /// The default pool's discovery source. `None` is permitted only + /// under `tenant_pools.on_unknown_tenant: reject` with at least + /// one tenant override — in that mode the default pool is never + /// selected (unmatched tenants are rejected before pool + /// selection), so requiring a discovery source would force + /// operators to configure a pool that cannot receive traffic. + pub discovery: Option, + /// Pre-shared token the gateway presents (as `authorization: + /// Bearer `) on every gateway→backend request. Applies to + /// all pools unless a tenant override carries its own + /// `backend_token`. `None` sends no credential — only safe when + /// the backends don't enforce `spark.connect.authenticate.token`. + pub backend_token: Option, + pub auth: AuthConfig, + /// `Some(addr)` to enable the admin HTTP server, `None` to skip it. + pub admin_addr: Option, + /// Distributed-tracing settings. `None` keeps tracing off (the + /// gateway only emits structured JSON logs in that case). + pub tracing: Option, + /// Where to keep affinity state. Defaults to in-memory; use + /// `redis` for multi-replica HA. + pub affinity_store: AffinityStoreConfig, + /// Active gRPC health-check probing for backends. Off by default. + pub health_check: HealthCheckSettings, + /// Graceful shutdown / drain settings. + pub shutdown: ShutdownSettings, + /// How to figure out the tenant for each inbound RPC. Defaults + /// to the back-compat behaviour (every RPC -> tenant="default"). + pub tenant_resolver: TenantResolverSettings, + /// Per-tenant backend pool overrides + unknown-tenant policy. + /// Empty `overrides` + `use_default` reproduces the single-pool + /// baseline. + pub tenant_pools: TenantPoolsSettings, + /// Per-tenant rate limiting. Disabled by default. + pub rate_limit: RateLimitSettings, + /// Structured audit logging. Enabled by default. + pub audit: AuditSettings, +} + +fn default_bind_addr() -> String { + ":15003".into() +} + +impl Config { + pub fn load(path: impl AsRef) -> Result { + let path_str = path.as_ref().display().to_string(); + let data = std::fs::read_to_string(path.as_ref()).map_err(|e| ConfigError::Io { + path: path_str.clone(), + source: e, + })?; + let raw: RawConfig = serde_yaml::from_str(&data).map_err(|e| ConfigError::Parse { + path: path_str, + source: e, + })?; + Self::from_raw(raw) + } + + fn from_raw(raw: RawConfig) -> Result { + let tenant_pools = raw.tenant_pools.unwrap_or_default(); + let discovery = match (raw.backends, raw.backend_discovery) { + (Some(_), Some(_)) => return Err(ConfigError::ConflictingDiscovery), + (None, None) => { + // No default pool is acceptable only when routing can + // never select it: strict multi-tenant deployments + // where every admitted tenant has its own pool and + // everything else is rejected. + let strict = matches!( + tenant_pools.on_unknown_tenant, + UnknownTenantPolicySetting::Reject + ) && !tenant_pools.overrides.is_empty(); + if !strict { + return Err(ConfigError::NoDiscoverySource); + } + None + } + (Some(addrs), None) => { + if addrs.is_empty() { + return Err(ConfigError::EmptyStatic); + } + Some(BackendDiscovery::Static { addresses: addrs }) + } + (None, Some(d)) => { + if let BackendDiscovery::Static { addresses } = &d { + if addresses.is_empty() { + return Err(ConfigError::EmptyStatic); + } + } + Some(d) + } + }; + let bind_addr = if raw.bind_addr.is_empty() { + default_bind_addr() + } else { + raw.bind_addr + }; + Ok(Self { + bind_addr, + discovery, + backend_token: raw.backend_token, + auth: raw.auth.unwrap_or_default(), + admin_addr: raw.admin_addr.filter(|s| !s.is_empty()), + tracing: raw.tracing, + affinity_store: raw.affinity_store.unwrap_or_default(), + health_check: raw.health_check.unwrap_or_default(), + shutdown: raw.shutdown.unwrap_or_default(), + tenant_resolver: raw.tenant_resolver.unwrap_or_default(), + tenant_pools, + rate_limit: raw.rate_limit.unwrap_or_default(), + audit: raw.audit.unwrap_or_default(), + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + + fn write(text: &str) -> tempfile::NamedTempFile { + let mut f = tempfile::NamedTempFile::new().unwrap(); + write!(f, "{}", text).unwrap(); + f + } + + #[test] + fn loads_legacy_backends_shorthand() { + let f = write( + r#" +bind_addr: ":15003" +backends: + - "127.0.0.1:15002" +"#, + ); + let c = Config::load(f.path()).unwrap(); + assert_eq!(c.bind_addr, ":15003"); + match c.discovery { + Some(BackendDiscovery::Static { addresses }) => { + assert_eq!(addresses, vec!["127.0.0.1:15002"]); + } + other => panic!("expected Static, got {:?}", other), + } + } + + #[test] + fn loads_tagged_static() { + let f = write( + r#" +backend_discovery: + type: static + addresses: ["a:1", "b:2"] +"#, + ); + let c = Config::load(f.path()).unwrap(); + match c.discovery { + Some(BackendDiscovery::Static { addresses }) => { + assert_eq!(addresses, vec!["a:1", "b:2"]); + } + other => panic!("expected Static, got {:?}", other), + } + } + + #[test] + fn loads_tagged_k8s() { + let f = write( + r#" +backend_discovery: + type: k8s + namespace: spark-connect + service_name: spark-connect + port: 15002 +"#, + ); + let c = Config::load(f.path()).unwrap(); + match c.discovery { + Some(BackendDiscovery::K8s { + namespace, + service_name, + port, + }) => { + assert_eq!(namespace, "spark-connect"); + assert_eq!(service_name, "spark-connect"); + assert_eq!(port, 15002); + } + other => panic!("expected K8s, got {:?}", other), + } + } + + #[test] + fn omitted_discovery_allowed_under_reject_with_overrides() { + // Strict multi-tenant mode: every admitted tenant has its own + // pool, unknown tenants are rejected — the default pool would + // never be selected, so requiring one would be dead config. + let f = write( + r#" +tenant_pools: + on_unknown_tenant: reject + overrides: + team-a: + type: static + addresses: ["a:15002"] +"#, + ); + let c = Config::load(f.path()).unwrap(); + assert!(c.discovery.is_none()); + assert!(matches!( + c.tenant_pools.on_unknown_tenant, + UnknownTenantPolicySetting::Reject + )); + assert_eq!(c.tenant_pools.overrides.len(), 1); + } + + #[test] + fn omitted_discovery_rejected_under_use_default() { + // With use_default (explicit or implicit), the default pool + // is reachable, so a discovery source is mandatory. + let f = write( + r#" +tenant_pools: + on_unknown_tenant: use_default + overrides: + team-a: + type: static + addresses: ["a:15002"] +"#, + ); + assert!(matches!( + Config::load(f.path()).unwrap_err(), + ConfigError::NoDiscoverySource + )); + } + + #[test] + fn omitted_discovery_rejected_without_overrides() { + // reject + zero overrides would reject every RPC — that's a + // config mistake, not a deployment shape; fail loudly. + let f = write( + r#" +tenant_pools: + on_unknown_tenant: reject +"#, + ); + assert!(matches!( + Config::load(f.path()).unwrap_err(), + ConfigError::NoDiscoverySource + )); + } + + #[test] + fn empty_backends_shorthand_rejected() { + let f = write("backends: []\n"); + assert!(matches!( + Config::load(f.path()).unwrap_err(), + ConfigError::EmptyStatic + )); + } + + #[test] + fn empty_static_in_tagged_form_rejected() { + let f = write( + r#" +backend_discovery: + type: static + addresses: [] +"#, + ); + assert!(matches!( + Config::load(f.path()).unwrap_err(), + ConfigError::EmptyStatic + )); + } + + #[test] + fn missing_discovery_rejected() { + let f = write("bind_addr: ':15003'\n"); + assert!(matches!( + Config::load(f.path()).unwrap_err(), + ConfigError::NoDiscoverySource + )); + } + + #[test] + fn conflicting_discovery_rejected() { + let f = write( + r#" +backends: ["a:1"] +backend_discovery: + type: static + addresses: ["b:2"] +"#, + ); + assert!(matches!( + Config::load(f.path()).unwrap_err(), + ConfigError::ConflictingDiscovery + )); + } + + #[test] + fn defaults_bind_addr() { + let f = write("backends: [\"a:1\"]\n"); + let c = Config::load(f.path()).unwrap(); + assert_eq!(c.bind_addr, ":15003"); + } + + #[test] + fn auth_defaults_to_none() { + let f = write("backends: [\"a:1\"]\n"); + let c = Config::load(f.path()).unwrap(); + assert!(matches!(c.auth, AuthConfig::None)); + } + + #[test] + fn loads_static_auth() { + let f = write( + r#" +backends: ["a:1"] +auth: + type: static + tokens: + - token: "alice-secret" + user_id: "alice" + tenant: "team-a" + groups: ["devs"] + - token: "bob-secret" + user_id: "bob" +"#, + ); + let c = Config::load(f.path()).unwrap(); + match c.auth { + AuthConfig::Static { tokens } => { + assert_eq!(tokens.len(), 2); + assert_eq!(tokens[0].user_id, "alice"); + assert_eq!(tokens[0].tenant.as_deref(), Some("team-a")); + assert_eq!(tokens[1].user_id, "bob"); + assert!(tokens[1].tenant.is_none()); + } + other => panic!("expected Static, got {:?}", other), + } + } + + #[test] + fn loads_jwt_auth() { + let f = write( + r#" +backends: ["a:1"] +auth: + type: jwt + algorithms: ["RS256"] + issuer: "https://idp.example.com" + audience: "spark-connect-gateway" + key: + kind: pem_file + path: "/etc/gateway/idp-pub.pem" +"#, + ); + let c = Config::load(f.path()).unwrap(); + match c.auth { + AuthConfig::Jwt(s) => { + assert_eq!(s.algorithms, vec!["RS256"]); + assert_eq!(s.issuer.as_deref(), Some("https://idp.example.com")); + match s.key { + KeySource::PemFile { path } => assert_eq!(path, "/etc/gateway/idp-pub.pem"), + other => panic!("expected PemFile, got {:?}", other), + } + } + other => panic!("expected Jwt, got {:?}", other), + } + } + + #[test] + fn tracing_defaults_to_off() { + let f = write("backends: [\"a:1\"]\n"); + let c = Config::load(f.path()).unwrap(); + assert!(c.tracing.is_none()); + } + + #[test] + fn loads_tracing_block() { + let f = write( + r#" +backends: ["a:1"] +tracing: + endpoint: "http://otel-collector:4317" + service_name: "scg-staging" + sample_ratio: 0.25 + export_timeout_secs: 5 +"#, + ); + let c = Config::load(f.path()).unwrap(); + let t = c.tracing.expect("tracing settings parsed"); + assert_eq!(t.endpoint.as_deref(), Some("http://otel-collector:4317")); + assert_eq!(t.service_name, "scg-staging"); + assert!((t.sample_ratio - 0.25).abs() < 1e-9); + assert_eq!(t.export_timeout_secs, 5); + } + + #[test] + fn tracing_block_endpoint_can_be_omitted_for_log_only() { + // A `tracing:` block without an endpoint is legal — the gateway + // skips OTLP export but still respects the other knobs (e.g. + // service_name) for when the user later sets an endpoint. + let f = write( + r#" +backends: ["a:1"] +tracing: + service_name: "scg-test" +"#, + ); + let c = Config::load(f.path()).unwrap(); + let t = c.tracing.expect("tracing settings parsed"); + assert!(t.endpoint.is_none()); + assert_eq!(t.service_name, "scg-test"); + // Defaults round-trip: + assert!((t.sample_ratio - 1.0).abs() < 1e-9); + assert_eq!(t.export_timeout_secs, 10); + } + + #[test] + fn affinity_store_defaults_to_memory() { + let f = write("backends: [\"a:1\"]\n"); + let c = Config::load(f.path()).unwrap(); + assert!(matches!(c.affinity_store, AffinityStoreConfig::Memory)); + } + + #[test] + fn loads_redis_affinity_store() { + let f = write( + r#" +backends: ["a:1"] +affinity_store: + type: redis + url: "redis://redis-cluster:6379" + key_prefix: "scg-prod" + session_ttl_secs: 7200 + op_ttl_secs: 600 +"#, + ); + let c = Config::load(f.path()).unwrap(); + match c.affinity_store { + AffinityStoreConfig::Redis(s) => { + assert_eq!(s.url, "redis://redis-cluster:6379"); + assert_eq!(s.key_prefix, "scg-prod"); + assert_eq!(s.session_ttl_secs, 7200); + assert_eq!(s.op_ttl_secs, 600); + } + other => panic!("expected Redis, got {:?}", other), + } + } + + #[test] + fn redis_affinity_store_has_sane_defaults() { + let f = write( + r#" +backends: ["a:1"] +affinity_store: + type: redis + url: "redis://localhost:6379" +"#, + ); + let c = Config::load(f.path()).unwrap(); + match c.affinity_store { + AffinityStoreConfig::Redis(s) => { + assert_eq!(s.key_prefix, "scg"); + assert_eq!(s.session_ttl_secs, 3600); + assert_eq!(s.op_ttl_secs, 900); + } + other => panic!("expected Redis, got {:?}", other), + } + } + + #[test] + fn health_check_defaults_to_disabled() { + let f = write("backends: [\"a:1\"]\n"); + let c = Config::load(f.path()).unwrap(); + assert!(!c.health_check.enabled); + // Default values present even when block omitted: + assert_eq!(c.health_check.interval_secs, 5); + assert_eq!(c.health_check.unhealthy_threshold, 3); + } + + #[test] + fn loads_health_check_block() { + let f = write( + r#" +backends: ["a:1"] +health_check: + enabled: true + interval_secs: 10 + timeout_secs: 3 + unhealthy_threshold: 5 + healthy_threshold: 3 +"#, + ); + let c = Config::load(f.path()).unwrap(); + assert!(c.health_check.enabled); + assert_eq!(c.health_check.interval_secs, 10); + assert_eq!(c.health_check.timeout_secs, 3); + assert_eq!(c.health_check.unhealthy_threshold, 5); + assert_eq!(c.health_check.healthy_threshold, 3); + } + + #[test] + fn shutdown_defaults() { + let f = write("backends: [\"a:1\"]\n"); + let c = Config::load(f.path()).unwrap(); + assert_eq!(c.shutdown.deadline_secs, 30); + } + + #[test] + fn loads_shutdown_block() { + let f = write( + r#" +backends: ["a:1"] +shutdown: + deadline_secs: 90 +"#, + ); + let c = Config::load(f.path()).unwrap(); + assert_eq!(c.shutdown.deadline_secs, 90); + } + + #[test] + fn tenant_resolver_defaults_to_from_claim_use_default() { + let f = write("backends: [\"a:1\"]\n"); + let c = Config::load(f.path()).unwrap(); + assert!(matches!( + c.tenant_resolver.source, + TenantResolverSource::FromClaim + )); + assert!(matches!( + c.tenant_resolver.on_missing, + TenantOnMissing::UseDefault + )); + assert_eq!(c.tenant_resolver.default_name, "default"); + } + + #[test] + fn loads_from_claim_reject() { + let f = write( + r#" +backends: ["a:1"] +tenant_resolver: + source: from_claim + on_missing: reject + default_name: "default" +"#, + ); + let c = Config::load(f.path()).unwrap(); + assert!(matches!( + c.tenant_resolver.source, + TenantResolverSource::FromClaim + )); + assert!(matches!( + c.tenant_resolver.on_missing, + TenantOnMissing::Reject + )); + } + + #[test] + fn loads_from_metadata_with_custom_header() { + let f = write( + r#" +backends: ["a:1"] +tenant_resolver: + source: from_metadata + header: "x-org" + on_missing: use_default + default_name: "shared" +"#, + ); + let c = Config::load(f.path()).unwrap(); + match c.tenant_resolver.source { + TenantResolverSource::FromMetadata { header } => assert_eq!(header, "x-org"), + other => panic!("expected FromMetadata, got {:?}", other), + } + assert_eq!(c.tenant_resolver.default_name, "shared"); + } + + #[test] + fn from_metadata_header_defaults_to_x_tenant() { + let f = write( + r#" +backends: ["a:1"] +tenant_resolver: + source: from_metadata +"#, + ); + let c = Config::load(f.path()).unwrap(); + match c.tenant_resolver.source { + TenantResolverSource::FromMetadata { header } => assert_eq!(header, "x-tenant"), + other => panic!("expected FromMetadata, got {:?}", other), + } + } + + #[test] + fn loads_always_default() { + let f = write( + r#" +backends: ["a:1"] +tenant_resolver: + source: always_default + default_name: "single-tenant" +"#, + ); + let c = Config::load(f.path()).unwrap(); + assert!(matches!( + c.tenant_resolver.source, + TenantResolverSource::AlwaysDefault + )); + assert_eq!(c.tenant_resolver.default_name, "single-tenant"); + } + + #[test] + fn tenant_pools_default_empty_use_default() { + let f = write("backends: [\"a:1\"]\n"); + let c = Config::load(f.path()).unwrap(); + assert!(c.tenant_pools.overrides.is_empty()); + assert!(matches!( + c.tenant_pools.on_unknown_tenant, + UnknownTenantPolicySetting::UseDefault + )); + } + + #[test] + fn loads_tenant_pools_with_overrides() { + let f = write( + r#" +backends: ["default-a:1", "default-b:1"] +tenant_pools: + on_unknown_tenant: reject + overrides: + team-a: + type: static + addresses: ["a-1:15002", "a-2:15002"] + team-b: + type: k8s + namespace: spark-b + service_name: spark-connect + port: 15002 +"#, + ); + let c = Config::load(f.path()).unwrap(); + assert_eq!(c.tenant_pools.overrides.len(), 2); + assert!(matches!( + c.tenant_pools.on_unknown_tenant, + UnknownTenantPolicySetting::Reject + )); + let team_a = c.tenant_pools.overrides.get("team-a").unwrap(); + assert!(team_a.backend_token.is_none()); + match &team_a.discovery { + BackendDiscovery::Static { addresses } => { + assert_eq!( + addresses, + &vec!["a-1:15002".to_string(), "a-2:15002".to_string()] + ) + } + other => panic!("expected Static, got {:?}", other), + } + match &c.tenant_pools.overrides.get("team-b").unwrap().discovery { + BackendDiscovery::K8s { + namespace, + service_name, + port, + } => { + assert_eq!(namespace, "spark-b"); + assert_eq!(service_name, "spark-connect"); + assert_eq!(*port, 15002); + } + other => panic!("expected K8s, got {:?}", other), + } + } + + #[test] + fn backend_token_defaults_to_none() { + let f = write("backends: [\"a:1\"]\n"); + let c = Config::load(f.path()).unwrap(); + assert!(c.backend_token.is_none()); + } + + #[test] + fn loads_inline_backend_token() { + let f = write( + r#" +backends: ["a:1"] +backend_token: + kind: inline + token: "deadbeef" +"#, + ); + let c = Config::load(f.path()).unwrap(); + let src = c.backend_token.expect("backend_token parsed"); + assert_eq!(src.resolve().unwrap(), "deadbeef"); + } + + #[test] + fn loads_env_backend_token() { + let f = write( + r#" +backends: ["a:1"] +backend_token: + kind: env + name: SCG_TEST_BACKEND_TOKEN +"#, + ); + let c = Config::load(f.path()).unwrap(); + let src = c.backend_token.expect("backend_token parsed"); + assert!(matches!( + src.resolve().unwrap_err(), + ConfigError::TokenEnvMissing { .. } + )); + // Env-var manipulation is process-global; this test relies on + // the name being unique to it, so set/remove is safe even + // with the parallel test runner. + std::env::set_var("SCG_TEST_BACKEND_TOKEN", "from-env"); + assert_eq!(src.resolve().unwrap(), "from-env"); + std::env::remove_var("SCG_TEST_BACKEND_TOKEN"); + } + + #[test] + fn loads_file_backend_token_and_trims_newline() { + let tok = write("s3cr3t\n"); + let f = write(&format!( + "backends: [\"a:1\"]\nbackend_token:\n kind: file\n path: \"{}\"\n", + tok.path().display() + )); + let c = Config::load(f.path()).unwrap(); + let src = c.backend_token.expect("backend_token parsed"); + assert_eq!(src.resolve().unwrap(), "s3cr3t"); + } + + #[test] + fn empty_backend_token_rejected_at_resolve() { + let src = BackendTokenSource::Inline { token: "".into() }; + assert!(matches!( + src.resolve().unwrap_err(), + ConfigError::TokenEmpty + )); + } + + #[test] + fn tenant_override_can_carry_its_own_backend_token() { + // `backend_token` sits at the same level as the flattened + // discovery fields inside an override entry. + let f = write( + r#" +backends: ["default:15002"] +backend_token: + kind: inline + token: "default-token" +tenant_pools: + overrides: + team-a: + type: static + addresses: ["a-1:15002"] + backend_token: + kind: inline + token: "team-a-token" + team-b: + type: static + addresses: ["b-1:15002"] +"#, + ); + let c = Config::load(f.path()).unwrap(); + let a = c.tenant_pools.overrides.get("team-a").unwrap(); + assert_eq!( + a.backend_token.as_ref().unwrap().resolve().unwrap(), + "team-a-token" + ); + // team-b has no token of its own — inheritance of the + // top-level token happens at wiring time, not parse time. + let b = c.tenant_pools.overrides.get("team-b").unwrap(); + assert!(b.backend_token.is_none()); + assert!(matches!( + &b.discovery, + BackendDiscovery::Static { addresses } if addresses == &vec!["b-1:15002".to_string()] + )); + } + + #[test] + fn rate_limit_defaults_to_disabled() { + let f = write("backends: [\"a:1\"]\n"); + let c = Config::load(f.path()).unwrap(); + assert!(!c.rate_limit.enabled); + assert_eq!(c.rate_limit.default.rpcs_per_second, 0.0); + assert!(c.rate_limit.overrides.is_empty()); + } + + #[test] + fn loads_rate_limit_with_overrides() { + let f = write( + r#" +backends: ["a:1"] +rate_limit: + enabled: true + default: + rpcs_per_second: 100 + burst: 200 + overrides: + team-a: + rpcs_per_second: 500 + burst: 1000 + per_user_rpcs_per_second: 50 + per_user_burst: 100 +"#, + ); + let c = Config::load(f.path()).unwrap(); + assert!(c.rate_limit.enabled); + assert_eq!(c.rate_limit.default.rpcs_per_second, 100.0); + assert_eq!(c.rate_limit.default.burst, 200); + let a = c.rate_limit.overrides.get("team-a").unwrap(); + assert_eq!(a.rpcs_per_second, 500.0); + assert_eq!(a.burst, 1000); + assert_eq!(a.per_user_rpcs_per_second, 50.0); + assert_eq!(a.per_user_burst, 100); + } + + #[test] + fn rate_limit_store_defaults_to_memory() { + let f = write( + r#" +backends: ["a:1"] +rate_limit: + enabled: true + default: + rpcs_per_second: 100 + burst: 200 +"#, + ); + let c = Config::load(f.path()).unwrap(); + assert_eq!(c.rate_limit.store, RateLimitStore::Memory); + // Redis defaults are present but irrelevant when store=memory. + assert_eq!(c.rate_limit.redis.url, "redis://localhost:6379"); + assert_eq!(c.rate_limit.redis.key_prefix, "scg-rl"); + assert_eq!(c.rate_limit.redis.on_failure, RateLimitFailMode::Open); + } + + #[test] + fn loads_rate_limit_with_redis_store() { + let f = write( + r#" +backends: ["a:1"] +rate_limit: + enabled: true + store: redis + redis: + url: "redis://shared.svc:6379" + key_prefix: "myrl" + key_ttl_secs: 7200 + on_failure: closed + default: + rpcs_per_second: 100 + burst: 200 +"#, + ); + let c = Config::load(f.path()).unwrap(); + assert_eq!(c.rate_limit.store, RateLimitStore::Redis); + assert_eq!(c.rate_limit.redis.url, "redis://shared.svc:6379"); + assert_eq!(c.rate_limit.redis.key_prefix, "myrl"); + assert_eq!(c.rate_limit.redis.key_ttl_secs, 7200); + assert_eq!(c.rate_limit.redis.on_failure, RateLimitFailMode::Closed); + } + + #[test] + fn audit_defaults_to_enabled_signals_only() { + let f = write("backends: [\"a:1\"]\n"); + let c = Config::load(f.path()).unwrap(); + assert!(c.audit.enabled); + assert!(!c.audit.log_successful_rpcs); + } + + #[test] + fn loads_audit_block_with_successful_rpcs_on() { + let f = write( + r#" +backends: ["a:1"] +audit: + enabled: true + log_successful_rpcs: true +"#, + ); + let c = Config::load(f.path()).unwrap(); + assert!(c.audit.enabled); + assert!(c.audit.log_successful_rpcs); + } + + #[test] + fn audit_can_be_disabled() { + let f = write( + r#" +backends: ["a:1"] +audit: + enabled: false +"#, + ); + let c = Config::load(f.path()).unwrap(); + assert!(!c.audit.enabled); + } + + #[test] + fn loads_oidc_auth() { + let f = write( + r#" +backends: ["a:1"] +auth: + type: oidc + algorithms: ["RS256"] + discovery_url: "https://idp.example.com/.well-known/openid-configuration" + audience: "spark-connect-gateway" +"#, + ); + let c = Config::load(f.path()).unwrap(); + match c.auth { + AuthConfig::Oidc(s) => { + assert_eq!( + s.discovery_url.as_deref(), + Some("https://idp.example.com/.well-known/openid-configuration") + ); + assert!(s.jwks_url.is_none()); + } + other => panic!("expected Oidc, got {:?}", other), + } + } +}