vector/sinks/azure_blob/
request_builder.rs

1use std::collections::{BTreeMap, HashMap};
2
3use bytes::Bytes;
4use chrono::Utc;
5use percent_encoding::{NON_ALPHANUMERIC, utf8_percent_encode};
6use uuid::Uuid;
7use vector_lib::{codecs::encoding::Framer, request_metadata::RequestMetadata};
8
9use crate::{
10    codecs::{Encoder, Transformer},
11    event::{Event, Finalizable},
12    sinks::{
13        azure_blob::config::{AzureBlobMetadata, AzureBlobRequest},
14        util::{
15            Compression, RequestBuilder, metadata::RequestMetadataBuilder,
16            request_builder::EncodeResult,
17        },
18    },
19};
20
21#[derive(Clone)]
22pub struct AzureBlobRequestOptions {
23    pub container_name: String,
24    pub blob_time_format: String,
25    pub blob_append_uuid: bool,
26    pub encoder: (Transformer, Encoder<Framer>),
27    pub compression: Compression,
28    pub tags: Option<BTreeMap<String, String>>,
29    pub metadata: Option<HashMap<String, String>>,
30}
31
32impl RequestBuilder<(String, Vec<Event>)> for AzureBlobRequestOptions {
33    type Metadata = AzureBlobMetadata;
34    type Events = Vec<Event>;
35    type Encoder = (Transformer, Encoder<Framer>);
36    type Payload = Bytes;
37    type Request = AzureBlobRequest;
38    type Error = std::io::Error;
39
40    fn compression(&self) -> Compression {
41        self.compression
42    }
43
44    fn encoder(&self) -> &Self::Encoder {
45        &self.encoder
46    }
47
48    fn split_input(
49        &self,
50        input: (String, Vec<Event>),
51    ) -> (Self::Metadata, RequestMetadataBuilder, Self::Events) {
52        let (partition_key, mut events) = input;
53        let finalizers = events.take_finalizers();
54        let azure_metadata = AzureBlobMetadata {
55            partition_key,
56            count: events.len(),
57            finalizers,
58        };
59
60        let builder = RequestMetadataBuilder::from_events(&events);
61
62        (azure_metadata, builder, events)
63    }
64
65    fn build_request(
66        &self,
67        mut azure_metadata: Self::Metadata,
68        request_metadata: RequestMetadata,
69        payload: EncodeResult<Self::Payload>,
70    ) -> Self::Request {
71        let formatted_ts = Utc::now().format(self.blob_time_format.as_str());
72        let blob_name = if self.blob_append_uuid {
73            format!("{formatted_ts}-{}", Uuid::new_v4().hyphenated())
74        } else {
75            formatted_ts.to_string()
76        };
77
78        let extension = self.compression.extension();
79        azure_metadata.partition_key = format!(
80            "{}{}.{}",
81            azure_metadata.partition_key, blob_name, extension
82        );
83
84        let blob_data = payload.into_payload();
85
86        debug!(
87            message = "Sending events.",
88            bytes = ?blob_data.len(),
89            events_len = ?azure_metadata.count,
90            blob = ?azure_metadata.partition_key,
91            container = ?self.container_name,
92        );
93
94        AzureBlobRequest {
95            blob_data,
96            content_encoding: self.compression.content_encoding(),
97            content_type: self.encoder.1.content_type(),
98            metadata: azure_metadata,
99            request_metadata,
100            // SDK 0.10.1 has no `with_tags()` helper, so we set the pre-encoded
101            // `x-ms-tags` header value directly. Match the Azure SDK's own encoding
102            // (`percent_encoding::NON_ALPHANUMERIC`, which emits `%20` for spaces) rather
103            // than `url::form_urlencoded` (which emits `+`).
104            tags: self
105                .tags
106                .as_ref()
107                .filter(|m| !m.is_empty())
108                .map(encode_tags),
109            blob_metadata: self.metadata.as_ref().filter(|m| !m.is_empty()).cloned(),
110        }
111    }
112}
113
114fn encode_tags(tags: &BTreeMap<String, String>) -> String {
115    let mut out = String::new();
116    for (k, v) in tags {
117        if !out.is_empty() {
118            out.push('&');
119        }
120        out.push_str(&utf8_percent_encode(k, NON_ALPHANUMERIC).to_string());
121        out.push('=');
122        out.push_str(&utf8_percent_encode(v, NON_ALPHANUMERIC).to_string());
123    }
124    out
125}
vector/sinks/azure_blob/request_builder.rs

vector/sinks/azure_blob/
request_builder.rs
