From 66c1daa4ca4d925d983b752ef424bcc2a6018fda Mon Sep 17 00:00:00 2001 From: Vincent Voyer Date: Thu, 9 Jul 2026 16:49:41 +0200 Subject: [PATCH 1/3] enhancement(aws_cloudwatch_logs sink): stop calling DescribeLogStreams for sequence tokens CloudWatch Logs removed the PutLogEvents sequence-token requirement in Jan 2023 and never returns InvalidSequenceToken. The sink was calling DescribeLogStreams once per batch (and again on every retry) solely to fetch that token. DescribeLogStreams is throttled account+region-wide at a low default quota (25 TPS), so large fleets writing to many streams could saturate it and stall the sink via retry amplification. Write straight to PutLogEvents without a token, and create the group/stream lazily only when PutLogEvents returns ResourceNotFoundException. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Vincent Voyer --- ...s_drop_describe_log_streams.enhancement.md | 3 + src/sinks/aws_cloudwatch_logs/request.rs | 170 ++++++++---------- src/sinks/aws_cloudwatch_logs/retry.rs | 12 +- src/sinks/aws_cloudwatch_logs/service.rs | 15 +- 4 files changed, 83 insertions(+), 117 deletions(-) create mode 100644 changelog.d/aws_cloudwatch_logs_drop_describe_log_streams.enhancement.md diff --git a/changelog.d/aws_cloudwatch_logs_drop_describe_log_streams.enhancement.md b/changelog.d/aws_cloudwatch_logs_drop_describe_log_streams.enhancement.md new file mode 100644 index 0000000000000..d44140a5a117f --- /dev/null +++ b/changelog.d/aws_cloudwatch_logs_drop_describe_log_streams.enhancement.md @@ -0,0 +1,3 @@ +The `aws_cloudwatch_logs` sink no longer calls `DescribeLogStreams` to fetch a sequence token before writing. Since January 2023 CloudWatch Logs accepts `PutLogEvents` without a sequence token and never returns `InvalidSequenceToken`, so the sink now writes directly and only calls `CreateLogGroup`/`CreateLogStream` when `PutLogEvents` returns `ResourceNotFoundException`. This removes one `DescribeLogStreams` request per batch (and one per retry), which is throttled account-and-region-wide at a low default quota (25 TPS) and could stall the sink under load on large fleets writing to many streams. + +authors: vvo diff --git a/src/sinks/aws_cloudwatch_logs/request.rs b/src/sinks/aws_cloudwatch_logs/request.rs index de3b03da25396..5ab408d921441 100644 --- a/src/sinks/aws_cloudwatch_logs/request.rs +++ b/src/sinks/aws_cloudwatch_logs/request.rs @@ -10,7 +10,6 @@ use aws_sdk_cloudwatchlogs::{ operation::{ create_log_group::CreateLogGroupError, create_log_stream::CreateLogStreamError, - describe_log_streams::{DescribeLogStreamsError, DescribeLogStreamsOutput}, put_log_events::{PutLogEventsError, PutLogEventsOutput}, put_retention_policy::PutRetentionPolicyError, }, @@ -30,7 +29,17 @@ pub struct CloudwatchFuture { create_missing_group: bool, create_missing_stream: bool, retention_enabled: bool, + // Batches still waiting to be sent after `current`. events: Vec>, + // The batch currently in flight. Retained so it can be resent after a + // missing log group/stream is created: `PutLogEvents` returns + // `ResourceNotFoundException` when the stream does not exist yet, and we + // need the events to replay once it does. + current: Vec, + // Set once we have created (or tried to create) the group/stream in + // response to a `ResourceNotFoundException`, so a second failure is a hard + // error rather than an infinite create loop. + created_missing: bool, token_tx: Option>>, } @@ -49,7 +58,6 @@ type ClientResult = BoxFuture<'static, Result enum State { CreateGroup(ClientResult<(), CreateLogGroupError>), CreateStream(ClientResult<(), CreateLogStreamError>), - DescribeStream(ClientResult), Put(ClientResult), PutRetentionPolicy(ClientResult<(), PutRetentionPolicyError>), } @@ -68,7 +76,9 @@ impl CloudwatchFuture { kms_key: Option, tags: Option>, mut events: Vec>, - token: Option, + // Kept for backwards compatibility with the caller; the sequence token + // is no longer used (see below) and is always ignored. + _token: Option, token_tx: oneshot::Sender>, ) -> Self { let retention_days = retention.days; @@ -82,17 +92,25 @@ impl CloudwatchFuture { tags, }; - let state = if let Some(token) = token { - State::Put(client.put_logs(Some(token), events.pop().expect("No Events to send"))) - } else { - State::DescribeStream(client.describe_stream()) - }; + // Since January 2023, CloudWatch Logs no longer requires a sequence + // token on `PutLogEvents` and never returns `InvalidSequenceToken`, so + // we write directly instead of first calling `DescribeLogStreams` to + // fetch a token. That describe call ran once per batch (and again on + // every retry), and is account+region-wide throttled at a low default + // quota (25 TPS for `DescribeLogStreams`), so on large fleets it became + // the bottleneck and starved the sink. Writing straight to + // `PutLogEvents` removes that dependency entirely. + // https://aws.amazon.com/about-aws/whats-new/2023/01/amazon-cloudwatch-logs-log-stream-transaction-quota-sequencetoken-requirement/ + let current = events.pop().expect("No Events to send"); + let state = State::Put(client.put_logs(current.clone())); let retention_enabled = retention.enabled; Self { client, events, + current, + created_missing: false, state, token_tx: Some(token_tx), create_missing_group, @@ -108,50 +126,49 @@ impl Future for CloudwatchFuture { fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll { loop { match &mut self.state { - State::DescribeStream(fut) => { - let response = match ready!(fut.poll_unpin(cx)) { - Ok(response) => response, + State::Put(fut) => { + match ready!(fut.poll_unpin(cx)) { + Ok(_response) => { + if let Some(events) = self.events.pop() { + self.current = events; + debug!(message = "Putting logs."); + self.state = + State::Put(self.client.put_logs(self.current.clone())); + } else { + info!(message = "Putting logs was successful."); + + self.token_tx + .take() + .expect("Put was polled after finishing.") + .send(None) + .expect("CloudwatchLogsSvc was dropped unexpectedly"); + + return Poll::Ready(Ok(())); + } + } Err(err) => { - if let SdkError::ServiceError(inner) = &err - && matches!( - inner.err(), - DescribeLogStreamsError::ResourceNotFoundException(_) - ) - && self.create_missing_group + // The stream (or its group) does not exist yet. + // Create it once, then replay the same batch. + if !self.created_missing + && is_resource_not_found(&err) + && (self.create_missing_group || self.create_missing_stream) { - info!("Log group provided does not exist; creating a new one."); - - self.state = State::CreateGroup(self.client.create_log_group()); + self.created_missing = true; + if self.create_missing_group { + info!( + "Log group provided does not exist; creating a new one." + ); + self.state = + State::CreateGroup(self.client.create_log_group()); + } else { + info!("Provided stream does not exist; creating a new one."); + self.state = + State::CreateStream(self.client.create_log_stream()); + } continue; } - return Poll::Ready(Err(CloudwatchError::DescribeLogStreams(err))); + return Poll::Ready(Err(CloudwatchError::Put(err))); } - }; - - let stream_name = &self.client.stream_name; - - if let Some(stream) = response - .log_streams - .ok_or(CloudwatchError::NoStreamsFound)? - .into_iter() - .find(|log_stream| log_stream.log_stream_name == Some(stream_name.clone())) - { - debug!(message = "Stream found.", stream = ?stream.log_stream_name); - - let events = self - .events - .pop() - .expect("Token got called multiple times, self is a bug!"); - - let token = stream.upload_sequence_token; - - info!(message = "Putting logs.", token = ?token); - self.state = State::Put(self.client.put_logs(token, events)); - } else if self.create_missing_stream { - info!("Provided stream does not exist; creating a new one."); - self.state = State::CreateStream(self.client.create_log_stream()); - } else { - return Poll::Ready(Err(CloudwatchError::NoStreamsFound)); } } @@ -175,13 +192,13 @@ impl Future for CloudwatchFuture { info!(message = "Group created.", name = %self.client.group_name); if self.retention_enabled { - self.state = State::PutRetentionPolicy(self.client.put_retention_policy()); + self.state = + State::PutRetentionPolicy(self.client.put_retention_policy()); continue; } - // self does not abide by `create_missing_stream` since a group - // never has any streams and thus we need to create one if a group - // is created no matter what. + // A newly created group never has any streams, so create + // one regardless of `create_missing_stream`. self.state = State::CreateStream(self.client.create_log_stream()); } @@ -204,29 +221,8 @@ impl Future for CloudwatchFuture { info!(message = "Stream created.", name = %self.client.stream_name); - self.state = State::DescribeStream(self.client.describe_stream()); - } - - State::Put(fut) => { - let next_token = match ready!(fut.poll_unpin(cx)) { - Ok(resp) => resp.next_sequence_token, - Err(err) => return Poll::Ready(Err(CloudwatchError::Put(err))), - }; - - if let Some(events) = self.events.pop() { - debug!(message = "Putting logs.", next_token = ?next_token); - self.state = State::Put(self.client.put_logs(next_token, events)); - } else { - info!(message = "Putting logs was successful.", next_token = ?next_token); - - self.token_tx - .take() - .expect("Put was polled after finishing.") - .send(next_token) - .expect("CloudwatchLogsSvc was dropped unexpectedly"); - - return Poll::Ready(Ok(())); - } + // Replay the batch that hit the missing group/stream. + self.state = State::Put(self.client.put_logs(self.current.clone())); } State::PutRetentionPolicy(fut) => { @@ -246,10 +242,17 @@ impl Future for CloudwatchFuture { } } +fn is_resource_not_found(err: &SdkError) -> bool { + matches!( + err, + SdkError::ServiceError(inner) + if matches!(inner.err(), PutLogEventsError::ResourceNotFoundException(_)) + ) +} + impl Client { pub fn put_logs( &self, - sequence_token: Option, log_events: Vec, ) -> ClientResult { let client = self.client.clone(); @@ -261,7 +264,6 @@ impl Client { client .put_log_events() .set_log_events(Some(log_events)) - .set_sequence_token(sequence_token) .log_group_name(group_name) .log_stream_name(stream_name) .customize() @@ -275,22 +277,6 @@ impl Client { }) } - pub fn describe_stream( - &self, - ) -> ClientResult { - let client = self.client.clone(); - let group_name = self.group_name.clone(); - let stream_name = self.stream_name.clone(); - Box::pin(async move { - client - .describe_log_streams() - .log_group_name(group_name) - .log_stream_name_prefix(stream_name) - .send() - .await - }) - } - pub fn create_log_group(&self) -> ClientResult<(), CreateLogGroupError> { let client = self.client.clone(); let group_name = self.group_name.clone(); diff --git a/src/sinks/aws_cloudwatch_logs/retry.rs b/src/sinks/aws_cloudwatch_logs/retry.rs index 11759f8f49976..608ac225c8e41 100644 --- a/src/sinks/aws_cloudwatch_logs/retry.rs +++ b/src/sinks/aws_cloudwatch_logs/retry.rs @@ -1,8 +1,7 @@ use std::marker::PhantomData; use aws_sdk_cloudwatchlogs::operation::{ - create_log_stream::CreateLogStreamError, describe_log_streams::DescribeLogStreamsError, - put_log_events::PutLogEventsError, + create_log_stream::CreateLogStreamError, put_log_events::PutLogEventsError, }; use aws_smithy_runtime_api::client::result::SdkError; @@ -54,15 +53,6 @@ impl RetryLogic } is_retriable_error(err) } - CloudwatchError::DescribeLogStreams(err) => { - if let SdkError::ServiceError(inner) = err { - let err = inner.err(); - if matches!(err, DescribeLogStreamsError::ServiceUnavailableException(_)) { - return true; - } - } - is_retriable_error(err) - } CloudwatchError::CreateStream(err) => { if let SdkError::ServiceError(inner) = err { let err = inner.err(); diff --git a/src/sinks/aws_cloudwatch_logs/service.rs b/src/sinks/aws_cloudwatch_logs/service.rs index cdb34702aa413..4d50a806a370f 100644 --- a/src/sinks/aws_cloudwatch_logs/service.rs +++ b/src/sinks/aws_cloudwatch_logs/service.rs @@ -8,8 +8,7 @@ use aws_sdk_cloudwatchlogs::{ Client as CloudwatchLogsClient, operation::{ create_log_group::CreateLogGroupError, create_log_stream::CreateLogStreamError, - describe_log_streams::DescribeLogStreamsError, put_log_events::PutLogEventsError, - put_retention_policy::PutRetentionPolicyError, + put_log_events::PutLogEventsError, put_retention_policy::PutRetentionPolicyError, }, types::InputLogEvent, }; @@ -66,27 +65,21 @@ type Svc = Buffer< #[derive(Debug)] pub enum CloudwatchError { Put(SdkError), - DescribeLogStreams(SdkError), CreateStream(SdkError), CreateGroup(SdkError), PutRetentionPolicy(SdkError), - NoStreamsFound, } impl fmt::Display for CloudwatchError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { CloudwatchError::Put(error) => write!(f, "CloudwatchError::Put: {error}"), - CloudwatchError::DescribeLogStreams(error) => { - write!(f, "CloudwatchError::DescribeLogStreams: {error}") - } CloudwatchError::CreateStream(error) => { write!(f, "CloudwatchError::CreateStream: {error}") } CloudwatchError::CreateGroup(error) => { write!(f, "CloudwatchError::CreateGroup: {error}") } - CloudwatchError::NoStreamsFound => write!(f, "CloudwatchError: No Streams Found"), CloudwatchError::PutRetentionPolicy(error) => { write!(f, "CloudwatchError::PutRetentionPolicy: {error}") } @@ -102,12 +95,6 @@ impl From> for CloudwatchError { } } -impl From> for CloudwatchError { - fn from(error: SdkError) -> Self { - CloudwatchError::DescribeLogStreams(error) - } -} - #[derive(Debug)] pub struct CloudwatchResponse { events_byte_size: GroupedCountByteSize, From ebb5392ed8fded87249c952c9944ded8c0e97db3 Mon Sep 17 00:00:00 2001 From: Vincent Voyer Date: Thu, 9 Jul 2026 17:36:34 +0200 Subject: [PATCH 2/3] enhancement(aws_cloudwatch_logs sink): drop dead sequence-token plumbing With no sequence token to fetch or pass, the oneshot channel between the service and its request future is dead weight. Remove it along with the token/token_rx fields and the _token param; concurrency_limit(1) already serializes per-stream writes. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Vincent Voyer --- src/sinks/aws_cloudwatch_logs/request.rs | 14 ------- src/sinks/aws_cloudwatch_logs/service.rs | 52 ++++++++---------------- 2 files changed, 17 insertions(+), 49 deletions(-) diff --git a/src/sinks/aws_cloudwatch_logs/request.rs b/src/sinks/aws_cloudwatch_logs/request.rs index 5ab408d921441..c0b1d6040fc3d 100644 --- a/src/sinks/aws_cloudwatch_logs/request.rs +++ b/src/sinks/aws_cloudwatch_logs/request.rs @@ -19,7 +19,6 @@ use aws_smithy_runtime_api::client::{orchestrator::HttpResponse, result::SdkErro use futures::{FutureExt, future::BoxFuture}; use http::{HeaderValue, header::HeaderName}; use indexmap::IndexMap; -use tokio::sync::oneshot; use crate::sinks::aws_cloudwatch_logs::{config::Retention, service::CloudwatchError}; @@ -40,7 +39,6 @@ pub struct CloudwatchFuture { // response to a `ResourceNotFoundException`, so a second failure is a hard // error rather than an infinite create loop. created_missing: bool, - token_tx: Option>>, } struct Client { @@ -76,10 +74,6 @@ impl CloudwatchFuture { kms_key: Option, tags: Option>, mut events: Vec>, - // Kept for backwards compatibility with the caller; the sequence token - // is no longer used (see below) and is always ignored. - _token: Option, - token_tx: oneshot::Sender>, ) -> Self { let retention_days = retention.days; let client = Client { @@ -112,7 +106,6 @@ impl CloudwatchFuture { current, created_missing: false, state, - token_tx: Some(token_tx), create_missing_group, create_missing_stream, retention_enabled, @@ -136,13 +129,6 @@ impl Future for CloudwatchFuture { State::Put(self.client.put_logs(self.current.clone())); } else { info!(message = "Putting logs was successful."); - - self.token_tx - .take() - .expect("Put was polled after finishing.") - .send(None) - .expect("CloudwatchLogsSvc was dropped unexpectedly"); - return Poll::Ready(Ok(())); } } diff --git a/src/sinks/aws_cloudwatch_logs/service.rs b/src/sinks/aws_cloudwatch_logs/service.rs index 4d50a806a370f..835c751bcbd47 100644 --- a/src/sinks/aws_cloudwatch_logs/service.rs +++ b/src/sinks/aws_cloudwatch_logs/service.rs @@ -1,7 +1,7 @@ use std::{ collections::HashMap, fmt, - task::{Context, Poll, ready}, + task::{Context, Poll}, }; use aws_sdk_cloudwatchlogs::{ @@ -22,7 +22,6 @@ use http::{ }; use indexmap::IndexMap; use snafu::{ResultExt, Snafu}; -use tokio::sync::oneshot; use tower::{ Service, ServiceBuilder, ServiceExt, buffer::Buffer, @@ -186,7 +185,7 @@ impl Service for CloudwatchLogsPartitionSvc { let svc = if let Some(svc) = &mut self.clients.get_mut(&key) { svc.clone() } else { - // Concurrency limit is 1 because we need token from previous request. + // Concurrency limit is 1 to keep writes to a single stream ordered. let svc = ServiceBuilder::new() .buffer(1) .concurrency_limit(1) @@ -246,8 +245,6 @@ impl CloudwatchLogsSvc { retention, kms_key, tags, - token: None, - token_rx: None, } } @@ -297,38 +294,25 @@ impl Service> for CloudwatchLogsSvc { type Error = CloudwatchError; type Future = request::CloudwatchFuture; - fn poll_ready(&mut self, cx: &mut Context) -> Poll> { - if let Some(rx) = &mut self.token_rx { - self.token = ready!(rx.poll_unpin(cx)).ok().flatten(); - self.token_rx = None; - } + fn poll_ready(&mut self, _cx: &mut Context) -> Poll> { Poll::Ready(Ok(())) } fn call(&mut self, req: Vec) -> Self::Future { - if self.token_rx.is_none() { - let event_batches = self.process_events(req); - - let (tx, rx) = oneshot::channel(); - self.token_rx = Some(rx); - - request::CloudwatchFuture::new( - self.client.clone(), - self.headers.clone(), - self.stream_name.clone(), - self.group_name.clone(), - self.create_missing_group, - self.create_missing_stream, - self.retention.clone(), - self.kms_key.clone(), - self.tags.clone(), - event_batches, - self.token.take(), - tx, - ) - } else { - panic!("poll_ready was not called; this is a bug!"); - } + let event_batches = self.process_events(req); + + request::CloudwatchFuture::new( + self.client.clone(), + self.headers.clone(), + self.stream_name.clone(), + self.group_name.clone(), + self.create_missing_group, + self.create_missing_stream, + self.retention.clone(), + self.kms_key.clone(), + self.tags.clone(), + event_batches, + ) } } @@ -342,8 +326,6 @@ pub struct CloudwatchLogsSvc { retention: Retention, kms_key: Option, tags: Option>, - token: Option, - token_rx: Option>>, } impl EncodedLength for InputLogEvent { From f87966bdc8a383b2fccc9cd9a688316fc551e2f6 Mon Sep 17 00:00:00 2001 From: Vincent Voyer Date: Thu, 9 Jul 2026 17:53:59 +0200 Subject: [PATCH 3/3] fix(aws_cloudwatch_logs sink): resolve missing group/stream via describe on ResourceNotFound Addresses review feedback. A PutLogEvents ResourceNotFoundException cannot tell a missing group from a missing stream, so unconditionally creating the group (then the stream) violated create_missing_stream=false and applied retention to preexisting groups. Fall back to a single DescribeLogStreams on ResourceNotFound to distinguish the two, restoring the original create/retention/flag semantics. Normal writes still go straight to PutLogEvents; the describe only runs when a stream is genuinely missing, not once per batch. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Vincent Voyer --- ...s_drop_describe_log_streams.enhancement.md | 2 +- src/sinks/aws_cloudwatch_logs/request.rs | 113 ++++++++++++------ src/sinks/aws_cloudwatch_logs/retry.rs | 12 +- src/sinks/aws_cloudwatch_logs/service.rs | 15 ++- 4 files changed, 104 insertions(+), 38 deletions(-) diff --git a/changelog.d/aws_cloudwatch_logs_drop_describe_log_streams.enhancement.md b/changelog.d/aws_cloudwatch_logs_drop_describe_log_streams.enhancement.md index d44140a5a117f..4f93a96f75b63 100644 --- a/changelog.d/aws_cloudwatch_logs_drop_describe_log_streams.enhancement.md +++ b/changelog.d/aws_cloudwatch_logs_drop_describe_log_streams.enhancement.md @@ -1,3 +1,3 @@ -The `aws_cloudwatch_logs` sink no longer calls `DescribeLogStreams` to fetch a sequence token before writing. Since January 2023 CloudWatch Logs accepts `PutLogEvents` without a sequence token and never returns `InvalidSequenceToken`, so the sink now writes directly and only calls `CreateLogGroup`/`CreateLogStream` when `PutLogEvents` returns `ResourceNotFoundException`. This removes one `DescribeLogStreams` request per batch (and one per retry), which is throttled account-and-region-wide at a low default quota (25 TPS) and could stall the sink under load on large fleets writing to many streams. +The `aws_cloudwatch_logs` sink no longer calls `DescribeLogStreams` before every batch to fetch a sequence token. Since January 2023 CloudWatch Logs accepts `PutLogEvents` without a sequence token and never returns `InvalidSequenceToken`, so the sink now writes directly and only falls back to `DescribeLogStreams` (to locate a missing group/stream) when `PutLogEvents` returns `ResourceNotFoundException`. This removes the per-batch (and per-retry) `DescribeLogStreams` request, which is throttled account-and-region-wide at a low default quota (25 TPS) and could stall the sink under load on large fleets writing to many streams. authors: vvo diff --git a/src/sinks/aws_cloudwatch_logs/request.rs b/src/sinks/aws_cloudwatch_logs/request.rs index c0b1d6040fc3d..a56793a485ed1 100644 --- a/src/sinks/aws_cloudwatch_logs/request.rs +++ b/src/sinks/aws_cloudwatch_logs/request.rs @@ -10,6 +10,7 @@ use aws_sdk_cloudwatchlogs::{ operation::{ create_log_group::CreateLogGroupError, create_log_stream::CreateLogStreamError, + describe_log_streams::{DescribeLogStreamsError, DescribeLogStreamsOutput}, put_log_events::{PutLogEventsError, PutLogEventsOutput}, put_retention_policy::PutRetentionPolicyError, }, @@ -31,14 +32,12 @@ pub struct CloudwatchFuture { // Batches still waiting to be sent after `current`. events: Vec>, // The batch currently in flight. Retained so it can be resent after a - // missing log group/stream is created: `PutLogEvents` returns - // `ResourceNotFoundException` when the stream does not exist yet, and we - // need the events to replay once it does. + // missing log group/stream is created. current: Vec, - // Set once we have created (or tried to create) the group/stream in - // response to a `ResourceNotFoundException`, so a second failure is a hard - // error rather than an infinite create loop. - created_missing: bool, + // Set once we've dropped into the describe/create path from a + // `ResourceNotFoundException`, so a second one is a hard error rather than + // an infinite resolve loop. + resolving: bool, } struct Client { @@ -56,6 +55,7 @@ type ClientResult = BoxFuture<'static, Result enum State { CreateGroup(ClientResult<(), CreateLogGroupError>), CreateStream(ClientResult<(), CreateLogStreamError>), + DescribeStream(ClientResult), Put(ClientResult), PutRetentionPolicy(ClientResult<(), PutRetentionPolicyError>), } @@ -88,12 +88,12 @@ impl CloudwatchFuture { // Since January 2023, CloudWatch Logs no longer requires a sequence // token on `PutLogEvents` and never returns `InvalidSequenceToken`, so - // we write directly instead of first calling `DescribeLogStreams` to - // fetch a token. That describe call ran once per batch (and again on - // every retry), and is account+region-wide throttled at a low default - // quota (25 TPS for `DescribeLogStreams`), so on large fleets it became - // the bottleneck and starved the sink. Writing straight to - // `PutLogEvents` removes that dependency entirely. + // we write directly. The old sink called `DescribeLogStreams` before + // every batch (and again on every retry) just to fetch that token, + // which is throttled account+region-wide at a low default quota + // (25 TPS) and starved the sink under load. We now only fall back to + // `DescribeLogStreams` when `PutLogEvents` reports the group/stream is + // missing, i.e. once per new stream, not once per batch. // https://aws.amazon.com/about-aws/whats-new/2023/01/amazon-cloudwatch-logs-log-stream-transaction-quota-sequencetoken-requirement/ let current = events.pop().expect("No Events to send"); let state = State::Put(client.put_logs(current.clone())); @@ -104,7 +104,7 @@ impl CloudwatchFuture { client, events, current, - created_missing: false, + resolving: false, state, create_missing_group, create_missing_stream, @@ -133,24 +133,12 @@ impl Future for CloudwatchFuture { } } Err(err) => { - // The stream (or its group) does not exist yet. - // Create it once, then replay the same batch. - if !self.created_missing - && is_resource_not_found(&err) - && (self.create_missing_group || self.create_missing_stream) - { - self.created_missing = true; - if self.create_missing_group { - info!( - "Log group provided does not exist; creating a new one." - ); - self.state = - State::CreateGroup(self.client.create_log_group()); - } else { - info!("Provided stream does not exist; creating a new one."); - self.state = - State::CreateStream(self.client.create_log_stream()); - } + // The group or stream is missing. Resolve it once + // with a describe (which distinguishes a missing + // group from a missing stream), then replay. + if !self.resolving && is_resource_not_found(&err) { + self.resolving = true; + self.state = State::DescribeStream(self.client.describe_stream()); continue; } return Poll::Ready(Err(CloudwatchError::Put(err))); @@ -158,6 +146,44 @@ impl Future for CloudwatchFuture { } } + State::DescribeStream(fut) => { + let response = match ready!(fut.poll_unpin(cx)) { + Ok(response) => response, + Err(err) => { + if let SdkError::ServiceError(inner) = &err + && matches!( + inner.err(), + DescribeLogStreamsError::ResourceNotFoundException(_) + ) + && self.create_missing_group + { + info!("Log group provided does not exist; creating a new one."); + + self.state = State::CreateGroup(self.client.create_log_group()); + continue; + } + return Poll::Ready(Err(CloudwatchError::DescribeLogStreams(err))); + } + }; + + let stream_name = &self.client.stream_name; + + if response + .log_streams + .ok_or(CloudwatchError::NoStreamsFound)? + .into_iter() + .any(|log_stream| log_stream.log_stream_name == Some(stream_name.clone())) + { + debug!(message = "Stream found.", stream = ?stream_name); + self.state = State::Put(self.client.put_logs(self.current.clone())); + } else if self.create_missing_stream { + info!("Provided stream does not exist; creating a new one."); + self.state = State::CreateStream(self.client.create_log_stream()); + } else { + return Poll::Ready(Err(CloudwatchError::NoStreamsFound)); + } + } + State::CreateGroup(fut) => { match ready!(fut.poll_unpin(cx)) { Ok(_) => {} @@ -183,8 +209,9 @@ impl Future for CloudwatchFuture { continue; } - // A newly created group never has any streams, so create - // one regardless of `create_missing_stream`. + // self does not abide by `create_missing_stream` since a group + // never has any streams and thus we need to create one if a group + // is created no matter what. self.state = State::CreateStream(self.client.create_log_stream()); } @@ -207,7 +234,7 @@ impl Future for CloudwatchFuture { info!(message = "Stream created.", name = %self.client.stream_name); - // Replay the batch that hit the missing group/stream. + // No sequence token needed, so replay the batch straight away. self.state = State::Put(self.client.put_logs(self.current.clone())); } @@ -263,6 +290,22 @@ impl Client { }) } + pub fn describe_stream( + &self, + ) -> ClientResult { + let client = self.client.clone(); + let group_name = self.group_name.clone(); + let stream_name = self.stream_name.clone(); + Box::pin(async move { + client + .describe_log_streams() + .log_group_name(group_name) + .log_stream_name_prefix(stream_name) + .send() + .await + }) + } + pub fn create_log_group(&self) -> ClientResult<(), CreateLogGroupError> { let client = self.client.clone(); let group_name = self.group_name.clone(); diff --git a/src/sinks/aws_cloudwatch_logs/retry.rs b/src/sinks/aws_cloudwatch_logs/retry.rs index 608ac225c8e41..11759f8f49976 100644 --- a/src/sinks/aws_cloudwatch_logs/retry.rs +++ b/src/sinks/aws_cloudwatch_logs/retry.rs @@ -1,7 +1,8 @@ use std::marker::PhantomData; use aws_sdk_cloudwatchlogs::operation::{ - create_log_stream::CreateLogStreamError, put_log_events::PutLogEventsError, + create_log_stream::CreateLogStreamError, describe_log_streams::DescribeLogStreamsError, + put_log_events::PutLogEventsError, }; use aws_smithy_runtime_api::client::result::SdkError; @@ -53,6 +54,15 @@ impl RetryLogic } is_retriable_error(err) } + CloudwatchError::DescribeLogStreams(err) => { + if let SdkError::ServiceError(inner) = err { + let err = inner.err(); + if matches!(err, DescribeLogStreamsError::ServiceUnavailableException(_)) { + return true; + } + } + is_retriable_error(err) + } CloudwatchError::CreateStream(err) => { if let SdkError::ServiceError(inner) = err { let err = inner.err(); diff --git a/src/sinks/aws_cloudwatch_logs/service.rs b/src/sinks/aws_cloudwatch_logs/service.rs index 835c751bcbd47..5f639b96809f7 100644 --- a/src/sinks/aws_cloudwatch_logs/service.rs +++ b/src/sinks/aws_cloudwatch_logs/service.rs @@ -8,7 +8,8 @@ use aws_sdk_cloudwatchlogs::{ Client as CloudwatchLogsClient, operation::{ create_log_group::CreateLogGroupError, create_log_stream::CreateLogStreamError, - put_log_events::PutLogEventsError, put_retention_policy::PutRetentionPolicyError, + describe_log_streams::DescribeLogStreamsError, put_log_events::PutLogEventsError, + put_retention_policy::PutRetentionPolicyError, }, types::InputLogEvent, }; @@ -64,21 +65,27 @@ type Svc = Buffer< #[derive(Debug)] pub enum CloudwatchError { Put(SdkError), + DescribeLogStreams(SdkError), CreateStream(SdkError), CreateGroup(SdkError), PutRetentionPolicy(SdkError), + NoStreamsFound, } impl fmt::Display for CloudwatchError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { CloudwatchError::Put(error) => write!(f, "CloudwatchError::Put: {error}"), + CloudwatchError::DescribeLogStreams(error) => { + write!(f, "CloudwatchError::DescribeLogStreams: {error}") + } CloudwatchError::CreateStream(error) => { write!(f, "CloudwatchError::CreateStream: {error}") } CloudwatchError::CreateGroup(error) => { write!(f, "CloudwatchError::CreateGroup: {error}") } + CloudwatchError::NoStreamsFound => write!(f, "CloudwatchError: No Streams Found"), CloudwatchError::PutRetentionPolicy(error) => { write!(f, "CloudwatchError::PutRetentionPolicy: {error}") } @@ -94,6 +101,12 @@ impl From> for CloudwatchError { } } +impl From> for CloudwatchError { + fn from(error: SdkError) -> Self { + CloudwatchError::DescribeLogStreams(error) + } +} + #[derive(Debug)] pub struct CloudwatchResponse { events_byte_size: GroupedCountByteSize,