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..4f93a96f75b63 --- /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` 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 de3b03da25396..a56793a485ed1 100644 --- a/src/sinks/aws_cloudwatch_logs/request.rs +++ b/src/sinks/aws_cloudwatch_logs/request.rs @@ -20,7 +20,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}; @@ -30,8 +29,15 @@ pub struct CloudwatchFuture { create_missing_group: bool, create_missing_stream: bool, retention_enabled: bool, + // Batches still waiting to be sent after `current`. events: Vec>, - token_tx: Option>>, + // The batch currently in flight. Retained so it can be resent after a + // missing log group/stream is created. + current: Vec, + // 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 { @@ -68,8 +74,6 @@ impl CloudwatchFuture { kms_key: Option, tags: Option>, mut events: Vec>, - token: Option, - token_tx: oneshot::Sender>, ) -> Self { let retention_days = retention.days; let client = Client { @@ -82,19 +86,26 @@ 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. 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())); let retention_enabled = retention.enabled; Self { client, events, + current, + resolving: false, state, - token_tx: Some(token_tx), create_missing_group, create_missing_stream, retention_enabled, @@ -108,6 +119,33 @@ impl Future for CloudwatchFuture { fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll { loop { match &mut self.state { + 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."); + return Poll::Ready(Ok(())); + } + } + Err(err) => { + // 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))); + } + } + } + State::DescribeStream(fut) => { let response = match ready!(fut.poll_unpin(cx)) { Ok(response) => response, @@ -130,23 +168,14 @@ impl Future for CloudwatchFuture { let stream_name = &self.client.stream_name; - if let Some(stream) = response + if response .log_streams .ok_or(CloudwatchError::NoStreamsFound)? .into_iter() - .find(|log_stream| log_stream.log_stream_name == Some(stream_name.clone())) + .any(|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)); + 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()); @@ -175,7 +204,8 @@ 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; } @@ -204,29 +234,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(())); - } + // No sequence token needed, so replay the batch straight away. + self.state = State::Put(self.client.put_logs(self.current.clone())); } State::PutRetentionPolicy(fut) => { @@ -246,10 +255,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 +277,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() diff --git a/src/sinks/aws_cloudwatch_logs/service.rs b/src/sinks/aws_cloudwatch_logs/service.rs index cdb34702aa413..5f639b96809f7 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::{ @@ -23,7 +23,6 @@ use http::{ }; use indexmap::IndexMap; use snafu::{ResultExt, Snafu}; -use tokio::sync::oneshot; use tower::{ Service, ServiceBuilder, ServiceExt, buffer::Buffer, @@ -199,7 +198,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) @@ -259,8 +258,6 @@ impl CloudwatchLogsSvc { retention, kms_key, tags, - token: None, - token_rx: None, } } @@ -310,38 +307,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, + ) } } @@ -355,8 +339,6 @@ pub struct CloudwatchLogsSvc { retention: Retention, kms_key: Option, tags: Option>, - token: Option, - token_rx: Option>>, } impl EncodedLength for InputLogEvent {