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

vector/sinks/humio/
logs.rs
