vector/sources/aws_kinesis_firehose/
mod.rs

1use std::{convert::Infallible, fmt, net::SocketAddr, time::Duration};
2
3use futures::FutureExt;
4use hyper::{Server, service::make_service_fn};
5use tokio::net::TcpStream;
6use tower::ServiceBuilder;
7use tracing::Span;
8use vector_lib::{
9    codecs::decoding::{DeserializerConfig, FramingConfig},
10    config::{LegacyKey, LogNamespace},
11    configurable::configurable_component,
12    lookup::owned_value_path,
13    sensitive_string::SensitiveString,
14    tls::MaybeTlsIncomingStream,
15};
16use vrl::value::{Kind, kind::Collection};
17
18use crate::{
19    codecs::DecodingConfig,
20    config::{
21        GenerateConfig, Resource, SourceAcknowledgementsConfig, SourceConfig, SourceContext,
22        SourceOutput,
23    },
24    http::{KeepaliveConfig, MaxConnectionAgeLayer, build_http_trace_layer},
25    serde::{bool_or_struct, default_decoding, default_framing_message_based},
26    sources::http_server::{build_param_matcher, remove_duplicates},
27    tls::{MaybeTlsSettings, TlsEnableableConfig},
28};
29
30pub mod errors;
31mod filters;
32mod handlers;
33mod models;
34
35/// Configuration for the `aws_kinesis_firehose` source.
36#[configurable_component(source(
37    "aws_kinesis_firehose",
38    "Collect logs from AWS Kinesis Firehose."
39))]
40#[derive(Clone, Debug)]
41pub struct AwsKinesisFirehoseConfig {
42    /// The socket address to listen for connections on.
43    #[configurable(metadata(docs::examples = "0.0.0.0:443"))]
44    #[configurable(metadata(docs::examples = "localhost:443"))]
45    address: SocketAddr,
46
47    /// An access key to authenticate requests against.
48    ///
49    /// AWS Kinesis Firehose can be configured to pass along a user-configurable access key with each request. If
50    /// configured, `access_key` should be set to the same value. Otherwise, all requests are allowed.
51    #[configurable(deprecated = "This option has been deprecated, use `access_keys` instead.")]
52    #[configurable(metadata(docs::examples = "A94A8FE5CCB19BA61C4C08"))]
53    access_key: Option<SensitiveString>,
54
55    /// A list of access keys to authenticate requests against.
56    ///
57    /// AWS Kinesis Firehose can be configured to pass along a user-configurable access key with each request. If
58    /// configured, `access_keys` should be set to the same value. Otherwise, all requests are allowed.
59    #[configurable(metadata(docs::examples = "access_keys_example()"))]
60    access_keys: Option<Vec<SensitiveString>>,
61
62    /// Whether or not to store the AWS Firehose Access Key in event secrets.
63    ///
64    /// If set to `true`, when incoming requests contains an access key sent by AWS Firehose, it is kept in the
65    /// event secrets as "aws_kinesis_firehose_access_key".
66    #[configurable(derived)]
67    store_access_key: bool,
68
69    /// The compression scheme to use for decompressing records within the Firehose message.
70    ///
71    /// Some services, like AWS CloudWatch Logs, [compresses the events with gzip][events_with_gzip],
72    /// before sending them AWS Kinesis Firehose. This option can be used to automatically decompress
73    /// them before forwarding them to the next component.
74    ///
75    /// Note that this is different from [Content encoding option][encoding_option] of the
76    /// Firehose HTTP endpoint destination. That option controls the content encoding of the entire HTTP request.
77    ///
78    /// [events_with_gzip]: https://docs.aws.amazon.com/firehose/latest/dev/writing-with-cloudwatch-logs.html
79    /// [encoding_option]: https://docs.aws.amazon.com/firehose/latest/dev/create-destination.html#create-destination-http
80    #[serde(default)]
81    record_compression: Compression,
82
83    #[configurable(derived)]
84    tls: Option<TlsEnableableConfig>,
85
86    #[configurable(derived)]
87    #[configurable(metadata(docs::advanced))]
88    #[serde(default = "default_framing_message_based")]
89    framing: FramingConfig,
90
91    #[configurable(derived)]
92    #[configurable(metadata(docs::advanced))]
93    #[serde(default = "default_decoding")]
94    decoding: DeserializerConfig,
95
96    #[configurable(derived)]
97    #[serde(default, deserialize_with = "bool_or_struct")]
98    acknowledgements: SourceAcknowledgementsConfig,
99
100    /// The namespace to use for logs. This overrides the global setting.
101    #[configurable(metadata(docs::hidden))]
102    #[serde(default)]
103    log_namespace: Option<bool>,
104
105    #[configurable(derived)]
106    #[serde(default)]
107    keepalive: KeepaliveConfig,
108
109    /// A list of attributes from X-Amz-Firehose-Common-Attributes header to include in the log event.
110    ///
111    /// Accepts the wildcard (`*`) character for attributes matching a specified pattern.
112    ///
113    /// Specifying "*" results in all common attributes included in the log event.
114    ///
115    /// Legacy namespace: selected attributes are added under the root `common_attributes` object
116    /// Vector namespace: selected attributes are added under the source metadata at `aws_kinesis_firehose.common_attributes`
117    #[serde(default)]
118    #[configurable(metadata(docs::examples = "environment"))]
119    #[configurable(metadata(docs::examples = "application_group"))]
120    #[configurable(metadata(docs::examples = "application_*"))]
121    #[configurable(metadata(docs::examples = "*"))]
122    common_attributes: Vec<String>,
123}
124
125const fn access_keys_example() -> [&'static str; 2] {
126    ["A94A8FE5CCB19BA61C4C08", "B94B8FE5CCB19BA61C4C12"]
127}
128
129/// Compression scheme for records in a Firehose message.
130#[configurable_component]
131#[configurable(metadata(docs::advanced))]
132#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
133#[serde(rename_all = "lowercase")]
134pub enum Compression {
135    /// Automatically attempt to determine the compression scheme.
136    ///
137    /// The compression scheme of the object is determined by looking at its file signature, also known
138    /// as [magic bytes][magic_bytes].
139    ///
140    /// If the record fails to decompress with the discovered format, the record is forwarded as is.
141    /// Thus, if you know the records are always gzip encoded (for example, if they are coming from AWS CloudWatch Logs),
142    /// set `gzip` in this field so that any records that are not-gzipped are rejected.
143    ///
144    /// [magic_bytes]: https://en.wikipedia.org/wiki/List_of_file_signatures
145    #[default]
146    Auto,
147
148    /// Uncompressed.
149    None,
150
151    /// GZIP.
152    Gzip,
153}
154
155impl fmt::Display for Compression {
156    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
157        match self {
158            Compression::Auto => write!(fmt, "auto"),
159            Compression::None => write!(fmt, "none"),
160            Compression::Gzip => write!(fmt, "gzip"),
161        }
162    }
163}
164
165#[async_trait::async_trait]
166#[typetag::serde(name = "aws_kinesis_firehose")]
167impl SourceConfig for AwsKinesisFirehoseConfig {
168    async fn build(&self, cx: SourceContext) -> crate::Result<super::Source> {
169        let log_namespace = cx.log_namespace(self.log_namespace);
170        let decoder =
171            DecodingConfig::new(self.framing.clone(), self.decoding.clone(), log_namespace)
172                .build()?;
173
174        let acknowledgements = cx.do_acknowledgements(self.acknowledgements);
175
176        if self.access_key.is_some() {
177            warn!("DEPRECATION `access_key`, use `access_keys` instead.")
178        }
179
180        // Merge with legacy `access_key`
181        let access_keys = self
182            .access_keys
183            .iter()
184            .flatten()
185            .chain(self.access_key.iter());
186
187        let common_attributes = build_param_matcher(&remove_duplicates(
188            self.common_attributes.clone(),
189            "common_attributes",
190        ))?;
191
192        let svc = filters::firehose(
193            access_keys.map(|key| key.inner().to_string()).collect(),
194            self.store_access_key,
195            self.record_compression,
196            decoder,
197            acknowledgements,
198            cx.out,
199            log_namespace,
200            common_attributes,
201        );
202
203        let tls = MaybeTlsSettings::from_config(self.tls.as_ref(), true)?;
204        let listener = tls.bind(&self.address).await?;
205
206        let keepalive_settings = self.keepalive.clone();
207        let shutdown = cx.shutdown;
208        Ok(Box::pin(async move {
209            let span = Span::current();
210            let make_svc = make_service_fn(move |conn: &MaybeTlsIncomingStream<TcpStream>| {
211                let svc = ServiceBuilder::new()
212                    .layer(build_http_trace_layer(span.clone()))
213                    .option_layer(keepalive_settings.max_connection_age_secs.map(|secs| {
214                        MaxConnectionAgeLayer::new(
215                            Duration::from_secs(secs),
216                            keepalive_settings.max_connection_age_jitter_factor,
217                            conn.peer_addr(),
218                        )
219                    }))
220                    .service(warp::service(svc.clone()));
221                futures_util::future::ok::<_, Infallible>(svc)
222            });
223
224            Server::builder(hyper::server::accept::from_stream(listener.accept_stream()))
225                .serve(make_svc)
226                .with_graceful_shutdown(shutdown.map(|_| ()))
227                .await
228                .map_err(|err| {
229                    error!("An error occurred: {:?}.", err);
230                })?;
231
232            Ok(())
233        }))
234    }
235
236    fn outputs(&self, global_log_namespace: LogNamespace) -> Vec<SourceOutput> {
237        let common_attributes_path = (!self.common_attributes.is_empty()).then_some(
238            LegacyKey::InsertIfEmpty(owned_value_path!("common_attributes")),
239        );
240        let schema_definition = self
241            .decoding
242            .schema_definition(global_log_namespace.merge(self.log_namespace))
243            .with_standard_vector_source_metadata()
244            .with_source_metadata(
245                Self::NAME,
246                Some(LegacyKey::InsertIfEmpty(owned_value_path!("request_id"))),
247                &owned_value_path!("request_id"),
248                Kind::bytes(),
249                None,
250            )
251            .with_source_metadata(
252                Self::NAME,
253                Some(LegacyKey::InsertIfEmpty(owned_value_path!("source_arn"))),
254                &owned_value_path!("source_arn"),
255                Kind::bytes(),
256                None,
257            )
258            // for common attributes dynamically added from X-Amz-Firehose-Common-Attributes header
259            .with_source_metadata(
260                Self::NAME,
261                common_attributes_path,
262                &owned_value_path!("common_attributes"),
263                Kind::object(Collection::from_unknown(
264                    Kind::bytes().or_null().or_undefined(),
265                ))
266                .or_undefined(),
267                None,
268            );
269
270        vec![SourceOutput::new_maybe_logs(
271            self.decoding.output_type(),
272            schema_definition,
273        )]
274    }
275
276    fn resources(&self) -> Vec<Resource> {
277        vec![Resource::tcp(self.address)]
278    }
279
280    fn can_acknowledge(&self) -> bool {
281        true
282    }
283}
284
285impl GenerateConfig for AwsKinesisFirehoseConfig {
286    fn generate_config() -> toml::Value {
287        toml::Value::try_from(Self {
288            address: "0.0.0.0:443".parse().unwrap(),
289            access_key: None,
290            access_keys: None,
291            store_access_key: false,
292            tls: None,
293            record_compression: Default::default(),
294            framing: default_framing_message_based(),
295            decoding: default_decoding(),
296            acknowledgements: Default::default(),
297            log_namespace: None,
298            keepalive: Default::default(),
299            common_attributes: vec![],
300        })
301        .unwrap()
302    }
303}
304
305#[cfg(test)]
306mod tests {
307    #![allow(clippy::print_stdout)] //tests
308
309    use std::{
310        io::{Cursor, Read},
311        net::SocketAddr,
312        sync::LazyLock,
313    };
314
315    use base64::prelude::{BASE64_STANDARD, Engine as _};
316    use bytes::Bytes;
317    use chrono::{DateTime, SubsecRound, Utc};
318    use flate2::read::GzEncoder;
319    use futures::Stream;
320    use similar_asserts::assert_eq;
321    use vector_lib::{assert_event_data_eq, lookup::path};
322    use vrl::{value, value::KeyString, value::ObjectMap, value::Value};
323
324    use super::*;
325    use crate::{
326        SourceSender,
327        event::{Event, EventStatus},
328        log_event,
329        test_util::{
330            addr::{PortGuard, next_addr},
331            collect_n,
332            components::{SOURCE_TAGS, assert_source_compliance},
333            wait_for_tcp,
334        },
335    };
336
337    const SOURCE_ARN: &str = "arn:aws:firehose:us-east-1:111111111111:deliverystream/test";
338    const REQUEST_ID: &str = "e17265d6-97af-4938-982e-90d5614c4242";
339    // example CloudWatch Logs subscription event
340    const RECORD: &str = r#"
341            {
342                "messageType": "DATA_MESSAGE",
343                "owner": "071959437513",
344                "logGroup": "/jesse/test",
345                "logStream": "test",
346                "subscriptionFilters": ["Destination"],
347                "logEvents": [
348                    {
349                        "id": "35683658089614582423604394983260738922885519999578275840",
350                        "timestamp": 1600110569039,
351                        "message": "{\"bytes\":26780,\"datetime\":\"14/Sep/2020:11:45:41 -0400\",\"host\":\"157.130.216.193\",\"method\":\"PUT\",\"protocol\":\"HTTP/1.0\",\"referer\":\"https://www.principalcross-platform.io/markets/ubiquitous\",\"request\":\"/expedite/convergence\",\"source_type\":\"stdin\",\"status\":301,\"user-identifier\":\"-\"}"
352                    },
353                    {
354                        "id": "35683658089659183914001456229543810359430816722590236673",
355                        "timestamp": 1600110569041,
356                        "message": "{\"bytes\":17707,\"datetime\":\"14/Sep/2020:11:45:41 -0400\",\"host\":\"109.81.244.252\",\"method\":\"GET\",\"protocol\":\"HTTP/2.0\",\"referer\":\"http://www.investormission-critical.io/24/7/vortals\",\"request\":\"/scale/functionalities/optimize\",\"source_type\":\"stdin\",\"status\":502,\"user-identifier\":\"feeney1708\"}"
357                    }
358                ]
359            }
360        "#;
361
362    const COMMON_ATTRIBUTES: &str = r#"{ "commonAttributes": { "environment": "testing", "application_group": "tymur_test" } }"#;
363
364    static COMMON_ATTRIBUTES_MAP: LazyLock<ObjectMap> = LazyLock::new(|| {
365        ObjectMap::from_iter([
366            (
367                KeyString::from("environment"),
368                Value::Bytes("testing".into()),
369            ),
370            (
371                KeyString::from("application_group"),
372                Value::Bytes("tymur_test".into()),
373            ),
374        ])
375    });
376
377    #[test]
378    fn generate_config() {
379        crate::test_util::test_generate_config::<AwsKinesisFirehoseConfig>();
380    }
381
382    async fn source(
383        access_key: Option<SensitiveString>,
384        access_keys: Option<Vec<SensitiveString>>,
385        store_access_key: bool,
386        record_compression: Compression,
387        delivered: bool,
388        log_namespace: bool,
389        common_attributes: Vec<String>,
390    ) -> (impl Stream<Item = Event> + Unpin, SocketAddr, PortGuard) {
391        use EventStatus::*;
392        let status = if delivered { Delivered } else { Rejected };
393        let (sender, recv) = SourceSender::new_test_finalize(status);
394        let (_guard, address) = next_addr();
395        let cx = SourceContext::new_test(sender, None);
396        tokio::spawn(async move {
397            AwsKinesisFirehoseConfig {
398                address,
399                tls: None,
400                access_key,
401                access_keys,
402                store_access_key,
403                record_compression,
404                framing: default_framing_message_based(),
405                decoding: default_decoding(),
406                acknowledgements: true.into(),
407                log_namespace: Some(log_namespace),
408                keepalive: Default::default(),
409                common_attributes,
410            }
411            .build(cx)
412            .await
413            .unwrap()
414            .await
415            .unwrap()
416        });
417        // Wait for the component to bind to the port
418        wait_for_tcp(address).await;
419        (recv, address, _guard)
420    }
421
422    /// Sends the body to the address with the appropriate Firehose headers
423    ///
424    /// https://docs.aws.amazon.com/firehose/latest/dev/httpdeliveryrequestresponse.html
425    async fn send(
426        address: SocketAddr,
427        timestamp: DateTime<Utc>,
428        records: Vec<&[u8]>,
429        key: Option<&str>,
430        gzip: bool,
431        record_compression: Compression,
432        common_attributes: Option<&str>,
433    ) -> reqwest::Result<reqwest::Response> {
434        let request = models::FirehoseRequest {
435            access_key: key.map(|s| s.to_string()),
436            request_id: REQUEST_ID.to_string(),
437            timestamp,
438            records: records
439                .into_iter()
440                .map(|record| models::EncodedFirehoseRecord {
441                    data: encode_record(record, record_compression).unwrap(),
442                })
443                .collect(),
444        };
445
446        let mut builder = reqwest::Client::new()
447            .post(format!("http://{address}"))
448            .header("host", address.to_string())
449            .header(
450                "x-amzn-trace-id",
451                "Root=1-5f5fbf1c-877c68cace58bea222ddbeec",
452            )
453            .header("x-amz-firehose-protocol-version", "1.0")
454            .header("x-amz-firehose-request-id", REQUEST_ID.to_string())
455            .header("x-amz-firehose-source-arn", SOURCE_ARN.to_string())
456            .header("user-agent", "Amazon Kinesis Data Firehose Agent/1.0")
457            .header("content-type", "application/json");
458
459        if let Some(key) = key {
460            builder = builder.header("x-amz-firehose-access-key", key);
461        }
462
463        if let Some(common_attributes) = common_attributes {
464            builder = builder.header("x-amz-firehose-common-attributes", common_attributes)
465        }
466
467        if gzip {
468            let mut gz = GzEncoder::new(
469                Cursor::new(serde_json::to_vec(&request).unwrap()),
470                flate2::Compression::fast(),
471            );
472            let mut buffer = Vec::new();
473            gz.read_to_end(&mut buffer).unwrap();
474            builder = builder.header("content-encoding", "gzip").body(buffer);
475        } else {
476            builder = builder.json(&request);
477        }
478
479        builder.send().await
480    }
481
482    async fn spawn_send(
483        address: SocketAddr,
484        timestamp: DateTime<Utc>,
485        records: Vec<&'static [u8]>,
486        key: Option<&'static str>,
487        gzip: bool,
488        record_compression: Compression,
489        common_attributes: Option<&'static str>,
490    ) -> tokio::task::JoinHandle<reqwest::Result<reqwest::Response>> {
491        tokio::spawn(async move {
492            send(
493                address,
494                timestamp,
495                records,
496                key,
497                gzip,
498                record_compression,
499                common_attributes,
500            )
501            .await
502        })
503    }
504
505    /// Encodes record data to mach AWS's representation: base64 encoded with an additional
506    /// compression
507    fn encode_record(record: &[u8], compression: Compression) -> std::io::Result<String> {
508        let compressed = match compression {
509            Compression::Auto => panic!("cannot encode records as Auto"),
510            Compression::Gzip => {
511                let mut buffer = Vec::new();
512                if !record.is_empty() {
513                    let mut gz = GzEncoder::new(record, flate2::Compression::fast());
514                    gz.read_to_end(&mut buffer)?;
515                }
516                buffer
517            }
518            Compression::None => record.to_vec(),
519        };
520
521        Ok(BASE64_STANDARD.encode(compressed))
522    }
523
524    #[tokio::test]
525    async fn aws_kinesis_firehose_forwards_events_legacy_namespace() {
526        let gzipped_record = {
527            let mut buf = Vec::new();
528            let mut gz = GzEncoder::new(RECORD.as_bytes(), flate2::Compression::fast());
529            gz.read_to_end(&mut buf).unwrap();
530            buf
531        };
532
533        for (source_record_compression, record_compression, success, record, expected) in [
534            (
535                Compression::Auto,
536                Compression::Gzip,
537                true,
538                RECORD.as_bytes(),
539                RECORD.as_bytes().to_owned(),
540            ),
541            (
542                Compression::Auto,
543                Compression::None,
544                true,
545                RECORD.as_bytes(),
546                RECORD.as_bytes().to_owned(),
547            ),
548            (
549                Compression::None,
550                Compression::Gzip,
551                true,
552                RECORD.as_bytes(),
553                gzipped_record,
554            ),
555            (
556                Compression::None,
557                Compression::None,
558                true,
559                RECORD.as_bytes(),
560                RECORD.as_bytes().to_owned(),
561            ),
562            (
563                Compression::Gzip,
564                Compression::Gzip,
565                true,
566                RECORD.as_bytes(),
567                RECORD.as_bytes().to_owned(),
568            ),
569            (
570                Compression::Gzip,
571                Compression::None,
572                false,
573                RECORD.as_bytes(),
574                RECORD.as_bytes().to_owned(),
575            ),
576            (
577                Compression::Gzip,
578                Compression::Gzip,
579                true,
580                "".as_bytes(),
581                Vec::new(),
582            ),
583        ] {
584            let (rx, addr, _guard) = source(
585                None,
586                None,
587                false,
588                source_record_compression,
589                true,
590                false,
591                vec![],
592            )
593            .await;
594
595            let timestamp: DateTime<Utc> = Utc::now();
596
597            let res = spawn_send(
598                addr,
599                timestamp,
600                vec![record],
601                None,
602                false,
603                record_compression,
604                None,
605            )
606            .await;
607
608            if success {
609                let events = collect_n(rx, 1).await;
610
611                let res = res.await.unwrap().unwrap();
612                assert_eq!(200, res.status().as_u16());
613
614                assert_event_data_eq!(
615                    events,
616                    vec![log_event! {
617                        "source_type" => Bytes::from("aws_kinesis_firehose"),
618                        "timestamp" => timestamp.trunc_subsecs(3), // AWS sends timestamps as ms
619                        "message" => Bytes::from(expected),
620                        "request_id" => REQUEST_ID,
621                        "source_arn" => SOURCE_ARN,
622                    },]
623                );
624
625                let response: models::FirehoseResponse = res.json().await.unwrap();
626                assert_eq!(response.request_id, REQUEST_ID);
627            } else {
628                let res = res.await.unwrap().unwrap();
629                assert_eq!(400, res.status().as_u16());
630            }
631        }
632    }
633
634    #[tokio::test]
635    async fn aws_kinesis_firehose_forwards_events_vector_namespace() {
636        let gzipped_record = {
637            let mut buf = Vec::new();
638            let mut gz = GzEncoder::new(RECORD.as_bytes(), flate2::Compression::fast());
639            gz.read_to_end(&mut buf).unwrap();
640            buf
641        };
642
643        for (source_record_compression, record_compression, success, record, expected) in [
644            (
645                Compression::Auto,
646                Compression::Gzip,
647                true,
648                RECORD.as_bytes(),
649                RECORD.as_bytes().to_owned(),
650            ),
651            (
652                Compression::Auto,
653                Compression::None,
654                true,
655                RECORD.as_bytes(),
656                RECORD.as_bytes().to_owned(),
657            ),
658            (
659                Compression::None,
660                Compression::Gzip,
661                true,
662                RECORD.as_bytes(),
663                gzipped_record,
664            ),
665            (
666                Compression::None,
667                Compression::None,
668                true,
669                RECORD.as_bytes(),
670                RECORD.as_bytes().to_owned(),
671            ),
672            (
673                Compression::Gzip,
674                Compression::Gzip,
675                true,
676                RECORD.as_bytes(),
677                RECORD.as_bytes().to_owned(),
678            ),
679            (
680                Compression::Gzip,
681                Compression::None,
682                false,
683                RECORD.as_bytes(),
684                RECORD.as_bytes().to_owned(),
685            ),
686            (
687                Compression::Gzip,
688                Compression::Gzip,
689                true,
690                "".as_bytes(),
691                Vec::new(),
692            ),
693        ] {
694            let (rx, addr, _guard) = source(
695                None,
696                None,
697                false,
698                source_record_compression,
699                true,
700                true,
701                vec![],
702            )
703            .await;
704
705            let timestamp: DateTime<Utc> = Utc::now();
706
707            let res = spawn_send(
708                addr,
709                timestamp,
710                vec![record],
711                None,
712                false,
713                record_compression,
714                None,
715            )
716            .await;
717
718            if success {
719                let events = collect_n(rx, 1).await;
720
721                let res = res.await.unwrap().unwrap();
722                assert_eq!(200, res.status().as_u16());
723
724                for event in events {
725                    let log = event.as_log();
726                    let meta = log.metadata();
727
728                    // event data, currently assumes default bytes deserializer
729                    assert_eq!(log.value(), &value!(Bytes::from(expected.to_owned())));
730
731                    // vector metadata
732                    assert_eq!(
733                        meta.value().get(path!("vector", "source_type")).unwrap(),
734                        &value!("aws_kinesis_firehose")
735                    );
736                    assert!(
737                        meta.value()
738                            .get(path!("vector", "ingest_timestamp"))
739                            .unwrap()
740                            .is_timestamp()
741                    );
742
743                    // source metadata
744                    assert_eq!(
745                        meta.value()
746                            .get(path!("aws_kinesis_firehose", "request_id"))
747                            .unwrap(),
748                        &value!(REQUEST_ID)
749                    );
750                    assert_eq!(
751                        meta.value()
752                            .get(path!("aws_kinesis_firehose", "source_arn"))
753                            .unwrap(),
754                        &value!(SOURCE_ARN)
755                    );
756                    assert_eq!(
757                        meta.value()
758                            .get(path!("aws_kinesis_firehose", "timestamp"))
759                            .unwrap(),
760                        &value!(timestamp.trunc_subsecs(3))
761                    );
762                    assert!(
763                        meta.value()
764                            .get(path!("aws_kinesis_firehose", "common_attributes"))
765                            .is_none()
766                    );
767                }
768
769                let response: models::FirehoseResponse = res.json().await.unwrap();
770                assert_eq!(response.request_id, REQUEST_ID);
771            } else {
772                let res = res.await.unwrap().unwrap();
773                assert_eq!(400, res.status().as_u16());
774            }
775        }
776    }
777
778    #[tokio::test]
779    async fn aws_kinesis_firehose_forwards_events_gzip_request() {
780        assert_source_compliance(&SOURCE_TAGS, async move {
781            let (rx, addr, _guard) =
782                source(None, None, false, Default::default(), true, false, vec![]).await;
783
784            let timestamp: DateTime<Utc> = Utc::now();
785
786            let res = spawn_send(
787                addr,
788                timestamp,
789                vec![RECORD.as_bytes()],
790                None,
791                true,
792                Compression::None,
793                None,
794            )
795            .await;
796
797            let events = collect_n(rx, 1).await;
798            let res = res.await.unwrap().unwrap();
799            assert_eq!(200, res.status().as_u16());
800
801            assert_event_data_eq!(
802                events,
803                vec![log_event! {
804                    "source_type" => Bytes::from("aws_kinesis_firehose"),
805                    "timestamp" => timestamp.trunc_subsecs(3), // AWS sends timestamps as ms
806                    "message"=> RECORD,
807                    "request_id" => REQUEST_ID,
808                    "source_arn" => SOURCE_ARN,
809                },]
810            );
811
812            let response: models::FirehoseResponse = res.json().await.unwrap();
813            assert_eq!(response.request_id, REQUEST_ID);
814        })
815        .await;
816    }
817
818    #[tokio::test]
819    async fn aws_kinesis_firehose_forwards_events_wildcard_common_attributes_legacy_namespace() {
820        assert_source_compliance(&SOURCE_TAGS, async move {
821            let (rx, addr, _guard) = source(
822                None,
823                None,
824                false,
825                Default::default(),
826                true,
827                false,
828                vec!["*".to_string()],
829            )
830            .await;
831
832            let timestamp: DateTime<Utc> = Utc::now();
833
834            let res = spawn_send(
835                addr,
836                timestamp,
837                vec![RECORD.as_bytes()],
838                None,
839                true,
840                Compression::None,
841                Some(COMMON_ATTRIBUTES),
842            )
843            .await;
844
845            let events = collect_n(rx, 1).await;
846            let res = res.await.unwrap().unwrap();
847            assert_eq!(200, res.status().as_u16());
848
849            assert_event_data_eq!(
850                events,
851                vec![log_event! {
852                    "source_type" => Bytes::from("aws_kinesis_firehose"),
853                    "timestamp" => timestamp.trunc_subsecs(3), // AWS sends timestamps as ms
854                    "message"=> RECORD,
855                    "request_id" => REQUEST_ID,
856                    "source_arn" => SOURCE_ARN,
857                    "common_attributes" => COMMON_ATTRIBUTES_MAP.clone(),
858                },]
859            );
860
861            let response: models::FirehoseResponse = res.json().await.unwrap();
862            assert_eq!(response.request_id, REQUEST_ID);
863        })
864        .await;
865    }
866
867    #[tokio::test]
868    async fn aws_kinesis_firehose_forwards_events_wildcard_common_attributes_vector_namespace() {
869        assert_source_compliance(&SOURCE_TAGS, async move {
870            let (rx, addr, _guard) = source(
871                None,
872                None,
873                false,
874                Default::default(),
875                true,
876                true,
877                vec!["*".to_string()],
878            )
879            .await;
880
881            let timestamp: DateTime<Utc> = Utc::now();
882
883            let res = spawn_send(
884                addr,
885                timestamp,
886                vec![RECORD.as_bytes()],
887                None,
888                true,
889                Compression::None,
890                Some(COMMON_ATTRIBUTES),
891            )
892            .await;
893
894            let mut events = collect_n(rx, 1).await;
895            let res = res.await.unwrap().unwrap();
896            assert_eq!(200, res.status().as_u16());
897
898            let event = events.pop().unwrap();
899            let log = event.as_log();
900            let meta = log.metadata();
901
902            // event data, currently assumes default bytes deserializer
903            assert_eq!(log.value(), &value!(Bytes::from(RECORD.to_owned())));
904
905            // vector metadata
906            assert_eq!(
907                meta.value().get(path!("vector", "source_type")).unwrap(),
908                &value!("aws_kinesis_firehose")
909            );
910            assert!(
911                meta.value()
912                    .get(path!("vector", "ingest_timestamp"))
913                    .unwrap()
914                    .is_timestamp()
915            );
916
917            // source metadata
918            assert_eq!(
919                meta.value()
920                    .get(path!("aws_kinesis_firehose", "request_id"))
921                    .unwrap(),
922                &value!(REQUEST_ID)
923            );
924            assert_eq!(
925                meta.value()
926                    .get(path!("aws_kinesis_firehose", "source_arn"))
927                    .unwrap(),
928                &value!(SOURCE_ARN)
929            );
930            assert_eq!(
931                meta.value()
932                    .get(path!("aws_kinesis_firehose", "timestamp"))
933                    .unwrap(),
934                &value!(timestamp.trunc_subsecs(3))
935            );
936            assert_eq!(
937                meta.value()
938                    .get(path!("aws_kinesis_firehose", "common_attributes"))
939                    .unwrap(),
940                &value!(COMMON_ATTRIBUTES_MAP.clone())
941            );
942
943            let response: models::FirehoseResponse = res.json().await.unwrap();
944            assert_eq!(response.request_id, REQUEST_ID);
945        })
946        .await;
947    }
948
949    #[tokio::test]
950    async fn aws_kinesis_firehose_forwards_events_common_attributes_legacy_namespace() {
951        assert_source_compliance(&SOURCE_TAGS, async move {
952            let mut expected_common_attributes = ObjectMap::new();
953            expected_common_attributes.insert(
954                KeyString::from("environment"),
955                COMMON_ATTRIBUTES_MAP["environment"].clone(),
956            );
957            expected_common_attributes.insert(KeyString::from("absent_attribute"), Value::Null);
958
959            let (rx, addr, _guard) = source(
960                None,
961                None,
962                false,
963                Default::default(),
964                true,
965                false,
966                vec!["environment".to_string(), "absent_attribute".to_string()],
967            )
968            .await;
969
970            let timestamp: DateTime<Utc> = Utc::now();
971
972            let res = spawn_send(
973                addr,
974                timestamp,
975                vec![RECORD.as_bytes()],
976                None,
977                true,
978                Compression::None,
979                Some(COMMON_ATTRIBUTES),
980            )
981            .await;
982
983            let events = collect_n(rx, 1).await;
984            let res = res.await.unwrap().unwrap();
985            assert_eq!(200, res.status().as_u16());
986
987            assert_event_data_eq!(
988                events,
989                vec![log_event! {
990                    "source_type" => Bytes::from("aws_kinesis_firehose"),
991                    "timestamp" => timestamp.trunc_subsecs(3), // AWS sends timestamps as ms
992                    "message"=> RECORD,
993                    "request_id" => REQUEST_ID,
994                    "source_arn" => SOURCE_ARN,
995                    "common_attributes" => expected_common_attributes,
996                },]
997            );
998
999            let response: models::FirehoseResponse = res.json().await.unwrap();
1000            assert_eq!(response.request_id, REQUEST_ID);
1001        })
1002        .await;
1003    }
1004
1005    #[tokio::test]
1006    async fn aws_kinesis_firehose_forwards_events_common_attributes_vector_namespace() {
1007        assert_source_compliance(&SOURCE_TAGS, async move {
1008            let mut expected_common_attributes = ObjectMap::new();
1009            expected_common_attributes.insert(
1010                KeyString::from("environment"),
1011                COMMON_ATTRIBUTES_MAP["environment"].clone(),
1012            );
1013            expected_common_attributes.insert(KeyString::from("absent_attribute"), Value::Null);
1014
1015            let (rx, addr, _guard) = source(
1016                None,
1017                None,
1018                false,
1019                Default::default(),
1020                true,
1021                true,
1022                vec!["environment".to_string(), "absent_attribute".to_string()],
1023            )
1024            .await;
1025
1026            let timestamp: DateTime<Utc> = Utc::now();
1027
1028            let res = spawn_send(
1029                addr,
1030                timestamp,
1031                vec![RECORD.as_bytes()],
1032                None,
1033                true,
1034                Compression::None,
1035                Some(COMMON_ATTRIBUTES),
1036            )
1037            .await;
1038
1039            let mut events = collect_n(rx, 1).await;
1040            let res = res.await.unwrap().unwrap();
1041            assert_eq!(200, res.status().as_u16());
1042
1043            let event = events.pop().unwrap();
1044            let log = event.as_log();
1045            let meta = log.metadata();
1046
1047            // event data, currently assumes default bytes deserializer
1048            assert_eq!(log.value(), &value!(Bytes::from(RECORD.to_owned())));
1049
1050            // vector metadata
1051            assert_eq!(
1052                meta.value().get(path!("vector", "source_type")).unwrap(),
1053                &value!("aws_kinesis_firehose")
1054            );
1055            assert!(
1056                meta.value()
1057                    .get(path!("vector", "ingest_timestamp"))
1058                    .unwrap()
1059                    .is_timestamp()
1060            );
1061
1062            // source metadata
1063            assert_eq!(
1064                meta.value()
1065                    .get(path!("aws_kinesis_firehose", "request_id"))
1066                    .unwrap(),
1067                &value!(REQUEST_ID)
1068            );
1069            assert_eq!(
1070                meta.value()
1071                    .get(path!("aws_kinesis_firehose", "source_arn"))
1072                    .unwrap(),
1073                &value!(SOURCE_ARN)
1074            );
1075            assert_eq!(
1076                meta.value()
1077                    .get(path!("aws_kinesis_firehose", "timestamp"))
1078                    .unwrap(),
1079                &value!(timestamp.trunc_subsecs(3))
1080            );
1081            assert_eq!(
1082                meta.value()
1083                    .get(path!("aws_kinesis_firehose", "common_attributes"))
1084                    .unwrap(),
1085                &value!(expected_common_attributes)
1086            );
1087
1088            let response: models::FirehoseResponse = res.json().await.unwrap();
1089            assert_eq!(response.request_id, REQUEST_ID);
1090        })
1091        .await;
1092    }
1093
1094    // Test there is no regression for existing setups and non-AWS test senders that previously
1095    // ignored X-Amz-Firehose-Common-Attributes header after firehose common attributes were introduced
1096    // (https://github.com/vectordotdev/vector/pull/24914#discussion_r3024341032)
1097    #[tokio::test]
1098    async fn aws_kinesis_firehose_ignores_malformed_common_attributes_if_none_configured() {
1099        assert_source_compliance(&SOURCE_TAGS, async move {
1100            let (rx, addr, _guard) =
1101                source(None, None, false, Default::default(), true, true, vec![]).await;
1102
1103            let timestamp: DateTime<Utc> = Utc::now();
1104
1105            let res = spawn_send(
1106                addr,
1107                timestamp,
1108                vec![RECORD.as_bytes()],
1109                None,
1110                true,
1111                Compression::None,
1112                Some("malformed-common-attributes"),
1113            )
1114            .await;
1115
1116            let mut events = collect_n(rx, 1).await;
1117            let res = res.await.unwrap().unwrap();
1118            assert_eq!(200, res.status().as_u16());
1119
1120            let event = events.pop().unwrap();
1121            let log = event.as_log();
1122            let meta = log.metadata();
1123
1124            // event data, currently assumes default bytes deserializer
1125            assert_eq!(log.value(), &value!(Bytes::from(RECORD.to_owned())));
1126
1127            // vector metadata
1128            assert_eq!(
1129                meta.value().get(path!("vector", "source_type")).unwrap(),
1130                &value!("aws_kinesis_firehose")
1131            );
1132            assert!(
1133                meta.value()
1134                    .get(path!("vector", "ingest_timestamp"))
1135                    .unwrap()
1136                    .is_timestamp()
1137            );
1138
1139            // source metadata
1140            assert_eq!(
1141                meta.value()
1142                    .get(path!("aws_kinesis_firehose", "request_id"))
1143                    .unwrap(),
1144                &value!(REQUEST_ID)
1145            );
1146            assert_eq!(
1147                meta.value()
1148                    .get(path!("aws_kinesis_firehose", "source_arn"))
1149                    .unwrap(),
1150                &value!(SOURCE_ARN)
1151            );
1152            assert_eq!(
1153                meta.value()
1154                    .get(path!("aws_kinesis_firehose", "timestamp"))
1155                    .unwrap(),
1156                &value!(timestamp.trunc_subsecs(3))
1157            );
1158            assert!(
1159                meta.value()
1160                    .get(path!("aws_kinesis_firehose", "common_attributes"))
1161                    .is_none()
1162            );
1163
1164            let response: models::FirehoseResponse = res.json().await.unwrap();
1165            assert_eq!(response.request_id, REQUEST_ID);
1166        })
1167        .await;
1168    }
1169
1170    #[tokio::test]
1171    async fn aws_kinesis_firehose_rejects_bad_access_key() {
1172        let (_rx, addr, _guard) = source(
1173            Some("an access key".to_string().into()),
1174            Some(vec!["an access key in list".to_string().into()]),
1175            Default::default(),
1176            Default::default(),
1177            true,
1178            false,
1179            vec![],
1180        )
1181        .await;
1182
1183        let res = send(
1184            addr,
1185            Utc::now(),
1186            vec![],
1187            Some("bad access key"),
1188            false,
1189            Compression::None,
1190            None,
1191        )
1192        .await
1193        .unwrap();
1194        assert_eq!(401, res.status().as_u16());
1195
1196        let response: models::FirehoseResponse = res.json().await.unwrap();
1197        assert_eq!(response.request_id, REQUEST_ID);
1198    }
1199
1200    #[tokio::test]
1201    async fn aws_kinesis_firehose_rejects_bad_access_key_from_list() {
1202        let (_rx, addr, _guard) = source(
1203            None,
1204            Some(vec!["an access key in list".to_string().into()]),
1205            Default::default(),
1206            Default::default(),
1207            true,
1208            false,
1209            vec![],
1210        )
1211        .await;
1212
1213        let res = send(
1214            addr,
1215            Utc::now(),
1216            vec![],
1217            Some("bad access key"),
1218            false,
1219            Compression::None,
1220            None,
1221        )
1222        .await
1223        .unwrap();
1224        assert_eq!(401, res.status().as_u16());
1225
1226        let response: models::FirehoseResponse = res.json().await.unwrap();
1227        assert_eq!(response.request_id, REQUEST_ID);
1228    }
1229
1230    #[tokio::test]
1231    async fn aws_kinesis_firehose_accepts_merged_access_keys() {
1232        let valid_access_key = SensitiveString::from(String::from("an access key in list"));
1233
1234        let (_rx, addr, _guard) = source(
1235            Some(valid_access_key.clone()),
1236            Some(vec!["valid access key 2".to_string().into()]),
1237            Default::default(),
1238            Default::default(),
1239            true,
1240            false,
1241            vec![],
1242        )
1243        .await;
1244
1245        let res = send(
1246            addr,
1247            Utc::now(),
1248            vec![],
1249            Some(valid_access_key.clone().inner()),
1250            false,
1251            Compression::None,
1252            None,
1253        )
1254        .await
1255        .unwrap();
1256
1257        assert_eq!(200, res.status().as_u16());
1258
1259        let response: models::FirehoseResponse = res.json().await.unwrap();
1260        assert_eq!(response.request_id, REQUEST_ID);
1261    }
1262
1263    #[tokio::test]
1264    async fn aws_kinesis_firehose_accepts_access_keys_from_list() {
1265        let valid_access_key = "an access key in list".to_string();
1266
1267        let (_rx, addr, _guard) = source(
1268            None,
1269            Some(vec![
1270                valid_access_key.clone().into(),
1271                "valid access key 2".to_string().into(),
1272            ]),
1273            Default::default(),
1274            Default::default(),
1275            true,
1276            false,
1277            vec![],
1278        )
1279        .await;
1280
1281        let res = send(
1282            addr,
1283            Utc::now(),
1284            vec![],
1285            Some(&valid_access_key),
1286            false,
1287            Compression::None,
1288            None,
1289        )
1290        .await
1291        .unwrap();
1292
1293        assert_eq!(200, res.status().as_u16());
1294
1295        let response: models::FirehoseResponse = res.json().await.unwrap();
1296        assert_eq!(response.request_id, REQUEST_ID);
1297    }
1298
1299    #[tokio::test]
1300    async fn handles_acknowledgement_failure() {
1301        let expected = RECORD.as_bytes().to_owned();
1302
1303        let (rx, addr, _guard) =
1304            source(None, None, false, Compression::None, false, false, vec![]).await;
1305
1306        let timestamp: DateTime<Utc> = Utc::now();
1307
1308        let res = spawn_send(
1309            addr,
1310            timestamp,
1311            vec![RECORD.as_bytes()],
1312            None,
1313            false,
1314            Compression::None,
1315            None,
1316        )
1317        .await;
1318
1319        let events = collect_n(rx, 1).await;
1320
1321        let res = res.await.unwrap().unwrap();
1322        assert_eq!(406, res.status().as_u16());
1323
1324        assert_event_data_eq!(
1325            events,
1326            vec![log_event! {
1327                "source_type" => Bytes::from("aws_kinesis_firehose"),
1328                "timestamp" => timestamp.trunc_subsecs(3), // AWS sends timestamps as ms
1329                "message"=> Bytes::from(expected),
1330                "request_id" => REQUEST_ID,
1331                "source_arn" => SOURCE_ARN,
1332            },]
1333        );
1334
1335        let response: models::FirehoseResponse = res.json().await.unwrap();
1336        assert_eq!(response.request_id, REQUEST_ID);
1337    }
1338
1339    #[tokio::test]
1340    async fn event_access_key_passthrough_enabled() {
1341        let (rx, address, _guard) = source(
1342            None,
1343            Some(vec!["an access key".to_string().into()]),
1344            true,
1345            Default::default(),
1346            true,
1347            true,
1348            vec![],
1349        )
1350        .await;
1351
1352        let timestamp: DateTime<Utc> = Utc::now();
1353
1354        spawn_send(
1355            address,
1356            timestamp,
1357            vec![RECORD.as_bytes()],
1358            Some("an access key"),
1359            false,
1360            Compression::None,
1361            None,
1362        )
1363        .await;
1364
1365        let events = collect_n(rx, 1).await;
1366        let access_key = events[0]
1367            .metadata()
1368            .secrets()
1369            .get("aws_kinesis_firehose_access_key")
1370            .unwrap();
1371        assert_eq!(access_key.to_string(), "an access key".to_string());
1372    }
1373
1374    #[tokio::test]
1375    async fn no_authorization_access_key_passthrough_enabled() {
1376        let (rx, address, _guard) =
1377            source(None, None, true, Default::default(), true, true, vec![]).await;
1378
1379        let timestamp: DateTime<Utc> = Utc::now();
1380
1381        spawn_send(
1382            address,
1383            timestamp,
1384            vec![RECORD.as_bytes()],
1385            None,
1386            false,
1387            Compression::None,
1388            None,
1389        )
1390        .await;
1391
1392        let events = collect_n(rx, 1).await;
1393
1394        assert!(
1395            events[0]
1396                .metadata()
1397                .secrets()
1398                .get("aws_kinesis_firehose_access_key")
1399                .is_none()
1400        );
1401    }
1402}
vector/sources/aws_kinesis_firehose/mod.rs

vector/sources/aws_kinesis_firehose/
mod.rs
