diff --git a/changelog.d/confinement_gauge_component_type_override.fix.md b/changelog.d/confinement_gauge_component_type_override.fix.md new file mode 100644 index 0000000000000..157709c74823b --- /dev/null +++ b/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 diff --git a/src/sinks/amqp/config.rs b/src/sinks/amqp/config.rs index b07319b2a069d..1c7e2bc9f4c91 100644 --- a/src/sinks/amqp/config.rs +++ b/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)) } diff --git a/src/sinks/aws_cloudwatch_logs/config.rs b/src/sinks/aws_cloudwatch_logs/config.rs index 281dd158c5678..31dd414d6b9b1 100644 --- a/src/sinks/aws_cloudwatch_logs/config.rs +++ b/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)) } diff --git a/src/sinks/aws_s3/config.rs b/src/sinks/aws_s3/config.rs index 7cebc471c207e..91535cadf1e96 100644 --- a/src/sinks/aws_s3/config.rs +++ b/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)) } diff --git a/src/sinks/axiom/config.rs b/src/sinks/axiom/config.rs index 449a4281b036b..84f4b06968287 100644 --- a/src/sinks/axiom/config.rs +++ b/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) } diff --git a/src/sinks/azure_blob/config.rs b/src/sinks/azure_blob/config.rs index e7ce79da94587..055741503846b 100644 --- a/src/sinks/azure_blob/config.rs +++ b/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)) } diff --git a/src/sinks/clickhouse/config.rs b/src/sinks/clickhouse/config.rs index 4f2fc15d7171f..483e40847da9e 100644 --- a/src/sinks/clickhouse/config.rs +++ b/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)) } diff --git a/src/sinks/doris/config.rs b/src/sinks/doris/config.rs index d54b1f5d25147..aa81005c0ee6a 100644 --- a/src/sinks/doris/config.rs +++ b/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)) } diff --git a/src/sinks/elasticsearch/config.rs b/src/sinks/elasticsearch/config.rs index 7c4d4cc4ab04f..6d2fc97d972ec 100644 --- a/src/sinks/elasticsearch/config.rs +++ b/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)) } diff --git a/src/sinks/file/mod.rs b/src/sinks/file/mod.rs index 65453c98fc218..0fce618be94f2 100644 --- a/src/sinks/file/mod.rs +++ b/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(), diff --git a/src/sinks/gcp/cloud_storage.rs b/src/sinks/gcp/cloud_storage.rs index edfbbbcca9e33..ca8ec6c986c4b 100644 --- a/src/sinks/gcp/cloud_storage.rs +++ b/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)) } diff --git a/src/sinks/gcp/stackdriver/logs/config.rs b/src/sinks/gcp/stackdriver/logs/config.rs index 375606cd6a032..965205a99d8bd 100644 --- a/src/sinks/gcp/stackdriver/logs/config.rs +++ b/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)) } diff --git a/src/sinks/greptimedb/logs/config.rs b/src/sinks/greptimedb/logs/config.rs index d41d46c340f18..99b97e6aafd8e 100644 --- a/src/sinks/greptimedb/logs/config.rs +++ b/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)) } diff --git a/src/sinks/http/config.rs b/src/sinks/http/config.rs index 4e2939b4227e2..e056f7390f8d3 100644 --- a/src/sinks/http/config.rs +++ b/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) } diff --git a/src/sinks/humio/logs.rs b/src/sinks/humio/logs.rs index fd7509d718dcd..a3c5208628f47 100644 --- a/src/sinks/humio/logs.rs +++ b/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::(); } + + // 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::(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)] diff --git a/src/sinks/humio/metrics.rs b/src/sinks/humio/metrics.rs index 8941eb567799e..378a961b0982c 100644 --- a/src/sinks/humio/metrics.rs +++ b/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)) } diff --git a/src/sinks/kafka/config.rs b/src/sinks/kafka/config.rs index 5aa91c947694b..17a2b11edb969 100644 --- a/src/sinks/kafka/config.rs +++ b/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)) } diff --git a/src/sinks/loki/config.rs b/src/sinks/loki/config.rs index 044438c837bd7..1c9fb62ceae70 100644 --- a/src/sinks/loki/config.rs +++ b/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)) } diff --git a/src/sinks/mqtt/config.rs b/src/sinks/mqtt/config.rs index aa03e7b52a512..5eddd64b3c965 100644 --- a/src/sinks/mqtt/config.rs +++ b/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 }), diff --git a/src/sinks/nats/config.rs b/src/sinks/nats/config.rs index 0d4f9fed0a86d..a7444a79c5834 100644 --- a/src/sinks/nats/config.rs +++ b/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)) } diff --git a/src/sinks/opentelemetry/mod.rs b/src/sinks/opentelemetry/mod.rs index 61e2202e939c4..411617133cb08 100644 --- a/src/sinks/opentelemetry/mod.rs +++ b/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) } } diff --git a/src/sinks/prometheus/remote_write/config.rs b/src/sinks/prometheus/remote_write/config.rs index 1435ad3e53b1f..a792ab13ff60a 100644 --- a/src/sinks/prometheus/remote_write/config.rs +++ b/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)) } diff --git a/src/sinks/pulsar/config.rs b/src/sinks/pulsar/config.rs index 9bd1713aff541..f074002c027dc 100644 --- a/src/sinks/pulsar/config.rs +++ b/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)) } diff --git a/src/sinks/redis/config.rs b/src/sinks/redis/config.rs index 952e44858fbb2..84e477b73c850 100644 --- a/src/sinks/redis/config.rs +++ b/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)) } diff --git a/src/sinks/splunk_hec/logs/config.rs b/src/sinks/splunk_hec/logs/config.rs index 3146569b13469..d86d8cce03f60 100644 --- a/src/sinks/splunk_hec/logs/config.rs +++ b/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) } diff --git a/src/sinks/splunk_hec/metrics/config.rs b/src/sinks/splunk_hec/metrics/config.rs index 131970d21824f..ed89111ba207f 100644 --- a/src/sinks/splunk_hec/metrics/config.rs +++ b/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)) } diff --git a/src/sinks/webhdfs/config.rs b/src/sinks/webhdfs/config.rs index c37684ffa1aee..beaa8080476dc 100644 --- a/src/sinks/webhdfs/config.rs +++ b/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)) } diff --git a/src/template.rs b/src/template.rs index d13e8a16a4e65..2277f90cfe5b3 100644 --- a/src/template.rs +++ b/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); } }