fix(topology): prevent security_confinement_disabled gauge from expiring

changelog.d/security_confinement_gauge_expiry.fix.md

@@ -0,0 +1,3 @@
+Fixed the `vector_security_confinement_disabled` internal metric disappearing after the metric idle timeout (300 seconds by default) while a sink was still running with `dangerously_allow_unconfined_template_resolution` enabled. The gauge is now owned by the topology and held for the lifetime of each sink, and refreshed on configuration reload, so alerts watching this metric no longer silently stop firing.
+
+authors: thomasqueirozb

src/config/sink.rs

@@ -266,6 +266,15 @@ pub trait SinkConfig: DynClone + NamedComponent + core::fmt::Debug + Send + Sync
         Vec::new()
     }
 
+    /// Returns this sink's template-confinement config, if it supports confinement.
+    ///
+    /// The topology uses this to own the `vector_security_confinement_disabled`
+    /// gauge for the sink's lifetime. `None` (the default) means the sink does
+    /// not participate in template confinement and emits no gauge.
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        None
+    }
+
     /// Gets the acknowledgements configuration for this sink.
     fn acknowledgements(&self) -> &AcknowledgementsConfig;
 }

src/sinks/amqp/config.rs

@@ -145,10 +145,13 @@ 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);
         Ok((VectorSink::from_event_streamsink(sink), hc))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         Input::new(DataType::Log)
     }

src/sinks/aws_cloudwatch_logs/config.rs

@@ -240,11 +240,13 @@ impl SinkConfig for CloudwatchLogsSinkConfig {
 
             service: svc,
         };
-
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         let requirement =
             schema::Requirement::empty().optional_meaning("timestamp", Kind::timestamp());

src/sinks/aws_s3/config.rs

@@ -232,10 +232,13 @@ 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);
         Ok((sink, healthcheck))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         #[cfg(feature = "codecs-parquet")]
         if let Some(batch_encoding) = &self.batch_encoding {

src/sinks/axiom/config.rs

@@ -193,14 +193,16 @@ impl SinkConfig for AxiomConfig {
             confinement: self.confinement.clone(),
         };
 
-        // Route through the HTTP builder that doesn't emit the gauge, so
-        // per-template warnings and the sink-level gauge both carry
-        // `component_type=axiom` — not `http`.
-        let result = http_sink_config
-            .build_without_confinement_gauge(cx, Self::NAME)
-            .await?;
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
-        Ok(result)
+        // Route through the HTTP builder threaded with our own component type,
+        // so per-template security warnings carry `component_type=axiom` rather
+        // than `http`.
+        http_sink_config
+            .build_with_component_type(cx, Self::NAME)
+            .await
+    }
+
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
     }
 
     fn input(&self) -> Input {

src/sinks/azure_blob/config.rs

@@ -275,10 +275,13 @@ 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);
         Ok((sink, healthcheck))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         Input::new(self.encoding.config().1.input_type() & DataType::Log)
     }

src/sinks/clickhouse/config.rs

@@ -280,11 +280,13 @@ impl SinkConfig for ClickhouseConfig {
         );
 
         let healthcheck = Box::pin(healthcheck(client, endpoint, auth));
-
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         Input::log()
     }

src/sinks/doris/config.rs

@@ -241,11 +241,13 @@ impl SinkConfig for DorisConfig {
         }))
         .map_ok(|((), _)| ())
         .boxed();
-
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
         Ok((sink, healthcheck))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         Input::log()
     }

src/sinks/elasticsearch/config.rs

@@ -753,10 +753,13 @@ impl SinkConfig for ElasticsearchConfig {
         )
         .map_ok(|((), _)| ())
         .boxed();
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
         Ok((stream, healthcheck))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         let requirements = Requirement::empty().optional_meaning("timestamp", Kind::timestamp());
 

src/sinks/file/mod.rs

@@ -248,13 +248,16 @@ 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);
         Ok((
             super::VectorSink::from_event_streamsink(sink),
             future::ok(()).boxed(),
         ))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         Input::new(self.encoding.config().1.input_type())
     }

src/sinks/gcp/cloud_storage.rs

@@ -281,11 +281,13 @@ 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);
         Ok((sink, healthcheck))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         Input::new(self.encoding.config().1.input_type() & DataType::Log)
     }

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

@@ -319,11 +319,13 @@ impl SinkConfig for StackdriverConfig {
         let healthcheck = healthcheck(client, auth.clone(), uri).boxed();
 
         auth.spawn_regenerate_token();
-
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         let requirement =
             schema::Requirement::empty().required_meaning("timestamp", Kind::timestamp());

src/sinks/greptimedb/logs/config.rs

@@ -207,10 +207,13 @@ impl SinkConfig for GreptimeDBLogsConfig {
             this.endpoint.clone(),
             auth.clone(),
         ));
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         Input::log()
     }

src/sinks/http/config.rs

@@ -241,9 +241,11 @@ pub(super) fn validate_payload_wrapper(
 #[typetag::serde(name = "http")]
 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);
-        Ok(result)
+        self.build_with_component_type(cx, Self::NAME).await
+    }
+
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
     }
 
     fn input(&self) -> Input {
@@ -269,12 +271,11 @@ impl SinkConfig for HttpSinkConfig {
 }
 
 impl HttpSinkConfig {
-    /// Confinement + sink construction without emitting the per-sink
-    /// confinement gauge. `component_name` is threaded through so both the
-    /// gauge (emitted by the caller) and per-template security warnings
-    /// carry the outer sink type — `http` when this is the top-level sink,
-    /// `opentelemetry` when [`OpenTelemetryConfig::build`] delegates here.
-    pub(crate) async fn build_without_confinement_gauge(
+    /// Confinement + sink construction. `component_name` is threaded through so
+    /// per-template security warnings carry the outer sink type — `http` when
+    /// this is the top-level sink, `opentelemetry` when
+    /// [`OpenTelemetryConfig::build`] delegates here.
+    pub(crate) async fn build_with_component_type(
         &self,
         cx: SinkContext,
         component_name: &'static str,

src/sinks/humio/logs.rs

@@ -180,9 +180,11 @@ impl GenerateConfig for HumioLogsConfig {
 #[typetag::serde(name = "humio_logs")]
 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);
-        Ok(result)
+        self.build_with_component_type(cx, Self::NAME)
+    }
+
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
     }
 
     fn input(&self) -> Input {
@@ -195,19 +197,17 @@ impl SinkConfig for HumioLogsConfig {
 }
 
 impl HumioLogsConfig {
-    /// Confinement + sink construction without emitting the per-sink
-    /// confinement gauge. `component_name` is threaded through so both the
-    /// gauge (emitted by the caller) and per-template security warnings
-    /// carry the outer sink type — `humio_logs` when this is the top-level
-    /// sink, `humio_metrics` when [`HumioMetricsConfig::build`] delegates
-    /// here.
-    pub(super) fn build_without_confinement_gauge(
+    /// Confinement + sink construction. `component_name` is threaded through so
+    /// per-template security warnings carry the outer sink type — `humio_logs`
+    /// when this is the top-level sink, `humio_metrics` when
+    /// [`HumioMetricsConfig::build`] delegates here.
+    pub(super) fn build_with_component_type(
         &self,
         cx: SinkContext,
         component_name: &'static str,
     ) -> crate::Result<(VectorSink, Healthcheck)> {
         self.build_hec_config()
-            .build_without_confinement_gauge(cx, component_name)
+            .build_with_component_type(cx, component_name)
     }
 
     fn build_hec_config(&self) -> HecLogsSinkConfig {

src/sinks/humio/metrics.rs

@@ -191,21 +191,22 @@ impl SinkConfig for HumioMetricsConfig {
             confinement: self.confinement.clone(),
         };
 
-        // Route through the inner Humio helper so the wrapped
-        // `humio_logs` sink's gauge isn't emitted alongside our own —
-        // operators watching `security_confinement_disabled` for a
-        // `humio_metrics` sink should only see one series.
-        let (sink, healthcheck) = sink.build_without_confinement_gauge(cx, Self::NAME)?;
+        // Route through the inner Humio helper threaded with our own component
+        // type, so per-template security warnings carry `humio_metrics` rather
+        // than the delegated `humio_logs`/`splunk_hec_logs`.
+        let (sink, healthcheck) = sink.build_with_component_type(cx, Self::NAME)?;
 
         let sink = HumioMetricsSink {
             inner: sink,
             transform,
         };
-
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
         Ok((VectorSink::Stream(Box::new(sink)), healthcheck))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         Input::metric()
     }

src/sinks/kafka/config.rs

@@ -292,10 +292,13 @@ 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);
         Ok((VectorSink::from_event_streamsink(sink), hc))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         let requirements = Requirement::empty().optional_meaning("timestamp", Kind::timestamp());
 

src/sinks/loki/config.rs

@@ -240,11 +240,13 @@ impl SinkConfig for LokiConfig {
         let sink = LokiSink::new(config.clone(), client.clone())?;
 
         let healthcheck = healthcheck(config, client).boxed();
-
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         let requirement =
             schema::Requirement::empty().optional_meaning("timestamp", Kind::timestamp());

src/sinks/mqtt/config.rs

@@ -122,14 +122,16 @@ impl SinkConfig for MqttSinkConfig {
             .confine(&self.confinement, Self::NAME, "topic")?;
         let connector = config.build_connector()?;
         let sink = MqttSink::new(&config, connector.clone())?;
-
-        self.confinement.set_confinement_gauge("sink", Self::NAME);
         Ok((
             VectorSink::from_event_streamsink(sink),
             Box::pin(async move { connector.healthcheck().await }),
         ))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         Input::log()
     }

src/sinks/nats/config.rs

@@ -191,10 +191,13 @@ 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);
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
 
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
     fn input(&self) -> Input {
         Input::new(self.encoding.config().input_type() & DataType::Log)
     }