From c1a34a00426e90d9389eeae72bcdb3177cee95ab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Thu, 3 Sep 2026 15:25:15 +0800 Subject: [PATCH 1/4] [gateway] Add Prometheus identity labels --- fluss-gateway/conf/gateway.yaml | 13 ++++++- fluss-gateway/src/config.rs | 58 ++++++++++++++++++++++++++-- fluss-gateway/src/lifecycle.rs | 2 +- fluss-gateway/src/observability.rs | 62 ++++++++++++++++++++++++++---- 4 files changed, 121 insertions(+), 14 deletions(-) diff --git a/fluss-gateway/conf/gateway.yaml b/fluss-gateway/conf/gateway.yaml index 4ead4175ec4..d89229a3289 100644 --- a/fluss-gateway/conf/gateway.yaml +++ b/fluss-gateway/conf/gateway.yaml @@ -28,10 +28,19 @@ # Do not store credentials in this file when it is baked into an image. Mount a # separate file or inject the corresponding FLUSS_GATEWAY__* variables. -# Optional stable identity used in logs and metrics. When omitted, the Gateway -# does not invent a persistent identity. +# Optional stable identity used in logs and diagnostics. When omitted, the +# Gateway does not invent a persistent identity. # gateway.instance-id: gateway-1 +# Optional stable identity shared by all processes serving one logical Gateway, +# exposed as the gateway_id label on metrics. This is separate from the backend +# Fluss cluster IDs under gateway.cluster..*. +# gateway.id: gateway-production + +# Optional advertised host identity exposed as the host label on metrics. +# In an orchestrated deployment this can be the Pod IP or a stable hostname. +# gateway.host: 192.0.2.10 + #============================================================================== # REST Server #============================================================================== diff --git a/fluss-gateway/src/config.rs b/fluss-gateway/src/config.rs index 847b30c3b1a..a5027a01095 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 advertised host identity, 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 unusable metric identities 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. + /// Metric identity remains optional because collectors may attach equivalent target labels. 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.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..54116d5fc29 100644 --- a/fluss-gateway/src/observability.rs +++ b/fluss-gateway/src/observability.rs @@ -20,9 +20,11 @@ //! [`METRIC_DEFINITIONS`] lists implemented metric families with their kinds, units, descriptions, //! 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. +//! Family-specific labels describe an operation or a bounded outcome. `cluster`, sourced from +//! validated configuration, is the only resource-name label the gateway itself emits. Configured +//! process identity is attached to every family as a global label. +use crate::config::ServerConfig; use log::{LevelFilter, Log, Metadata, Record}; use metrics::Unit; use metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle}; @@ -52,6 +54,11 @@ impl Log for StderrLogger { static LOGGER: StderrLogger = StderrLogger; static METRICS_HANDLE: OnceLock = OnceLock::new(); +const GATEWAY_ID_LABEL: &str = "gateway_id"; +const HOST_LABEL: &str = "host"; +#[cfg(test)] +const IDENTITY_LABELS: &[&str] = &[GATEWAY_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 +204,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 +217,21 @@ 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()), + (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 @@ -429,6 +448,10 @@ fn parse_level(value: &str) -> LevelFilter { #[cfg(test)] mod tests { use super::*; + use metrics::{Key, Recorder}; + + static METADATA: metrics::Metadata = + metrics::Metadata::new(module_path!(), metrics::Level::INFO, Some(module_path!())); #[test] fn parses_supported_global_levels() { @@ -441,6 +464,24 @@ mod tests { assert_eq!(parse_level("module=debug"), LevelFilter::Info); } + #[test] + fn configured_gateway_identity_is_applied_to_every_metric() { + let server = ServerConfig { + gateway_id: Some("gateway-production".to_string()), + host: Some("2001:db8::10".to_string()), + ..ServerConfig::default() + }; + let recorder = prometheus_builder(&server).unwrap().build_recorder(); + recorder + .register_counter(&Key::from_name("test_counter"), &METADATA) + .increment(1); + + let rendered = recorder.handle().render(); + for label in ["gateway_id=\"gateway-production\"", "host=\"2001:db8::10\""] { + assert!(rendered.contains(label), "missing {label} in {rendered}"); + } + } + /// Allowed metric families, including those reserved for future capabilities. const ALLOWED_METRIC_FAMILIES: &[&str] = &[ "fluss_gateway_rest_requests_total", @@ -520,6 +561,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 From c248f4120d6f21c30d629d94a40d87e1fe9edc71 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Thu, 3 Sep 2026 20:43:33 +0800 Subject: [PATCH 2/4] [gateway] Reuse instance ID for Prometheus identity --- fluss-gateway/conf/gateway.yaml | 13 +++----- fluss-gateway/src/config.rs | 50 ++++++++++-------------------- fluss-gateway/src/observability.rs | 34 ++++---------------- fluss-gateway/tests/http_api.rs | 8 ++++- fluss-gateway/tests/support/mod.rs | 17 +++++++--- 5 files changed, 47 insertions(+), 75 deletions(-) diff --git a/fluss-gateway/conf/gateway.yaml b/fluss-gateway/conf/gateway.yaml index d89229a3289..dfe75b43225 100644 --- a/fluss-gateway/conf/gateway.yaml +++ b/fluss-gateway/conf/gateway.yaml @@ -28,19 +28,14 @@ # Do not store credentials in this file when it is baked into an image. Mount a # separate file or inject the corresponding FLUSS_GATEWAY__* variables. -# Optional stable identity used in logs and diagnostics. When omitted, the -# Gateway does not invent a persistent identity. +# Optional stable identity used in logs and metrics. When omitted, the Gateway +# does not invent a persistent identity. # gateway.instance-id: gateway-1 -# Optional stable identity shared by all processes serving one logical Gateway, -# exposed as the gateway_id label on metrics. This is separate from the backend -# Fluss cluster IDs under gateway.cluster..*. +# Optional identity shared by all processes serving one logical Gateway, +# exposed as the gateway_id label on metrics. # gateway.id: gateway-production -# Optional advertised host identity exposed as the host label on metrics. -# In an orchestrated deployment this can be the Pod IP or a stable hostname. -# gateway.host: 192.0.2.10 - #============================================================================== # REST Server #============================================================================== diff --git a/fluss-gateway/src/config.rs b/fluss-gateway/src/config.rs index a5027a01095..6963eb9e228 100644 --- a/fluss-gateway/src/config.rs +++ b/fluss-gateway/src/config.rs @@ -214,7 +214,6 @@ 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"; @@ -432,7 +431,6 @@ 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, @@ -580,8 +578,6 @@ pub struct ServerConfig { pub instance_id: Option, /// Optional identity shared by all processes serving one logical Gateway. pub gateway_id: Option, - /// Optional advertised host identity, exposed as the `host` metric label. - pub host: Option, pub rest: RestServerConfig, pub metrics: MetricsServerConfig, } @@ -1075,9 +1071,9 @@ impl GatewayConfig { } } - /// Rejects unusable metric identities or a port clash between the two listeners. + /// Rejects an unusable instance identity, Gateway identity, or a port clash between the two listeners. /// - /// Metric identity remains optional because collectors may attach equivalent target labels. + /// 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; @@ -1094,17 +1090,12 @@ 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.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.gateway_id.as_deref().is_some_and(|value| { + value.trim().is_empty() || value.len() > 253 || value.chars().any(char::is_control) + }) { + problems.push(format!( + "{GATEWAY_ID_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!( @@ -1808,7 +1799,6 @@ mod tests { 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 @@ -1826,7 +1816,6 @@ mod tests { 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() @@ -2021,10 +2010,6 @@ mod tests { "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(), @@ -2065,7 +2050,6 @@ mod tests { 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() @@ -2460,15 +2444,15 @@ 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:?}" - ); - } + fn malformed_gateway_id_rejected() { + for value in ["", " ", "line\\u0007break"] { + let error = load_file(&format!("{GATEWAY_ID_KEY}: \"{value}\"\n")).unwrap_err(); + assert!( + problems(error) + .iter() + .any(|problem| problem.contains(GATEWAY_ID_KEY)), + "{GATEWAY_ID_KEY} accepted {value:?}" + ); } } diff --git a/fluss-gateway/src/observability.rs b/fluss-gateway/src/observability.rs index 54116d5fc29..8199feac285 100644 --- a/fluss-gateway/src/observability.rs +++ b/fluss-gateway/src/observability.rs @@ -20,9 +20,9 @@ //! [`METRIC_DEFINITIONS`] lists implemented metric families with their kinds, units, descriptions, //! and label sets. Families for future capabilities are added alongside their implementations. //! -//! Family-specific labels describe an operation or a bounded outcome. `cluster`, sourced from -//! validated configuration, is the only resource-name label the gateway itself emits. Configured -//! process identity is attached to every family as a global label. +//! Labels describe an operation or a bounded outcome. `cluster`, sourced from validated configuration, is the +//! 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}; @@ -55,9 +55,9 @@ static LOGGER: StderrLogger = StderrLogger; static METRICS_HANDLE: OnceLock = OnceLock::new(); const GATEWAY_ID_LABEL: &str = "gateway_id"; -const HOST_LABEL: &str = "host"; +const INSTANCE_ID_LABEL: &str = "instance_id"; #[cfg(test)] -const IDENTITY_LABELS: &[&str] = &[GATEWAY_ID_LABEL, HOST_LABEL]; +const IDENTITY_LABELS: &[&str] = &[GATEWAY_ID_LABEL, INSTANCE_ID_LABEL]; /// Buckets for the duration histograms, spanning a fast local answer to a request that runs into the /// configured deadline. @@ -223,7 +223,7 @@ fn prometheus_builder(server: &ServerConfig) -> Result LevelFilter { #[cfg(test)] mod tests { use super::*; - use metrics::{Key, Recorder}; - - static METADATA: metrics::Metadata = - metrics::Metadata::new(module_path!(), metrics::Level::INFO, Some(module_path!())); #[test] fn parses_supported_global_levels() { @@ -464,24 +460,6 @@ mod tests { assert_eq!(parse_level("module=debug"), LevelFilter::Info); } - #[test] - fn configured_gateway_identity_is_applied_to_every_metric() { - let server = ServerConfig { - gateway_id: Some("gateway-production".to_string()), - host: Some("2001:db8::10".to_string()), - ..ServerConfig::default() - }; - let recorder = prometheus_builder(&server).unwrap().build_recorder(); - recorder - .register_counter(&Key::from_name("test_counter"), &METADATA) - .increment(1); - - let rendered = recorder.handle().render(); - for label in ["gateway_id=\"gateway-production\"", "host=\"2001:db8::10\""] { - assert!(rendered.contains(label), "missing {label} in {rendered}"); - } - } - /// Allowed metric families, including those reserved for future capabilities. const ALLOWED_METRIC_FAMILIES: &[&str] = &[ "fluss_gateway_rest_requests_total", diff --git a/fluss-gateway/tests/http_api.rs b/fluss-gateway/tests/http_api.rs index 6ef32f71a1a..2dbb4f1f527 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 @@ -102,6 +102,12 @@ async fn request_durations_are_exported_as_histograms() { exposition.contains("fluss_gateway_rest_request_duration_seconds_bucket"), "histogram buckets are exported: {exposition}" ); + for label in [ + "gateway_id=\"gateway-production\"", + "instance_id=\"gateway-1\"", + ] { + assert!(exposition.contains(label), "missing {label}: {exposition}"); + } gateway.shutdown().await.expect("clean shutdown"); } diff --git a/fluss-gateway/tests/support/mod.rs b/fluss-gateway/tests/support/mod.rs index 32934ff6a4f..b2ae338a590 100644 --- a/fluss-gateway/tests/support/mod.rs +++ b/fluss-gateway/tests/support/mod.rs @@ -151,17 +151,26 @@ 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()); + 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 From 45ef8f6dc2fefbeef83e9a880009f29a5d8ea972 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Fri, 4 Sep 2026 08:10:53 +0800 Subject: [PATCH 3/4] [gateway] Add host to Prometheus identity --- fluss-gateway/conf/gateway.yaml | 3 ++ fluss-gateway/src/config.rs | 46 ++++++++++++++++++++---------- fluss-gateway/src/observability.rs | 4 ++- fluss-gateway/tests/http_api.rs | 1 + fluss-gateway/tests/support/mod.rs | 1 + 5 files changed, 39 insertions(+), 16 deletions(-) diff --git a/fluss-gateway/conf/gateway.yaml b/fluss-gateway/conf/gateway.yaml index dfe75b43225..68b24bbe5b9 100644 --- a/fluss-gateway/conf/gateway.yaml +++ b/fluss-gateway/conf/gateway.yaml @@ -36,6 +36,9 @@ # 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 6963eb9e228..41adb2ca3d5 100644 --- a/fluss-gateway/src/config.rs +++ b/fluss-gateway/src/config.rs @@ -214,6 +214,7 @@ 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"; @@ -431,6 +432,7 @@ 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, @@ -578,6 +580,8 @@ pub struct ServerConfig { 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, } @@ -1090,12 +1094,17 @@ impl GatewayConfig { )); } } - if server.gateway_id.as_deref().is_some_and(|value| { - value.trim().is_empty() || value.len() > 253 || value.chars().any(char::is_control) - }) { - problems.push(format!( - "{GATEWAY_ID_KEY} must be 1-253 bytes and contain no control characters" - )); + 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!( @@ -1799,6 +1808,7 @@ mod tests { 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 @@ -1816,6 +1826,7 @@ mod tests { 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() @@ -2010,6 +2021,10 @@ mod tests { "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(), @@ -2050,6 +2065,7 @@ mod tests { 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() @@ -2444,15 +2460,15 @@ mod tests { } #[test] - fn malformed_gateway_id_rejected() { - for value in ["", " ", "line\\u0007break"] { - let error = load_file(&format!("{GATEWAY_ID_KEY}: \"{value}\"\n")).unwrap_err(); - assert!( - problems(error) - .iter() - .any(|problem| problem.contains(GATEWAY_ID_KEY)), - "{GATEWAY_ID_KEY} accepted {value:?}" - ); + 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:?}" + ); + } } } diff --git a/fluss-gateway/src/observability.rs b/fluss-gateway/src/observability.rs index 8199feac285..9c1871c7072 100644 --- a/fluss-gateway/src/observability.rs +++ b/fluss-gateway/src/observability.rs @@ -56,8 +56,9 @@ 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]; +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. @@ -224,6 +225,7 @@ fn prometheus_builder(server: &ServerConfig) -> Result RunningGateway { 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") From 5346cb75d23246d137916d2fc0a5fc0e25264d03 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Fri, 4 Sep 2026 09:41:33 +0800 Subject: [PATCH 4/4] [gateway] Strengthen metrics identity coverage --- fluss-gateway/tests/http_api.rs | 20 +++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/fluss-gateway/tests/http_api.rs b/fluss-gateway/tests/http_api.rs index 780b631d405..4192f6733b9 100644 --- a/fluss-gateway/tests/http_api.rs +++ b/fluss-gateway/tests/http_api.rs @@ -87,6 +87,7 @@ async fn metrics_endpoint_exports_histograms_and_gateway_identity() { .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,12 +103,25 @@ async fn metrics_endpoint_exports_histograms_and_gateway_identity() { exposition.contains("fluss_gateway_rest_request_duration_seconds_bucket"), "histogram buckets are exported: {exposition}" ); - for label in [ + let identity_labels = [ "gateway_id=\"gateway-production\"", "instance_id=\"gateway-1\"", "host=\"192.0.2.10\"", - ] { - assert!(exposition.contains(label), "missing {label}: {exposition}"); + ]; + 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");