fix(amqp sink): attempt one reconnect when channel has errored #22971

aramperes · 2025-05-01T02:54:32Z

Summary

Improves the resilience of the amqp sink by attempting to re-create a broken Channel when sending new events.

This is a simple implementation that could be extended later or in this PR, depending on the Vector team's preference & vision. The current implementation:

only attempts a reconnect when sending an event fails, and will not attempt multiple times or with exponential back-off. This could be better integrated with the tower retry mechanism in some way;
does not make the re-connection behavior configurable;
does not implement the re-connection for the amqp source, which would be more complex to re-create the listeners.

Lapin does not have a built-in reconnection mechanism and it is up to the client to implement this. See amqp-rs/lapin#70, amqp-rs/lapin#389

Implementation

A new AmqpChannel struct wraps a tokio::sync::RwLock<lapin::Channel>. This read-write lock is used to lock during the re-connection to prevent concurrent attempts.

The API could be further improved by making crate::amqp::AmqpConfig::connect() return an AmqpChannel directly, but that would also affect the amqp source.

Change Type

Bug fix
New feature
Non-functional (chore, refactoring, docs)
Performance

Is this a breaking change?

Yes
No

How did you test this PR?

Run RabbitMQ in a container or remote server.
Run Vector with an amqp sink. Example:

sources:
  in:
    type: http_server
    address: 0.0.0.0:8000
    codec: json

sinks:
  sink_amqp:
    type: amqp
    inputs:
      - in
    connection_string: amqp://user:password@127.0.0.1:5672/%2f?timeout=10
    routing_key: example-key
    exchange: example-exchange
    encoding:
      codec: json

Restart the RabbitMQ container/process. The ioloop in Lapin will show as broken.
Send a new event through Vector. Vector will try to reconnect the channel since it is in Error state. If the RabbitMQ process hasn't booted up and the re-connection failed, the event is dropped. Otherwise, the channel is re-established and the event is sent to the new channel.

Example of the re-connection succeeding:

2025-05-01T02:40:39.687291Z  INFO vector::app: Log level is enabled. level="info"
2025-05-01T02:40:39.701858Z  INFO vector::app: Loading configs. paths=["vector.yaml"]
2025-05-01T02:40:39.736801Z  INFO vector::topology::running: Running healthchecks.
2025-05-01T02:40:39.737012Z  INFO vector: Vector has started. debug="false" version="0.47.0" arch="x86_64" revision=""
2025-05-01T02:40:39.737014Z  INFO source{component_kind="source" component_id=in component_type=http_server}: vector::sources::util::http::prelude: Building HTTP server. address=0.0.0.0:8000
2025-05-01T02:40:39.736933Z  INFO vector::topology::builder: Healthcheck passed.
2025-05-01T02:40:43.868982Z ERROR lapin::io_loop: Socket was readable but we read 0. This usually means that the connection is half closed this mark it as broken
2025-05-01T02:40:43.869422Z ERROR lapin::io_loop: error doing IO error=IOError(Kind(ConnectionAborted))
2025-05-01T02:40:43.869447Z ERROR lapin::channels: Connection error error=IO error: connection aborted
2025-05-01T02:40:53.114888Z  INFO sink{component_kind="sink" component_id=sink_amqp component_type=amqp}:request{request_id=2}: vector::sinks::amqp::channel: Recovering broken connection to the AMQP broker. internal_log_rate_limit=true
2025-05-01T02:40:53.120366Z  INFO sink{component_kind="sink" component_id=sink_amqp component_type=amqp}:request{request_id=2}: vector::sinks::amqp::channel: Recovered connection to the AMQP broker. internal_log_rate_limit=true
^C2025-05-01T02:40:58.323475Z  INFO vector::signal: Signal received. signal="SIGINT"
2025-05-01T02:40:58.323566Z  INFO vector: Vector has stopped.
2025-05-01T02:40:58.325213Z  INFO vector::topology::running: Shutting down... Waiting on running components. remaining_components="sink_amqp" time_remaining="59 seconds left"

Example of re-connection failing:

2025-05-01T02:45:58.663389Z  INFO vector::app: Log level is enabled. level="info"
2025-05-01T02:45:58.676135Z  INFO vector::app: Loading configs. paths=["vector.yaml"]
2025-05-01T02:45:58.705614Z  INFO vector::topology::running: Running healthchecks.
2025-05-01T02:45:58.705760Z  INFO vector: Vector has started. debug="false" version="0.47.0" arch="x86_64" revision=""
2025-05-01T02:45:58.705784Z  INFO vector::app: API is disabled, enable by setting `api.enabled` to `true` and use commands like `vector top`.
2025-05-01T02:45:58.705778Z  INFO source{component_kind="source" component_id=in component_type=http_server}: vector::sources::util::http::prelude: Building HTTP server. address=0.0.0.0:8000
2025-05-01T02:45:58.705744Z  INFO vector::topology::builder: Healthcheck passed.
2025-05-01T02:46:01.964157Z ERROR lapin::channel: Connection closed channel=0 method=Close { reply_code: 320, reply_text: ShortString("CONNECTION_FORCED - broker forced connection closure with reason 'shutdown'"), class_id: 0, method_id: 0 } error=AMQPError { kind: Hard(CONNECTIONFORCED), message: ShortString("CONNECTION_FORCED - broker forced connection closure with reason 'shutdown'") }
2025-05-01T02:46:01.964395Z ERROR lapin::channels: Connection error error=protocol error: AMQP hard error: CONNECTION-FORCED: CONNECTION_FORCED - broker forced connection closure with reason 'shutdown'
2025-05-01T02:46:04.668137Z  INFO sink{component_kind="sink" component_id=sink_amqp component_type=amqp}:request{request_id=1}: vector::sinks::amqp::channel: Recovering broken connection to the AMQP broker. internal_log_rate_limit=true
2025-05-01T02:46:04.668767Z ERROR sink{component_kind="sink" component_id=sink_amqp component_type=amqp}:request{request_id=1}: vector_common::internal_event::service: Service call failed. No retries or retries exhausted. error=Some(ConnectFailed { error: IOError(Os { code: 111, kind: ConnectionRefused, message: "Connection refused" }) }) request_id=1 error_type="request_failed" stage="sending" internal_log_rate_limit=true
2025-05-01T02:46:04.668831Z ERROR sink{component_kind="sink" component_id=sink_amqp component_type=amqp}:request{request_id=1}: vector_common::internal_event::component_events_dropped: Events dropped intentional=false count=1 reason="Service call failed. No retries or retries exhausted." internal_log_rate_limit=true

Does this PR include user facing changes?

Yes. Please add a changelog fragment based on our guidelines.
No. A maintainer will apply the "no-changelog" label to this PR.

Notes

Please read our Vector contributor resources.
Do not hesitate to use @vectordotdev/vector to reach out to us regarding this PR.
The CI checks run only after we manually approve them.
- We recommend adding a pre-push hook, please see this template.
- Alternatively, we recommend running the following locally before pushing to the remote branch:
  - cargo fmt --all
  - cargo clippy --workspace --all-targets -- -D warnings
  - cargo nextest run --workspace (alternatively, you can run cargo test --all)
  - ./scripts/check_changelog_fragments.sh
After a review is requested, please avoid force pushes to help us review incrementally.
- Feel free to push as many commits as you want. They will be squashed into one before merging.
- For example, you can run git merge origin master and git push.
If this PR introduces changes Vector dependencies (modifies Cargo.lock), please
run cargo vdev build licenses to regenerate the license inventory and commit the changes (if any). More details here.

References

Closes #22313

pront

Thanks @aramperes

pront · 2025-05-09T18:27:52Z

src/sinks/amqp/channel.rs

+            }
+
+            info!(
+                message = "Recovering broken connection to the AMQP broker.",


Nit: We can add a URL tag here

I'm not sure that is a good idea since it could print the password:

INFO sink{component_kind="sink" component_id=sink_amqp component_type=amqp}:request{request_id=1}: vector::sinks::amqp::channel: Recovering broken connection to the AMQP broker. uri="amqp://user:password@127.0.0.1:5672/%2f?timeout=10" internal_log_rate_limit=true INFO sink{component_kind="sink" component_id=sink_amqp component_type=amqp}:request{request_id=1}: vector::sinks::amqp::channel: Recovered connection to the AMQP broker. uri="amqp://user:password@127.0.0.1:5672/%2f?timeout=10" internal_log_rate_limit=true

src/sinks/amqp/channel.rs

src/sinks/amqp/service.rs

aramperes · 2025-05-10T17:36:21Z

@pront Any concerns with potentially integrating the deadpool crate into Vector? This could provide a cleaner interface for recycling AMQP channels while also supporting a pool of connections & channels instead of a single one.

pront · 2025-05-14T19:49:07Z

@pront Any concerns with potentially integrating the deadpool crate into Vector? This could provide a cleaner interface for recycling AMQP channels while also supporting a pool of connections & channels instead of a single one.

No objections. It seems like light and widely used dependency.

aramperes · 2025-05-14T22:32:55Z

@pront Any concerns with potentially integrating the deadpool crate into Vector? This could provide a cleaner interface for recycling AMQP channels while also supporting a pool of connections & channels instead of a single one.

No objections. It seems like light and widely used dependency.

Ok sounds good, I will try a proof of concept and see if it provides value. The official deadpool-lapin implementation is not quite what we need here, but the interface is pretty easy to implement.

aramperes · 2025-05-22T03:11:31Z

@pront I've integrated deadpool to maintain and recycle the amqp sink channel using a pool. I hard-coded the pool size to 4 but it will only grow beyond 1 channel if there are multiple messages being received fast enough for this to be needed. 4 channels should be lightweight enough to not cause any regressions but we could make this configurable in the sink as well.

I tested the recovery behavior and it is the same as with my previous implementation. When picking a Channel from the pool, it will check its current state. If it is not Connected, the pool attempts to create a new Channel. If that fails an error bubbles up and the message is dropped. If the new Channel works, the message is published.

Logs:

>>>>>>>> STARTUP <<<<<<<<
 INFO vector::sinks::amqp::channel: Created a new channel to the AMQP broker. id=1
 INFO vector::topology::builder: Healthcheck passed.
>>>>>>>> Rabbit killed <<<<<<<<
ERROR lapin::io_loop: Socket was readable but we read 0. This usually means that the connection is half closed this mark it as broken
ERROR lapin::io_loop: error doing IO error=IOError(Kind(ConnectionAborted))
ERROR lapin::channels: Connection error error=IO error: connection aborted
>>>>>>>> Reconnect attempted but Rabbit still dead <<<<<<<<
ERROR lapin::io_loop: error doing IO error=IOError(Os { code: 104, kind: ConnectionReset, message: "Connection reset by peer" })
ERROR lapin::channels: Connection error error=IO error: Connection reset by peer (os error 104)
ERROR lapin::channels: Connection error error=IO error: Connection reset by peer (os error 104)
ERROR sink{component_kind="sink" component_id=sink_amqp component_type=amqp}:request{request_id=2}: vector_common::internal_event::service: Service call failed. No retries or retries exhausted. error=Some(PoolError { error: Backend(ConnectFailed { error: IOError(Os { code: 104, kind: ConnectionReset, message: "Connection reset by peer" }) }) }) request_id=2 error_type="request_failed" stage="sending" internal_log_rate_limit=true
ERROR sink{component_kind="sink" component_id=sink_amqp component_type=amqp}:request{request_id=2}: vector_common::internal_event::component_events_dropped: Events dropped intentional=false count=1 reason="Service call failed. No retries or retries exhausted." internal_log_rate_limit=true
>>>>>>>> Next message, Rabbit recovered <<<<<<<<
 INFO sink{component_kind="sink" component_id=sink_amqp component_type=amqp}:request{request_id=3}: vector::sinks::amqp::channel: Created a new channel to the AMQP broker. id=1

aramperes · 2025-06-04T11:48:34Z

Hi @pront this is ready for review when you have some time, thanks!

pront

Hi @aramperes, this looks great

pront · 2025-06-11T13:11:14Z

src/sinks/amqp/channel.rs

+pub(super) fn new_channel_pool(config: &AmqpSinkConfig) -> crate::Result<AmqpSinkChannels> {
+    let channel_manager = AmqpSinkChannelManager::new(&config.connection);
+    let channels = Pool::builder(channel_manager)
+        .max_size(4)


Let's make this configurable and default to 4. Also, let's mention it in the changelog.

…changelog

aramperes · 2025-06-12T16:22:50Z

@pront There was a formatting issue with the docs, could you re-trigger the CI? 🙏

pront · 2025-06-12T17:53:51Z

@pront There was a formatting issue with the docs, could you re-trigger the CI? 🙏

Done. But I wonder if this should have been auto-generated with make generate-component-docs.

…econnect

aramperes · 2025-06-12T23:49:51Z

Ahhh. I tried running cargo vdev build component-docs directly and it wasn't working... my branch is now properly formatted.

I updated the DEVELOPING.md doc as well, and I updated my branch with the fix for the spell-check.

aramperes requested a review from a team as a code owner May 1, 2025 02:54

github-actions bot added the domain: sinks Anything related to the Vector's sinks label May 1, 2025

pront reviewed May 9, 2025

View reviewed changes

pront added the sink: amqp Anything `amqp` sink related label May 9, 2025

pront approved these changes Jun 11, 2025

View reviewed changes

aramperes requested a review from a team as a code owner June 12, 2025 01:58

github-actions bot added the domain: external docs Anything related to Vector's external, public documentation label Jun 12, 2025

aramperes added 8 commits June 11, 2025 22:07

fix(amqp sink): attempt one reconnect when channel has errored

ca5f5f8

add changelog.d

1295c1f

fix changelog.d

a199e35

Use vector_common::Error

a0cde40

use deadpool to manage channel recovery and multiple channels

cdd660d

Rename AmqpSink

db7723f

Update integration tests

c5d29f5

make amqp max_channels configurable; default to 4; added to docs and …

d1c0726

…changelog

aramperes force-pushed the bugfix-22313-amqp-reconnect branch from e4e33f3 to d1c0726 Compare June 12, 2025 02:13

aramperes added 2 commits June 11, 2025 23:24

dupe word

6f46416

Update 3rd party licenses for deadpool

e19ab8b

estherk15 approved these changes Jun 12, 2025

View reviewed changes

pront enabled auto-merge June 12, 2025 13:39

Fix formatting in amqp.cue

12b59e5

auto-merge was automatically disabled June 12, 2025 15:20
Head branch was pushed to by a user without write access

Update DEVELOPING.md

b6180e5

aramperes requested a review from a team as a code owner June 12, 2025 23:47

github-actions bot added domain: sources Anything related to the Vector's sources domain: transforms Anything related to Vector's transform components labels Jun 12, 2025

Merge remote-tracking branch 'origin/master' into bugfix-22313-amqp-r…

01c5278

…econnect

aramperes force-pushed the bugfix-22313-amqp-reconnect branch from db683a1 to 01c5278 Compare June 12, 2025 23:49

github-actions bot removed domain: sources Anything related to the Vector's sources domain: transforms Anything related to Vector's transform components labels Jun 12, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

fix(amqp sink): attempt one reconnect when channel has errored #22971

fix(amqp sink): attempt one reconnect when channel has errored #22971

aramperes commented May 1, 2025

Uh oh!

pront left a comment

Uh oh!

pront May 9, 2025

Uh oh!

aramperes May 10, 2025

Uh oh!

Uh oh!

Uh oh!

aramperes commented May 10, 2025

Uh oh!

pront commented May 14, 2025

Uh oh!

aramperes commented May 14, 2025

Uh oh!

aramperes commented May 22, 2025 •

edited

Loading

Uh oh!

aramperes commented Jun 4, 2025

Uh oh!

pront left a comment

Uh oh!

pront Jun 11, 2025

Uh oh!

aramperes commented Jun 12, 2025

Uh oh!

pront commented Jun 12, 2025

Uh oh!

aramperes commented Jun 12, 2025

Uh oh!

Uh oh!

fix(amqp sink): attempt one reconnect when channel has errored #22971

Are you sure you want to change the base?

fix(amqp sink): attempt one reconnect when channel has errored #22971

Conversation

aramperes commented May 1, 2025

Summary

Implementation

Change Type

Is this a breaking change?

How did you test this PR?

Does this PR include user facing changes?

Notes

References

Uh oh!

pront left a comment

Choose a reason for hiding this comment

Uh oh!

pront May 9, 2025

Choose a reason for hiding this comment

Uh oh!

aramperes May 10, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

aramperes commented May 10, 2025

Uh oh!

pront commented May 14, 2025

Uh oh!

aramperes commented May 14, 2025

Uh oh!

aramperes commented May 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

aramperes commented Jun 4, 2025

Uh oh!

pront left a comment

Choose a reason for hiding this comment

Uh oh!

pront Jun 11, 2025

Choose a reason for hiding this comment

Uh oh!

aramperes commented Jun 12, 2025

Uh oh!

pront commented Jun 12, 2025

Uh oh!

aramperes commented Jun 12, 2025

Uh oh!

Uh oh!

aramperes commented May 22, 2025 •

edited

Loading