vector/sinks/splunk_hec/logs/
config.rs

1use std::sync::Arc;
2
3use vector_lib::{
4    codecs::TextSerializerConfig,
5    lookup::lookup_v2::{ConfigValuePath, OptionalTargetPath},
6    sensitive_string::SensitiveString,
7};
8
9use super::{encoder::HecLogsEncoder, request_builder::HecLogsRequestBuilder, sink::HecLogsSink};
10use crate::{
11    config::{DynValidatedSink, ValidatedSink},
12    http::HttpClient,
13    sinks::{
14        prelude::*,
15        splunk_hec::common::{
16            EndpointTarget, SplunkHecDefaultBatchSettings,
17            acknowledgements::HecClientAcknowledgementsConfig,
18            build_healthcheck, build_http_batch_service, create_client,
19            service::{HecService, HttpRequestBuilder},
20        },
21        util::{HttpEndpoint, http::HttpRetryLogic},
22    },
23    template::ConfinementConfig,
24};
25
26/// Configuration for the `splunk_hec_logs` sink.
27#[configurable_component(sink(
28    "splunk_hec_logs",
29    "Deliver log data to Splunk's HTTP Event Collector."
30))]
31#[derive(Clone, Debug)]
32#[serde(deny_unknown_fields)]
33pub struct HecLogsSinkConfig {
34    /// Default Splunk HEC token.
35    ///
36    /// If an event has a token set in its secrets (`splunk_hec_token`), it prevails over the one set here.
37    #[serde(alias = "token")]
38    pub default_token: SensitiveString,
39
40    /// The base URL of the Splunk instance.
41    ///
42    /// The scheme (`http` or `https`) must be specified. No path should be included since the paths defined
43    /// by the [`Splunk`][splunk] API are used.
44    ///
45    /// [splunk]: https://docs.splunk.com/Documentation/Splunk/8.0.0/Data/HECRESTendpoints
46    #[configurable(metadata(
47        docs::examples = "https://http-inputs-hec.splunkcloud.com",
48        docs::examples = "https://hec.splunk.com:8088",
49        docs::examples = "http://example.com"
50    ))]
51    #[configurable(validation(format = "uri"))]
52    pub endpoint: HttpEndpoint,
53
54    /// Overrides the name of the log field used to retrieve the hostname to send to Splunk HEC.
55    ///
56    /// By default, the [global `log_schema.host_key` option][global_host_key] is used if log
57    /// events are Legacy namespaced, or the semantic meaning of "host" is used, if defined.
58    ///
59    /// [global_host_key]: https://vector.dev/docs/reference/configuration/global-options/#log_schema.host_key
60    // NOTE: The `OptionalTargetPath` is wrapped in an `Option` in order to distinguish between a true
61    //       `None` type and an empty string. This is necessary because `OptionalTargetPath` deserializes an
62    //       empty string to a `None` path internally.
63    pub host_key: Option<OptionalTargetPath>,
64
65    /// Fields to be [added to Splunk index][splunk_field_index_docs].
66    ///
67    /// [splunk_field_index_docs]: https://docs.splunk.com/Documentation/Splunk/8.0.0/Data/IFXandHEC
68    #[serde(default)]
69    #[configurable(metadata(docs::examples = "field1", docs::examples = "field2"))]
70    pub indexed_fields: Vec<ConfigValuePath>,
71
72    /// The name of the index to send events to.
73    ///
74    /// If not specified, the default index defined within Splunk is used.
75    #[configurable(metadata(
76        docs::examples = "index-{{ host }}",
77        docs::examples = "custom_index"
78    ))]
79    pub index: Option<Template>,
80
81    /// The sourcetype of events sent to this sink.
82    ///
83    /// If unset, Splunk defaults to `httpevent`.
84    #[configurable(metadata(
85        docs::examples = "sourcetype-{{ sourcetype }}",
86        docs::examples = "_json",
87    ))]
88    pub sourcetype: Option<Template>,
89
90    /// The source of events sent to this sink.
91    ///
92    /// This is typically the filename the logs originated from.
93    ///
94    /// If unset, the Splunk collector sets it.
95    #[configurable(metadata(
96        docs::examples = "source-{{ file }}",
97        docs::examples = "/var/log/syslog",
98        docs::examples = "UDP:514"
99    ))]
100    pub source: Option<Template>,
101
102    #[configurable(derived)]
103    pub encoding: EncodingConfig,
104
105    #[configurable(derived)]
106    #[serde(default)]
107    pub compression: Compression,
108
109    #[configurable(derived)]
110    #[serde(default)]
111    pub batch: BatchConfig<SplunkHecDefaultBatchSettings>,
112
113    #[configurable(derived)]
114    #[serde(default)]
115    pub request: TowerRequestConfig,
116
117    #[configurable(derived)]
118    pub tls: Option<TlsConfig>,
119
120    #[configurable(derived)]
121    #[serde(default)]
122    pub acknowledgements: HecClientAcknowledgementsConfig,
123
124    // This settings is relevant only for the `humio_logs` sink and should be left as `None`
125    // everywhere else.
126    #[serde(skip)]
127    pub timestamp_nanos_key: Option<String>,
128
129    /// Overrides the name of the log field used to retrieve the timestamp to send to Splunk HEC.
130    /// When set to `“”`, a timestamp is not set in the events sent to Splunk HEC.
131    ///
132    /// By default, either the [global `log_schema.timestamp_key` option][global_timestamp_key] is used
133    /// if log events are Legacy namespaced, or the semantic meaning of "timestamp" is used, if defined.
134    ///
135    /// [global_timestamp_key]: https://vector.dev/docs/reference/configuration/global-options/#log_schema.timestamp_key
136    #[configurable(metadata(docs::examples = "timestamp", docs::examples = ""))]
137    // NOTE: The `OptionalTargetPath` is wrapped in an `Option` in order to distinguish between a true
138    //       `None` type and an empty string. This is necessary because `OptionalTargetPath` deserializes an
139    //       empty string to a `None` path internally.
140    pub timestamp_key: Option<OptionalTargetPath>,
141
142    /// Passes the `auto_extract_timestamp` option to Splunk.
143    ///
144    /// This option is only relevant to Splunk v8.x and above, and is only applied when
145    /// `endpoint_target` is set to `event`.
146    ///
147    /// Setting this to `true` causes Splunk to extract the timestamp from the message text
148    /// rather than use the timestamp embedded in the event. The timestamp must be in the format
149    /// `yyyy-mm-dd hh:mm:ss`.
150    #[serde(default)]
151    pub auto_extract_timestamp: Option<bool>,
152
153    #[configurable(derived)]
154    #[serde(default = "default_endpoint_target")]
155    pub endpoint_target: EndpointTarget,
156
157    #[configurable(derived)]
158    #[serde(flatten)]
159    pub confinement: ConfinementConfig,
160}
161
162const fn default_endpoint_target() -> EndpointTarget {
163    EndpointTarget::Event
164}
165
166impl GenerateConfig for HecLogsSinkConfig {
167    fn generate_config() -> serde_json::Value {
168        serde_json::to_value(Self {
169            default_token: "${VECTOR_SPLUNK_HEC_TOKEN}".to_owned().into(),
170            endpoint: HttpEndpoint::parse("http://example.com").unwrap(),
171            host_key: None,
172            indexed_fields: vec![],
173            index: None,
174            sourcetype: None,
175            source: None,
176            encoding: TextSerializerConfig::default().into(),
177            compression: Compression::default(),
178            batch: BatchConfig::default(),
179            request: TowerRequestConfig::default(),
180            tls: None,
181            acknowledgements: Default::default(),
182            timestamp_nanos_key: None,
183            timestamp_key: None,
184            auto_extract_timestamp: None,
185            endpoint_target: EndpointTarget::Event,
186            confinement: ConfinementConfig::default(),
187        })
188        .unwrap()
189    }
190}
191
192impl HecLogsSinkConfig {
193    /// Pure structural validation. `component_name` is threaded into the
194    /// per-template security warnings emitted from `Template::confine`, so
195    /// wrapping sinks (Humio) see their own type in logs rather than the
196    /// delegated `splunk_hec_logs`.
197    pub(crate) fn validate_with_component_name(
198        &self,
199        component_name: &'static str,
200    ) -> crate::Result<ValidatedHecLogsSink> {
201        if self.auto_extract_timestamp.is_some() && self.endpoint_target == EndpointTarget::Raw {
202            return Err("`auto_extract_timestamp` cannot be set for the `raw` endpoint.".into());
203        }
204
205        // The endpoint is validated at config load as an absolute http(s) URL.
206
207        let index = self
208            .index
209            .clone()
210            .map(|t| t.confine(&self.confinement, component_name, "index"))
211            .transpose()?;
212        let source = self
213            .source
214            .clone()
215            .map(|t| t.confine(&self.confinement, component_name, "source"))
216            .transpose()?;
217        let sourcetype = self
218            .sourcetype
219            .clone()
220            .map(|t| t.confine(&self.confinement, component_name, "sourcetype"))
221            .transpose()?;
222
223        let batch_settings = self.batch.into_batcher_settings()?;
224
225        Ok(ValidatedHecLogsSink {
226            index,
227            source,
228            sourcetype,
229            batch_settings,
230        })
231    }
232
233    /// Build the sink from validated state. Only environment-dependent work
234    /// (client creation, healthcheck) happens here; the validated state is
235    /// consumed without recomputing any pure validation.
236    pub(crate) fn build_from_validated(
237        &self,
238        cx: SinkContext,
239        validated: &ValidatedHecLogsSink,
240    ) -> crate::Result<(VectorSink, Healthcheck)> {
241        let client = create_client(self.tls.as_ref(), cx.proxy())?;
242        let healthcheck = build_healthcheck(
243            self.endpoint.clone().into(),
244            self.default_token.inner().to_owned(),
245            client.clone(),
246        )
247        .boxed();
248        let sink = self.build_processor(client, cx, validated)?;
249
250        Ok((sink, healthcheck))
251    }
252}
253
254#[async_trait::async_trait]
255#[typetag::serde(name = "splunk_hec_logs")]
256impl SinkConfig for HecLogsSinkConfig {
257    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
258        Some(&self.confinement)
259    }
260
261    fn input(&self) -> Input {
262        Input::new(self.encoding.config().input_type() & DataType::Log)
263    }
264
265    fn acknowledgements(&self) -> &AcknowledgementsConfig {
266        &self.acknowledgements.inner
267    }
268
269    fn as_dyn_validated(&self) -> Option<&dyn DynValidatedSink> {
270        Some(self)
271    }
272}
273
274#[derive(Clone, Debug)]
275pub struct ValidatedHecLogsSink {
276    index: Option<ConfinedTemplate>,
277    source: Option<ConfinedTemplate>,
278    sourcetype: Option<ConfinedTemplate>,
279    batch_settings: BatcherSettings,
280}
281
282#[async_trait::async_trait]
283impl ValidatedSink for HecLogsSinkConfig {
284    type Validated = ValidatedHecLogsSink;
285
286    fn validate(&self) -> crate::Result<ValidatedHecLogsSink> {
287        self.validate_with_component_name(Self::NAME)
288    }
289
290    async fn build(
291        &self,
292        validated: &ValidatedHecLogsSink,
293        cx: SinkContext,
294    ) -> crate::Result<(VectorSink, Healthcheck)> {
295        self.build_from_validated(cx, validated)
296    }
297}
298
299impl HecLogsSinkConfig {
300    pub fn build_processor(
301        &self,
302        client: HttpClient,
303        _: SinkContext,
304        validated: &ValidatedHecLogsSink,
305    ) -> crate::Result<VectorSink> {
306        let ack_client = if self.acknowledgements.indexer_acknowledgements_enabled {
307            Some(client.clone())
308        } else {
309            None
310        };
311
312        let transformer = self.encoding.transformer();
313        let serializer = self.encoding.build()?;
314        let encoder = HecLogsEncoder {
315            transformer,
316            encoder: Encoder::<()>::new(serializer),
317            auto_extract_timestamp: self.auto_extract_timestamp.unwrap_or_default(),
318        };
319        let request_builder = HecLogsRequestBuilder {
320            encoder,
321            compression: self.compression,
322        };
323
324        let request_settings = self.request.into_settings();
325        let http_request_builder = Arc::new(HttpRequestBuilder::new(
326            self.endpoint.clone().into(),
327            self.endpoint_target,
328            self.default_token.inner().to_owned(),
329            self.compression,
330        ));
331        let http_service = ServiceBuilder::new()
332            .settings(request_settings, HttpRetryLogic::default())
333            .service(build_http_batch_service(
334                client,
335                Arc::clone(&http_request_builder),
336                self.endpoint_target,
337                self.auto_extract_timestamp.unwrap_or_default(),
338            ));
339
340        let service = HecService::new(
341            http_service,
342            ack_client,
343            http_request_builder,
344            self.acknowledgements.clone(),
345        );
346
347        let sink = HecLogsSink {
348            service,
349            request_builder,
350            batch_settings: validated.batch_settings,
351            sourcetype: validated.sourcetype.clone(),
352            source: validated.source.clone(),
353            index: validated.index.clone(),
354            indexed_fields: self
355                .indexed_fields
356                .iter()
357                .map(|config_path| config_path.0.clone())
358                .collect(),
359            host_key: self.host_key.clone(),
360            timestamp_nanos_key: self.timestamp_nanos_key.clone(),
361            timestamp_key: self.timestamp_key.clone(),
362            endpoint_target: self.endpoint_target,
363            auto_extract_timestamp: self.auto_extract_timestamp.unwrap_or_default(),
364        };
365
366        Ok(VectorSink::from_event_streamsink(sink))
367    }
368}
369
370#[cfg(test)]
371mod tests {
372    use vector_lib::{
373        codecs::{JsonSerializerConfig, MetricTagValues, encoding::format::JsonSerializerOptions},
374        config::LogNamespace,
375    };
376
377    use super::*;
378    use crate::components::validation::prelude::*;
379    use crate::template::{ConfinementConfig, Template};
380
381    #[test]
382    fn generate_config() {
383        crate::test_util::test_generate_config::<HecLogsSinkConfig>();
384    }
385
386    #[test]
387    fn validate_produces_usable_state() {
388        use crate::config::ValidatedSink;
389
390        let config = HecLogsSinkConfig {
391            default_token: "token".to_string().into(),
392            endpoint: HttpEndpoint::parse("http://localhost:8088").unwrap(),
393            host_key: None,
394            indexed_fields: vec![],
395            index: Some("custom_index".try_into().unwrap()),
396            sourcetype: None,
397            source: None,
398            encoding: JsonSerializerConfig::default().into(),
399            compression: Compression::default(),
400            batch: Default::default(),
401            request: Default::default(),
402            tls: None,
403            acknowledgements: Default::default(),
404            timestamp_nanos_key: None,
405            timestamp_key: None,
406            auto_extract_timestamp: None,
407            endpoint_target: EndpointTarget::Event,
408            confinement: Default::default(),
409        };
410
411        let validated = config.validate().expect("validation should succeed");
412        assert_eq!(
413            validated.index.as_ref().unwrap().to_string(),
414            "custom_index"
415        );
416        assert!(validated.source.is_none());
417        assert!(validated.sourcetype.is_none());
418    }
419
420    #[test]
421    fn confinement_rejects_unconfined_index() {
422        let template = Template::try_from("{{ index }}").unwrap();
423        let config = ConfinementConfig::default();
424        let result = template.confine(&config, "splunk_hec_logs", "index");
425        assert!(result.is_err());
426    }
427
428    #[test]
429    fn confinement_opt_out_allows_unconfined_index() {
430        let template = Template::try_from("{{ index }}").unwrap();
431        let config = ConfinementConfig {
432            dangerously_allow_unconfined_template_resolution: true,
433        };
434        let result = template.confine(&config, "splunk_hec_logs", "index");
435        assert!(result.is_ok());
436    }
437
438    #[test]
439    fn confinement_allows_prefixed_index() {
440        let template = Template::try_from("events-{{ env }}").unwrap();
441        let config = ConfinementConfig::default();
442        let result = template.confine(&config, "splunk_hec_logs", "index");
443        assert!(result.is_ok());
444    }
445
446    impl ValidatableComponent for HecLogsSinkConfig {
447        fn validation_configuration() -> ValidationConfiguration {
448            let endpoint = HttpEndpoint::parse("http://127.0.0.1:9001").unwrap();
449
450            let mut batch = BatchConfig::default();
451            batch.max_events = Some(1);
452
453            let config = Self {
454                endpoint: endpoint.clone(),
455                default_token: "i_am_an_island".to_string().into(),
456                host_key: None,
457                indexed_fields: vec![],
458                index: None,
459                sourcetype: None,
460                source: None,
461                encoding: EncodingConfig::new(
462                    JsonSerializerConfig::new(
463                        MetricTagValues::Full,
464                        JsonSerializerOptions::default(),
465                    )
466                    .into(),
467                    Transformer::default(),
468                ),
469                compression: Compression::default(),
470                batch,
471                request: TowerRequestConfig {
472                    timeout_secs: 2,
473                    retry_attempts: 0,
474                    ..Default::default()
475                },
476                tls: None,
477                acknowledgements: HecClientAcknowledgementsConfig {
478                    indexer_acknowledgements_enabled: false,
479                    ..Default::default()
480                },
481                timestamp_nanos_key: None,
482                timestamp_key: None,
483                auto_extract_timestamp: None,
484                endpoint_target: EndpointTarget::Raw,
485                confinement: ConfinementConfig::default(),
486            };
487
488            let endpoint = endpoint
489                .append_path("services/collector/raw")
490                .unwrap()
491                .into_uri();
492
493            let external_resource = ExternalResource::new(
494                ResourceDirection::Push,
495                HttpResourceConfig::from_parts(endpoint, None),
496                config.encoding.clone(),
497            );
498
499            ValidationConfiguration::from_sink(
500                Self::NAME,
501                LogNamespace::Legacy,
502                vec![ComponentTestCaseConfig::from_sink(
503                    config,
504                    None,
505                    Some(external_resource),
506                )],
507            )
508        }
509    }
510
511    register_validatable_component!(HecLogsSinkConfig);
512}
vector/sinks/splunk_hec/logs/config.rs

vector/sinks/splunk_hec/logs/
config.rs
