fix(security): inherit component_type on confinement gauge from ambient span

changelog.d/confinement_gauge_component_type_override.fix.md

@@ -0,0 +1,3 @@
+Fixed the `vector_security_confinement_disabled` gauge reporting the wrong `component_type` when a sink delegates its build to another sink's `SinkConfig::build` internally (e.g. `humio_metrics` → `humio_logs`), or is wrapped by an external sink that constructs and builds a Vector sink directly. The gauge now inherits `component_type`/`component_kind` from the ambient tracing span like every other per-component metric, instead of hardcoding the delegate's own type.
+
+authors: vladimir-dd

src/sinks/amqp/config.rs

@@ -145,7 +145,7 @@ impl SinkConfig for AmqpSinkConfig {
             .transpose()?;
         let sink = AmqpSink::new(config).await?;
         let hc = healthcheck(sink.channels.clone()).boxed();
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((VectorSink::from_event_streamsink(sink), hc))
     }
 

src/sinks/aws_cloudwatch_logs/config.rs

@@ -241,7 +241,7 @@ impl SinkConfig for CloudwatchLogsSinkConfig {
             service: svc,
         };
 
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
 

src/sinks/aws_s3/config.rs

@@ -232,7 +232,7 @@ impl SinkConfig for S3SinkConfig {
         let service = self.create_service(&cx.proxy).await?;
         let healthcheck = self.build_healthcheck(service.client())?;
         let sink = self.build_processor(service, cx)?;
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((sink, healthcheck))
     }
 

src/sinks/axiom/config.rs

@@ -199,7 +199,7 @@ impl SinkConfig for AxiomConfig {
         let result = http_sink_config
             .build_without_confinement_gauge(cx, Self::NAME)
             .await?;
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok(result)
     }
 

src/sinks/azure_blob/config.rs

@@ -275,7 +275,7 @@ impl SinkConfig for AzureBlobSinkConfig {
 
         let healthcheck = build_healthcheck(self.container_name.clone(), Arc::clone(&client))?;
         let sink = self.build_processor(client)?;
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((sink, healthcheck))
     }
 

src/sinks/clickhouse/config.rs

@@ -281,7 +281,7 @@ impl SinkConfig for ClickhouseConfig {
 
         let healthcheck = Box::pin(healthcheck(client, endpoint, auth));
 
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
 

src/sinks/doris/config.rs

@@ -242,7 +242,7 @@ impl SinkConfig for DorisConfig {
         .map_ok(|((), _)| ())
         .boxed();
 
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((sink, healthcheck))
     }
 

src/sinks/elasticsearch/config.rs

@@ -753,7 +753,7 @@ impl SinkConfig for ElasticsearchConfig {
         )
         .map_ok(|((), _)| ())
         .boxed();
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((stream, healthcheck))
     }
 

src/sinks/file/mod.rs

@@ -248,7 +248,7 @@ impl SinkConfig for FileSinkConfig {
         cx: SinkContext,
     ) -> crate::Result<(super::VectorSink, super::Healthcheck)> {
         let sink = FileSink::new(self, cx)?;
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((
             super::VectorSink::from_event_streamsink(sink),
             future::ok(()).boxed(),

src/sinks/gcp/cloud_storage.rs

@@ -282,7 +282,7 @@ impl SinkConfig for GcsSinkConfig {
         auth.spawn_regenerate_token();
         let sink = self.build_sink(client, base_url, auth, cx)?;
 
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((sink, healthcheck))
     }
 

src/sinks/gcp/stackdriver/logs/config.rs

@@ -320,7 +320,7 @@ impl SinkConfig for StackdriverConfig {
 
         auth.spawn_regenerate_token();
 
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
 

src/sinks/greptimedb/logs/config.rs

@@ -207,7 +207,7 @@ impl SinkConfig for GreptimeDBLogsConfig {
             this.endpoint.clone(),
             auth.clone(),
         ));
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
 

src/sinks/http/config.rs

@@ -242,7 +242,7 @@ pub(super) fn validate_payload_wrapper(
 impl SinkConfig for HttpSinkConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
         let result = self.build_without_confinement_gauge(cx, Self::NAME).await?;
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok(result)
     }
 

src/sinks/humio/logs.rs

@@ -181,7 +181,7 @@ impl GenerateConfig for HumioLogsConfig {
 impl SinkConfig for HumioLogsConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
         let result = self.build_without_confinement_gauge(cx, Self::NAME)?;
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok(result)
     }
 
@@ -239,12 +239,65 @@ impl HumioLogsConfig {
 
 #[cfg(test)]
 mod tests {
+    use indoc::indoc;
+    use tracing::Instrument;
+
     use super::*;
+    use crate::sinks::util::test::load_sink;
 
     #[test]
     fn generate_config() {
         crate::test_util::test_generate_config::<HumioLogsConfig>();
     }
+
+    // Regression test for the confinement gauge reporting the wrong
+    // `component_type` when this sink is built by a wrapping caller
+    // outside this crate that constructs a `HumioLogsConfig` directly and
+    // calls `SinkConfig::build` on it.
+    //
+    // In production, the topology builder opens a span tagged with the
+    // *outer* (wrapping) sink's component_type around the whole
+    // `SinkConfig::build` call; this test opens an equivalent span by hand
+    // to simulate that outer wrapper, since no such wrapper exists in this
+    // crate.
+    #[tokio::test]
+    async fn confinement_gauge_reports_wrapping_sink_component_type() {
+        crate::test_util::trace_init();
+
+        let (config, cx) = load_sink::<HumioLogsConfig>(indoc! {r#"
+            token = "atoken"
+            encoding.codec = "json"
+        "#})
+        .unwrap();
+
+        let span = error_span!(
+            "sink",
+            component_kind = "sink",
+            component_id = "confinement_gauge_wrapper_test",
+            component_type = "wrapping_sink",
+        );
+        let (_sink, _healthcheck) = config.build(cx).instrument(span).await.unwrap();
+
+        let metrics = vector_lib::metrics::Controller::get()
+            .expect("metrics controller not initialized")
+            .capture_metrics();
+
+        let gauge = metrics
+            .iter()
+            .find(|m| {
+                m.name() == "security_confinement_disabled"
+                    && m.tag_value("component_id")
+                        == Some("confinement_gauge_wrapper_test".to_string())
+            })
+            .expect("security_confinement_disabled gauge was not emitted");
+
+        assert_eq!(
+            gauge.tag_value("component_type"),
+            Some("wrapping_sink".to_string()),
+            "confinement gauge must report the wrapping sink's component_type, not the \
+             delegated humio_logs sink's own type"
+        );
+    }
 }
 
 #[cfg(test)]

src/sinks/humio/metrics.rs

@@ -202,7 +202,7 @@ impl SinkConfig for HumioMetricsConfig {
             transform,
         };
 
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((VectorSink::Stream(Box::new(sink)), healthcheck))
     }
 

src/sinks/kafka/config.rs

@@ -292,7 +292,7 @@ impl SinkConfig for KafkaSinkConfig {
             .confine(&self.confinement, Self::NAME, "topic")?;
         let sink = KafkaSink::new(config.clone())?;
         let hc = healthcheck(config, cx.healthcheck.clone()).boxed();
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((VectorSink::from_event_streamsink(sink), hc))
     }
 

src/sinks/loki/config.rs

@@ -241,7 +241,7 @@ impl SinkConfig for LokiConfig {
 
         let healthcheck = healthcheck(config, client).boxed();
 
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
 

src/sinks/mqtt/config.rs

@@ -123,7 +123,7 @@ impl SinkConfig for MqttSinkConfig {
         let connector = config.build_connector()?;
         let sink = MqttSink::new(&config, connector.clone())?;
 
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((
             VectorSink::from_event_streamsink(sink),
             Box::pin(async move { connector.healthcheck().await }),

src/sinks/nats/config.rs

@@ -191,7 +191,7 @@ impl SinkConfig for NatsSinkConfig {
             .confine(&self.confinement, Self::NAME, "subject")?;
         let sink = NatsSink::new(config.clone()).await?;
         let healthcheck = healthcheck(config).boxed();
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
 

src/sinks/opentelemetry/mod.rs

@@ -84,7 +84,7 @@ impl SinkConfig for OpenTelemetryConfig {
                 let result = config
                     .build_without_confinement_gauge(cx, Self::NAME)
                     .await?;
-                config.confinement.set_confinement_gauge("sink", Self::NAME);
+                config.confinement.set_confinement_gauge();
                 Ok(result)
             }
         }

src/sinks/prometheus/remote_write/config.rs

@@ -290,7 +290,7 @@ impl SinkConfig for RemoteWriteConfig {
             service,
         };
 
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
 

src/sinks/pulsar/config.rs

@@ -407,7 +407,7 @@ impl SinkConfig for PulsarSinkConfig {
         let sink = PulsarSink::new(client, config.clone())?;
         let hc = healthcheck(config).boxed();
 
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((VectorSink::from_event_streamsink(sink), hc))
     }
 

src/sinks/redis/config.rs

@@ -209,7 +209,7 @@ impl SinkConfig for RedisSinkConfig {
         let conn = config.build_connection().await?;
         let healthcheck = RedisSinkConfig::healthcheck(conn.clone()).boxed();
         let sink = RedisSink::new(&config, conn)?;
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((super::VectorSink::from_event_streamsink(sink), healthcheck))
     }
 

src/sinks/splunk_hec/logs/config.rs

@@ -235,7 +235,7 @@ impl HecLogsSinkConfig {
 impl SinkConfig for HecLogsSinkConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
         let result = self.build_without_confinement_gauge(cx, Self::NAME)?;
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok(result)
     }
 

src/sinks/splunk_hec/metrics/config.rs

@@ -172,7 +172,7 @@ impl SinkConfig for HecMetricsSinkConfig {
         )
         .boxed();
         let sink = config.build_processor(client, cx)?;
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((sink, healthcheck))
     }
 

src/sinks/webhdfs/config.rs

@@ -111,7 +111,7 @@ impl SinkConfig for WebHdfsConfig {
         let healthcheck = Box::pin(async move { Ok(check_op.check().await?) });
 
         let sink = self.build_processor(op)?;
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
+        self.confinement.set_confinement_gauge();
         Ok((sink, healthcheck))
     }
 

src/template.rs

@@ -1287,22 +1287,26 @@ impl ConfinementConfig {
     /// build, the topology manager keeps the old sink active but the
     /// failed replacement's write would silently misreport the live
     /// sink's confinement state.
-    pub fn set_confinement_gauge(
-        &self,
-        component_kind: &'static str,
-        component_type: &'static str,
-    ) {
+    ///
+    /// `component_kind`/`component_type` are deliberately *not* parameters
+    /// here: like every other per-component metric (e.g.
+    /// `component_sent_events_total`), they're picked up automatically from
+    /// the ambient tracing span the topology builder opens around the whole
+    /// `SinkConfig::build` call (see `build_sinks` in
+    /// `src/topology/builder.rs`), tagged with the outer sink's own
+    /// registered component type. A sink that delegates to another
+    /// `SinkConfig::build` internally, or is itself wrapped by an external
+    /// sink that constructs and delegates to it, gets the correct
+    /// outer/wrapping component type for free, with no plumbing required.
+    /// Passing them explicitly here would just override that with
+    /// whichever component happens to own this particular gauge call site.
+    pub fn set_confinement_gauge(&self) {
         let value = if self.dangerously_allow_unconfined_template_resolution {
             1.0
         } else {
             0.0
         };
-        gauge!(
-            GaugeName::SecurityConfinementDisabled,
-            "component_kind" => component_kind,
-            "component_type" => component_type,
-        )
-        .set(value);
+        gauge!(GaugeName::SecurityConfinementDisabled).set(value);
     }
 }