feat(pusher): Add concurrent log publishing with circuit breaker and retry heap

[the-mann]

Summary

Add multi-threaded log publishing to CloudWatch Logs with circuit breaker isolation and retry heap for failed batches.

Problem

The CloudWatch agent publishes logs synchronously — one batch at a time per target. When a target encounters persistent failures (e.g., AccessDenied), the retry loop blocks the entire pipeline, starving healthy log groups.

Solution

Architecture

Queue → SenderPool → WorkerPool (N workers) → Sender → PutLogEvents
                                                  ↓ (on failure)
                                            RetryHeap (unbounded min-heap)
                                                  ↓ (on retry time)
                                         RetryHeapProcessor → WorkerPool

Key Components

• WorkerPool: Shared pool of N goroutines for concurrent PutLogEvents calls
• RetryHeap: Unbounded min-heap ordered by next retry time for failed batches
• RetryHeapProcessor: Periodically pops ready batches and resubmits to worker pool
• Circuit Breaker: Per-target halt/resume — failing target blocks its own queue without affecting others

Circuit Breaker Flow

1. Batch fails → batch.fail() → queue.halt() — queue stops sending new batches
2. Failed batch pushed to RetryHeap with backoff
3. RetryHeapProcessor retries batch later
4. Success → batch.done() → queue.resume() — queue resumes
5. Expired (14d) → batch.done() → queue resumes, batch dropped

Poison Pill Fix

The retry heap is unbounded. A previous bounded implementation (size = concurrency) caused deadlock when failing log groups exceeded the heap size — workers blocked on Push(), starving all targets.

Changes

• batch.go: Add retry metadata, state/fail/done callbacks, expiry logic
• queue.go: Add channel-based halt/resume circuit breaker with mutex protection
• sender.go: Push to retry heap on failure, call batch.fail() for circuit breaker
• retryheap.go: Unbounded min-heap + RetryHeapProcessor with periodic retry loop
• pool.go: WorkerPool and SenderPool for concurrent dispatch
• pusher.go: Wire up retry heap and worker pool when concurrency enabled
• cloudwatchlogs.go: Create shared WorkerPool and RetryHeap

Testing

• Unit tests: 27 new tests covering retry heap, circuit breaker, poison pill, state callbacks, shutdown deadlock
• Race detector: All tests pass with -race
• Key scenarios tested:
  • 10 denied + 1 allowed log group with concurrency=2 (poison pill)
  • Shutdown while queue is halted (no deadlock)
  • Batch expiry after 14 days resumes circuit breaker and persists state
  • State callbacks fire on success and expiry but not on shutdown

Related PRs

• feat(pusher): Add circuit breaker to halt queue on target failure #1975 — Circuit breaker (sender-block-on-failure)
• Reapply "Add default log publishing concurrency (#1770)" (#1819) #1977 — Enable multi-threaded logging by default
• fix(clean): Delete inline policies before IAM role deletion #2022 — IAM role cleanup fix (unrelated, merged from main)

[the-mann]

Incorporated review feedback. Integration tests passing:

• Full suite: https://github.com/aws/amazon-cloudwatch-agent/actions/runs/22458358617
• Concurrency tests (recovery-test branch): https://github.com/aws/amazon-cloudwatch-agent/actions/runs/22458610567

[the-mann]

Updated PR with review feedback and merged latest from main. Commit to review: 9aabc04

Integration tests passing:

• Full suite: https://github.com/aws/amazon-cloudwatch-agent/actions/runs/22458358617
• Concurrency tests (recovery-test branch): https://github.com/aws/amazon-cloudwatch-agent/actions/runs/22458610567

[github-actions]

This PR was marked stale due to lack of activity.