vector/sinks/humio/
logs.rs

1use vector_lib::{
2    codecs::JsonSerializerConfig,
3    configurable::configurable_component,
4    lookup::lookup_v2::{ConfigValuePath, OptionalTargetPath},
5    sensitive_string::SensitiveString,
6};
7
8use super::config_host_key_target_path;
9use crate::{
10    codecs::EncodingConfig,
11    config::{
12        AcknowledgementsConfig, DataType, DynValidatedSink, GenerateConfig, Input, SinkConfig,
13        SinkContext, ValidatedSink,
14    },
15    sinks::{
16        Healthcheck, VectorSink,
17        splunk_hec::{
18            common::{
19                EndpointTarget, SplunkHecDefaultBatchSettings,
20                acknowledgements::HecClientAcknowledgementsConfig,
21                config_timestamp_key_target_path,
22            },
23            logs::config::{HecLogsSinkConfig, ValidatedHecLogsSink},
24        },
25        util::{BatchConfig, Compression, HttpEndpoint, TowerRequestConfig},
26    },
27    template::Template,
28    tls::TlsConfig,
29};
30
31pub(super) const HOST: &str = "https://cloud.humio.com";
32
33/// Configuration for the `humio_logs` sink.
34#[configurable_component(sink("humio_logs", "Deliver log event data to Humio."))]
35#[derive(Clone, Debug)]
36#[serde(deny_unknown_fields)]
37pub struct HumioLogsConfig {
38    /// The Humio ingestion token.
39    #[configurable(metadata(
40        docs::examples = "${HUMIO_TOKEN}",
41        docs::examples = "A94A8FE5CCB19BA61C4C08"
42    ))]
43    pub token: SensitiveString,
44
45    /// The base URL of the Humio instance.
46    ///
47    /// The scheme (`http` or `https`) must be specified. No path should be included since the paths defined
48    /// by the [`Splunk`][splunk] API are used.
49    ///
50    /// [splunk]: https://docs.splunk.com/Documentation/Splunk/8.0.0/Data/HECRESTendpoints
51    #[serde(alias = "host")]
52    #[serde(default = "default_endpoint")]
53    #[configurable(metadata(
54        docs::examples = "http://127.0.0.1",
55        docs::examples = "https://example.com",
56    ))]
57    pub endpoint: HttpEndpoint,
58
59    /// The source of events sent to this sink.
60    ///
61    /// Typically the filename the logs originated from. Maps to `@source` in Humio.
62    pub source: Option<Template>,
63
64    #[configurable(derived)]
65    pub encoding: EncodingConfig,
66
67    /// The type of events sent to this sink. Humio uses this as the name of the parser to use to ingest the data.
68    ///
69    /// If unset, Humio defaults it to none.
70    #[configurable(metadata(
71        docs::examples = "json",
72        docs::examples = "none",
73        docs::examples = "event_type-{{ event_type }}"
74    ))]
75    pub event_type: Option<Template>,
76
77    /// Overrides the name of the log field used to retrieve the hostname to send to Humio.
78    ///
79    /// By default, the [global `log_schema.host_key` option][global_host_key] is used if log
80    /// events are Legacy namespaced, or the semantic meaning of "host" is used, if defined.
81    ///
82    /// [global_host_key]: https://vector.dev/docs/reference/configuration/global-options/#log_schema.host_key
83    #[serde(default = "config_host_key_target_path")]
84    pub host_key: OptionalTargetPath,
85
86    /// Event fields to be added to Humio’s extra fields.
87    ///
88    /// Can be used to tag events by specifying fields starting with `#`.
89    ///
90    /// For more information, see [Humio’s Format of Data][humio_data_format].
91    ///
92    /// [humio_data_format]: https://docs.humio.com/integrations/data-shippers/hec/#format-of-data
93    #[serde(default)]
94    pub indexed_fields: Vec<ConfigValuePath>,
95
96    /// Optional name of the repository to ingest into.
97    ///
98    /// In public-facing APIs, this must (if present) be equal to the repository used to create the ingest token used for authentication.
99    ///
100    /// In private cluster setups, Humio can be configured to allow these to be different.
101    ///
102    /// For more information, see [Humio’s Format of Data][humio_data_format].
103    ///
104    /// [humio_data_format]: https://docs.humio.com/integrations/data-shippers/hec/#format-of-data
105    #[serde(default)]
106    #[configurable(metadata(
107        docs::examples = "index-{{ host }}",
108        docs::examples = "custom_index"
109    ))]
110    pub index: Option<Template>,
111
112    #[configurable(derived)]
113    #[serde(default)]
114    pub compression: Compression,
115
116    #[configurable(derived)]
117    #[serde(default)]
118    pub request: TowerRequestConfig,
119
120    #[configurable(derived)]
121    #[serde(default)]
122    pub batch: BatchConfig<SplunkHecDefaultBatchSettings>,
123
124    #[configurable(derived)]
125    pub tls: Option<TlsConfig>,
126
127    /// Overrides the name of the log field used to retrieve the nanosecond-enabled timestamp to send to Humio.
128    #[serde(default = "timestamp_nanos_key")]
129    pub timestamp_nanos_key: Option<String>,
130
131    #[configurable(derived)]
132    #[serde(
133        default,
134        deserialize_with = "crate::serde::bool_or_struct",
135        skip_serializing_if = "crate::serde::is_default"
136    )]
137    pub acknowledgements: AcknowledgementsConfig,
138
139    /// Overrides the name of the log field used to retrieve the timestamp to send to Humio.
140    /// When set to `“”`, a timestamp is not set in the events sent to Humio.
141    ///
142    /// By default, either the [global `log_schema.timestamp_key` option][global_timestamp_key] is used
143    /// if log events are Legacy namespaced, or the semantic meaning of "timestamp" is used, if defined.
144    ///
145    /// [global_timestamp_key]: https://vector.dev/docs/reference/configuration/global-options/#log_schema.timestamp_key
146    #[serde(default = "config_timestamp_key_target_path")]
147    pub timestamp_key: OptionalTargetPath,
148
149    #[serde(flatten)]
150    pub confinement: crate::template::ConfinementConfig,
151}
152
153fn default_endpoint() -> HttpEndpoint {
154    HttpEndpoint::parse(HOST).expect("static default endpoint should be a valid http(s) URL")
155}
156
157pub fn timestamp_nanos_key() -> Option<String> {
158    Some("@timestamp.nanos".to_string())
159}
160
161impl GenerateConfig for HumioLogsConfig {
162    fn generate_config() -> serde_json::Value {
163        serde_json::to_value(Self {
164            token: "${HUMIO_TOKEN}".to_owned().into(),
165            endpoint: default_endpoint(),
166            source: None,
167            encoding: JsonSerializerConfig::default().into(),
168            event_type: None,
169            indexed_fields: vec![],
170            index: None,
171            host_key: config_host_key_target_path(),
172            compression: Compression::default(),
173            request: TowerRequestConfig::default(),
174            batch: BatchConfig::default(),
175            tls: None,
176            timestamp_nanos_key: None,
177            acknowledgements: Default::default(),
178            timestamp_key: config_timestamp_key_target_path(),
179            confinement: Default::default(),
180        })
181        .unwrap()
182    }
183}
184
185#[async_trait::async_trait]
186#[typetag::serde(name = "humio_logs")]
187impl SinkConfig for HumioLogsConfig {
188    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
189        Some(&self.confinement)
190    }
191
192    fn input(&self) -> Input {
193        Input::new(self.encoding.config().input_type() & DataType::Log)
194    }
195
196    fn acknowledgements(&self) -> &AcknowledgementsConfig {
197        &self.acknowledgements
198    }
199
200    fn as_dyn_validated(&self) -> Option<&dyn DynValidatedSink> {
201        Some(self)
202    }
203}
204
205#[derive(Clone, Debug)]
206pub struct ValidatedHumioLogs {
207    pub(super) hec: ValidatedHecLogsSink,
208}
209
210#[async_trait::async_trait]
211impl ValidatedSink for HumioLogsConfig {
212    type Validated = ValidatedHumioLogs;
213
214    fn validate(&self) -> crate::Result<ValidatedHumioLogs> {
215        self.validate_with_component_name(Self::NAME)
216    }
217
218    async fn build(
219        &self,
220        validated: &ValidatedHumioLogs,
221        cx: SinkContext,
222    ) -> crate::Result<(VectorSink, Healthcheck)> {
223        self.build_from_validated(cx, &validated.hec)
224    }
225}
226
227impl HumioLogsConfig {
228    /// Pure structural validation. `component_name` is threaded through so
229    /// per-template security warnings carry the outer sink type — `humio_logs`
230    /// when this is the top-level sink, `humio_metrics` when
231    /// [`HumioMetricsConfig`] delegates here.
232    pub(super) fn validate_with_component_name(
233        &self,
234        component_name: &'static str,
235    ) -> crate::Result<ValidatedHumioLogs> {
236        Ok(ValidatedHumioLogs {
237            hec: self
238                .build_hec_config()
239                .validate_with_component_name(component_name)?,
240        })
241    }
242
243    /// Build the sink from validated state, delegating to the underlying
244    /// Splunk HEC logs sink without redoing any pure validation.
245    pub(super) fn build_from_validated(
246        &self,
247        cx: SinkContext,
248        validated: &ValidatedHecLogsSink,
249    ) -> crate::Result<(VectorSink, Healthcheck)> {
250        self.build_hec_config().build_from_validated(cx, validated)
251    }
252
253    fn build_hec_config(&self) -> HecLogsSinkConfig {
254        HecLogsSinkConfig {
255            default_token: self.token.clone(),
256            endpoint: self.endpoint.clone(),
257            host_key: Some(self.host_key.clone()),
258            indexed_fields: self.indexed_fields.clone(),
259            index: self.index.clone(),
260            sourcetype: self.event_type.clone(),
261            source: self.source.clone(),
262            timestamp_nanos_key: self.timestamp_nanos_key.clone(),
263            encoding: self.encoding.clone(),
264            compression: self.compression,
265            batch: self.batch,
266            request: self.request,
267            tls: self.tls.clone(),
268            acknowledgements: HecClientAcknowledgementsConfig {
269                indexer_acknowledgements_enabled: false,
270                ..Default::default()
271            },
272            timestamp_key: Some(config_timestamp_key_target_path()),
273            endpoint_target: EndpointTarget::Event,
274            auto_extract_timestamp: None,
275            confinement: self.confinement.clone(),
276        }
277    }
278}
279
280#[cfg(test)]
281mod tests {
282    use super::*;
283
284    #[test]
285    fn generate_config() {
286        crate::test_util::test_generate_config::<HumioLogsConfig>();
287    }
288
289    #[test]
290    fn validate_rejects_unconfined_template() {
291        use crate::config::ValidatedSink;
292
293        let config = HumioLogsConfig {
294            token: "token".to_string().into(),
295            endpoint: default_endpoint(),
296            source: None,
297            encoding: JsonSerializerConfig::default().into(),
298            event_type: None,
299            host_key: config_host_key_target_path(),
300            indexed_fields: vec![],
301            index: Some("{{ index }}".try_into().unwrap()),
302            compression: Compression::default(),
303            request: TowerRequestConfig::default(),
304            batch: BatchConfig::default(),
305            tls: None,
306            timestamp_nanos_key: None,
307            acknowledgements: Default::default(),
308            timestamp_key: config_timestamp_key_target_path(),
309            confinement: Default::default(),
310        };
311
312        let result = config.validate();
313        assert!(result.is_err());
314    }
315}
316
317#[cfg(test)]
318#[cfg(feature = "humio-integration-tests")]
319mod integration_tests {
320    use std::{collections::HashMap, convert::TryFrom};
321
322    use chrono::{TimeZone, Utc};
323    use futures::{future::ready, stream};
324    use indoc::indoc;
325    use serde::Deserialize;
326    use serde_json::{Value as JsonValue, json};
327    use tokio::time::Duration;
328
329    use vrl::event_path;
330
331    use super::*;
332    use crate::{
333        config::{SinkConfig, SinkContext, log_schema},
334        event::LogEvent,
335        sinks::util::Compression,
336        test_util::{
337            components::{HTTP_SINK_TAGS, run_and_assert_sink_compliance},
338            random_string,
339        },
340    };
341
342    fn humio_address() -> String {
343        std::env::var("HUMIO_ADDRESS").unwrap_or_else(|_| "http://localhost:8080".into())
344    }
345
346    #[tokio::test]
347    async fn humio_insert_message() {
348        wait_ready().await;
349
350        let cx = SinkContext::default();
351
352        let repo = create_repository().await;
353
354        let config = config(&repo.default_ingest_token);
355
356        let (sink, _) = SinkConfig::build(&config, cx).await.unwrap();
357
358        let message = random_string(100);
359        let host = "192.168.1.1".to_string();
360        let mut event = LogEvent::from(message.clone());
361        event.insert(log_schema().host_key_target_path().unwrap(), host.clone());
362
363        let ts = Utc.timestamp_nanos(Utc::now().timestamp_millis() * 1_000_000 + 132_456);
364        event.insert(log_schema().timestamp_key_target_path().unwrap(), ts);
365
366        run_and_assert_sink_compliance(sink, stream::once(ready(event)), &HTTP_SINK_TAGS).await;
367
368        let entry = find_entry(repo.name.as_str(), message.as_str()).await;
369
370        assert_eq!(
371            message,
372            entry
373                .fields
374                .get("message")
375                .expect("no message key")
376                .as_str()
377                .unwrap()
378        );
379        assert!(
380            entry.error.is_none(),
381            "Humio encountered an error parsing this message: {}",
382            entry
383                .error_msg
384                .unwrap_or_else(|| "no error message".to_string())
385        );
386        assert_eq!(Some(host), entry.host);
387        assert_eq!("132456", entry.timestamp_nanos);
388    }
389
390    #[tokio::test]
391    async fn humio_insert_source() {
392        wait_ready().await;
393
394        let cx = SinkContext::default();
395
396        let repo = create_repository().await;
397
398        let mut config = config(&repo.default_ingest_token);
399        config.source = Template::try_from("/var/log/syslog".to_string()).ok();
400
401        let (sink, _) = SinkConfig::build(&config, cx).await.unwrap();
402
403        let message = random_string(100);
404        let event = LogEvent::from(message.clone());
405        run_and_assert_sink_compliance(sink, stream::once(ready(event)), &HTTP_SINK_TAGS).await;
406
407        let entry = find_entry(repo.name.as_str(), message.as_str()).await;
408
409        assert_eq!(entry.source, Some("/var/log/syslog".to_owned()));
410        assert!(
411            entry.error.is_none(),
412            "Humio encountered an error parsing this message: {}",
413            entry
414                .error_msg
415                .unwrap_or_else(|| "no error message".to_string())
416        );
417    }
418
419    #[tokio::test]
420    async fn humio_type() {
421        wait_ready().await;
422
423        let repo = create_repository().await;
424
425        // sets type
426        {
427            let mut config = config(&repo.default_ingest_token);
428            config.event_type = Template::try_from("json".to_string()).ok();
429
430            let (sink, _) = SinkConfig::build(&config, SinkContext::default())
431                .await
432                .unwrap();
433
434            let message = random_string(100);
435            let mut event = LogEvent::from(message.clone());
436            // Humio expects to find an @timestamp field for JSON lines
437            // https://docs.humio.com/ingesting-data/parsers/built-in-parsers/#json
438            event.insert(event_path!("@timestamp"), Utc::now().to_rfc3339());
439
440            run_and_assert_sink_compliance(sink, stream::once(ready(event)), &HTTP_SINK_TAGS).await;
441
442            let entry = find_entry(repo.name.as_str(), message.as_str()).await;
443
444            assert_eq!(entry.humio_type, "json");
445            assert!(
446                entry.error.is_none(),
447                "Humio encountered an error parsing this message: {}",
448                entry
449                    .error_msg
450                    .unwrap_or_else(|| "no error message".to_string())
451            );
452        }
453
454        // defaults to none
455        {
456            let config = config(&repo.default_ingest_token);
457
458            let (sink, _) = SinkConfig::build(&config, SinkContext::default())
459                .await
460                .unwrap();
461
462            let message = random_string(100);
463            let event = LogEvent::from(message.clone());
464
465            run_and_assert_sink_compliance(sink, stream::once(ready(event)), &HTTP_SINK_TAGS).await;
466
467            let entry = find_entry(repo.name.as_str(), message.as_str()).await;
468
469            assert_eq!(entry.humio_type, "none");
470        }
471    }
472
473    /// create a new test config with the given ingest token
474    fn config(token: &str) -> super::HumioLogsConfig {
475        let mut batch = BatchConfig::default();
476        batch.max_events = Some(1);
477
478        HumioLogsConfig {
479            token: token.to_string().into(),
480            endpoint: HttpEndpoint::parse(&humio_address()).unwrap(),
481            source: None,
482            encoding: JsonSerializerConfig::default().into(),
483            event_type: None,
484            host_key: OptionalTargetPath {
485                path: log_schema().host_key_target_path().cloned(),
486            },
487            indexed_fields: vec![],
488            index: None,
489            compression: Compression::None,
490            request: TowerRequestConfig::default(),
491            batch,
492            tls: None,
493            timestamp_nanos_key: timestamp_nanos_key(),
494            acknowledgements: Default::default(),
495            timestamp_key: Default::default(),
496            confinement: Default::default(),
497        }
498    }
499
500    async fn wait_ready() {
501        crate::test_util::retry_until(
502            || async {
503                reqwest::get(format!("{}/api/v1/status", humio_address()))
504                    .await
505                    .map_err(|err| err.to_string())
506                    .and_then(|res| {
507                        if res.status().is_success() {
508                            Ok(())
509                        } else {
510                            Err("server not ready...".into())
511                        }
512                    })
513            },
514            Duration::from_secs(1),
515            Duration::from_secs(30),
516        )
517        .await;
518    }
519
520    /// create a new test humio repository to publish to
521    async fn create_repository() -> HumioRepository {
522        let client = reqwest::Client::builder().build().unwrap();
523
524        // https://docs.humio.com/api/graphql/
525        let graphql_url = format!("{}/graphql", humio_address());
526
527        let name = random_string(50);
528
529        let params = json!({
530        "query": format!(
531            indoc!{ r#"
532                mutation {{
533                  createRepository(name:"{}") {{
534                    repository {{
535                      name
536                      type
537                      ingestTokens {{
538                        name
539                        token
540                      }}
541                    }}
542                  }}
543                }}
544            "#},
545            name
546        ),
547        });
548
549        let res = client
550            .post(&graphql_url)
551            .json(&params)
552            .send()
553            .await
554            .unwrap();
555
556        let json: JsonValue = res.json().await.unwrap();
557        let repository = &json["data"]["createRepository"]["repository"];
558
559        let token = repository["ingestTokens"].as_array().unwrap()[0]["token"]
560            .as_str()
561            .unwrap()
562            .to_string();
563
564        HumioRepository {
565            name: repository["name"].as_str().unwrap().to_string(),
566            default_ingest_token: token,
567        }
568    }
569
570    /// fetch event from the repository that has a matching message value
571    async fn find_entry(repository_name: &str, message: &str) -> HumioLog {
572        let client = reqwest::Client::builder().build().unwrap();
573
574        // https://docs.humio.com/api/using-the-search-api-with-humio
575        let search_url = format!(
576            "{}/api/v1/repositories/{}/query",
577            humio_address(),
578            repository_name
579        );
580        let search_query = format!(r#"message="{message}""#);
581
582        // events are not available to search API immediately
583        // poll up 200 times for event to show up
584        for _ in 0..200usize {
585            let res = client
586                .post(&search_url)
587                .json(&json!({
588                    "queryString": search_query,
589                }))
590                .header(reqwest::header::ACCEPT, "application/json")
591                .send()
592                .await
593                .unwrap();
594
595            let logs: Vec<HumioLog> = res.json().await.unwrap();
596
597            if !logs.is_empty() {
598                return logs[0].clone();
599            }
600        }
601        panic!("did not find event in Humio repository {repository_name} with message {message}");
602    }
603
604    #[derive(Debug)]
605    struct HumioRepository {
606        name: String,
607        default_ingest_token: String,
608    }
609
610    #[derive(Clone, Deserialize)]
611    #[allow(dead_code)] // deserialize all fields
612    struct HumioLog {
613        #[serde(rename = "#repo")]
614        humio_repo: String,
615
616        #[serde(rename = "#type")]
617        humio_type: String,
618
619        #[serde(rename = "@error")]
620        error: Option<String>,
621
622        #[serde(rename = "@error_msg")]
623        error_msg: Option<String>,
624
625        #[serde(rename = "@rawstring")]
626        rawstring: String,
627
628        #[serde(rename = "@id")]
629        id: String,
630
631        #[serde(rename = "@timestamp")]
632        timestamp_millis: u64,
633
634        #[serde(rename = "@timestamp.nanos")]
635        timestamp_nanos: String,
636
637        #[serde(rename = "@timezone")]
638        timezone: String,
639
640        #[serde(rename = "@source")]
641        source: Option<String>,
642
643        #[serde(rename = "@host")]
644        host: Option<String>,
645
646        // fields parsed from ingested log
647        #[serde(flatten)]
648        fields: HashMap<String, JsonValue>,
649    }
650}
vector/sinks/humio/logs.rs

vector/sinks/humio/
logs.rs
