vector/sinks/s3_common/
partitioner.rs

1use vector_lib::{event::Event, partition::Partitioner};
2
3use crate::{
4    internal_events::TemplateRenderingError, sinks::util::partitioner::render_key_with_fallback,
5    template::Template,
6};
7
8#[derive(Clone, Debug, Eq, Hash, PartialEq)]
9pub struct S3PartitionKey {
10    pub key_prefix: String,
11    pub ssekms_key_id: Option<String>,
12}
13
14/// Partitions items based on the generated key for the given event.
15///
16/// If the template was built with a confinement check (via
17/// [`Template::confine`][crate::template::Template::confine]),
18/// keys that escape the base prefix are dropped as intentional security discards.
19pub struct S3KeyPartitioner {
20    key_prefix_template: Template,
21    ssekms_key_id_template: Option<Template>,
22    dead_letter_key_prefix: Option<String>,
23}
24
25impl S3KeyPartitioner {
26    pub const fn new(
27        key_prefix_template: Template,
28        ssekms_key_id_template: Option<Template>,
29        dead_letter_key_prefix: Option<String>,
30    ) -> Self {
31        Self {
32            key_prefix_template,
33            ssekms_key_id_template,
34            dead_letter_key_prefix,
35        }
36    }
37}
38
39impl Partitioner for S3KeyPartitioner {
40    type Item = Event;
41    type Key = Option<S3PartitionKey>;
42
43    fn partition(&self, item: &Self::Item) -> Self::Key {
44        let key_prefix = render_key_with_fallback(
45            &self.key_prefix_template,
46            item,
47            self.dead_letter_key_prefix.as_deref(),
48        )?;
49
50        let ssekms_key_id = self
51            .ssekms_key_id_template
52            .as_ref()
53            .map(|t| {
54                t.render_string(item).map_err(|error| {
55                    emit!(TemplateRenderingError {
56                        error,
57                        field: Some("ssekms_key_id"),
58                        drop_event: true,
59                    });
60                })
61            })
62            .transpose()
63            .ok()?;
64
65        Some(S3PartitionKey {
66            key_prefix,
67            ssekms_key_id,
68        })
69    }
70}
vector/sinks/s3_common/partitioner.rs

vector/sinks/s3_common/
partitioner.rs
