vector/sinks/azure_logs_ingestion/
config.rs

1use std::sync::Arc;
2
3use azure_core::credentials::TokenCredential;
4
5use http::uri::PathAndQuery;
6
7use vector_lib::{configurable::configurable_component, schema};
8use vrl::value::Kind;
9
10use crate::{
11    config::{DynValidatedSink, ValidatedSink},
12    http::{HttpClient, get_http_scheme_from_uri},
13    sinks::{
14        azure_common::config::AzureAuthentication,
15        prelude::*,
16        util::{
17            HttpEndpoint, RealtimeSizeBasedDefaultBatchSettings,
18            http::{HttpStatusRetryLogic, RetryStrategy},
19        },
20    },
21};
22
23use super::{
24    service::{AzureLogsIngestionResponse, AzureLogsIngestionService, request_path},
25    sink::AzureLogsIngestionSink,
26};
27
28/// Max number of bytes in request body
29const MAX_BATCH_SIZE: usize = 30 * 1024 * 1024;
30
31pub(super) fn default_scope() -> String {
32    "https://monitor.azure.com/.default".into()
33}
34
35pub(super) fn default_timestamp_field() -> String {
36    "TimeGenerated".into()
37}
38
39/// Configuration for the `azure_logs_ingestion` sink.
40#[configurable_component(sink(
41    "azure_logs_ingestion",
42    "Publish log events to the Azure Monitor Logs Ingestion API."
43))]
44#[derive(Clone, Debug)]
45#[serde(deny_unknown_fields)]
46pub struct AzureLogsIngestionConfig {
47    /// The [Data collection endpoint URI][endpoint] associated with the Log Analytics workspace.
48    ///
49    /// [endpoint]: https://learn.microsoft.com/en-us/azure/azure-monitor/logs/logs-ingestion-api-overview
50    #[configurable(metadata(
51        docs::examples = "https://my-dce-5kyl.eastus-1.ingest.monitor.azure.com"
52    ))]
53    pub endpoint: HttpEndpoint,
54
55    /// The [Data collection rule immutable ID][dcr_immutable_id] for the Data collection endpoint.
56    ///
57    /// [dcr_immutable_id]: https://learn.microsoft.com/en-us/azure/azure-monitor/logs/logs-ingestion-api-overview
58    #[configurable(metadata(docs::examples = "dcr-000a00a000a00000a000000aa000a0aa"))]
59    pub dcr_immutable_id: String,
60
61    /// The [Stream name][stream_name] for the Data collection rule.
62    ///
63    /// [stream_name]: https://learn.microsoft.com/en-us/azure/azure-monitor/logs/logs-ingestion-api-overview
64    #[configurable(metadata(docs::examples = "Custom-MyTable"))]
65    pub stream_name: String,
66
67    #[configurable(derived)]
68    pub auth: AzureAuthentication,
69
70    /// [Token scope][token_scope] for dedicated Azure regions.
71    ///
72    /// [token_scope]: https://learn.microsoft.com/en-us/azure/azure-monitor/logs/logs-ingestion-api-overview
73    #[configurable(metadata(docs::examples = "https://monitor.azure.us/.default"))]
74    #[configurable(metadata(docs::examples = "https://monitor.azure.cn/.default"))]
75    #[serde(default = "default_scope")]
76    pub(super) token_scope: String,
77
78    /// The destination field (column) for the timestamp.
79    ///
80    /// The setting of `log_schema.timestamp_key`, usually `timestamp`, is used as the source.
81    /// Most schemas use `TimeGenerated`, but some use `Timestamp` (legacy) or `EventStartTime` (ASIM) [std_columns].
82    ///
83    /// [std_columns]: https://learn.microsoft.com/en-us/azure/azure-monitor/logs/log-standard-columns#timegenerated
84    #[configurable(metadata(docs::examples = "EventStartTime"))]
85    #[configurable(metadata(docs::examples = "Timestamp"))]
86    #[serde(default = "default_timestamp_field")]
87    pub timestamp_field: String,
88
89    #[configurable(derived)]
90    #[serde(default, skip_serializing_if = "crate::serde::is_default")]
91    pub encoding: Transformer,
92
93    #[configurable(derived)]
94    #[serde(default)]
95    pub batch: BatchConfig<RealtimeSizeBasedDefaultBatchSettings>,
96
97    #[configurable(derived)]
98    #[serde(default)]
99    pub request: TowerRequestConfig,
100
101    #[configurable(derived)]
102    pub tls: Option<TlsConfig>,
103
104    #[configurable(derived)]
105    #[serde(
106        default,
107        deserialize_with = "crate::serde::bool_or_struct",
108        skip_serializing_if = "crate::serde::is_default"
109    )]
110    pub acknowledgements: AcknowledgementsConfig,
111
112    #[configurable(derived)]
113    #[serde(default)]
114    pub retry_strategy: RetryStrategy,
115}
116
117impl Default for AzureLogsIngestionConfig {
118    fn default() -> Self {
119        Self {
120            endpoint: HttpEndpoint::parse("http://localhost:8080").unwrap(),
121            dcr_immutable_id: Default::default(),
122            stream_name: Default::default(),
123            auth: Default::default(),
124            token_scope: default_scope(),
125            timestamp_field: default_timestamp_field(),
126            encoding: Default::default(),
127            batch: Default::default(),
128            request: Default::default(),
129            tls: None,
130            acknowledgements: Default::default(),
131            retry_strategy: Default::default(),
132        }
133    }
134}
135
136impl AzureLogsIngestionConfig {
137    #[allow(clippy::too_many_arguments)]
138    pub(super) async fn build_inner(
139        &self,
140        cx: SinkContext,
141        validated: &ValidatedAzureLogsIngestion,
142        endpoint: HttpEndpoint,
143        credential: Arc<dyn TokenCredential>,
144        token_scope: String,
145        timestamp_field: String,
146    ) -> crate::Result<(VectorSink, Healthcheck)> {
147        let endpoint = endpoint.into_uri();
148        let protocol = get_http_scheme_from_uri(&endpoint).to_string();
149
150        let tls_settings = TlsSettings::from_options(self.tls.as_ref())?;
151        let client = HttpClient::new(Some(tls_settings), &cx.proxy)?;
152
153        let service = AzureLogsIngestionService::new(
154            client,
155            endpoint,
156            validated.path_and_query.clone(),
157            credential,
158            token_scope,
159        )?;
160        let healthcheck = service.healthcheck();
161
162        let retry_logic = HttpStatusRetryLogic::new(
163            |res: &AzureLogsIngestionResponse| res.http_status,
164            self.retry_strategy.clone(),
165        );
166        let request_settings = self.request.into_settings();
167        let service = ServiceBuilder::new()
168            .settings(request_settings, retry_logic)
169            .service(service);
170
171        let sink = AzureLogsIngestionSink::new(
172            validated.batch_settings,
173            self.encoding.clone(),
174            service,
175            timestamp_field,
176            protocol,
177        );
178
179        Ok((VectorSink::from_event_streamsink(sink), healthcheck))
180    }
181}
182
183impl_generate_config_from_default!(AzureLogsIngestionConfig);
184
185#[derive(Clone, Debug)]
186pub struct ValidatedAzureLogsIngestion {
187    batch_settings: BatcherSettings,
188    path_and_query: PathAndQuery,
189}
190
191#[async_trait::async_trait]
192#[typetag::serde(name = "azure_logs_ingestion")]
193impl SinkConfig for AzureLogsIngestionConfig {
194    fn input(&self) -> Input {
195        let requirements =
196            schema::Requirement::empty().optional_meaning("timestamp", Kind::timestamp());
197
198        Input::log().with_schema_requirement(requirements)
199    }
200
201    fn acknowledgements(&self) -> &AcknowledgementsConfig {
202        &self.acknowledgements
203    }
204
205    fn as_dyn_validated(&self) -> Option<&dyn DynValidatedSink> {
206        Some(self)
207    }
208}
209
210#[async_trait::async_trait]
211impl ValidatedSink for AzureLogsIngestionConfig {
212    type Validated = ValidatedAzureLogsIngestion;
213
214    fn validate(&self) -> crate::Result<ValidatedAzureLogsIngestion> {
215        let batch_settings = self
216            .batch
217            .validate()?
218            .limit_max_bytes(MAX_BATCH_SIZE)?
219            .into_batcher_settings()?;
220
221        let path_and_query = request_path(&self.dcr_immutable_id, &self.stream_name)?;
222
223        Ok(ValidatedAzureLogsIngestion {
224            batch_settings,
225            path_and_query,
226        })
227    }
228
229    async fn build(
230        &self,
231        validated: &ValidatedAzureLogsIngestion,
232        cx: SinkContext,
233    ) -> crate::Result<(VectorSink, Healthcheck)> {
234        let credential: Arc<dyn TokenCredential> = self.auth.credential().await?;
235
236        self.build_inner(
237            cx,
238            validated,
239            self.endpoint.clone(),
240            credential,
241            self.token_scope.clone(),
242            self.timestamp_field.clone(),
243        )
244        .await
245    }
246}
vector/sinks/azure_logs_ingestion/config.rs

vector/sinks/azure_logs_ingestion/
config.rs
