feat(docker_logs source): add exponential retry to docker client

[titaneric]

Summary

Present docker_logs source would not retry the request after the event watcher stop (which would cause all container logs streamer stop immediately), and this PR would like to add the exponential backoff retry to reset the event watcher.

Vector configuration

api:
  enabled: true
  graphql: true
  playground: true
  address: "127.0.0.1:8686"
sources:
  my_source_id:
    type: docker_logs
    docker_host: http://127.0.0.1:2376
sinks:
  my_sink_id:
    type: console
    inputs:
    - my_source_id
    encoding:
      codec: raw_message

How did you test this PR?

download docker proxy and launch it to simulate the slow docker request (bind to 127.0.0.1:2376)

wget https://gist.githubusercontent.com/titaneric/e1b35ccf37f3907ee9accb426d8c2d48/raw/54e711bfe91f399889b71fc96c50ec68cd4491b3/gistfile1.txt main.py
python3 -m pip install aiohttp
python3 main.py --delay 30

open another terminal, and run the vector

cargo build --no-default-features --features sources-docker_logs --features sinks-console --features api
./target/debug/vector -v --config docker-source.yaml

AS IS

2025-10-23T17:24:22.020374Z ERROR source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::internal_events::docker_logs: Error in communication with Docker daemon. error=RequestTimeoutError error_type="connection_failed" stage="receiving" container_id=None internal_log_rate_limit=true
2025-10-23T17:24:22.020499Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::topology::builder: Source finished normally.
2025-10-23T17:24:46.766150Z DEBUG vector::utilization: component_id=my_sink_id utilization=0
2025-10-23T17:24:52.044724Z ERROR source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::internal_events::docker_logs: Error in communication with Docker daemon. error=RequestTimeoutError error_type="connection_failed" stage="receiving" container_id=Some("9902c3061be53eb8a8fde424be98a7a192a1530c4b956d136781754eb7341f4f") internal_log_rate_limit=true
2025-10-23T17:24:52.044827Z  INFO source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::internal_events::docker_logs: Stopped watching for container logs. container_id=9902c3061be53eb8a8fde424be98a7a192a1530c4b956d136781754eb7341f4f
2025-10-23T17:24:52.045001Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::topology::builder: Source pump finished normally.
2025-10-23T17:24:52.045505Z  INFO vector_common::shutdown: All sources have finished.
2025-10-23T17:24:52.045956Z  INFO vector_common::shutdown: Internal log [All sources have finished.] is being suppressed to avoid flooding.
2025-10-23T17:24:52.045977Z  INFO vector::app: All sources have finished.
2025-10-23T17:24:52.046028Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::topology::builder: Source pump supervisor task finished normally.
2025-10-23T17:24:52.046081Z  INFO vector: Vector has stopped.
2025-10-23T17:24:52.046217Z DEBUG sink{component_kind="sink" component_id=my_sink_id component_type=console}: vector::utilization: Couldn't send utilization start wait message. component_id=ComponentKey { id: "my_sink_id" } error="Closed(..)"
2025-10-23T17:24:52.046250Z DEBUG sink{component_kind="sink" component_id=my_sink_id component_type=console}: vector::utilization: Couldn't send utilization stop wait message. component_id=ComponentKey { id: "my_sink_id" } error="Closed(..)"
2025-10-23T17:24:52.046287Z DEBUG sink{component_kind="sink" component_id=my_sink_id component_type=console}: vector::topology::builder: Sink finished normally.
2025-10-23T17:24:52.046407Z DEBUG hyper::server::shutdown: signal received, starting graceful shutdown

TO BE

2025-10-23T17:13:18.572773Z ERROR source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::internal_events::docker_logs: Error in communication with Docker daemon. error=RequestTimeoutError error_type="connection_failed" stage="receiving" container_id=None internal_log_rate_limit=true
2025-10-23T17:13:18.573134Z  WARN source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::sources::docker_logs: Docker events stream failed action="retrying with backoff" delay_ms=2000
2025-10-23T17:13:20.574785Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: log: Internal log [http://127.0.0.1:2376/events?since=1761239448&filters=%7B%22type%22%3A%5B%22container%22%5D%2C%22event%22%3A%5B%22start%22%2C%22unpause%22%2C%22die%22%2C%22pause%22%5D%7D] has been suppressed 1 times.
2025-10-23T17:13:20.574931Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: bollard::docker: http://127.0.0.1:2376/events?since=1761239448&filters=%7B%22type%22%3A%5B%22container%22%5D%2C%22event%22%3A%5B%22start%22%2C%22unpause%22%2C%22die%22%2C%22pause%22%5D%7D
2025-10-23T17:13:20.575291Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: hyper_util::client::legacy::connect::http: connecting to 127.0.0.1:2376
2025-10-23T17:13:20.576224Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: hyper_util::client::legacy::connect::http: connected to 127.0.0.1:2376
2025-10-23T17:13:43.285606Z DEBUG vector::utilization: component_id=my_sink_id utilization=0
2025-10-23T17:13:48.612973Z ERROR source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::internal_events::docker_logs: Error in communication with Docker daemon. error=RequestTimeoutError error_type="connection_failed" stage="receiving" container_id=Some("9902c3061be53eb8a8fde424be98a7a192a1530c4b956d136781754eb7341f4f") internal_log_rate_limit=true
2025-10-23T17:13:48.613166Z  INFO source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::internal_events::docker_logs: Stopped watching for container logs. container_id=9902c3061be53eb8a8fde424be98a7a192a1530c4b956d136781754eb7341f4f
2025-10-23T17:13:50.615800Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: bollard::docker: http://127.0.0.1:2376/containers/9902c3061be53eb8a8fde424be98a7a192a1530c4b956d136781754eb7341f4f/json
2025-10-23T17:13:50.615893Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: hyper_util::client::legacy::connect::http: connecting to 127.0.0.1:2376
2025-10-23T17:13:50.616135Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: hyper_util::client::legacy::connect::http: connected to 127.0.0.1:2376
...
2025-10-23T17:14:20.642094Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: log: Internal log [http://127.0.0.1:2376/events?since=1761239448&filters=%7B%22type%22%3A%5B%22container%22%5D%2C%22event%22%3A%5B%22start%22%2C%22unpause%22%2C%22die%22%2C%22pause%22%5D%7D] is being suppressed to avoid flooding.
2025-10-23T17:14:20.642214Z  INFO source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::internal_events::docker_logs: Started watching for container logs. container_id=9902c3061be53eb8a8fde424be98a7a192a1530c4b956d136781754eb7341f4f
2025-10-23T17:14:20.642518Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: hyper_util::client::legacy::connect::http: connecting to 127.0.0.1:2376
2025-10-23T17:14:20.643017Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: hyper_util::client::legacy::connect::http: connected to 127.0.0.1:2376
2025-10-23T17:14:33.285477Z DEBUG vector::utilization: component_id=my_sink_id utilization=0
2025-10-23T17:14:58.284977Z DEBUG vector::utilization: component_id=my_sink_id utilization=0
2025-10-23T17:15:20.578451Z ERROR source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::internal_events::docker_logs: Error in communication with Docker daemon. error=RequestTimeoutError error_type="connection_failed" stage="receiving" container_id=None internal_log_rate_limit=true
2025-10-23T17:15:20.578880Z  WARN source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::sources::docker_logs: Docker events stream failed action="retrying with backoff" delay_ms=300000

Notice that the event listener would retry after backoff time.

Change Type

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

Is this a breaking change?

• Yes
• No

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.

References

Closes: #23608 and closes #23847

Notes

• Please read our Vector contributor resources.
• Do not hesitate to use @vectordotdev/vector to reach out to us regarding this PR.
• Some 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:
    • make fmt
    • make check-clippy (if there are failures it's possible some of them can be fixed with make clippy-fix)
    • make test
• 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 make build-licenses to regenerate the license inventory and commit the changes (if any). More details here.

[pront]

Hey @titaneric some small checks are failing. Also, worth testing this locally and see that the actual backoff duration is what we expected.

[titaneric]

First retry, please note that event API is requested after 2 seconds

2025-10-31T11:10:12.726723Z ERROR source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::internal_events::docker_logs: Error in communication with Docker daemon. error=RequestTimeoutError error_type="connection_failed" stage="receiving" container_id=None
2025-10-31T11:10:12.726891Z  WARN source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::sources::docker_logs: Docker events stream failed action="retrying with backoff" delay_ms=2000
2025-10-31T11:10:14.729423Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: log: Internal log [http://127.0.0.1:2376/events?since=1761908862&filters=%7B%22event%22%3A%5B%22start%22%2C%22unpause%22%2C%22die%22%2C%22pause%22%5D%2C%22type%22%3A%5B%22container%22%5D%7D] has been suppressed 3 times.
2025-10-31T11:10:14.729581Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: bollard::docker: http://127.0.0.1:2376/events?since=1761908862&filters=%7B%22type%22%3A%5B%22container%22%5D%2C%22event%22%3A%5B%22start%22%2C%22unpause%22%2C%22die%22%2C%22pause%22%5D%7D

Second retry after 4 seconds

2025-10-31T11:12:14.732886Z ERROR source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::internal_events::docker_logs: Internal log [Error in communication with Docker daemon.] has been suppressed 1 times.
2025-10-31T11:12:14.733018Z ERROR source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::internal_events::docker_logs: Error in communication with Docker daemon. error=RequestTimeoutError error_type="connection_failed" stage="receiving" container_id=None
2025-10-31T11:12:14.733394Z  WARN source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::sources::docker_logs: Docker events stream failed action="retrying with backoff" delay_ms=4000
2025-10-31T11:12:17.691618Z DEBUG vector::utilization: component_id=my_sink_id utilization=0
2025-10-31T11:12:18.736935Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: log: Internal log [http://127.0.0.1:2376/events?since=1761908862&filters=%7B%22event%22%3A%5B%22start%22%2C%22unpause%22%2C%22die%22%2C%22pause%22%5D%2C%22type%22%3A%5B%22container%22%5D%7D] has been suppressed 3 times.
2025-10-31T11:12:18.737034Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: bollard::docker: http://127.0.0.1:2376/events?since=1761908862&filters=%7B%22type%22%3A%5B%22container%22%5D%2C%22event%22%3A%5B%22start%22%2C%22unpause%22%2C%22die%22%2C%22pause%22%5D%7D

Third retry after 8 seconds

2025-10-31T11:14:18.743472Z ERROR source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::internal_events::docker_logs: Internal log [Error in communication with Docker daemon.] has been suppressed 1 times.
2025-10-31T11:14:18.743549Z ERROR source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::internal_events::docker_logs: Error in communication with Docker daemon. error=RequestTimeoutError error_type="connection_failed" stage="receiving" container_id=None
2025-10-31T11:14:18.743796Z  WARN source{component_kind="source" component_id=my_source_id component_type=docker_logs}: vector::sources::docker_logs: Docker events stream failed action="retrying with backoff" delay_ms=8000
2025-10-31T11:14:22.696631Z DEBUG vector::utilization: component_id=my_sink_id utilization=0
2025-10-31T11:14:26.747244Z DEBUG source{component_kind="source" component_id=my_source_id component_type=docker_logs}: log: Internal log [http://127.0.0.1:2376/events?since=1761908862&filters=%7B%22event%22%3A%5B%22start%22%2C%22unpause%22%2C%22die%22%2C%22pause%22%5D%2C%22type%22%3A%5B%22container%22%5D%7D] has been suppressed 3 times.