vector/sinks/elasticsearch/
mod.rs

1mod common;
2mod config;
3pub mod encoder;
4pub mod health;
5pub mod request_builder;
6pub mod retry;
7pub mod service;
8pub mod sink;
9
10#[cfg(test)]
11mod tests;
12
13#[cfg(test)]
14#[cfg(feature = "es-integration-tests")]
15mod integration_tests;
16
17use std::{convert::TryFrom, fmt};
18
19pub use common::*;
20pub use config::*;
21pub use encoder::ElasticsearchEncoder;
22use http::{Request, uri::InvalidUri};
23use snafu::Snafu;
24use vector_lib::{
25    NamedInternalEvent, configurable::configurable_component, internal_event::InternalEvent,
26    sensitive_string::SensitiveString,
27};
28
29use crate::{
30    event::{EventRef, LogEvent},
31    internal_events::TemplateRenderingError,
32    template::{Template, TemplateParseError},
33};
34
35/// Elasticsearch Authentication strategies.
36#[configurable_component]
37#[derive(Clone, Debug)]
38#[serde(deny_unknown_fields, rename_all = "snake_case", tag = "strategy")]
39#[configurable(metadata(
40    docs::enum_tag_description = "The authentication strategy to use.\n\nAmazon OpenSearch Serverless requires this option to be set to `aws`."
41))]
42pub enum ElasticsearchAuthConfig {
43    /// HTTP Basic Authentication.
44    Basic {
45        /// Basic authentication username.
46        #[configurable(metadata(docs::examples = "${ELASTICSEARCH_USERNAME}"))]
47        #[configurable(metadata(docs::examples = "username"))]
48        user: String,
49
50        /// Basic authentication password.
51        #[configurable(metadata(docs::examples = "${ELASTICSEARCH_PASSWORD}"))]
52        #[configurable(metadata(docs::examples = "password"))]
53        password: SensitiveString,
54    },
55
56    #[cfg(feature = "aws-core")]
57    /// Amazon OpenSearch Service-specific authentication.
58    Aws(crate::aws::AwsAuthentication),
59}
60
61/// Elasticsearch Indexing mode.
62#[configurable_component]
63#[derive(Clone, Debug, Eq, PartialEq)]
64#[serde(deny_unknown_fields, rename_all = "snake_case")]
65#[derive(Default)]
66pub enum ElasticsearchMode {
67    /// Ingests documents in bulk, using the bulk API `index` action.
68    #[serde(alias = "normal")]
69    #[default]
70    Bulk,
71
72    /// Ingests documents in bulk, using the bulk API `create` action.
73    ///
74    /// Elasticsearch Data Streams only support the `create` action.
75    ///
76    /// If the mode is set to `data_stream` and a `timestamp` field is present in a message,
77    /// Vector renames this field to the expected `@timestamp` to comply with the Elastic Common Schema.
78    DataStream,
79}
80
81/// Bulk API actions.
82#[configurable_component]
83#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
84#[serde(deny_unknown_fields, rename_all = "snake_case")]
85pub enum BulkAction {
86    /// The `index` action.
87    Index,
88
89    /// The `create` action.
90    Create,
91
92    /// The `update` action.
93    Update,
94}
95
96#[allow(clippy::trivially_copy_pass_by_ref)]
97impl BulkAction {
98    pub const fn as_str(&self) -> &'static str {
99        match self {
100            BulkAction::Index => "index",
101            BulkAction::Create => "create",
102            BulkAction::Update => "update",
103        }
104    }
105
106    pub const fn as_json_pointer(&self) -> &'static str {
107        match self {
108            BulkAction::Index => "/index",
109            BulkAction::Create => "/create",
110            BulkAction::Update => "/update",
111        }
112    }
113}
114
115impl TryFrom<&str> for BulkAction {
116    type Error = String;
117
118    fn try_from(input: &str) -> Result<Self, Self::Error> {
119        match input {
120            "index" => Ok(BulkAction::Index),
121            "create" => Ok(BulkAction::Create),
122            "update" => Ok(BulkAction::Update),
123            _ => Err(format!("Invalid bulk action: {input}")),
124        }
125    }
126}
127
128/// Elasticsearch version types.
129#[configurable_component]
130#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
131#[serde(deny_unknown_fields, rename_all = "snake_case")]
132pub enum VersionType {
133    /// The `internal` type.
134    Internal,
135
136    /// The `external` or `external_gt` type.
137    External,
138
139    /// The `external_gte` type.
140    ExternalGte,
141}
142
143#[allow(clippy::trivially_copy_pass_by_ref)]
144impl VersionType {
145    pub const fn as_str(&self) -> &'static str {
146        match self {
147            Self::Internal => "internal",
148            Self::External => "external",
149            Self::ExternalGte => "external_gte",
150        }
151    }
152}
153
154impl TryFrom<&str> for VersionType {
155    type Error = String;
156
157    fn try_from(input: &str) -> Result<Self, Self::Error> {
158        match input {
159            "internal" => Ok(VersionType::Internal),
160            "external" | "external_gt" => Ok(VersionType::External),
161            "external_gte" => Ok(VersionType::ExternalGte),
162            _ => Err(format!("Invalid versioning mode: {input}")),
163        }
164    }
165}
166
167impl_generate_config_from_default!(ElasticsearchConfig);
168
169#[derive(Debug, Clone)]
170pub enum ElasticsearchCommonMode {
171    Bulk {
172        index: Template,
173        template_fallback_index: Option<String>,
174        action: Template,
175        version: Option<Template>,
176        version_type: VersionType,
177    },
178    DataStream(DataStreamConfig),
179}
180
181#[derive(NamedInternalEvent)]
182struct VersionValueParseError<'a> {
183    value: &'a str,
184}
185
186impl InternalEvent for VersionValueParseError<'_> {
187    fn emit(self) {
188        warn!("{self}")
189    }
190}
191
192impl fmt::Display for VersionValueParseError<'_> {
193    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
194        write!(f, "Cannot parse version \"{}\" as integer", self.value)
195    }
196}
197
198impl ElasticsearchCommonMode {
199    fn index(&self, log: &LogEvent) -> Option<String> {
200        match self {
201            Self::Bulk {
202                index,
203                template_fallback_index,
204                ..
205            } => index
206                .render_string(log)
207                .or_else(|error| {
208                    // Confined errors are intentional security drops — never fall back
209                    // to the default index, as that would silently accept an attack event.
210                    if matches!(
211                        error,
212                        crate::template::TemplateRenderingError::Confined { .. }
213                    ) {
214                        emit!(TemplateRenderingError {
215                            error,
216                            field: Some("index"),
217                            drop_event: true,
218                        });
219                        return Err(());
220                    }
221                    if let Some(fallback) = template_fallback_index {
222                        emit!(TemplateRenderingError {
223                            error,
224                            field: Some("index"),
225                            drop_event: false,
226                        });
227                        Ok(fallback.clone())
228                    } else {
229                        emit!(TemplateRenderingError {
230                            error,
231                            field: Some("index"),
232                            drop_event: true,
233                        });
234                        Err(())
235                    }
236                })
237                .ok(),
238            Self::DataStream(ds) => ds.index(log),
239        }
240    }
241
242    fn bulk_action<'a>(&self, event: impl Into<EventRef<'a>>) -> Option<BulkAction> {
243        match self {
244            ElasticsearchCommonMode::Bulk {
245                action: bulk_action_template,
246                ..
247            } => bulk_action_template
248                .render_string(event)
249                .map_err(|error| {
250                    emit!(TemplateRenderingError {
251                        error,
252                        field: Some("bulk_action"),
253                        drop_event: true,
254                    });
255                })
256                .ok()
257                .and_then(|value| BulkAction::try_from(value.as_str()).ok()),
258            // avoid the interpolation
259            ElasticsearchCommonMode::DataStream(_) => Some(BulkAction::Create),
260        }
261    }
262
263    fn version<'a>(&self, event: impl Into<EventRef<'a>>) -> Option<u64> {
264        match self {
265            ElasticsearchCommonMode::Bulk {
266                version: Some(version),
267                ..
268            } => version
269                .render_string(event)
270                .map_err(|error| {
271                    emit!(TemplateRenderingError {
272                        error,
273                        field: Some("version"),
274                        drop_event: true,
275                    });
276                })
277                .ok()
278                .as_ref()
279                .and_then(|value| {
280                    value
281                        .parse()
282                        .map_err(|_| emit!(VersionValueParseError { value }))
283                        .ok()
284                }),
285            _ => None,
286        }
287    }
288
289    const fn version_type(&self) -> Option<VersionType> {
290        match self {
291            ElasticsearchCommonMode::Bulk { version_type, .. } => Some(*version_type),
292            _ => Some(VersionType::Internal),
293        }
294    }
295
296    const fn as_data_stream_config(&self) -> Option<&DataStreamConfig> {
297        match self {
298            Self::DataStream(value) => Some(value),
299            _ => None,
300        }
301    }
302}
303
304/// Configuration for Elasticsearch API version.
305#[configurable_component]
306#[derive(Clone, Debug, Eq, PartialEq)]
307#[cfg_attr(feature = "proptest", derive(proptest_derive::Arbitrary))]
308#[serde(deny_unknown_fields, rename_all = "snake_case")]
309#[derive(Default)]
310pub enum ElasticsearchApiVersion {
311    /// Auto-detect the API version.
312    ///
313    /// If the [cluster state version endpoint][es_version] isn't reachable, a warning is logged to
314    /// stdout, and the version is assumed to be V6 if the `suppress_type_name` option is set to
315    /// `true`. Otherwise, the version is assumed to be V8. In the future, the sink instead
316    /// returns an error during configuration parsing, since a wrongly assumed version could lead to
317    /// incorrect API calls.
318    ///
319    /// [es_version]: https://www.elastic.co/guide/en/elasticsearch/reference/current/cluster-state.html#cluster-state-api-path-params
320    #[default]
321    Auto,
322    /// Use the Elasticsearch 6.x API.
323    V6,
324    /// Use the Elasticsearch 7.x API.
325    V7,
326    /// Use the Elasticsearch 8.x API.
327    V8,
328}
329
330#[derive(Debug, Snafu)]
331#[snafu(visibility(pub))]
332pub enum ParseError {
333    #[snafu(display("Invalid host {:?}: {:?}", host, source))]
334    InvalidHost { host: String, source: InvalidUri },
335    #[snafu(display("Host {:?} must include hostname", host))]
336    HostMustIncludeHostname { host: String },
337    #[snafu(display("Index template parse error: {}", source))]
338    IndexTemplate { source: TemplateParseError },
339    #[snafu(display("Batch action template parse error: {}", source))]
340    BatchActionTemplate { source: TemplateParseError },
341    #[cfg(feature = "aws-core")]
342    #[snafu(display("aws.region required when AWS authentication is in use"))]
343    RegionRequired,
344    #[snafu(display("Endpoints option must be specified"))]
345    EndpointRequired,
346    #[snafu(display(
347        "`endpoint` and `endpoints` options are mutually exclusive. Please use `endpoints` option."
348    ))]
349    EndpointsExclusive,
350    #[snafu(display("Tried to use external versioning without specifying the version itself"))]
351    ExternalVersioningWithoutVersion,
352    #[snafu(display("Cannot use external versioning without specifying a document ID"))]
353    ExternalVersioningWithoutDocumentID,
354    #[snafu(display("Your version field will be ignored because you use internal versioning"))]
355    ExternalVersionIgnoredWithInternalVersioning,
356    #[snafu(display("Amazon OpenSearch Serverless requires `api_version` value to be `auto`"))]
357    ServerlessElasticsearchApiVersionMustBeAuto,
358    #[snafu(display("Amazon OpenSearch Serverless requires `auth.strategy` value to be `aws`"))]
359    OpenSearchServerlessRequiresAwsAuth,
360}
vector/sinks/elasticsearch/mod.rs

vector/sinks/elasticsearch/
mod.rs
