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

vector/sinks/splunk_hec/logs/
config.rs
