vector/sinks/aws_cloudwatch_metrics/
mod.rs

1#[cfg(all(test, feature = "aws-cloudwatch-metrics-integration-tests"))]
2mod integration_tests;
3#[cfg(test)]
4mod tests;
5
6use std::{
7    fmt,
8    task::{Context, Poll},
9};
10
11use aws_config::Region;
12use aws_sdk_cloudwatch::{
13    Client as CloudwatchClient,
14    error::SdkError,
15    operation::put_metric_data::PutMetricDataError,
16    types::{Dimension, MetricDatum},
17};
18use aws_smithy_types::DateTime as AwsDateTime;
19use futures::{FutureExt, SinkExt, stream};
20use futures_util::{future, future::BoxFuture};
21use indexmap::IndexMap;
22use tower::Service;
23use vector_lib::{
24    ByteSizeOf, EstimatedJsonEncodedSizeOf, configurable::configurable_component, sink::VectorSink,
25};
26
27use super::util::service::TowerRequestConfigDefaults;
28use crate::{
29    aws::{
30        ClientBuilder, RegionOrEndpoint, auth::AwsAuthentication, create_client, is_retriable_error,
31    },
32    config::{
33        AcknowledgementsConfig, DynValidatedSink, Input, ProxyConfig, SinkConfig, SinkContext,
34        ValidatedSink,
35    },
36    event::{
37        Event,
38        metric::{Metric, MetricTags, MetricValue},
39    },
40    sinks::util::{
41        Compression, EncodedEvent, PartitionBuffer, PartitionInnerBuffer, SinkBatchSettings,
42        TowerRequestConfig,
43        batch::{BatchConfig, BatchSettings},
44        buffer::metrics::{MetricNormalize, MetricNormalizer, MetricSet, MetricsBuffer},
45        retries::RetryLogic,
46    },
47    tls::TlsConfig,
48};
49
50#[derive(Clone, Copy, Debug, Default)]
51pub struct CloudWatchMetricsDefaultBatchSettings;
52
53impl SinkBatchSettings for CloudWatchMetricsDefaultBatchSettings {
54    const MAX_EVENTS: Option<usize> = Some(20);
55    const MAX_BYTES: Option<usize> = None;
56    const TIMEOUT_SECS: f64 = 1.0;
57}
58
59#[derive(Clone, Copy, Debug)]
60pub struct CloudWatchMetricsTowerRequestConfigDefaults;
61
62impl TowerRequestConfigDefaults for CloudWatchMetricsTowerRequestConfigDefaults {
63    const RATE_LIMIT_NUM: u64 = 150;
64}
65
66/// Configuration for the `aws_cloudwatch_metrics` sink.
67#[configurable_component(sink(
68    "aws_cloudwatch_metrics",
69    "Publish metric events to AWS CloudWatch Metrics."
70))]
71#[derive(Clone, Debug, Default)]
72#[serde(deny_unknown_fields)]
73pub struct CloudWatchMetricsSinkConfig {
74    /// The default [namespace][namespace] to use for metrics that do not have one.
75    ///
76    /// Metrics with the same name can only be differentiated by their namespace, and not all
77    /// metrics have their own namespace.
78    ///
79    /// [namespace]: https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/cloudwatch_concepts.html#Namespace
80    #[serde(alias = "namespace")]
81    #[configurable(metadata(docs::examples = "service"))]
82    pub default_namespace: String,
83
84    /// The [AWS region][aws_region] of the target service.
85    ///
86    /// [aws_region]: https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Concepts.RegionsAndAvailabilityZones.html
87    #[serde(flatten)]
88    pub region: RegionOrEndpoint,
89
90    #[configurable(derived)]
91    #[serde(default)]
92    pub compression: Compression,
93
94    #[configurable(derived)]
95    #[serde(default)]
96    pub batch: BatchConfig<CloudWatchMetricsDefaultBatchSettings>,
97
98    #[configurable(derived)]
99    #[serde(default)]
100    pub request: TowerRequestConfig<CloudWatchMetricsTowerRequestConfigDefaults>,
101
102    #[configurable(derived)]
103    pub tls: Option<TlsConfig>,
104
105    /// The ARN of an [IAM role][iam_role] to assume at startup.
106    ///
107    /// [iam_role]: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html
108    #[configurable(deprecated)]
109    #[configurable(metadata(docs::hidden))]
110    assume_role: Option<String>,
111
112    #[configurable(derived)]
113    #[serde(default)]
114    pub auth: AwsAuthentication,
115
116    #[configurable(derived)]
117    #[serde(
118        default,
119        deserialize_with = "crate::serde::bool_or_struct",
120        skip_serializing_if = "crate::serde::is_default"
121    )]
122    acknowledgements: AcknowledgementsConfig,
123
124    /// A map from metric name to AWS storage resolution.
125    /// Valid values are 1 (high resolution) and 60 (standard resolution).
126    /// If unset, the AWS SDK default of 60 (standard resolution) is used.
127    /// See [AWS Metrics Resolution](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/cloudwatch_concepts.html#Resolution_definition)
128    /// See [MetricDatum::storage_resolution](https://docs.rs/aws-sdk-cloudwatch/1.91.0/aws_sdk_cloudwatch/types/struct.MetricDatum.html#structfield.storage_resolution)
129    #[configurable(metadata(docs::additional_props_description = "An AWS storage resolution."))]
130    #[serde(default)]
131    pub storage_resolution: IndexMap<String, i32>,
132}
133
134impl_generate_config_from_default!(CloudWatchMetricsSinkConfig);
135
136struct CloudwatchMetricsClientBuilder;
137
138impl ClientBuilder for CloudwatchMetricsClientBuilder {
139    type Client = aws_sdk_cloudwatch::client::Client;
140
141    fn build(&self, config: &aws_types::SdkConfig) -> Self::Client {
142        aws_sdk_cloudwatch::client::Client::new(config)
143    }
144}
145
146#[async_trait::async_trait]
147#[typetag::serde(name = "aws_cloudwatch_metrics")]
148impl SinkConfig for CloudWatchMetricsSinkConfig {
149    fn input(&self) -> Input {
150        Input::metric()
151    }
152
153    fn acknowledgements(&self) -> &AcknowledgementsConfig {
154        &self.acknowledgements
155    }
156
157    fn as_dyn_validated(&self) -> Option<&dyn DynValidatedSink> {
158        Some(self)
159    }
160}
161
162pub struct ValidatedCloudWatchMetrics {
163    batch: BatchSettings<MetricsBuffer>,
164    storage_resolution: IndexMap<String, i32>,
165}
166
167impl fmt::Debug for ValidatedCloudWatchMetrics {
168    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
169        f.debug_struct("ValidatedCloudWatchMetrics")
170            .field("batch_size_bytes", &self.batch.size.bytes)
171            .field("batch_size_events", &self.batch.size.events)
172            .field("batch_timeout", &self.batch.timeout)
173            .field("storage_resolution", &self.storage_resolution)
174            .finish()
175    }
176}
177
178#[async_trait::async_trait]
179impl ValidatedSink for CloudWatchMetricsSinkConfig {
180    type Validated = ValidatedCloudWatchMetrics;
181
182    fn validate(&self) -> crate::Result<ValidatedCloudWatchMetrics> {
183        let batch = self.batch.into_batch_settings()?;
184        let storage_resolution = validate_storage_resolutions(self.storage_resolution.clone())?;
185        Ok(ValidatedCloudWatchMetrics {
186            batch,
187            storage_resolution,
188        })
189    }
190
191    async fn build(
192        &self,
193        validated: &ValidatedCloudWatchMetrics,
194        cx: SinkContext,
195    ) -> crate::Result<(super::VectorSink, super::Healthcheck)> {
196        let client = self.create_client(&cx.proxy).await?;
197        let healthcheck = self.clone().healthcheck(client.clone()).boxed();
198        let sink = CloudWatchMetricsSvc::new(self.clone(), client, validated)?;
199        Ok((sink, healthcheck))
200    }
201}
202
203impl CloudWatchMetricsSinkConfig {
204    async fn healthcheck(self, client: CloudwatchClient) -> crate::Result<()> {
205        client
206            .put_metric_data()
207            .metric_data(
208                MetricDatum::builder()
209                    .metric_name("healthcheck")
210                    .value(1.0)
211                    .build(),
212            )
213            .namespace(&self.default_namespace)
214            .send()
215            .await?;
216
217        Ok(())
218    }
219
220    async fn create_client(&self, proxy: &ProxyConfig) -> crate::Result<CloudwatchClient> {
221        let region = if cfg!(test) {
222            // Moto (used for mocking AWS) doesn't recognize 'custom' as valid region name
223            Some(Region::new("us-east-1"))
224        } else {
225            self.region.region()
226        };
227
228        create_client::<CloudwatchMetricsClientBuilder>(
229            &CloudwatchMetricsClientBuilder {},
230            &self.auth,
231            region,
232            self.region.endpoint(),
233            proxy,
234            self.tls.as_ref(),
235            None,
236        )
237        .await
238    }
239}
240
241#[derive(Default)]
242struct AwsCloudwatchMetricNormalize;
243
244impl MetricNormalize for AwsCloudwatchMetricNormalize {
245    fn normalize(&mut self, state: &mut MetricSet, metric: Metric) -> Option<Metric> {
246        match metric.value() {
247            MetricValue::Gauge { .. } => state.make_absolute(metric),
248            _ => state.make_incremental(metric),
249        }
250    }
251}
252
253#[derive(Debug, Clone)]
254struct CloudWatchMetricsRetryLogic;
255
256impl RetryLogic for CloudWatchMetricsRetryLogic {
257    type Error = SdkError<PutMetricDataError>;
258    type Request = PartitionInnerBuffer<Vec<Metric>, String>;
259    type Response = ();
260
261    fn is_retriable_error(&self, error: &Self::Error) -> bool {
262        is_retriable_error(error)
263    }
264}
265
266fn tags_to_dimensions(tags: &MetricTags) -> Vec<Dimension> {
267    // according to the API, up to 30 dimensions per metric can be provided
268    tags.iter_single()
269        .take(30)
270        .map(|(k, v)| Dimension::builder().name(k).value(v).build())
271        .collect()
272}
273
274#[derive(Clone)]
275pub struct CloudWatchMetricsSvc {
276    client: CloudwatchClient,
277    storage_resolution: IndexMap<String, i32>,
278}
279
280impl CloudWatchMetricsSvc {
281    pub fn new(
282        config: CloudWatchMetricsSinkConfig,
283        client: CloudwatchClient,
284        validated: &ValidatedCloudWatchMetrics,
285    ) -> crate::Result<VectorSink> {
286        let default_namespace = config.default_namespace.clone();
287        let batch = &validated.batch;
288        let request_settings = config.request.into_settings();
289
290        let service = CloudWatchMetricsSvc {
291            client,
292            storage_resolution: validated.storage_resolution.clone(),
293        };
294        let buffer = PartitionBuffer::new(MetricsBuffer::new(batch.size));
295        let mut normalizer = MetricNormalizer::<AwsCloudwatchMetricNormalize>::default();
296
297        let sink = request_settings
298            .partition_sink(CloudWatchMetricsRetryLogic, service, buffer, batch.timeout)
299            .sink_map_err(|error| error!(message = "Fatal CloudwatchMetrics sink error.", %error, internal_log_rate_limit = false))
300            .with_flat_map(move |event: Event| {
301                stream::iter({
302                    let byte_size = event.allocated_bytes();
303                    let json_byte_size = event.estimated_json_encoded_size_of();
304                    normalizer.normalize(event.into_metric()).map(|mut metric| {
305                        let namespace = metric
306                            .take_namespace()
307                            .unwrap_or_else(|| default_namespace.clone());
308                        Ok(EncodedEvent::new(
309                            PartitionInnerBuffer::new(metric, namespace),
310                            byte_size,
311                            json_byte_size,
312                        ))
313                    })
314                })
315            });
316
317        #[allow(deprecated)]
318        Ok(VectorSink::from_event_sink(sink))
319    }
320
321    fn encode_events(&mut self, events: Vec<Metric>) -> Vec<MetricDatum> {
322        let resolutions = &self.storage_resolution;
323        events
324            .into_iter()
325            .filter_map(|event| {
326                let metric_name = event.name().to_string();
327                let timestamp = event
328                    .timestamp()
329                    .map(|x| AwsDateTime::from_millis(x.timestamp_millis()));
330                let dimensions = event.tags().map(tags_to_dimensions);
331                let resolution = resolutions.get(&metric_name).copied();
332                // AwsCloudwatchMetricNormalize converts these to the right MetricKind
333                match event.value() {
334                    MetricValue::Counter { value } => Some(
335                        MetricDatum::builder()
336                            .metric_name(metric_name)
337                            .value(*value)
338                            .set_timestamp(timestamp)
339                            .set_dimensions(dimensions)
340                            .set_storage_resolution(resolution)
341                            .build(),
342                    ),
343                    MetricValue::Distribution {
344                        samples,
345                        statistic: _,
346                    } => Some(
347                        MetricDatum::builder()
348                            .metric_name(metric_name)
349                            .set_values(Some(samples.iter().map(|s| s.value).collect()))
350                            .set_counts(Some(samples.iter().map(|s| s.rate as f64).collect()))
351                            .set_timestamp(timestamp)
352                            .set_dimensions(dimensions)
353                            .set_storage_resolution(resolution)
354                            .build(),
355                    ),
356                    MetricValue::Set { values } => Some(
357                        MetricDatum::builder()
358                            .metric_name(metric_name)
359                            .value(values.len() as f64)
360                            .set_timestamp(timestamp)
361                            .set_dimensions(dimensions)
362                            .set_storage_resolution(resolution)
363                            .build(),
364                    ),
365                    MetricValue::Gauge { value } => Some(
366                        MetricDatum::builder()
367                            .metric_name(metric_name)
368                            .value(*value)
369                            .set_timestamp(timestamp)
370                            .set_dimensions(dimensions)
371                            .set_storage_resolution(resolution)
372                            .build(),
373                    ),
374                    _ => None,
375                }
376            })
377            .collect()
378    }
379}
380
381impl Service<PartitionInnerBuffer<Vec<Metric>, String>> for CloudWatchMetricsSvc {
382    type Response = ();
383    type Error = SdkError<PutMetricDataError>;
384    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
385
386    // Emission of an internal event in case of errors is handled upstream by the caller.
387    fn poll_ready(&mut self, _cx: &mut Context) -> Poll<Result<(), Self::Error>> {
388        Poll::Ready(Ok(()))
389    }
390
391    // Emission of internal events for errors and dropped events is handled upstream by the caller.
392    fn call(&mut self, items: PartitionInnerBuffer<Vec<Metric>, String>) -> Self::Future {
393        let (items, namespace) = items.into_parts();
394        let metric_data = self.encode_events(items);
395        if metric_data.is_empty() {
396            return future::ok(()).boxed();
397        }
398
399        let client = self.client.clone();
400
401        Box::pin(async move {
402            client
403                .put_metric_data()
404                .namespace(namespace)
405                .set_metric_data(Some(metric_data))
406                .send()
407                .await?;
408            Ok(())
409        })
410    }
411}
412
413fn validate_storage_resolutions(
414    storage_resolutions: IndexMap<String, i32>,
415) -> crate::Result<IndexMap<String, i32>> {
416    for (metric_name, storage_resolution) in storage_resolutions.iter() {
417        if !matches!(storage_resolution, 1 | 60) {
418            return Err(
419                format!("Storage resolution for {metric_name} should be '1' or '60'").into(),
420            );
421        }
422    }
423    Ok(storage_resolutions)
424}
vector/sinks/aws_cloudwatch_metrics/mod.rs

vector/sinks/aws_cloudwatch_metrics/
mod.rs
