[gateway] Add Prometheus identity labels

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
 #==============================================================================

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<String>,
+    /// Optional identity shared by all processes serving one logical Gateway.
+    pub gateway_id: Option<String>,
+    /// Optional network location exposed as the `host` metric label.
+    pub host: Option<String>,
     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<String>) {
         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!(

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

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<PrometheusHandle> = 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<PrometheusBuilder, String> {
+    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

fluss-gateway/tests/http_api.rs

@@ -79,14 +79,15 @@ 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
         .metrics_addr()
         .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::<Vec<_>>();
+    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");
 }

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