diff --git a/fluss-gateway/conf/gateway.yaml b/fluss-gateway/conf/gateway.yaml index 4ead4175ec4..68b24bbe5b9 100644 --- a/fluss-gateway/conf/gateway.yaml +++ b/fluss-gateway/conf/gateway.yaml @@ -32,6 +32,13 @@ # does not invent a persistent identity. # gateway.instance-id: gateway-1 +# Optional identity shared by all processes serving one logical Gateway, +# exposed as the gateway_id label on metrics. +# gateway.id: gateway-production + +# Optional network location exposed as the host label on metrics. +# gateway.host: 192.0.2.10 + #============================================================================== # REST Server #============================================================================== diff --git a/fluss-gateway/src/config.rs b/fluss-gateway/src/config.rs index 847b30c3b1a..41adb2ca3d5 100644 --- a/fluss-gateway/src/config.rs +++ b/fluss-gateway/src/config.rs @@ -213,6 +213,8 @@ impl<'de> Deserialize<'de> for ByteSize { } const INSTANCE_ID_KEY: &str = "gateway.instance-id"; +const GATEWAY_ID_KEY: &str = "gateway.id"; +const HOST_KEY: &str = "gateway.host"; const REST_LISTEN_KEY: &str = "gateway.rest.listen"; const REST_HEADER_READ_TIMEOUT_KEY: &str = "gateway.rest.header-read-timeout"; const REST_REQUEST_TIMEOUT_KEY: &str = "gateway.rest.request-timeout"; @@ -429,6 +431,8 @@ macro_rules! typed_entry { const CONFIG_ENTRIES: &[GatewayConfigEntry] = &[ typed_entry!(GatewayConfigEntry, INSTANCE_ID_KEY, optional server.instance_id), + typed_entry!(GatewayConfigEntry, GATEWAY_ID_KEY, optional server.gateway_id), + typed_entry!(GatewayConfigEntry, HOST_KEY, optional server.host), typed_entry!( GatewayConfigEntry, REST_LISTEN_KEY, @@ -574,6 +578,10 @@ const CLUSTER_ENTRIES: &[ClusterConfigEntry] = &[ pub struct ServerConfig { /// Optional identity for logs and diagnostics. pub instance_id: Option, + /// Optional identity shared by all processes serving one logical Gateway. + pub gateway_id: Option, + /// Optional network location exposed as the `host` metric label. + pub host: Option, pub rest: RestServerConfig, pub metrics: MetricsServerConfig, } @@ -1067,10 +1075,9 @@ impl GatewayConfig { } } - /// Rejects an unusable instance identity or a port clash between the two listeners. + /// Rejects an unusable instance identity, Gateway identity, or a port clash between the two listeners. /// - /// A non-loopback listener does **not** require an instance ID. Nothing the gateway returns is scoped to an - /// instance, so there is no identity to pin. + /// A non-loopback listener does not require either identity. fn validate_identity(&self, problems: &mut Vec) { let server = &self.server; let rest_address = server.rest.bind_address; @@ -1087,6 +1094,18 @@ impl GatewayConfig { )); } } + for (key, value) in [ + (GATEWAY_ID_KEY, server.gateway_id.as_deref()), + (HOST_KEY, server.host.as_deref()), + ] { + if value.is_some_and(|value| { + value.trim().is_empty() || value.len() > 253 || value.chars().any(char::is_control) + }) { + problems.push(format!( + "{key} must be 1-253 bytes and contain no control characters" + )); + } + } if server.metrics.enabled && addresses_overlap(rest_address, server.metrics.bind_address) { problems.push(format!( "{} ({}) must differ from {} ({})", @@ -1788,6 +1807,8 @@ mod tests { let config = load_file( r#" gateway.instance-id: gateway-1 + gateway.id: gateway-production + gateway.host: 192.0.2.10 gateway.rest.listen: 0.0.0.0:8080 gateway.rest.write.max-request-bytes: 32MiB gateway.rest.request-timeout: 30s @@ -1801,6 +1822,11 @@ mod tests { ) .unwrap(); assert_eq!(config.server.instance_id.as_deref(), Some("gateway-1")); + assert_eq!( + config.server.gateway_id.as_deref(), + Some("gateway-production") + ); + assert_eq!(config.server.host.as_deref(), Some("192.0.2.10")); assert_eq!( config.server.rest.bind_address, "0.0.0.0:8080".parse().unwrap() @@ -1991,6 +2017,14 @@ mod tests { let mut env = no_env(); env.extend([ ("FLUSS_GATEWAY__INSTANCE_ID".to_string(), "123".to_string()), + ( + "FLUSS_GATEWAY__ID".to_string(), + "gateway-production".to_string(), + ), + ( + "FLUSS_GATEWAY__HOST".to_string(), + "2001:db8::10".to_string(), + ), ( "FLUSS_GATEWAY__REST__LISTEN".to_string(), "127.0.0.1:18080".to_string(), @@ -2027,6 +2061,11 @@ mod tests { let config = load(None, &env, &CliOverrides::default()).unwrap(); assert_eq!(config.server.instance_id.as_deref(), Some("123")); + assert_eq!( + config.server.gateway_id.as_deref(), + Some("gateway-production") + ); + assert_eq!(config.server.host.as_deref(), Some("2001:db8::10")); assert_eq!( config.server.rest.bind_address, "127.0.0.1:18080".parse().unwrap() @@ -2420,6 +2459,19 @@ mod tests { ); } + #[test] + fn malformed_metric_identity_rejected() { + for key in [GATEWAY_ID_KEY, HOST_KEY] { + for value in ["", " ", "line\\u0007break"] { + let error = load_file(&format!("{key}: \"{value}\"\n")).unwrap_err(); + assert!( + problems(error).iter().any(|problem| problem.contains(key)), + "{key} accepted {value:?}" + ); + } + } + } + #[test] fn duration_units() { assert_eq!( diff --git a/fluss-gateway/src/lifecycle.rs b/fluss-gateway/src/lifecycle.rs index fe1aba5dcb8..be28e49a532 100644 --- a/fluss-gateway/src/lifecycle.rs +++ b/fluss-gateway/src/lifecycle.rs @@ -280,7 +280,7 @@ async fn start_internal( for warning in config.warnings() { log::warn!("{warning}"); } - observability::init_metrics(config.server.metrics.enabled)?; + observability::init_metrics(&config.server)?; let listener = bind_listener(config.server.rest.bind_address, "REST").await?; let local_addr = listener diff --git a/fluss-gateway/src/observability.rs b/fluss-gateway/src/observability.rs index 60083cb53eb..9c1871c7072 100644 --- a/fluss-gateway/src/observability.rs +++ b/fluss-gateway/src/observability.rs @@ -21,8 +21,10 @@ //! and label sets. Families for future capabilities are added alongside their implementations. //! //! Labels describe an operation or a bounded outcome. `cluster`, sourced from validated configuration, is the -//! only resource-name label the gateway itself emits. +//! only resource-name label the gateway itself emits. Configured Gateway and instance identities are attached +//! to every family as global labels. +use crate::config::ServerConfig; use log::{LevelFilter, Log, Metadata, Record}; use metrics::Unit; use metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle}; @@ -52,6 +54,12 @@ impl Log for StderrLogger { static LOGGER: StderrLogger = StderrLogger; static METRICS_HANDLE: OnceLock = OnceLock::new(); +const GATEWAY_ID_LABEL: &str = "gateway_id"; +const INSTANCE_ID_LABEL: &str = "instance_id"; +const HOST_LABEL: &str = "host"; +#[cfg(test)] +const IDENTITY_LABELS: &[&str] = &[GATEWAY_ID_LABEL, INSTANCE_ID_LABEL, HOST_LABEL]; + /// Buckets for the duration histograms, spanning a fast local answer to a request that runs into the /// configured deadline. /// @@ -197,14 +205,11 @@ pub fn init_logging() { } /// Installs the process-wide Prometheus recorder before the Fluss client creates metric handles. -pub fn init_metrics(enabled: bool) -> Result<(), String> { - if !enabled || METRICS_HANDLE.get().is_some() { +pub fn init_metrics(server: &ServerConfig) -> Result<(), String> { + if !server.metrics.enabled || METRICS_HANDLE.get().is_some() { return Ok(()); } - let recorder = PrometheusBuilder::new() - .set_buckets(DURATION_BUCKETS) - .map_err(|error| format!("failed to configure histogram buckets: {error}"))? - .build_recorder(); + let recorder = prometheus_builder(server)?.build_recorder(); let handle = recorder.handle(); metrics::set_global_recorder(recorder) .map_err(|error| format!("failed to install Prometheus recorder: {error}"))?; @@ -213,6 +218,22 @@ pub fn init_metrics(enabled: bool) -> Result<(), String> { Ok(()) } +fn prometheus_builder(server: &ServerConfig) -> Result { + let mut builder = PrometheusBuilder::new() + .set_buckets(DURATION_BUCKETS) + .map_err(|error| format!("failed to configure histogram buckets: {error}"))?; + for (key, value) in [ + (GATEWAY_ID_LABEL, server.gateway_id.as_deref()), + (INSTANCE_ID_LABEL, server.instance_id.as_deref()), + (HOST_LABEL, server.host.as_deref()), + ] { + if let Some(value) = value { + builder = builder.add_global_label(key, value.to_string()); + } + } + Ok(builder) +} + /// Records one completed REST request against the matched route template, never the raw URI. /// /// `operation` is the matched route template; `code` is the HTTP status. The caller bounds @@ -520,6 +541,11 @@ mod tests { "metric {} has forbidden label {label}", definition.name ); + assert!( + !IDENTITY_LABELS.contains(label), + "metric {} shadows global identity label {label}", + definition.name + ); } let resource_labels = definition .labels diff --git a/fluss-gateway/tests/http_api.rs b/fluss-gateway/tests/http_api.rs index 6ef32f71a1a..4192f6733b9 100644 --- a/fluss-gateway/tests/http_api.rs +++ b/fluss-gateway/tests/http_api.rs @@ -79,7 +79,7 @@ async fn an_unknown_route_returns_the_shared_error_envelope() { /// The duration families are exported as Prometheus histograms, which aggregate across gateway instances. /// Without explicit buckets the exporter emits pre-computed summary quantiles instead, which do not. #[tokio::test] -async fn request_durations_are_exported_as_histograms() { +async fn metrics_endpoint_exports_histograms_and_gateway_identity() { let gateway = support::start_gateway_with_metrics().await; let api = Api::new(format!("http://{}", gateway.local_addr())); let metrics_address = gateway @@ -87,6 +87,7 @@ async fn request_durations_are_exported_as_histograms() { .expect("the metrics listener is bound"); api.get_ok("/health").await; + metrics::counter!("test_external_component_requests_total").increment(1); let exposition = Api::new(format!("http://{metrics_address}")) .get("/metrics") .await @@ -102,6 +103,26 @@ async fn request_durations_are_exported_as_histograms() { exposition.contains("fluss_gateway_rest_request_duration_seconds_bucket"), "histogram buckets are exported: {exposition}" ); + let identity_labels = [ + "gateway_id=\"gateway-production\"", + "instance_id=\"gateway-1\"", + "host=\"192.0.2.10\"", + ]; + let samples = exposition + .lines() + .filter(|line| !line.is_empty() && !line.starts_with('#')) + .collect::>(); + assert!( + samples + .iter() + .any(|line| line.starts_with("test_external_component_requests_total")), + "external component metric is exported: {exposition}" + ); + for sample in samples { + for label in identity_labels { + assert!(sample.contains(label), "missing {label}: {sample}"); + } + } gateway.shutdown().await.expect("clean shutdown"); } diff --git a/fluss-gateway/tests/support/mod.rs b/fluss-gateway/tests/support/mod.rs index 32934ff6a4f..b2d9a1d377f 100644 --- a/fluss-gateway/tests/support/mod.rs +++ b/fluss-gateway/tests/support/mod.rs @@ -151,17 +151,27 @@ pub async fn start_gateway() -> RunningGateway { /// Starts an in-process gateway with the Prometheus listener bound to an ephemeral port. pub async fn start_gateway_with_metrics() -> RunningGateway { - start(true).await + let mut config = gateway_config(true); + config.server.gateway_id = Some("gateway-production".to_string()); + config.server.instance_id = Some("gateway-1".to_string()); + config.server.host = Some("192.0.2.10".to_string()); + fluss_gateway::lifecycle::start(config) + .await + .expect("gateway starts") } async fn start(metrics: bool) -> RunningGateway { + fluss_gateway::lifecycle::start(gateway_config(metrics)) + .await + .expect("gateway starts") +} + +fn gateway_config(metrics: bool) -> GatewayConfig { let mut config = GatewayConfig::default(); config.server.rest.bind_address = "127.0.0.1:0".parse().expect("valid"); config.server.metrics.enabled = metrics; config.server.metrics.bind_address = "127.0.0.1:0".parse().expect("valid"); - fluss_gateway::lifecycle::start(config) - .await - .expect("gateway starts") + config } /// A command that runs the compiled gateway executable, so the suites exercise CLI parsing, configuration