enhancement(aws_cloudwatch_logs sink): stop calling DescribeLogStreams for sequence tokens

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

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<Vec<InputLogEvent>>,
-    token_tx: Option<oneshot::Sender<Option<String>>>,
+    // The batch currently in flight. Retained so it can be resent after a
+    // missing log group/stream is created.
+    current: Vec<InputLogEvent>,
+    // 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<String>,
         tags: Option<HashMap<String, String>>,
         mut events: Vec<Vec<InputLogEvent>>,
-        token: Option<String>,
-        token_tx: oneshot::Sender<Option<String>>,
     ) -> 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<Self::Output> {
         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<PutLogEventsError, HttpResponse>) -> bool {
+    matches!(
+        err,
+        SdkError::ServiceError(inner)
+            if matches!(inner.err(), PutLogEventsError::ResourceNotFoundException(_))
+    )
+}
+
 impl Client {
     pub fn put_logs(
         &self,
-        sequence_token: Option<String>,
         log_events: Vec<InputLogEvent>,
     ) -> ClientResult<PutLogEventsOutput, PutLogEventsError> {
         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()

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<BatchCloudwatchRequest> 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<Vec<InputLogEvent>> for CloudwatchLogsSvc {
     type Error = CloudwatchError;
     type Future = request::CloudwatchFuture;
 
-    fn poll_ready(&mut self, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
-        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<Result<(), Self::Error>> {
         Poll::Ready(Ok(()))
     }
 
     fn call(&mut self, req: Vec<InputLogEvent>) -> 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<String>,
     tags: Option<HashMap<String, String>>,
-    token: Option<String>,
-    token_rx: Option<oneshot::Receiver<Option<String>>>,
 }
 
 impl EncodedLength for InputLogEvent {