vector/config/
sink.rs

1#![allow(clippy::let_underscore_must_use)]
2
3use std::{any::Any, cell::RefCell, path::PathBuf, sync::Arc, time::Duration};
4
5use async_trait::async_trait;
6use derivative::Derivative;
7use dyn_clone::DynClone;
8use serde::Serialize;
9use serde_with::serde_as;
10use vector_lib::{
11    buffers::{BufferConfig, BufferType},
12    config::{AcknowledgementsConfig, GlobalOptions, Input},
13    configurable::{
14        Configurable, GenerateError, Metadata, NamedComponent,
15        attributes::CustomAttribute,
16        configurable_component,
17        schema::{SchemaGenerator, SchemaObject},
18    },
19    id::Inputs,
20    sink::VectorSink,
21};
22use vector_vrl_metrics::MetricsStorage;
23
24use super::{
25    ComponentKey, DynValidatedSink, ProxyConfig, Resource, dot_graph::GraphConfig, schema,
26};
27use crate::{
28    extra_context::ExtraContext,
29    sinks::{Healthcheck, util::UriSerde},
30};
31
32pub type BoxedSink = Box<dyn SinkConfig>;
33
34impl Configurable for BoxedSink {
35    fn referenceable_name() -> Option<&'static str> {
36        Some("vector::sinks::Sinks")
37    }
38
39    fn metadata() -> Metadata {
40        let mut metadata = Metadata::default();
41        metadata.set_description("Configurable sinks in Vector.");
42        metadata.add_custom_attribute(CustomAttribute::kv("docs::enum_tagging", "internal"));
43        metadata.add_custom_attribute(CustomAttribute::kv("docs::enum_tag_field", "type"));
44        metadata
45    }
46
47    fn generate_schema(
48        generator: &RefCell<SchemaGenerator>,
49    ) -> Result<SchemaObject, GenerateError> {
50        vector_lib::configurable::component::SinkDescription::generate_schemas(generator)
51    }
52}
53
54impl<T: SinkConfig + 'static> From<T> for BoxedSink {
55    fn from(value: T) -> Self {
56        Box::new(value)
57    }
58}
59
60/// Fully resolved sink component.
61#[configurable_component]
62#[configurable(metadata(docs::component_base_type = "sink"))]
63#[derive(Clone, Derivative)]
64#[derivative(Debug)]
65pub struct SinkOuter<T>
66where
67    T: Configurable + Serialize + 'static,
68{
69    #[configurable(derived)]
70    #[serde(default, skip_serializing_if = "vector_lib::serde::is_default")]
71    pub graph: GraphConfig,
72
73    #[configurable(derived)]
74    pub inputs: Inputs<T>,
75
76    /// The full URI to make HTTP healthcheck requests to.
77    ///
78    /// This must be a valid URI, which requires at least the scheme and host. All other
79    /// components -- port, path, etc -- are allowed as well.
80    #[configurable(deprecated, metadata(docs::hidden), validation(format = "uri"))]
81    pub healthcheck_uri: Option<UriSerde>,
82
83    #[configurable(derived)]
84    #[serde(default, deserialize_with = "crate::serde::bool_or_struct")]
85    pub healthcheck: SinkHealthcheckOptions,
86
87    #[configurable(derived)]
88    #[serde(default, skip_serializing_if = "vector_lib::serde::is_default")]
89    pub buffer: BufferConfig,
90
91    #[configurable(derived)]
92    #[serde(default, skip_serializing_if = "vector_lib::serde::is_default")]
93    pub proxy: ProxyConfig,
94
95    #[serde(flatten)]
96    #[configurable(metadata(docs::hidden))]
97    pub inner: BoxedSink,
98
99    /// Validated state, filled in during config compilation.
100    ///
101    /// This is the erased state produced by `ValidatedSink::validate`, retained so
102    /// the topology builder can build the sink without re-validating. It is never serialized
103    /// or diffed (see `#[serde(skip)]`), and is shared (via `Arc`) so enrichment-table-derived
104    /// sinks can carry it without cloning the underlying value.
105    #[serde(skip)]
106    #[derivative(Debug = "ignore")]
107    pub(crate) validated: Option<Arc<dyn Any + Send + Sync>>,
108}
109
110impl<T> SinkOuter<T>
111where
112    T: Configurable + Serialize,
113{
114    /// Builds the sink, dispatching through the erased validated boundary when the sink
115    /// has been migrated, or falling back to the raw config for legacy sinks.
116    pub async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
117        match self.inner.as_dyn_validated() {
118            Some(p) => {
119                let validated = self
120                    .validated
121                    .as_ref()
122                    .expect("validated state missing for migrated sink");
123                p.build_dyn(validated.as_ref(), cx).await
124            }
125            None => self.inner.build(cx).await,
126        }
127    }
128    pub fn new<I, IS>(inputs: I, inner: IS) -> SinkOuter<T>
129    where
130        I: IntoIterator<Item = T>,
131        IS: Into<BoxedSink>,
132    {
133        SinkOuter {
134            inputs: Inputs::from_iter(inputs),
135            buffer: Default::default(),
136            healthcheck: SinkHealthcheckOptions::default(),
137            healthcheck_uri: None,
138            inner: inner.into(),
139            proxy: Default::default(),
140            graph: Default::default(),
141            validated: None,
142        }
143    }
144
145    pub fn resources(&self, id: &ComponentKey) -> Vec<Resource> {
146        let mut resources = self.inner.resources();
147        for stage in self.buffer.stages() {
148            match stage {
149                BufferType::Memory { .. } => {}
150                BufferType::DiskV2 { .. } => resources.push(Resource::DiskBuffer(id.to_string())),
151            }
152        }
153        resources
154    }
155
156    pub fn healthcheck(&self) -> SinkHealthcheckOptions {
157        if self.healthcheck_uri.is_some() && self.healthcheck.uri.is_some() {
158            warn!(
159                "Both `healthcheck.uri` and `healthcheck_uri` options are specified. Using value of `healthcheck.uri`."
160            )
161        } else if self.healthcheck_uri.is_some() {
162            warn!(
163                "The `healthcheck_uri` option has been deprecated, use `healthcheck.uri` instead."
164            )
165        }
166        SinkHealthcheckOptions {
167            uri: self
168                .healthcheck
169                .uri
170                .clone()
171                .or_else(|| self.healthcheck_uri.clone()),
172            ..self.healthcheck.clone()
173        }
174    }
175
176    pub const fn proxy(&self) -> &ProxyConfig {
177        &self.proxy
178    }
179
180    pub(super) fn map_inputs<U>(self, f: impl Fn(&T) -> U) -> SinkOuter<U>
181    where
182        U: Configurable + Serialize,
183    {
184        let inputs = self.inputs.iter().map(f).collect::<Vec<_>>();
185        self.with_inputs(inputs)
186    }
187
188    pub(crate) fn with_inputs<I, U>(self, inputs: I) -> SinkOuter<U>
189    where
190        I: IntoIterator<Item = U>,
191        U: Configurable + Serialize,
192    {
193        SinkOuter {
194            inputs: Inputs::from_iter(inputs),
195            inner: self.inner,
196            buffer: self.buffer,
197            healthcheck: self.healthcheck,
198            healthcheck_uri: self.healthcheck_uri,
199            proxy: self.proxy,
200            graph: self.graph,
201            validated: self.validated,
202        }
203    }
204}
205
206/// Healthcheck configuration.
207#[serde_as]
208#[configurable_component]
209#[derive(Clone, Debug)]
210#[serde(default)]
211pub struct SinkHealthcheckOptions {
212    /// Whether or not to check the health of the sink when Vector starts up.
213    pub enabled: bool,
214
215    /// Timeout duration for healthcheck in seconds.
216    #[serde_as(as = "serde_with::DurationSecondsWithFrac<f64>")]
217    #[serde(
218        default = "default_healthcheck_timeout",
219        skip_serializing_if = "is_default_healthcheck_timeout"
220    )]
221    pub timeout: Duration,
222
223    /// The full URI to make HTTP healthcheck requests to.
224    ///
225    /// This must be a valid URI, which requires at least the scheme and host. All other
226    /// components -- port, path, etc -- are allowed as well.
227    #[configurable(validation(format = "uri"))]
228    pub uri: Option<UriSerde>,
229}
230
231const fn default_healthcheck_timeout() -> Duration {
232    Duration::from_secs(10)
233}
234
235fn is_default_healthcheck_timeout(timeout: &Duration) -> bool {
236    timeout == &default_healthcheck_timeout()
237}
238
239impl Default for SinkHealthcheckOptions {
240    fn default() -> Self {
241        Self {
242            enabled: true,
243            uri: None,
244            timeout: default_healthcheck_timeout(),
245        }
246    }
247}
248
249impl From<bool> for SinkHealthcheckOptions {
250    fn from(enabled: bool) -> Self {
251        Self {
252            enabled,
253            ..Default::default()
254        }
255    }
256}
257
258impl From<UriSerde> for SinkHealthcheckOptions {
259    fn from(uri: UriSerde) -> Self {
260        Self {
261            uri: Some(uri),
262            ..Default::default()
263        }
264    }
265}
266
267/// Generalized interface for describing and building sink components.
268#[async_trait]
269#[typetag::serde(tag = "type")]
270pub trait SinkConfig: DynClone + NamedComponent + core::fmt::Debug + Send + Sync {
271    /// Builds the sink with the given context.
272    ///
273    /// Migrated sinks route through the validated lifecycle: this default validates
274    /// the structure and builds from the retained state. Legacy sinks override this
275    /// with their own build logic.
276    ///
277    /// # Errors
278    ///
279    /// If an error occurs while building the sink, an error variant explaining the issue is
280    /// returned.
281    async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
282        match self.as_dyn_validated() {
283            Some(dyn_sink) => {
284                let validated = dyn_sink.validate_dyn()?;
285                dyn_sink.build_dyn(&*validated, cx).await
286            }
287            None => Err("sink does not implement a build method".into()),
288        }
289    }
290
291    /// Gets the input configuration for this sink.
292    fn input(&self) -> Input;
293
294    /// Gets the files to watch to trigger reload
295    fn files_to_watch(&self) -> Vec<&PathBuf> {
296        Vec::new()
297    }
298
299    /// Gets the list of resources, if any, used by this sink.
300    ///
301    /// Resources represent dependencies -- network ports, file descriptors, and so on -- that
302    /// cannot be shared between components at runtime. This ensures that components can not be
303    /// configured in a way that would deadlock the spawning of a topology, and as well, allows
304    /// Vector to determine the correct order for rebuilding a topology during configuration reload
305    /// when resources must first be reclaimed before being reassigned, and so on.
306    fn resources(&self) -> Vec<Resource> {
307        Vec::new()
308    }
309
310    /// Returns this sink's template-confinement config, if it supports confinement.
311    ///
312    /// The topology uses this to own the `vector_security_confinement_disabled`
313    /// gauge for the sink's lifetime. `None` (the default) means the sink does
314    /// not participate in template confinement and emits no gauge.
315    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
316        None
317    }
318
319    /// Gets the acknowledgements configuration for this sink.
320    fn acknowledgements(&self) -> &AcknowledgementsConfig;
321
322    /// Returns this sink as a `DynValidatedSink` if it has been migrated to the
323    /// validated lifecycle, or `None` for legacy sinks.
324    ///
325    /// Migrated sinks implement `ValidatedSink` (working with their concrete validated
326    /// type) and override this to return `Some(self)`. The framework then routes
327    /// validation and building through the dynamic boundary automatically.
328    fn as_dyn_validated(&self) -> Option<&dyn DynValidatedSink> {
329        None
330    }
331}
332
333dyn_clone::clone_trait_object!(SinkConfig);
334
335#[derive(Clone, Debug)]
336pub struct SinkContext {
337    pub healthcheck: SinkHealthcheckOptions,
338    pub globals: GlobalOptions,
339    pub enrichment_tables: vector_lib::enrichment::TableRegistry,
340    pub metrics_storage: MetricsStorage,
341    pub proxy: ProxyConfig,
342    pub schema: schema::Options,
343    pub app_name: String,
344    pub app_name_slug: String,
345
346    /// Extra context data provided by the running app and shared across all components. This can be
347    /// used to pass shared settings or other data from outside the components.
348    pub extra_context: ExtraContext,
349}
350
351impl Default for SinkContext {
352    fn default() -> Self {
353        Self {
354            healthcheck: Default::default(),
355            globals: Default::default(),
356            enrichment_tables: Default::default(),
357            metrics_storage: Default::default(),
358            proxy: Default::default(),
359            schema: Default::default(),
360            app_name: crate::get_app_name().to_string(),
361            app_name_slug: crate::get_slugified_app_name(),
362            extra_context: Default::default(),
363        }
364    }
365}
366
367impl SinkContext {
368    pub const fn globals(&self) -> &GlobalOptions {
369        &self.globals
370    }
371
372    pub const fn proxy(&self) -> &ProxyConfig {
373        &self.proxy
374    }
375}
vector/config/sink.rs

vector/config/
sink.rs
