out_cloudwatch_logs: isolate stream caches per output worker

[ParakhJaggi]

cloudwatch_logs output workers share ctx->streams and mutate it without synchronization. After an entry's four-hour expiration, one worker can unlink/free it while another worker traverses or unlinks it. This crashes collectors that use multiple output workers.

Allocate one stream-cache list per configured output worker and select it by the existing thread-pool worker ID. Keep the instance cache for workers 0. Destroy all caches after the output worker pool stops. Stream expiry, timestamps, and entity state then belong to one worker. This preserves parallel output processing and avoids holding a mutex across asynchronous CloudWatch requests.

Compatibility: configuration and record contents stay unchanged. Each worker can cache the same CloudWatch stream independently, so cache memory and initial CreateLogStream calls can increase with worker count. The existing ResourceAlreadyExists handling covers duplicate stream creation.

Evidence

A local reproduction used synthetic dummy inputs, a HTTPS CloudWatch mock, static dummy AWS credentials, add_entity false, and workers 2. A clock shim advanced time() by 14,405 seconds every five real seconds to exercise expiry repeatedly. It did not change the system clock or contact AWS.

• AWS image 3.3.0 / Fluent Bit 5.0.3: SIGSEGV and exit 133 after 45 seconds, without OOM.
• AWS image 3.4.12 / Fluent Bit 5.0.9: SIGSEGV and exit 133 after 21 seconds, without OOM.
• The one-worker control and the two-worker control without clock acceleration survived more than five minutes, with zero output errors/retries/drops.
• An unfixed debug build stopped in GDB with SIGSEGV, si_code=SEGV_MAPERR, address 0x8. mk_list_del() received an entry with both list pointers already null:

__mk_list_del(prev=0x0, next=0x0)
mk_list_del(entry=...)
get_or_create_log_stream(...)
get_log_stream(...)
process_log_events(...)
process_and_send(...)
cb_cloudwatch_flush(...)

The fixed two-worker build completed a bounded 180-second accelerated-expiry run without a crash. A snapshot before the run ended showed 40,005 delivered records and zero output errors, retries, or drops. The test harness then stopped the process with SIGTERM.

Tests

Three new runtime cases cover workers 0, workers 1, and independent ownership/expiry for two workers. The isolation test selects worker identities on an otherwise idle real output context. It deterministically fails against the original shared-cache lookup (first != second), then passes with this change. It expires one worker's entry repeatedly while the other worker's entry remains valid. Shutdown exercises cache cleanup.

ctest --test-dir /build -R '^flb-rt-out_cloudwatch$' --output-on-failure

PASS: all 19 CloudWatch runtime cases, 46.26 seconds.

valgrind --error-exitcode=99 --leak-check=full \
  --show-leak-kinds=definite,indirect \
  --errors-for-leak-kinds=definite,indirect \
  --suppressions=/src/valgrind.supp \
  /build/bin/flb-rt-out_cloudwatch --no-exec \
  stream_cache_no_workers stream_cache_one_worker stream_cache_worker_expiry

PASS on Linux aarch64: zero errors, zero bytes allocated at exit, 18,252 allocations and frees. No suppressions were used. No CloudWatch scenario exists under tests/integration; the focused runtime suite and the local HTTPS mock provide coverage. Full PR-range commit-prefix lint and git diff --check also pass.

This is separate from the entity-buffer memory leak in #12426. It does not fix the AWS credential-refresh race reported in #12206.

Summary by CodeRabbit

• Bug Fixes

  • Improved CloudWatch Logs stream handling for configurations using multiple workers.
  • Stream state is now maintained independently for each worker, preventing unrelated workers from affecting one another.
  • Expired streams are refreshed correctly without unnecessarily increasing the cache.

• Tests

  • Added coverage for shared and worker-specific stream caching, including expiration and refresh scenarios across multiple workers.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Advanced

Run ID: 9d6158b9-f79c-4fbc-ad01-9db48e6beb70

📥 Commits

Reviewing files that changed from the base of the PR and between 69e5f07 and 0069bd6.

📒 Files selected for processing (4)

• plugins/out_cloudwatch_logs/cloudwatch_api.c
• plugins/out_cloudwatch_logs/cloudwatch_logs.c
• plugins/out_cloudwatch_logs/cloudwatch_logs.h
• tests/runtime/out_cloudwatch.c

Included review availability: Your plan provides up to 8 included reviews per hour; 7 remain after this review.

---

📝 Walkthrough

Walkthrough

Changes

CloudWatch stream cache isolation

Layer / File(s)	Summary
Per-worker cache lifecycle plugins/out_cloudwatch_logs/cloudwatch_logs.h, plugins/out_cloudwatch_logs/cloudwatch_logs.c	The CloudWatch context adds per-worker stream lists, initializes them from tp_workers, and cleans up their streams during destruction.
Runtime cache selection plugins/out_cloudwatch_logs/cloudwatch_api.c	Threaded instances use the active worker's stream cache. Non-threaded instances continue to use the shared cache. Invalid worker resolution returns an error.
Cache behavior validation tests/runtime/out_cloudwatch.c	Runtime tests cover non-threaded caching, single-worker caching, and independent stream expiry across two workers.

Priority: ⬇️ Low

Estimated code review effort: 3 (Moderate) | ~25 minutes

Change: Bug fix

Sequence Diagram(s)

sequenceDiagram
  participant CloudWatchOutput
  participant OutputThreadInstance
  participant StreamCache
  CloudWatchOutput->>OutputThreadInstance: Resolve active worker
  OutputThreadInstance-->>CloudWatchOutput: Return worker thread id
  CloudWatchOutput->>StreamCache: Select worker-local cache
  StreamCache-->>CloudWatchOutput: Return or refresh log stream

Loading

Suggested reviewers: cosmo0920

Merge Risk: ⚪ Minimal · up to 0069b

The worker-local cache selection and cleanup paths are consistent with the output worker lifecycle. No current merge-blocking risk was identified.

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 22.22% which is insufficient. The required threshold is 80.00%. Docstring coverage is scoped to functions touched by this diff. Analyzed 9 functions across 4 files.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly and concisely describes the main change: isolating CloudWatch Logs stream caches for each output worker.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

• Fix all pre-merge checks with AI

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 0069bd6448

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".