feat(influxdb3_server): include node_id label on exported metrics

Add `with_default_attributes` to `PrometheusTextEncoder` so the same
label set can be applied to every reported observation. `handle_metrics`
uses it to attach `node_id` to each series, sparing operators from
relabeling on hostname. `node_modes` is more relevant to Enterprise and
is left for a follow-up.

Author: o6ivp

core/metric_exporters/src/lib.rs

@@ -29,6 +29,11 @@ pub struct PrometheusTextEncoder<'a, W: Write> {
 
     encoder: TextEncoder,
     writer: &'a mut W,
+
+    /// Labels prepended to every reported observation. Per-metric attributes
+    /// must not reuse these label names — Prometheus rejects duplicates within
+    /// a series.
+    default_attributes: Attributes,
 }
 
 impl<'a, W: Write> PrometheusTextEncoder<'a, W> {
@@ -37,8 +42,15 @@ impl<'a, W: Write> PrometheusTextEncoder<'a, W> {
             metric: None,
             encoder: TextEncoder::new(),
             writer,
+            default_attributes: Attributes::from(&[]),
         }
     }
+
+    /// Set labels that will be added to every reported observation.
+    pub fn with_default_attributes(mut self, attributes: Attributes) -> Self {
+        self.default_attributes = attributes;
+        self
+    }
 }
 
 impl<W: Write> metric::Reporter for PrometheusTextEncoder<'_, W> {
@@ -79,8 +91,9 @@ impl<W: Write> metric::Reporter for PrometheusTextEncoder<'_, W> {
         let mut metric = Metric::default();
 
         metric.set_label(
-            attributes
+            self.default_attributes
                 .iter()
+                .chain(attributes.iter())
                 .map(|(name, value)| {
                     let mut pair = LabelPair::default();
                     pair.set_name(name.to_string());
@@ -355,6 +368,29 @@ foo_total{tag1="value",tag2="value2"} 7
         assert_not_contains!(tracing_capture.to_string(), "error");
     }
 
+    #[test]
+    fn test_encode_with_default_attributes() {
+        let registry = Registry::new();
+        let counter: Metric<U64Counter> = registry.register_metric("foo", "a counter metric");
+        counter.recorder(&[("tag1", "value")]).inc(3);
+
+        let mut buffer = Vec::new();
+        let mut default_attrs = Attributes::from(&[]);
+        default_attrs.insert("node_id", "node-1".to_string());
+        let mut encoder =
+            PrometheusTextEncoder::new(&mut buffer).with_default_attributes(default_attrs);
+        registry.report(&mut encoder);
+
+        let buffer = String::from_utf8(buffer).unwrap();
+        let expected = r#"
+# HELP foo_total a counter metric
+# TYPE foo_total counter
+foo_total{node_id="node-1",tag1="value"} 3
+"#
+        .trim_start();
+        assert_eq!(&buffer, expected, "{buffer}");
+    }
+
     #[test]
     fn pushgateway_path_behaviour() {
         let client = PushGatewayClient::new(

influxdb3/src/commands/serve.rs

@@ -1118,6 +1118,7 @@ pub async fn command(config: Config, user_params: HashMap<String, String>) -> Re
         trace_exporter,
         trace_header_parser,
         Arc::clone(&telemetry_store),
+        Arc::from(node_id.as_str()),
     );
 
     if config.without_auth {

influxdb3/tests/server/metrics.rs

@@ -0,0 +1,32 @@
+use crate::server::TestServer;
+
+#[tokio::test]
+async fn metrics_endpoint_includes_node_id_label() {
+    let server = TestServer::spawn().await;
+    let body = server
+        .http_client()
+        .get(format!("{}/metrics", server.client_addr()))
+        .send()
+        .await
+        .unwrap()
+        .text()
+        .await
+        .unwrap();
+
+    let series_lines: Vec<&str> = body
+        .lines()
+        .filter(|l| !l.is_empty() && !l.starts_with('#'))
+        .collect();
+
+    assert!(
+        !series_lines.is_empty(),
+        "expected /metrics to expose at least one series, got:\n{body}"
+    );
+
+    for line in &series_lines {
+        assert!(
+            line.contains(r#"node_id="test-server""#),
+            "expected every series to carry node_id label, missing on:\n  {line}\nfull body:\n{body}"
+        );
+    }
+}

influxdb3/tests/server/mod.rs

@@ -37,6 +37,7 @@ mod flight;
 mod gen1_lookback_guard;
 mod limits;
 mod logs;
+mod metrics;
 mod packages;
 mod ping;
 mod plugin_restriction;

influxdb3_server/src/http.rs

@@ -1204,7 +1204,10 @@ impl HttpApi {
 
     fn handle_metrics(&self) -> Result<Response> {
         let mut body: Vec<u8> = Default::default();
-        let mut reporter = metric_exporters::PrometheusTextEncoder::new(&mut body);
+        let mut default_attributes = metric::Attributes::from(&[]);
+        default_attributes.insert("node_id", self.common_state.node_id().to_string());
+        let mut reporter = metric_exporters::PrometheusTextEncoder::new(&mut body)
+            .with_default_attributes(default_attributes);
         self.common_state.metrics.report(&mut reporter);
 
         // Add required OpenMetrics EOF marker

influxdb3_server/src/lib.rs

@@ -96,6 +96,7 @@ pub struct CommonServerState {
     trace_exporter: Option<Arc<trace_exporters::export::AsyncExporter>>,
     trace_header_parser: TraceHeaderParser,
     telemetry_store: Arc<TelemetryStore>,
+    node_id: Arc<str>,
 }
 
 impl CommonServerState {
@@ -105,16 +106,22 @@ impl CommonServerState {
         trace_exporter: Option<Arc<trace_exporters::export::AsyncExporter>>,
         trace_header_parser: TraceHeaderParser,
         telemetry_store: Arc<TelemetryStore>,
+        node_id: Arc<str>,
     ) -> Self {
         Self {
             catalog,
             metrics,
             trace_exporter,
             trace_header_parser,
             telemetry_store,
+            node_id,
         }
     }
 
+    pub fn node_id(&self) -> &str {
+        &self.node_id
+    }
+
     pub fn trace_exporter(&self) -> Option<Arc<trace_exporters::export::AsyncExporter>> {
         self.trace_exporter.clone()
     }

influxdb3_server/src/tests.rs

@@ -1382,6 +1382,7 @@ async fn setup_server(start_time: i64) -> (String, CancellationToken, Arc<dyn Wr
         None,
         trace_header_parser,
         Arc::clone(&sample_telem_store),
+        Arc::from("test-node"),
     );
     let query_executor = Arc::new(QueryExecutorImpl::new(CreateQueryExecutorArgs {
         catalog: write_buffer.catalog(),