vector/sinks/influxdb/
logs.rs

1use std::collections::{HashMap, HashSet};
2
3use bytes::{Bytes, BytesMut};
4use futures::SinkExt;
5use http::{Request, Uri};
6use indoc::indoc;
7use vector_lib::{
8    config::log_schema,
9    configurable::configurable_component,
10    lookup::{PathPrefix, lookup_v2::OptionalValuePath},
11    schema,
12};
13use vrl::{event_path, path::OwnedValuePath, value::Kind};
14
15use super::{
16    Field, InfluxDb1Settings, InfluxDb2Settings, ProtocolVersion, encode_timestamp, healthcheck,
17    influx_line_protocol, influxdb_settings,
18};
19use crate::{
20    codecs::Transformer,
21    config::{AcknowledgementsConfig, GenerateConfig, Input, SinkConfig, SinkContext},
22    event::{Event, KeyString, MetricTags, Value},
23    http::HttpClient,
24    internal_events::InfluxdbEncodingError,
25    sinks::{
26        Healthcheck, VectorSink,
27        util::{
28            BatchConfig, Buffer, Compression, SinkBatchSettings, TowerRequestConfig,
29            http::{BatchedHttpSink, HttpEventEncoder, HttpSink},
30        },
31    },
32    tls::{TlsConfig, TlsSettings},
33};
34
35#[derive(Clone, Copy, Debug, Default)]
36pub struct InfluxDbLogsDefaultBatchSettings;
37
38impl SinkBatchSettings for InfluxDbLogsDefaultBatchSettings {
39    const MAX_EVENTS: Option<usize> = None;
40    const MAX_BYTES: Option<usize> = Some(1_000_000);
41    const TIMEOUT_SECS: f64 = 1.0;
42}
43
44/// Configuration for the `influxdb_logs` sink.
45#[configurable_component(sink("influxdb_logs", "Deliver log event data to InfluxDB."))]
46#[derive(Clone, Debug, Default)]
47#[serde(deny_unknown_fields)]
48pub struct InfluxDbLogsConfig {
49    /// The namespace of the measurement name to use.
50    ///
51    /// When specified, the measurement name is `<namespace>.vector`.
52    ///
53    #[configurable(
54        deprecated = "This field is deprecated, and `measurement` should be used instead."
55    )]
56    #[configurable(metadata(docs::examples = "service"))]
57    pub namespace: Option<String>,
58
59    /// The name of the InfluxDB measurement that is written to.
60    #[configurable(metadata(docs::examples = "vector-logs"))]
61    pub measurement: Option<String>,
62
63    /// The endpoint to send data to.
64    ///
65    /// This should be a full HTTP URI, including the scheme, host, and port.
66    #[configurable(metadata(docs::examples = "http://localhost:8086"))]
67    pub endpoint: String,
68
69    /// The list of names of log fields that should be added as tags to each measurement.
70    ///
71    /// By default Vector adds `metric_type` as well as the configured `log_schema.host_key` and
72    /// `log_schema.source_type_key` options.
73    #[serde(default)]
74    #[configurable(metadata(docs::examples = "field1"))]
75    #[configurable(metadata(docs::examples = "parent.child_field"))]
76    pub tags: Vec<KeyString>,
77
78    #[serde(flatten)]
79    pub influxdb1_settings: Option<InfluxDb1Settings>,
80
81    #[serde(flatten)]
82    pub influxdb2_settings: Option<InfluxDb2Settings>,
83
84    #[configurable(derived)]
85    #[serde(skip_serializing_if = "crate::serde::is_default", default)]
86    pub encoding: Transformer,
87
88    #[configurable(derived)]
89    #[serde(default)]
90    pub batch: BatchConfig<InfluxDbLogsDefaultBatchSettings>,
91
92    #[configurable(derived)]
93    #[serde(default)]
94    pub request: TowerRequestConfig,
95
96    #[configurable(derived)]
97    pub tls: Option<TlsConfig>,
98
99    #[configurable(derived)]
100    #[serde(
101        default,
102        deserialize_with = "crate::serde::bool_or_struct",
103        skip_serializing_if = "crate::serde::is_default"
104    )]
105    acknowledgements: AcknowledgementsConfig,
106
107    // `host_key`, `message_key`, and `source_type_key` are `Option` as we want `vector generate`
108    // to produce a config with these as `None`, to not accidentally override a users configured
109    // `log_schema`. Generating is constrained by build-time and can't account for changes to the
110    // default `log_schema`.
111    /// Use this option to customize the key containing the hostname.
112    ///
113    /// The setting of `log_schema.host_key`, usually `host`, is used here by default.
114    #[configurable(metadata(docs::examples = "hostname"))]
115    pub host_key: Option<OptionalValuePath>,
116
117    /// Use this option to customize the key containing the message.
118    ///
119    /// The setting of `log_schema.message_key`, usually `message`, is used here by default.
120    #[configurable(metadata(docs::examples = "text"))]
121    pub message_key: Option<OptionalValuePath>,
122
123    /// Use this option to customize the key containing the source_type.
124    ///
125    /// The setting of `log_schema.source_type_key`, usually `source_type`, is used here by default.
126    #[configurable(metadata(docs::examples = "source"))]
127    pub source_type_key: Option<OptionalValuePath>,
128}
129
130#[derive(Debug)]
131struct InfluxDbLogsSink {
132    uri: Uri,
133    token: String,
134    protocol_version: ProtocolVersion,
135    measurement: String,
136    tags: HashSet<KeyString>,
137    transformer: Transformer,
138    host_key: OwnedValuePath,
139    message_key: OwnedValuePath,
140    source_type_key: OwnedValuePath,
141}
142
143impl GenerateConfig for InfluxDbLogsConfig {
144    fn generate_config() -> toml::Value {
145        toml::from_str(indoc! {r#"
146            endpoint = "http://localhost:8086/"
147            namespace = "my-namespace"
148            tags = []
149            org = "my-org"
150            bucket = "my-bucket"
151            token = "${INFLUXDB_TOKEN}"
152        "#})
153        .unwrap()
154    }
155}
156
157#[async_trait::async_trait]
158#[typetag::serde(name = "influxdb_logs")]
159impl SinkConfig for InfluxDbLogsConfig {
160    async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
161        let measurement = self.get_measurement()?;
162        let tags: HashSet<KeyString> = self.tags.iter().cloned().collect();
163
164        let tls_settings = TlsSettings::from_options(self.tls.as_ref())?;
165        let client = HttpClient::new(tls_settings, cx.proxy())?;
166        let healthcheck = self.healthcheck(client.clone())?;
167
168        let batch = self.batch.into_batch_settings()?;
169        let request = self.request.into_settings();
170
171        let settings = influxdb_settings(
172            self.influxdb1_settings.clone(),
173            self.influxdb2_settings.clone(),
174        )
175        .unwrap();
176
177        let endpoint = self.endpoint.clone();
178        let uri = settings.write_uri(endpoint).unwrap();
179
180        let token = settings.token();
181        let protocol_version = settings.protocol_version();
182
183        let host_key = self
184            .host_key
185            .as_ref()
186            .and_then(|k| k.path.clone())
187            .or_else(|| log_schema().host_key().cloned())
188            .expect("global log_schema.host_key to be valid path");
189
190        let message_key = self
191            .message_key
192            .as_ref()
193            .and_then(|k| k.path.clone())
194            .or_else(|| log_schema().message_key().cloned())
195            .expect("global log_schema.message_key to be valid path");
196
197        let source_type_key = self
198            .source_type_key
199            .as_ref()
200            .and_then(|k| k.path.clone())
201            .or_else(|| log_schema().source_type_key().cloned())
202            .expect("global log_schema.source_type_key to be valid path");
203
204        let sink = InfluxDbLogsSink {
205            uri,
206            token: token.inner().to_owned(),
207            protocol_version,
208            measurement,
209            tags,
210            transformer: self.encoding.clone(),
211            host_key,
212            message_key,
213            source_type_key,
214        };
215
216        let sink = BatchedHttpSink::new(
217            sink,
218            Buffer::new(batch.size, Compression::None),
219            request,
220            batch.timeout,
221            client,
222        )
223        .sink_map_err(|error| error!(message = "Fatal influxdb_logs sink error.", %error, internal_log_rate_limit = false));
224
225        #[allow(deprecated)]
226        Ok((VectorSink::from_event_sink(sink), healthcheck))
227    }
228
229    fn input(&self) -> Input {
230        let requirements = schema::Requirement::empty()
231            .optional_meaning("message", Kind::bytes())
232            .optional_meaning("host", Kind::bytes())
233            .optional_meaning("timestamp", Kind::timestamp());
234
235        Input::log().with_schema_requirement(requirements)
236    }
237
238    fn acknowledgements(&self) -> &AcknowledgementsConfig {
239        &self.acknowledgements
240    }
241}
242
243struct InfluxDbLogsEncoder {
244    protocol_version: ProtocolVersion,
245    measurement: String,
246    tags: HashSet<KeyString>,
247    transformer: Transformer,
248    host_key: OwnedValuePath,
249    message_key: OwnedValuePath,
250    source_type_key: OwnedValuePath,
251}
252
253impl HttpEventEncoder<BytesMut> for InfluxDbLogsEncoder {
254    fn encode_event(&mut self, event: Event) -> Option<BytesMut> {
255        let mut log = event.into_log();
256        // If the event isn't an object (`. = "foo"`), inserting or renaming will result in losing
257        // the original value that was assigned to the root. To avoid this we intentionally rename
258        // the path that points to "message" such that it has a dedicated key.
259        // TODO: add a `TargetPath::is_event_root()` to conditionally rename?
260        if let Some(message_path) = log.message_path().cloned().as_ref() {
261            log.rename_key(message_path, (PathPrefix::Event, &self.message_key));
262        }
263        // Add the `host` and `source_type` to the HashSet of tags to include
264        // Ensure those paths are on the event to be encoded, rather than metadata
265        if let Some(host_path) = log.host_path().cloned().as_ref() {
266            self.tags.replace(host_path.path.to_string().into());
267            log.rename_key(host_path, (PathPrefix::Event, &self.host_key));
268        }
269
270        if let Some(source_type_path) = log.source_type_path().cloned().as_ref() {
271            self.tags.replace(source_type_path.path.to_string().into());
272            log.rename_key(source_type_path, (PathPrefix::Event, &self.source_type_key));
273        }
274
275        self.tags.replace("metric_type".into());
276        log.insert(event_path!("metric_type"), "logs");
277
278        // Timestamp
279        let timestamp = encode_timestamp(match log.remove_timestamp() {
280            Some(Value::Timestamp(ts)) => Some(ts),
281            _ => None,
282        });
283
284        let log = {
285            let mut event = Event::from(log);
286            self.transformer.transform(&mut event);
287            event.into_log()
288        };
289
290        // Tags + Fields
291        let mut tags = MetricTags::default();
292        let mut fields: HashMap<KeyString, Field> = HashMap::new();
293        log.convert_to_fields().for_each(|(key, value)| {
294            if self.tags.contains(key.as_str()) {
295                tags.replace(key.into(), value.to_string_lossy().into_owned());
296            } else {
297                fields.insert(key, to_field(value));
298            }
299        });
300
301        let mut output = BytesMut::new();
302        if let Err(error_message) = influx_line_protocol(
303            self.protocol_version,
304            &self.measurement,
305            Some(tags),
306            Some(fields),
307            timestamp,
308            &mut output,
309        ) {
310            emit!(InfluxdbEncodingError {
311                error_message,
312                count: 1
313            });
314            return None;
315        };
316
317        Some(output)
318    }
319}
320
321impl HttpSink for InfluxDbLogsSink {
322    type Input = BytesMut;
323    type Output = BytesMut;
324    type Encoder = InfluxDbLogsEncoder;
325
326    fn build_encoder(&self) -> Self::Encoder {
327        InfluxDbLogsEncoder {
328            protocol_version: self.protocol_version,
329            measurement: self.measurement.clone(),
330            tags: self.tags.clone(),
331            transformer: self.transformer.clone(),
332            host_key: self.host_key.clone(),
333            message_key: self.message_key.clone(),
334            source_type_key: self.source_type_key.clone(),
335        }
336    }
337
338    async fn build_request(&self, events: Self::Output) -> crate::Result<Request<Bytes>> {
339        Request::post(&self.uri)
340            .header("Content-Type", "text/plain")
341            .header("Authorization", format!("Token {}", &self.token))
342            .body(events.freeze())
343            .map_err(Into::into)
344    }
345}
346
347impl InfluxDbLogsConfig {
348    fn get_measurement(&self) -> Result<String, &'static str> {
349        match (self.measurement.as_ref(), self.namespace.as_ref()) {
350            (Some(measure), Some(_)) => {
351                warn!("Option `namespace` has been superseded by `measurement`.");
352                Ok(measure.clone())
353            }
354            (Some(measure), None) => Ok(measure.clone()),
355            (None, Some(namespace)) => {
356                warn!(
357                    "Option `namespace` has been deprecated. Use `measurement` instead. \
358                       For example, you can use `measurement=<namespace>.vector` for the \
359                       same effect."
360                );
361                Ok(format!("{namespace}.vector"))
362            }
363            (None, None) => Err("The `measurement` option is required."),
364        }
365    }
366
367    fn healthcheck(&self, client: HttpClient) -> crate::Result<Healthcheck> {
368        let config = self.clone();
369
370        let healthcheck = healthcheck(
371            config.endpoint,
372            config.influxdb1_settings,
373            config.influxdb2_settings,
374            client,
375        )?;
376
377        Ok(healthcheck)
378    }
379}
380
381fn to_field(value: &Value) -> Field {
382    match value {
383        Value::Integer(num) => Field::Int(*num),
384        Value::Float(num) => Field::Float(num.into_inner()),
385        Value::Boolean(b) => Field::Bool(*b),
386        _ => Field::String(value.to_string_lossy().into_owned()),
387    }
388}
389
390#[cfg(test)]
391mod tests {
392    use chrono::{Utc, offset::TimeZone};
393    use futures::{StreamExt, channel::mpsc, stream};
394    use http::{StatusCode, request::Parts};
395    use indoc::indoc;
396    use vector_lib::{
397        event::{BatchNotifier, BatchStatus, Event, LogEvent},
398        lookup::owned_value_path,
399    };
400
401    use super::*;
402    use crate::{
403        sinks::{
404            influxdb::test_util::{assert_fields, split_line_protocol, ts},
405            util::test::{build_test_server_status, load_sink},
406        },
407        test_util::{
408            addr::next_addr,
409            components::{
410                COMPONENT_ERROR_TAGS, HTTP_SINK_TAGS, run_and_assert_sink_compliance,
411                run_and_assert_sink_error,
412            },
413        },
414    };
415
416    type Receiver = mpsc::Receiver<(Parts, bytes::Bytes)>;
417
418    #[test]
419    fn generate_config() {
420        crate::test_util::test_generate_config::<InfluxDbLogsConfig>();
421    }
422
423    #[test]
424    fn test_config_without_tags() {
425        let config = indoc! {r#"
426            namespace: "vector-logs"
427            endpoint: "http://localhost:9999"
428            bucket: "my-bucket"
429            org: "my-org"
430            token: "my-token"
431        "#};
432
433        serde_yaml::from_str::<InfluxDbLogsConfig>(config).unwrap();
434    }
435
436    #[test]
437    fn test_config_measurement_from_namespace() {
438        let config = indoc! {r#"
439            namespace: "ns"
440            endpoint: "http://localhost:9999"
441        "#};
442
443        let sink_config = serde_yaml::from_str::<InfluxDbLogsConfig>(config).unwrap();
444        assert_eq!("ns.vector", sink_config.get_measurement().unwrap());
445    }
446
447    #[test]
448    fn test_encode_event_apply_rules() {
449        let mut event = Event::Log(LogEvent::from("hello"));
450        event
451            .as_mut_log()
452            .insert(event_path!("host"), "aws.cloud.eur");
453        event.as_mut_log().insert(event_path!("timestamp"), ts());
454
455        let mut sink = create_sink(
456            "http://localhost:9999",
457            "my-token",
458            ProtocolVersion::V1,
459            "vector",
460            ["metric_type", "host"].to_vec(),
461        );
462        sink.transformer
463            .set_except_fields(Some(vec!["host".into()]))
464            .unwrap();
465        let mut encoder = sink.build_encoder();
466
467        let bytes = encoder.encode_event(event.clone()).unwrap();
468        let string = std::str::from_utf8(&bytes).unwrap();
469
470        let line_protocol = split_line_protocol(string);
471        assert_eq!("vector", line_protocol.0);
472        assert_eq!("metric_type=logs", line_protocol.1);
473        assert_fields(line_protocol.2.to_string(), ["message=\"hello\""].to_vec());
474        assert_eq!("1542182950000000011\n", line_protocol.3);
475
476        sink.transformer
477            .set_except_fields(Some(vec!["metric_type".into()]))
478            .unwrap();
479        let mut encoder = sink.build_encoder();
480        let bytes = encoder.encode_event(event.clone()).unwrap();
481        let string = std::str::from_utf8(&bytes).unwrap();
482        let line_protocol = split_line_protocol(string);
483        assert_eq!(
484            "host=aws.cloud.eur", line_protocol.1,
485            "metric_type tag should be excluded"
486        );
487        assert_fields(line_protocol.2, ["message=\"hello\""].to_vec());
488    }
489
490    #[test]
491    fn test_encode_event_v1() {
492        let mut event = Event::Log(LogEvent::from("hello"));
493        event
494            .as_mut_log()
495            .insert(event_path!("host"), "aws.cloud.eur");
496        event
497            .as_mut_log()
498            .insert(event_path!("source_type"), "file");
499
500        event.as_mut_log().insert(event_path!("int"), 4i32);
501        event.as_mut_log().insert(event_path!("float"), 5.5);
502        event.as_mut_log().insert(event_path!("bool"), true);
503        event
504            .as_mut_log()
505            .insert(event_path!("string"), "thisisastring");
506        event.as_mut_log().insert(event_path!("timestamp"), ts());
507
508        let sink = create_sink(
509            "http://localhost:9999",
510            "my-token",
511            ProtocolVersion::V1,
512            "vector",
513            ["source_type", "host", "metric_type"].to_vec(),
514        );
515        let mut encoder = sink.build_encoder();
516
517        let bytes = encoder.encode_event(event).unwrap();
518        let string = std::str::from_utf8(&bytes).unwrap();
519
520        let line_protocol = split_line_protocol(string);
521        assert_eq!("vector", line_protocol.0);
522        assert_eq!(
523            "host=aws.cloud.eur,metric_type=logs,source_type=file",
524            line_protocol.1
525        );
526        assert_fields(
527            line_protocol.2.to_string(),
528            [
529                "int=4i",
530                "float=5.5",
531                "bool=true",
532                "string=\"thisisastring\"",
533                "message=\"hello\"",
534            ]
535            .to_vec(),
536        );
537
538        assert_eq!("1542182950000000011\n", line_protocol.3);
539    }
540
541    #[test]
542    fn test_encode_event() {
543        let mut event = Event::Log(LogEvent::from("hello"));
544        event
545            .as_mut_log()
546            .insert(event_path!("host"), "aws.cloud.eur");
547        event
548            .as_mut_log()
549            .insert(event_path!("source_type"), "file");
550
551        event.as_mut_log().insert(event_path!("int"), 4i32);
552        event.as_mut_log().insert(event_path!("float"), 5.5);
553        event.as_mut_log().insert(event_path!("bool"), true);
554        event
555            .as_mut_log()
556            .insert(event_path!("string"), "thisisastring");
557        event.as_mut_log().insert(event_path!("timestamp"), ts());
558
559        let sink = create_sink(
560            "http://localhost:9999",
561            "my-token",
562            ProtocolVersion::V2,
563            "vector",
564            ["source_type", "host", "metric_type"].to_vec(),
565        );
566        let mut encoder = sink.build_encoder();
567
568        let bytes = encoder.encode_event(event).unwrap();
569        let string = std::str::from_utf8(&bytes).unwrap();
570
571        let line_protocol = split_line_protocol(string);
572        assert_eq!("vector", line_protocol.0);
573        assert_eq!(
574            "host=aws.cloud.eur,metric_type=logs,source_type=file",
575            line_protocol.1
576        );
577        assert_fields(
578            line_protocol.2.to_string(),
579            [
580                "int=4i",
581                "float=5.5",
582                "bool=true",
583                "string=\"thisisastring\"",
584                "message=\"hello\"",
585            ]
586            .to_vec(),
587        );
588
589        assert_eq!("1542182950000000011\n", line_protocol.3);
590    }
591
592    #[test]
593    fn test_encode_event_without_tags() {
594        let mut event = Event::Log(LogEvent::from("hello"));
595
596        event.as_mut_log().insert(event_path!("value"), 100);
597        event.as_mut_log().insert(event_path!("timestamp"), ts());
598
599        let mut sink = create_sink(
600            "http://localhost:9999",
601            "my-token",
602            ProtocolVersion::V2,
603            "vector",
604            [].to_vec(),
605        );
606        // exclude default metric_type tag so to emit empty tags
607        sink.transformer
608            .set_except_fields(Some(vec!["metric_type".into()]))
609            .unwrap();
610        let mut encoder = sink.build_encoder();
611
612        let bytes = encoder.encode_event(event).unwrap();
613        let line = std::str::from_utf8(&bytes).unwrap();
614        assert!(
615            line.starts_with("vector "),
616            "measurement (without tags) should ends with space ' '"
617        );
618
619        let line_protocol = split_line_protocol(line);
620        assert_eq!("vector", line_protocol.0);
621        assert_eq!("", line_protocol.1, "tags should be empty");
622        assert_fields(
623            line_protocol.2,
624            ["value=100i", "message=\"hello\""].to_vec(),
625        );
626
627        assert_eq!("1542182950000000011\n", line_protocol.3);
628    }
629
630    #[test]
631    fn test_encode_nested_fields() {
632        let mut event = LogEvent::default();
633
634        event.insert(event_path!("a"), 1);
635        event.insert(event_path!("nested", "field"), "2");
636        event.insert(event_path!("nested", "bool"), true);
637        event.insert(event_path!("nested", "array", 0isize), "example-value");
638        event.insert(event_path!("nested", "array", 2isize), "another-value");
639        event.insert(event_path!("nested", "array", 3isize), 15);
640
641        let sink = create_sink(
642            "http://localhost:9999",
643            "my-token",
644            ProtocolVersion::V2,
645            "vector",
646            ["metric_type"].to_vec(),
647        );
648        let mut encoder = sink.build_encoder();
649
650        let bytes = encoder.encode_event(event.into()).unwrap();
651        let string = std::str::from_utf8(&bytes).unwrap();
652
653        let line_protocol = split_line_protocol(string);
654        assert_eq!("vector", line_protocol.0);
655        assert_eq!("metric_type=logs", line_protocol.1);
656        assert_fields(
657            line_protocol.2,
658            [
659                "a=1i",
660                "nested.array[0]=\"example-value\"",
661                "nested.array[1]=\"<null>\"",
662                "nested.array[2]=\"another-value\"",
663                "nested.array[3]=15i",
664                "nested.bool=true",
665                "nested.field=\"2\"",
666            ]
667            .to_vec(),
668        );
669    }
670
671    #[test]
672    fn test_add_tag() {
673        let mut event = Event::Log(LogEvent::from("hello"));
674        event
675            .as_mut_log()
676            .insert(event_path!("source_type"), "file");
677
678        event.as_mut_log().insert(event_path!("as_a_tag"), 10);
679        event.as_mut_log().insert(event_path!("timestamp"), ts());
680
681        let sink = create_sink(
682            "http://localhost:9999",
683            "my-token",
684            ProtocolVersion::V2,
685            "vector",
686            ["as_a_tag", "not_exists_field", "source_type", "metric_type"].to_vec(),
687        );
688        let mut encoder = sink.build_encoder();
689
690        let bytes = encoder.encode_event(event).unwrap();
691        let string = std::str::from_utf8(&bytes).unwrap();
692
693        let line_protocol = split_line_protocol(string);
694        assert_eq!("vector", line_protocol.0);
695        assert_eq!(
696            "as_a_tag=10,metric_type=logs,source_type=file",
697            line_protocol.1
698        );
699        assert_fields(line_protocol.2.to_string(), ["message=\"hello\""].to_vec());
700
701        assert_eq!("1542182950000000011\n", line_protocol.3);
702    }
703
704    #[tokio::test]
705    async fn smoke_v1() {
706        let rx = smoke_test(
707            r#"database = "my-database""#,
708            StatusCode::OK,
709            BatchStatus::Delivered,
710        )
711        .await;
712
713        let query = receive_response(rx).await;
714        assert!(query.contains("db=my-database"));
715        assert!(query.contains("precision=ns"));
716    }
717
718    #[tokio::test]
719    async fn smoke_v1_failure() {
720        smoke_test(
721            r#"database = "my-database""#,
722            StatusCode::BAD_REQUEST,
723            BatchStatus::Rejected,
724        )
725        .await;
726    }
727
728    #[tokio::test]
729    async fn smoke_v2() {
730        let rx = smoke_test(
731            indoc! {r#"
732            bucket = "my-bucket"
733            org = "my-org"
734            token = "my-token"
735        "#},
736            StatusCode::OK,
737            BatchStatus::Delivered,
738        )
739        .await;
740
741        let query = receive_response(rx).await;
742        assert!(query.contains("org=my-org"));
743        assert!(query.contains("bucket=my-bucket"));
744        assert!(query.contains("precision=ns"));
745    }
746
747    #[tokio::test]
748    async fn smoke_v2_failure() {
749        smoke_test(
750            indoc! {r#"
751            bucket = "my-bucket"
752            org = "my-org"
753            token = "my-token"
754        "#},
755            StatusCode::BAD_REQUEST,
756            BatchStatus::Rejected,
757        )
758        .await;
759    }
760
761    async fn smoke_test(
762        config: &str,
763        status_code: StatusCode,
764        batch_status: BatchStatus,
765    ) -> Receiver {
766        let config = format!(
767            indoc! {r#"
768            measurement = "vector"
769            endpoint = "http://localhost:9999"
770            {}
771        "#},
772            config
773        );
774        let (mut config, cx) = load_sink::<InfluxDbLogsConfig>(&config).unwrap();
775
776        // Make sure we can build the config
777        _ = config.build(cx.clone()).await.unwrap();
778
779        let (_guard, addr) = next_addr();
780        // Swap out the host so we can force send it
781        // to our local server
782        let host = format!("http://{addr}");
783        config.endpoint = host;
784
785        let (sink, _) = config.build(cx).await.unwrap();
786
787        let (rx, _trigger, server) = build_test_server_status(addr, status_code);
788        tokio::spawn(server);
789
790        let (batch, mut receiver) = BatchNotifier::new_with_receiver();
791
792        let lines = std::iter::repeat(())
793            .map(move |_| "message_value")
794            .take(5)
795            .collect::<Vec<_>>();
796        let mut events = Vec::new();
797
798        // Create 5 events with custom field
799        for (i, line) in lines.iter().enumerate() {
800            let mut event = LogEvent::from(line.to_string()).with_batch_notifier(&batch);
801            event.insert(event_path!(format!("key{i}").as_str()), format!("value{i}"));
802
803            let timestamp = Utc
804                .with_ymd_and_hms(1970, 1, 1, 0, 0, (i as u32) + 1)
805                .single()
806                .expect("invalid timestamp");
807            event.insert(event_path!("timestamp"), timestamp);
808            event.insert(event_path!("source_type"), "file");
809
810            events.push(Event::Log(event));
811        }
812        drop(batch);
813
814        if batch_status == BatchStatus::Delivered {
815            run_and_assert_sink_compliance(sink, stream::iter(events), &HTTP_SINK_TAGS).await;
816        } else {
817            run_and_assert_sink_error(sink, stream::iter(events), &COMPONENT_ERROR_TAGS).await;
818        }
819
820        assert_eq!(receiver.try_recv(), Ok(batch_status));
821
822        rx
823    }
824
825    async fn receive_response(mut rx: Receiver) -> String {
826        let output = rx.next().await.unwrap();
827
828        let request = &output.0;
829        let query = request.uri.query().unwrap();
830
831        let body = std::str::from_utf8(&output.1[..]).unwrap();
832        let mut lines = body.lines();
833
834        assert_eq!(5, lines.clone().count());
835        assert_line_protocol(0, lines.next());
836
837        query.into()
838    }
839
840    fn assert_line_protocol(i: i64, value: Option<&str>) {
841        //vector,metric_type=logs key0="value0",message="message_value" 1000000000
842        let line_protocol = split_line_protocol(value.unwrap());
843        assert_eq!("vector", line_protocol.0);
844        assert_eq!("metric_type=logs,source_type=file", line_protocol.1);
845        assert_fields(
846            line_protocol.2.to_string(),
847            [
848                &*format!("key{i}=\"value{i}\""),
849                "message=\"message_value\"",
850            ]
851            .to_vec(),
852        );
853
854        assert_eq!(((i + 1) * 1000000000).to_string(), line_protocol.3);
855    }
856
857    fn create_sink(
858        uri: &str,
859        token: &str,
860        protocol_version: ProtocolVersion,
861        measurement: &str,
862        tags: Vec<&str>,
863    ) -> InfluxDbLogsSink {
864        let uri = uri.parse::<Uri>().unwrap();
865        let token = token.to_string();
866        let measurement = measurement.to_string();
867        let tags: HashSet<_> = tags.into_iter().map(|tag| tag.into()).collect();
868        InfluxDbLogsSink {
869            uri,
870            token,
871            protocol_version,
872            measurement,
873            tags,
874            transformer: Default::default(),
875            host_key: owned_value_path!("host"),
876            message_key: owned_value_path!("message"),
877            source_type_key: owned_value_path!("source_type"),
878        }
879    }
880}
881
882#[cfg(feature = "influxdb-integration-tests")]
883#[cfg(test)]
884mod integration_tests {
885    use std::sync::Arc;
886
887    use chrono::Utc;
888    use futures::stream;
889    use vector_lib::{
890        codecs::BytesDeserializerConfig,
891        config::{LegacyKey, LogNamespace},
892        event::{BatchNotifier, BatchStatus, Event, LogEvent},
893        lookup::{owned_value_path, path},
894    };
895    use vrl::value;
896
897    use super::*;
898    use crate::{
899        config::SinkContext,
900        sinks::influxdb::{
901            InfluxDb2Settings,
902            logs::InfluxDbLogsConfig,
903            test_util::{BUCKET, ORG, TOKEN, address_v2, onboarding_v2},
904        },
905        test_util::components::{HTTP_SINK_TAGS, run_and_assert_sink_compliance},
906    };
907
908    #[tokio::test]
909    async fn influxdb2_logs_put_data() {
910        let endpoint = address_v2();
911        onboarding_v2(&endpoint).await;
912
913        let now = Utc::now();
914        let measure = format!(
915            "vector-{}",
916            now.timestamp_nanos_opt().expect("Timestamp out of range")
917        );
918
919        let cx = SinkContext::default();
920
921        let config = InfluxDbLogsConfig {
922            namespace: None,
923            measurement: Some(measure.clone()),
924            endpoint: endpoint.clone(),
925            tags: Default::default(),
926            influxdb1_settings: None,
927            influxdb2_settings: Some(InfluxDb2Settings {
928                org: ORG.to_string(),
929                bucket: BUCKET.to_string(),
930                token: TOKEN.to_string().into(),
931            }),
932            encoding: Default::default(),
933            batch: Default::default(),
934            request: Default::default(),
935            tls: None,
936            acknowledgements: Default::default(),
937            host_key: None,
938            message_key: None,
939            source_type_key: None,
940        };
941
942        let (sink, _) = config.build(cx).await.unwrap();
943
944        let (batch, mut receiver) = BatchNotifier::new_with_receiver();
945
946        let mut event1 = LogEvent::from("message_1").with_batch_notifier(&batch);
947        event1.insert(event_path!("host"), "aws.cloud.eur");
948        event1.insert(event_path!("source_type"), "file");
949
950        let mut event2 = LogEvent::from("message_2").with_batch_notifier(&batch);
951        event2.insert(event_path!("host"), "aws.cloud.eur");
952        event2.insert(event_path!("source_type"), "file");
953
954        let mut namespaced_log =
955            LogEvent::from(value!("namespaced message")).with_batch_notifier(&batch);
956        LogNamespace::Vector.insert_source_metadata(
957            "file",
958            &mut namespaced_log,
959            Some(LegacyKey::Overwrite(path!("host"))),
960            path!("host"),
961            "aws.cloud.eur",
962        );
963        LogNamespace::Vector.insert_standard_vector_source_metadata(
964            &mut namespaced_log,
965            "file",
966            now,
967        );
968        let schema = BytesDeserializerConfig
969            .schema_definition(LogNamespace::Vector)
970            .with_metadata_field(
971                &owned_value_path!("file", "host"),
972                Kind::bytes(),
973                Some("host"),
974            );
975        namespaced_log
976            .metadata_mut()
977            .set_schema_definition(&Arc::new(schema));
978
979        drop(batch);
980
981        let events = vec![
982            Event::Log(event1),
983            Event::Log(event2),
984            Event::Log(namespaced_log),
985        ];
986
987        run_and_assert_sink_compliance(sink, stream::iter(events), &HTTP_SINK_TAGS).await;
988
989        assert_eq!(receiver.try_recv(), Ok(BatchStatus::Delivered));
990
991        let mut body = std::collections::HashMap::new();
992        body.insert("query", format!("from(bucket:\"my-bucket\") |> range(start: 0) |> filter(fn: (r) => r._measurement == \"{}\")", measure.clone()));
993        body.insert("type", "flux".to_owned());
994
995        let client = reqwest::Client::builder()
996            .danger_accept_invalid_certs(true)
997            .build()
998            .unwrap();
999
1000        let res = client
1001            .post(format!("{endpoint}/api/v2/query?org=my-org"))
1002            .json(&body)
1003            .header("accept", "application/json")
1004            .header("Authorization", "Token my-token")
1005            .send()
1006            .await
1007            .unwrap();
1008        let string = res.text().await.unwrap();
1009
1010        let lines = string.split('\n').collect::<Vec<&str>>();
1011        let header = lines[0].split(',').collect::<Vec<&str>>();
1012        let record1 = lines[1].split(',').collect::<Vec<&str>>();
1013        let record2 = lines[2].split(',').collect::<Vec<&str>>();
1014        let record_ns = lines[3].split(',').collect::<Vec<&str>>();
1015
1016        // measurement
1017        assert_eq!(
1018            record1[header
1019                .iter()
1020                .position(|&r| r.trim() == "_measurement")
1021                .unwrap()]
1022            .trim(),
1023            measure.clone()
1024        );
1025        assert_eq!(
1026            record2[header
1027                .iter()
1028                .position(|&r| r.trim() == "_measurement")
1029                .unwrap()]
1030            .trim(),
1031            measure.clone()
1032        );
1033        assert_eq!(
1034            record_ns[header
1035                .iter()
1036                .position(|&r| r.trim() == "_measurement")
1037                .unwrap()]
1038            .trim(),
1039            measure.clone()
1040        );
1041
1042        // tags
1043        assert_eq!(
1044            record1[header
1045                .iter()
1046                .position(|&r| r.trim() == "metric_type")
1047                .unwrap()]
1048            .trim(),
1049            "logs"
1050        );
1051        assert_eq!(
1052            record2[header
1053                .iter()
1054                .position(|&r| r.trim() == "metric_type")
1055                .unwrap()]
1056            .trim(),
1057            "logs"
1058        );
1059        assert_eq!(
1060            record_ns[header
1061                .iter()
1062                .position(|&r| r.trim() == "metric_type")
1063                .unwrap()]
1064            .trim(),
1065            "logs"
1066        );
1067        assert_eq!(
1068            record1[header.iter().position(|&r| r.trim() == "host").unwrap()].trim(),
1069            "aws.cloud.eur"
1070        );
1071        assert_eq!(
1072            record2[header.iter().position(|&r| r.trim() == "host").unwrap()].trim(),
1073            "aws.cloud.eur"
1074        );
1075        assert_eq!(
1076            record_ns[header.iter().position(|&r| r.trim() == "host").unwrap()].trim(),
1077            "aws.cloud.eur"
1078        );
1079        assert_eq!(
1080            record1[header
1081                .iter()
1082                .position(|&r| r.trim() == "source_type")
1083                .unwrap()]
1084            .trim(),
1085            "file"
1086        );
1087        assert_eq!(
1088            record2[header
1089                .iter()
1090                .position(|&r| r.trim() == "source_type")
1091                .unwrap()]
1092            .trim(),
1093            "file"
1094        );
1095        assert_eq!(
1096            record_ns[header
1097                .iter()
1098                .position(|&r| r.trim() == "source_type")
1099                .unwrap()]
1100            .trim(),
1101            "file"
1102        );
1103
1104        // field
1105        assert_eq!(
1106            record1[header.iter().position(|&r| r.trim() == "_field").unwrap()].trim(),
1107            "message"
1108        );
1109        assert_eq!(
1110            record2[header.iter().position(|&r| r.trim() == "_field").unwrap()].trim(),
1111            "message"
1112        );
1113        assert_eq!(
1114            record_ns[header.iter().position(|&r| r.trim() == "_field").unwrap()].trim(),
1115            "message"
1116        );
1117        assert_eq!(
1118            record1[header.iter().position(|&r| r.trim() == "_value").unwrap()].trim(),
1119            "message_1"
1120        );
1121        assert_eq!(
1122            record2[header.iter().position(|&r| r.trim() == "_value").unwrap()].trim(),
1123            "message_2"
1124        );
1125        assert_eq!(
1126            record_ns[header.iter().position(|&r| r.trim() == "_value").unwrap()].trim(),
1127            "namespaced message"
1128        );
1129    }
1130}
vector/sinks/influxdb/logs.rs

vector/sinks/influxdb/
logs.rs
