feat(redis source): add reconnecting pubsub session for channel data_type#25892
feat(redis source): add reconnecting pubsub session for channel data_type#25892artusmode wants to merge 2 commits into
data_type#25892Conversation
…_type`
- Rework channel source to maintain a pubsub session that auto-reconnects
and re-subscribes after Redis disconnects.
- Add shutdown-aware backoff and best-effort unsubscribe on exit.
- Log when the connection is re-established ("re-established and resubscribed").
Resolves vectordotdev#22615
Follow-up to the reconnecting pub/sub work: - Replace the ad-hoc `backoff_exponential` helper with the shared `common::backoff::ExponentialBackoff` (from_millis(2).factor(250), capped at 30s), matching the strategy used by other reconnecting sources such as `aws_s3`/`sqs`. Backoff resets on a successful connect. - Emit internal events for reconnection so operators can observe/alert: `RedisConnectionError` (increments `component_errors_total` with error_type=connection_failed) on connect/subscribe failure, and `RedisConnectionEstablished` (increments `connection_established_total`) on initial connect and recovery. - Unify the `list` data_type source (`list.rs`) onto the same shared `ExponentialBackoff`, removing its duplicate `backoff_exponential` helper and making its retry sleep shutdown-aware. - Add a changelog fragment. Co-authored-by: Gibran <gibran.zaman@finaccel.co>
github-actions
Bot
added
the
domain: sources
|
Thank you for your contribution! Before we can merge this PR, please sign our Contributor License Agreement. To sign, copy and post the phrase below as a new comment on this PR.
I have read the CLA Document and I hereby sign the CLA 1 out of 2 committers have signed the CLA. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: aa85aeb2f1
ℹ️ 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".
| loop { | ||
| // connect + SUBSCRIBE | ||
| let mut pubsub_conn = | ||
| match connect_and_subscribe(&client, &endpoint, &channel).await { |
There was a problem hiding this comment.
When the channel source is in the reconnect loop and a Redis connect/subscribe attempt stalls, for example against a black-holed TCP/TLS endpoint, this await is not raced with shutdown. A shutdown signal that arrives after the backoff completes or during the connect will not be observed until the Redis client/OS connect attempt returns, so the source misses graceful shutdown and Vector waits for the force-shutdown deadline. Please wrap connect_and_subscribe in a tokio::select! with the shutdown signal and/or apply a per-attempt timeout.
Useful? React with 👍 / 👎.
| // `aws_s3`/`sqs`. Reset once a value is successfully received. | ||
| let mut backoff = ExponentialBackoff::from_millis(2) | ||
| .factor(250) | ||
| .max_delay(Duration::from_secs(30)); |
There was a problem hiding this comment.
For existing data_type = "list" configurations, this reused backoff now grows to 30 seconds after repeated Redis I/O errors. During a Redis outage the list source can reach that cap, and once Redis is back it may still wait up to 30 seconds before the next BLPOP/BRPOP; the removed helper capped retries at 1 second, so this unrelated list change is a significant recovery regression. Keep the list source's prior cap or avoid changing its retry policy while adding channel reconnects.
Useful? React with 👍 / 👎.
| match end_reason { | ||
| SessionEnd::Shutdown | SessionEnd::DownstreamClosed => { | ||
| // shutting down cleanly, or downstream closed: stop for good. | ||
| let _ = pubsub_conn.unsubscribe(&channel).await; |
There was a problem hiding this comment.
When shutdown is selected while the Pub/Sub session is still connected, this branch now waits for Redis to acknowledge UNSUBSCRIBE before the source can exit. If Redis is slow or the TCP connection is half-open during shutdown, graceful shutdown can block until the force-shutdown deadline; the previous take_until(shutdown) path stopped immediately by dropping the connection. Since the component is already stopping, avoid a network round trip here or race it with the shutdown/force path.
Useful? React with 👍 / 👎.
2 participants
Supersedes #24100 (rebased onto master; original work by @gibranbadrul, preserved as the first commit). Closes #22615.
The
redissource withdata_type = "channel"currently stops receiving after any Redis connection drop and only recovers on a Vector restart. This adds a reconnect loop that re-connects and re-subscribes automatically, with exponential backoff.Addresses the review feedback on #24100:
common::backoff::ExponentialBackoff(from_millis(2).factor(250), capped at 30s), matchingaws_s3/sqs, instead of a bespoke helper. Backoff resets on a successful connect.RedisConnectionError→component_errors_total(error_type=connection_failed) on connect/subscribe failure, andRedisConnectionEstablished→connection_established_totalon initial connect and recovery.Also unifies the
listdata_type source onto the same sharedExponentialBackoff(removing its duplicatebackoff_exponential) and makes its retry sleep shutdown-aware.