vector/sinks/gcp/
cloud_storage.rs

1use std::{collections::HashMap, convert::TryFrom, io};
2
3use bytes::Bytes;
4use chrono::{FixedOffset, Utc};
5use http::{
6    Uri,
7    header::{HeaderName, HeaderValue},
8};
9use indoc::indoc;
10use snafu::{ResultExt, Snafu};
11use tower::ServiceBuilder;
12use uuid::Uuid;
13use vector_lib::{
14    TimeZone,
15    codecs::encoding::Framer,
16    configurable::configurable_component,
17    event::{EventFinalizers, Finalizable},
18    request_metadata::RequestMetadata,
19};
20
21use crate::{
22    codecs::{Encoder, EncodingConfigWithFraming, SinkType, Transformer},
23    config::{AcknowledgementsConfig, DataType, GenerateConfig, Input, SinkConfig, SinkContext},
24    event::Event,
25    gcp::{GcpAuthConfig, GcpAuthenticator, Scope},
26    http::{HttpClient, get_http_scheme_from_uri},
27    serde::json::to_string,
28    sinks::{
29        Healthcheck, VectorSink,
30        gcs_common::{
31            config::{
32                GcsPredefinedAcl, GcsRetryLogic, GcsStorageClass, build_healthcheck,
33                default_endpoint,
34            },
35            service::{GcsRequest, GcsRequestSettings, GcsService},
36            sink::GcsSink,
37        },
38        util::{
39            BulkSizeBasedDefaultBatchSettings, Compression, RequestBuilder, ServiceBuilderExt,
40            TowerRequestConfig, batch::BatchConfig, metadata::RequestMetadataBuilder,
41            partitioner::KeyPartitioner, request_builder::EncodeResult,
42            service::TowerRequestConfigDefaults, timezone_to_offset,
43        },
44    },
45    template::{ConfinementConfig, Template, TemplateParseError},
46    tls::{TlsConfig, TlsSettings},
47};
48
49#[derive(Debug, Snafu)]
50#[snafu(visibility(pub))]
51pub enum GcsHealthcheckError {
52    #[snafu(display("key_prefix template parse error: {}", source))]
53    KeyPrefixTemplate { source: TemplateParseError },
54}
55
56#[derive(Clone, Copy, Debug)]
57pub struct GcsTowerRequestConfigDefaults;
58
59impl TowerRequestConfigDefaults for GcsTowerRequestConfigDefaults {
60    const RATE_LIMIT_NUM: u64 = 1_000;
61}
62
63/// Configuration for the `gcp_cloud_storage` sink.
64#[configurable_component(sink(
65    "gcp_cloud_storage",
66    "Store observability events in GCP Cloud Storage."
67))]
68#[derive(Clone, Debug)]
69#[serde(deny_unknown_fields)]
70pub struct GcsSinkConfig {
71    /// The GCS bucket name.
72    #[configurable(metadata(docs::examples = "my-bucket"))]
73    bucket: String,
74
75    /// The Predefined ACL to apply to created objects.
76    ///
77    /// For more information, see [Predefined ACLs][predefined_acls].
78    ///
79    /// [predefined_acls]: https://cloud.google.com/storage/docs/access-control/lists#predefined-acl
80    acl: Option<GcsPredefinedAcl>,
81
82    /// The storage class for created objects.
83    ///
84    /// For more information, see the [storage classes][storage_classes] documentation.
85    ///
86    /// [storage_classes]: https://cloud.google.com/storage/docs/storage-classes
87    storage_class: Option<GcsStorageClass>,
88
89    /// The set of metadata `key:value` pairs for the created objects.
90    ///
91    /// For more information, see the [custom metadata][custom_metadata] documentation.
92    ///
93    /// [custom_metadata]: https://cloud.google.com/storage/docs/metadata#custom-metadata
94    #[configurable(metadata(docs::additional_props_description = "A key/value pair."))]
95    #[configurable(metadata(docs::advanced))]
96    metadata: Option<HashMap<String, String>>,
97
98    /// A prefix to apply to all object keys.
99    ///
100    /// Prefixes are useful for partitioning objects, such as by creating an object key that
101    /// stores objects under a particular directory. If using a prefix for this purpose, it must end
102    /// in `/` in order to act as a directory path. A trailing `/` is **not** automatically added.
103    #[configurable(metadata(docs::templateable))]
104    #[configurable(metadata(
105        docs::examples = "date=%F/",
106        docs::examples = "date=%F/hour=%H/",
107        docs::examples = "year=%Y/month=%m/day=%d/",
108        docs::examples = "application_id={{ application_id }}/date=%F/"
109    ))]
110    #[configurable(metadata(docs::advanced))]
111    key_prefix: Option<String>,
112
113    /// The timestamp format for the time component of the object key.
114    ///
115    /// By default, object keys are appended with a timestamp that reflects when the objects are
116    /// sent to S3, such that the resulting object key is functionally equivalent to joining the key
117    /// prefix with the formatted timestamp, such as `date=2022-07-18/1658176486`.
118    ///
119    /// This would represent a `key_prefix` set to `date=%F/` and the timestamp of Mon Jul 18 2022
120    /// 20:34:44 GMT+0000, with the `filename_time_format` being set to `%s`, which renders
121    /// timestamps in seconds since the Unix epoch.
122    ///
123    /// Supports the common [`strftime`][chrono_strftime_specifiers] specifiers found in most
124    /// languages.
125    ///
126    /// When set to an empty string, no timestamp is appended to the key prefix.
127    ///
128    /// [chrono_strftime_specifiers]: https://docs.rs/chrono/latest/chrono/format/strftime/index.html#specifiers
129    #[serde(default = "default_time_format")]
130    #[configurable(metadata(docs::advanced))]
131    filename_time_format: String,
132
133    /// Whether or not to append a UUID v4 token to the end of the object key.
134    ///
135    /// The UUID is appended to the timestamp portion of the object key, such that if the object key
136    /// generated is `date=2022-07-18/1658176486`, setting this field to `true` results
137    /// in an object key that looks like `date=2022-07-18/1658176486-30f6652c-71da-4f9f-800d-a1189c47c547`.
138    ///
139    /// This ensures there are no name collisions, and can be useful in high-volume workloads where
140    /// object keys must be unique.
141    #[serde(default = "crate::serde::default_true")]
142    #[configurable(metadata(docs::advanced))]
143    filename_append_uuid: bool,
144
145    /// The filename extension to use in the object key.
146    ///
147    /// If not specified, the extension is determined by the compression scheme used.
148    #[configurable(metadata(docs::advanced))]
149    filename_extension: Option<String>,
150
151    #[serde(flatten)]
152    encoding: EncodingConfigWithFraming,
153
154    /// Compression configuration.
155    ///
156    /// All compression algorithms use the default compression level unless otherwise specified.
157    ///
158    /// Some cloud storage API clients and browsers handle decompression transparently, so
159    /// depending on how they are accessed, files may not always appear to be compressed.
160    #[configurable(derived)]
161    #[serde(default)]
162    compression: Compression,
163
164    /// Overrides the MIME type of the created objects.
165    ///
166    /// Directly comparable to the `Content-Type` HTTP header.
167    ///
168    /// If not specified, defaults to the encoder's content type.
169    #[configurable(metadata(
170        docs::examples = "text/plain; charset=utf-8",
171        docs::examples = "application/gzip"
172    ))]
173    content_type: Option<String>,
174
175    /// Overrides what content encoding has been applied to the object.
176    ///
177    /// Directly comparable to the `Content-Encoding` HTTP header.
178    ///
179    /// If not specified, the compression scheme used dictates this value.
180    #[configurable(metadata(docs::examples = "gzip", docs::examples = "zstd"))]
181    content_encoding: Option<String>,
182
183    /// Sets the `Cache-Control` header for the created objects.
184    ///
185    /// Directly comparable to the `Cache-Control` HTTP header.
186    #[configurable(metadata(docs::examples = "no-transform"))]
187    cache_control: Option<String>,
188
189    #[configurable(derived)]
190    #[serde(default)]
191    batch: BatchConfig<BulkSizeBasedDefaultBatchSettings>,
192
193    /// API endpoint for Google Cloud Storage
194    #[configurable(metadata(docs::examples = "http://localhost:9000"))]
195    #[configurable(validation(format = "uri"))]
196    #[serde(default = "default_endpoint")]
197    endpoint: String,
198
199    #[configurable(derived)]
200    #[serde(default)]
201    request: TowerRequestConfig<GcsTowerRequestConfigDefaults>,
202
203    #[serde(flatten)]
204    auth: GcpAuthConfig,
205
206    #[configurable(derived)]
207    tls: Option<TlsConfig>,
208
209    #[configurable(derived)]
210    #[serde(
211        default,
212        deserialize_with = "crate::serde::bool_or_struct",
213        skip_serializing_if = "crate::serde::is_default"
214    )]
215    acknowledgements: AcknowledgementsConfig,
216
217    #[configurable(derived)]
218    #[serde(default)]
219    pub timezone: Option<TimeZone>,
220
221    #[serde(flatten)]
222    pub confinement: ConfinementConfig,
223}
224
225fn default_time_format() -> String {
226    "%s".to_string()
227}
228
229#[cfg(test)]
230fn default_config(encoding: EncodingConfigWithFraming) -> GcsSinkConfig {
231    GcsSinkConfig {
232        bucket: Default::default(),
233        acl: Default::default(),
234        storage_class: Default::default(),
235        metadata: Default::default(),
236        key_prefix: Default::default(),
237        filename_time_format: default_time_format(),
238        filename_append_uuid: true,
239        filename_extension: Default::default(),
240        content_type: Default::default(),
241        content_encoding: Default::default(),
242        cache_control: Default::default(),
243        encoding,
244        compression: Compression::gzip_default(),
245        batch: Default::default(),
246        endpoint: Default::default(),
247        request: Default::default(),
248        auth: Default::default(),
249        tls: Default::default(),
250        acknowledgements: Default::default(),
251        timezone: Default::default(),
252        confinement: ConfinementConfig::default(),
253    }
254}
255
256impl GenerateConfig for GcsSinkConfig {
257    fn generate_config() -> toml::Value {
258        toml::from_str(indoc! {r#"
259            bucket = "my-bucket"
260            credentials_path = "/path/to/credentials.json"
261            framing.method = "newline_delimited"
262            encoding.codec = "json"
263        "#})
264        .unwrap()
265    }
266}
267
268#[async_trait::async_trait]
269#[typetag::serde(name = "gcp_cloud_storage")]
270impl SinkConfig for GcsSinkConfig {
271    async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
272        let auth = self.auth.build(Scope::DevStorageReadWrite).await?;
273        let base_url = format!("{}/{}/", self.endpoint, self.bucket);
274        let tls = TlsSettings::from_options(self.tls.as_ref())?;
275        let client = HttpClient::new(tls, cx.proxy())?;
276        let healthcheck = build_healthcheck(
277            self.bucket.clone(),
278            client.clone(),
279            base_url.clone(),
280            auth.clone(),
281        )?;
282        auth.spawn_regenerate_token();
283        let sink = self.build_sink(client, base_url, auth, cx)?;
284
285        self.confinement.set_confinement_gauge("sink", Self::NAME);
286        Ok((sink, healthcheck))
287    }
288
289    fn input(&self) -> Input {
290        Input::new(self.encoding.config().1.input_type() & DataType::Log)
291    }
292
293    fn acknowledgements(&self) -> &AcknowledgementsConfig {
294        &self.acknowledgements
295    }
296}
297
298impl GcsSinkConfig {
299    fn build_sink(
300        &self,
301        client: HttpClient,
302        base_url: String,
303        auth: GcpAuthenticator,
304        cx: SinkContext,
305    ) -> crate::Result<VectorSink> {
306        let request = self.request.into_settings();
307
308        let batch_settings = self.batch.into_batcher_settings()?;
309
310        let partitioner = self.key_partitioner()?;
311
312        let protocol = get_http_scheme_from_uri(&base_url.parse::<Uri>().unwrap());
313
314        let svc = ServiceBuilder::new()
315            .settings(request, GcsRetryLogic::default())
316            .service(GcsService::new(client, base_url, auth));
317
318        let request_settings = RequestSettings::new(self, cx)?;
319
320        let sink = GcsSink::new(svc, request_settings, partitioner, batch_settings, protocol);
321
322        Ok(VectorSink::from_event_streamsink(sink))
323    }
324
325    fn key_partitioner(&self) -> crate::Result<KeyPartitioner> {
326        let tpl = Template::try_from(self.key_prefix.as_deref().unwrap_or("date=%F/"))
327            .context(KeyPrefixTemplateSnafu)?;
328        let tpl = tpl.confine(&self.confinement, Self::NAME, "key_prefix")?;
329        Ok(KeyPartitioner::new(tpl, None))
330    }
331}
332
333// Settings required to produce a request that do not change per
334// request. All possible values are pre-computed for direct use in
335// producing a request.
336#[derive(Clone, Debug)]
337struct RequestSettings {
338    acl: Option<HeaderValue>,
339    content_type: HeaderValue,
340    content_encoding: Option<HeaderValue>,
341    storage_class: HeaderValue,
342    cache_control: Option<HeaderValue>,
343    headers: Vec<(HeaderName, HeaderValue)>,
344    extension: String,
345    time_format: String,
346    append_uuid: bool,
347    encoder: (Transformer, Encoder<Framer>),
348    compression: Compression,
349    tz_offset: Option<FixedOffset>,
350}
351
352impl RequestBuilder<(String, Vec<Event>)> for RequestSettings {
353    type Metadata = (String, EventFinalizers);
354    type Events = Vec<Event>;
355    type Encoder = (Transformer, Encoder<Framer>);
356    type Payload = Bytes;
357    type Request = GcsRequest;
358    type Error = io::Error;
359
360    fn compression(&self) -> Compression {
361        self.compression
362    }
363
364    fn encoder(&self) -> &Self::Encoder {
365        &self.encoder
366    }
367
368    fn split_input(
369        &self,
370        input: (String, Vec<Event>),
371    ) -> (Self::Metadata, RequestMetadataBuilder, Self::Events) {
372        let (partition_key, mut events) = input;
373        let finalizers = events.take_finalizers();
374        let builder = RequestMetadataBuilder::from_events(&events);
375
376        ((partition_key, finalizers), builder, events)
377    }
378
379    fn build_request(
380        &self,
381        gcp_metadata: Self::Metadata,
382        metadata: RequestMetadata,
383        payload: EncodeResult<Self::Payload>,
384    ) -> Self::Request {
385        let (key, finalizers) = gcp_metadata;
386        // TODO: pull the seconds from the last event
387        let filename = {
388            let seconds = match self.tz_offset {
389                Some(offset) => Utc::now().with_timezone(&offset).format(&self.time_format),
390                None => Utc::now()
391                    .with_timezone(&chrono::Utc)
392                    .format(&self.time_format),
393            };
394
395            if self.append_uuid {
396                let uuid = Uuid::new_v4();
397                format!("{}-{}", seconds, uuid.hyphenated())
398            } else {
399                seconds.to_string()
400            }
401        };
402
403        let key = format!("{}{}.{}", key, filename, self.extension);
404        let body = payload.into_payload();
405
406        GcsRequest {
407            key,
408            body,
409            finalizers,
410            settings: GcsRequestSettings {
411                acl: self.acl.clone(),
412                content_type: self.content_type.clone(),
413                content_encoding: self.content_encoding.clone(),
414                storage_class: self.storage_class.clone(),
415                cache_control: self.cache_control.clone(),
416                headers: self.headers.clone(),
417            },
418            metadata,
419        }
420    }
421}
422
423impl RequestSettings {
424    fn new(config: &GcsSinkConfig, cx: SinkContext) -> crate::Result<Self> {
425        let transformer = config.encoding.transformer();
426        let (framer, serializer) = config.encoding.build(SinkType::MessageBased)?;
427        let encoder = Encoder::<Framer>::new(framer, serializer);
428        let acl = config
429            .acl
430            .map(|acl| HeaderValue::from_str(&to_string(acl)).unwrap());
431        let content_type_str = config
432            .content_type
433            .as_deref()
434            .unwrap_or_else(|| encoder.content_type());
435        let content_type = HeaderValue::from_str(content_type_str)?;
436        let content_encoding = match &config.content_encoding {
437            Some(ce) => Some(HeaderValue::from_str(ce)?),
438            None => config
439                .compression
440                .content_encoding()
441                .map(|ce| HeaderValue::from_str(&to_string(ce)).unwrap()),
442        };
443        let storage_class = config.storage_class.unwrap_or_default();
444        let storage_class = HeaderValue::from_str(&to_string(storage_class)).unwrap();
445        let cache_control = config
446            .cache_control
447            .as_ref()
448            .map(|cc| HeaderValue::from_str(cc))
449            .transpose()?;
450        let metadata = config
451            .metadata
452            .as_ref()
453            .map(|metadata| {
454                metadata
455                    .iter()
456                    .map(make_header)
457                    .collect::<Result<Vec<_>, _>>()
458            })
459            .unwrap_or_else(|| Ok(vec![]))?;
460        let extension = config
461            .filename_extension
462            .clone()
463            .unwrap_or_else(|| config.compression.extension().into());
464        let time_format = config.filename_time_format.clone();
465        let append_uuid = config.filename_append_uuid;
466        let offset = config
467            .timezone
468            .or(cx.globals.timezone)
469            .and_then(timezone_to_offset);
470
471        Ok(Self {
472            acl,
473            content_type,
474            content_encoding,
475            storage_class,
476            cache_control,
477            headers: metadata,
478            extension,
479            time_format,
480            append_uuid,
481            compression: config.compression,
482            encoder: (transformer, encoder),
483            tz_offset: offset,
484        })
485    }
486}
487
488// Make a header pair from a key-value string pair
489fn make_header((name, value): (&String, &String)) -> crate::Result<(HeaderName, HeaderValue)> {
490    Ok((
491        HeaderName::from_bytes(name.as_bytes())?,
492        HeaderValue::from_str(value)?,
493    ))
494}
495
496#[cfg(test)]
497mod tests {
498    use futures_util::{future::ready, stream};
499    use vector_lib::{
500        EstimatedJsonEncodedSizeOf,
501        codecs::{
502            JsonSerializerConfig, NewlineDelimitedEncoderConfig, TextSerializerConfig,
503            encoding::FramingConfig,
504        },
505        partition::Partitioner,
506        request_metadata::GroupedCountByteSize,
507    };
508    use vrl::event_path;
509
510    use super::*;
511    use crate::{
512        event::LogEvent,
513        template::{ConfinementConfig, Template},
514        test_util::{
515            components::{SINK_TAGS, run_and_assert_sink_compliance},
516            http::{always_200_response, spawn_blackhole_http_server},
517        },
518    };
519
520    #[test]
521    fn generate_config() {
522        crate::test_util::test_generate_config::<GcsSinkConfig>();
523    }
524
525    #[tokio::test]
526    async fn component_spec_compliance() {
527        let mock_endpoint = spawn_blackhole_http_server(always_200_response).await;
528
529        let context = SinkContext::default();
530
531        let tls = TlsSettings::default();
532        let client =
533            HttpClient::new(tls, context.proxy()).expect("should not fail to create HTTP client");
534
535        let config =
536            default_config((None::<FramingConfig>, JsonSerializerConfig::default()).into());
537        let sink = config
538            .build_sink(
539                client,
540                mock_endpoint.to_string(),
541                GcpAuthenticator::None,
542                context,
543            )
544            .expect("failed to build sink");
545
546        let event = Event::Log(LogEvent::from("simple message"));
547        run_and_assert_sink_compliance(sink, stream::once(ready(event)), &SINK_TAGS).await;
548    }
549
550    #[test]
551    fn gcs_encode_event_apply_rules() {
552        crate::test_util::trace_init();
553
554        let message = "hello world".to_string();
555        let mut event = LogEvent::from(message);
556        event.insert(event_path!("key"), "value");
557
558        let sink_config = GcsSinkConfig {
559            key_prefix: Some("key: {{ key }}".into()),
560            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
561        };
562        let key = sink_config
563            .key_partitioner()
564            .unwrap()
565            .partition(&Event::Log(event))
566            .expect("key wasn't provided");
567
568        assert_eq!(key, "key: value");
569    }
570
571    fn request_settings(sink_config: &GcsSinkConfig, context: SinkContext) -> RequestSettings {
572        RequestSettings::new(sink_config, context).expect("Could not create request settings")
573    }
574
575    fn build_request(extension: Option<&str>, uuid: bool, compression: Compression) -> GcsRequest {
576        let context = SinkContext::default();
577        let sink_config = GcsSinkConfig {
578            key_prefix: Some("key/".into()),
579            filename_time_format: "date".into(),
580            filename_extension: extension.map(Into::into),
581            filename_append_uuid: uuid,
582            compression,
583            ..default_config(
584                (
585                    Some(NewlineDelimitedEncoderConfig::new()),
586                    JsonSerializerConfig::default(),
587                )
588                    .into(),
589            )
590        };
591        let log = LogEvent::default().into();
592        let key = sink_config
593            .key_partitioner()
594            .unwrap()
595            .partition(&log)
596            .expect("key wasn't provided");
597
598        let mut byte_size = GroupedCountByteSize::new_untagged();
599        byte_size.add_event(&log, log.estimated_json_encoded_size_of());
600
601        let request_settings = request_settings(&sink_config, context);
602        let (metadata, metadata_request_builder, _events) =
603            request_settings.split_input((key, vec![log]));
604        let payload = EncodeResult::uncompressed(Bytes::new(), byte_size);
605        let request_metadata = metadata_request_builder.build(&payload);
606
607        request_settings.build_request(metadata, request_metadata, payload)
608    }
609
610    #[test]
611    fn gcs_build_request() {
612        let req = build_request(Some("ext"), false, Compression::None);
613        assert_eq!(req.key, "key/date.ext".to_string());
614
615        let req = build_request(None, false, Compression::None);
616        assert_eq!(req.key, "key/date.log".to_string());
617
618        let req = build_request(None, false, Compression::gzip_default());
619        assert_eq!(req.key, "key/date.log.gz".to_string());
620
621        let req = build_request(None, true, Compression::gzip_default());
622        assert_ne!(req.key, "key/date.log.gz".to_string());
623    }
624
625    #[test]
626    fn gcs_content_type_default() {
627        let context = SinkContext::default();
628        let sink_config = GcsSinkConfig {
629            content_type: None,
630            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
631        };
632
633        let request_settings = request_settings(&sink_config, context);
634        // Should default to encoder's content type which is "text/plain" for text codec
635        assert_eq!(
636            request_settings.content_type.to_str().unwrap(),
637            "text/plain"
638        );
639    }
640
641    #[test]
642    fn gcs_content_type_custom() {
643        let context = SinkContext::default();
644        let sink_config = GcsSinkConfig {
645            content_type: Some("text/plain; charset=utf-8".to_string()),
646            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
647        };
648
649        let request_settings = request_settings(&sink_config, context);
650        // Should use custom content type
651        assert_eq!(
652            request_settings.content_type.to_str().unwrap(),
653            "text/plain; charset=utf-8"
654        );
655    }
656
657    #[test]
658    fn gcs_content_type_invalid() {
659        let context = SinkContext::default();
660        let sink_config = GcsSinkConfig {
661            // Invalid header value with newline character
662            content_type: Some("text/plain\nInvalid".to_string()),
663            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
664        };
665
666        let result = RequestSettings::new(&sink_config, context);
667        // Should return an error, not panic
668        assert!(result.is_err());
669    }
670
671    #[test]
672    fn gcs_content_encoding_default() {
673        let context = SinkContext::default();
674        let sink_config = GcsSinkConfig {
675            content_encoding: None,
676            compression: Compression::gzip_default(),
677            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
678        };
679
680        let request_settings = request_settings(&sink_config, context);
681        // Should default to compression's content encoding which is "gzip"
682        assert_eq!(
683            request_settings.content_encoding.unwrap().to_str().unwrap(),
684            "gzip"
685        );
686    }
687
688    #[test]
689    fn gcs_content_encoding_none_when_no_compression() {
690        let context = SinkContext::default();
691        let sink_config = GcsSinkConfig {
692            content_encoding: None,
693            compression: Compression::None,
694            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
695        };
696
697        let request_settings = request_settings(&sink_config, context);
698        // Should be None when compression is None
699        assert!(request_settings.content_encoding.is_none());
700    }
701
702    #[test]
703    fn gcs_content_encoding_custom() {
704        let context = SinkContext::default();
705        let sink_config = GcsSinkConfig {
706            content_encoding: Some("gzip".to_string()),
707            compression: Compression::None,
708            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
709        };
710
711        let request_settings = request_settings(&sink_config, context);
712        // Should use custom content encoding
713        assert_eq!(
714            request_settings.content_encoding.unwrap().to_str().unwrap(),
715            "gzip"
716        );
717    }
718
719    #[test]
720    fn gcs_content_encoding_invalid() {
721        let context = SinkContext::default();
722        let sink_config = GcsSinkConfig {
723            // Invalid header value with newline character
724            content_encoding: Some("gzip\nInvalid".to_string()),
725            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
726        };
727
728        let result = RequestSettings::new(&sink_config, context);
729        // Should return an error, not panic
730        assert!(result.is_err());
731    }
732
733    #[test]
734    fn gcs_content_encoding_empty() {
735        let context = SinkContext::default();
736        let sink_config = GcsSinkConfig {
737            // Empty string to disable content encoding header even with compression
738            content_encoding: Some("".to_string()),
739            compression: Compression::gzip_default(),
740            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
741        };
742
743        let request_settings = request_settings(&sink_config, context);
744        // Should use empty content encoding (overriding the compression default)
745        assert_eq!(
746            request_settings.content_encoding.unwrap().to_str().unwrap(),
747            ""
748        );
749    }
750
751    #[test]
752    fn gcs_cache_control_default() {
753        let context = SinkContext::default();
754        let sink_config = GcsSinkConfig {
755            cache_control: None,
756            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
757        };
758
759        let request_settings = request_settings(&sink_config, context);
760        // Should be None by default
761        assert!(request_settings.cache_control.is_none());
762    }
763
764    #[test]
765    fn gcs_cache_control_custom() {
766        let context = SinkContext::default();
767        let sink_config = GcsSinkConfig {
768            cache_control: Some("no-transform".to_string()),
769            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
770        };
771
772        let request_settings = request_settings(&sink_config, context);
773        assert_eq!(
774            request_settings.cache_control.unwrap().to_str().unwrap(),
775            "no-transform"
776        );
777    }
778
779    #[test]
780    fn gcs_cache_control_invalid() {
781        let context = SinkContext::default();
782        let sink_config = GcsSinkConfig {
783            // Invalid header value with newline character
784            cache_control: Some("no-cache\nInvalid".to_string()),
785            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
786        };
787
788        let result = RequestSettings::new(&sink_config, context);
789        // Should return an error, not panic
790        assert!(result.is_err());
791    }
792
793    #[test]
794    fn confinement_rejects_unconfined_key_prefix() {
795        let config = GcsSinkConfig {
796            key_prefix: Some("{{ tenant }}".into()),
797            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
798        };
799        match config.key_partitioner() {
800            Err(err) => assert!(
801                err.to_string().contains("no literal string prefix"),
802                "unexpected error: {err}"
803            ),
804            Ok(_) => panic!("expected confinement error"),
805        }
806    }
807
808    #[test]
809    fn confinement_opt_out_allows_unconfined_key_prefix() {
810        let config = GcsSinkConfig {
811            key_prefix: Some("{{ tenant }}".into()),
812            confinement: ConfinementConfig {
813                dangerously_allow_unconfined_template_resolution: true,
814            },
815            ..default_config((None::<FramingConfig>, TextSerializerConfig::default()).into())
816        };
817        assert!(config.key_partitioner().is_ok());
818    }
819
820    #[test]
821    fn confinement_blocks_dotdot_escape_at_render() {
822        use crate::event::Event;
823
824        let template: Template = "safe/{{ tenant }}/".try_into().unwrap();
825        let template = template
826            .confine(
827                &ConfinementConfig::default(),
828                "gcp_cloud_storage",
829                "key_prefix",
830            )
831            .unwrap();
832        let mut event = Event::Log(LogEvent::from("x"));
833        event
834            .as_mut_log()
835            .insert(event_path!("tenant"), "../../escape");
836        assert!(template.render_string(&event).is_err());
837    }
838}
vector/sinks/gcp/cloud_storage.rs

vector/sinks/gcp/
cloud_storage.rs
