socket source: passed UDP file descriptor not processed until shutdown

[aagor]

A note for the community

• Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
• If you are interested in working on this issue or have submitted a pull request, please leave a comment

Problem

When using socket activation (via systemd) for an UDP socket as described in the docs here, vector does not process it correctly.

The logs show, that events are received, however, they are not forwarded to the sink.
When quitting vector via SIGINT (Ctrl+C) or SIGTERM, vector doesn't terminate in a timely manner, as the socket source doesn't terminate correctly.
Even after waiting the graceful shutdown limit, vector still doesn't terminate.

Only when sending new data to the UDP socket, vector finally ends.
When that happens, previous events are finally forwarded to the relevant sources.

All of this does not happen when using UDP without file descriptor passing (comment out address: systemd#1, comment in address: 127.0.0.1:12345).
I expect no observable difference when letting vector open an UDP socket vs passing an already existing UDP socket file descriptor in to vector.

Configuration

sources:
  systemd_udp_socket_activation:
    type: socket
    #address: 127.0.0.1:12345
    address: systemd#1
    mode: udp
    decoding:
      codec: "json"

sinks:
  stdout:
    type: console
    inputs:
      - systemd_udp_socket_activation
    encoding:
      codec: json
      json:
        pretty: true

Version

vector 0.51.1 (x86_64-unknown-linux-gnu 44c8f1c 2025-11-13 15:16:05.303418529)

Debug Output

systemd-socket-activate --datagram --listen 127.0.0.1:12345 -E RUST_BACKTRACE=full -E VECTOR_INTERNAL_LOG_RATE_LIMIT=0 vector -vvv --config vector.yaml


Listening on 127.0.0.1:12345 as 3.
Communication attempt on fd 3.
Execing vector (vector -vvv --config vector.yaml)
2025-11-25T16:15:40.466594Z DEBUG vector::app: Internal log rate limit configured. internal_log_rate_secs=0
2025-11-25T16:15:40.466684Z  INFO vector::app: Log level is enabled. level="trace"
2025-11-25T16:15:40.466735Z DEBUG vector::app: Building runtime. worker_threads=8
2025-11-25T16:15:40.466874Z TRACE mio::poll: registering event source with poller: token=Token(1), interests=READABLE
2025-11-25T16:15:40.494958Z  INFO vector::app: Loading configs. paths=["vector.yaml"]
2025-11-25T16:15:40.506131Z DEBUG vector::config::loading: No secret placeholder found, skipping secret resolution.
2025-11-25T16:15:40.507217Z DEBUG vector::topology::builder: Building new source. component_id=systemd_udp_socket_activation
2025-11-25T16:15:40.508562Z DEBUG vector::topology::builder: Building new sink. component_id=stdout
2025-11-25T16:15:40.509465Z  INFO vector::topology::running: Running healthchecks.
2025-11-25T16:15:40.509631Z DEBUG vector::topology::running: Connecting changed/added component(s).
2025-11-25T16:15:40.509657Z DEBUG vector::topology::running: Configuring outputs for source. component_id=systemd_udp_socket_activation
2025-11-25T16:15:40.509669Z DEBUG vector::topology::running: Configuring output for component. component_id=systemd_udp_socket_activation output_id=None
2025-11-25T16:15:40.509675Z DEBUG vector::topology::running: Connecting inputs for sink. component_id=stdout
2025-11-25T16:15:40.509683Z DEBUG vector::topology::running: Adding component input to fanout. component_id=stdout fanout_id=systemd_udp_socket_activation
2025-11-25T16:15:40.509717Z DEBUG vector::topology::running: Spawning new source. component_id=systemd_udp_socket_activation
2025-11-25T16:15:40.509794Z  INFO vector::topology::builder: Healthcheck passed.
2025-11-25T16:15:40.509830Z TRACE vector::topology::running: Spawning new sink. component_id=stdout
2025-11-25T16:15:40.509859Z  INFO vector: Vector has started. debug="false" version="0.51.1" arch="x86_64" revision="44c8f1c 2025-11-13 15:16:05.303418529"
2025-11-25T16:15:40.509928Z  INFO vector::app: API is disabled, enable by setting `api.enabled` to `true` and use commands like `vector top`.
2025-11-25T16:15:40.509855Z DEBUG source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector::topology::builder: Source pump supervisor starting.
2025-11-25T16:15:40.510001Z DEBUG source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector::topology::builder: Source pump starting.
2025-11-25T16:15:40.510033Z DEBUG source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector::topology::builder: Source starting.
2025-11-25T16:15:40.510039Z DEBUG sink{component_kind="sink" component_id=stdout component_type=console}: vector::topology::builder: Sink starting.
2025-11-25T16:15:40.510155Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: mio::poll: registering event source with poller: token=Token(140597450375296), interests=READABLE | WRITABLE
2025-11-25T16:15:40.510257Z  INFO source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector::sources::socket::udp: Listening. address=systemd socket #0
2025-11-25T16:15:40.510521Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector_common::internal_event::bytes_received: Bytes received. byte_size=19 protocol=udp
2025-11-25T16:15:40.511161Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector::internal_events::socket: Events received. count=1 byte_size=62 mode=udp
2025-11-25T16:15:40.511258Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector_buffers::topology::channel::limited_queue: Sent item.
2025-11-25T16:15:40.511269Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector_common::internal_event::events_sent: Events sent. count=1 byte_size=119 output=_default
2025-11-25T16:15:40.513552Z TRACE vector: Beep.
2025-11-25T16:15:41.933236Z TRACE vector: Beep.
2025-11-25T16:15:42.951368Z TRACE vector: Beep.
2025-11-25T16:15:43.731512Z TRACE vector: Beep.
2025-11-25T16:15:44.149704Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector_common::internal_event::bytes_received: Bytes received. byte_size=25 protocol=udp
2025-11-25T16:15:44.149792Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector::internal_events::socket: Events received. count=1 byte_size=68 mode=udp
2025-11-25T16:15:44.149828Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector_buffers::topology::channel::limited_queue: Sent item.
2025-11-25T16:15:44.149832Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector_common::internal_event::events_sent: Events sent. count=1 byte_size=125 output=_default
2025-11-25T16:15:44.686167Z TRACE vector: Beep.
2025-11-25T16:15:45.513219Z TRACE vector: Beep.
2025-11-25T16:15:46.874931Z TRACE vector: Beep.
2025-11-25T16:15:47.511333Z TRACE vector: Beep.
2025-11-25T16:15:48.535573Z TRACE vector: Beep.
2025-11-25T16:15:49.528906Z TRACE vector: Beep.
2025-11-25T16:15:49.919276Z  INFO vector::signal: Signal received. signal="SIGTERM"
2025-11-25T16:15:49.919592Z  INFO vector: Vector has stopped.
2025-11-25T16:15:49.919942Z DEBUG sink{component_kind="sink" component_id=stdout component_type=console}: vector::utilization: Couldn't send utilization start wait message. component_id=ComponentKey { id: "stdout" } error="Closed(..)"
2025-11-25T16:15:49.920012Z DEBUG sink{component_kind="sink" component_id=stdout component_type=console}: vector::utilization: Couldn't send utilization start wait message. component_id=ComponentKey { id: "stdout" } error="Closed(..)"
2025-11-25T16:15:49.920075Z  INFO vector::topology::running: Shutting down... Waiting on running components. remaining_components="stdout, systemd_udp_socket_activation" time_remaining="59 seconds left"
2025-11-25T16:15:50.523278Z DEBUG sink{component_kind="sink" component_id=stdout component_type=console}: vector::utilization: Couldn't send utilization start wait message. component_id=ComponentKey { id: "stdout" } error="Closed(..)"
2025-11-25T16:15:50.523364Z TRACE vector: Beep.
2025-11-25T16:15:51.531153Z TRACE vector: Beep.
2025-11-25T16:15:52.560313Z TRACE vector: Beep.
2025-11-25T16:15:53.515193Z TRACE vector: Beep.
2025-11-25T16:15:54.518591Z TRACE vector: Beep.
2025-11-25T16:15:54.929487Z  INFO vector::topology::running: Shutting down... Waiting on running components. remaining_components="stdout, systemd_udp_socket_activation" time_remaining="54 seconds left"
2025-11-25T16:15:54.936062Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector_common::internal_event::bytes_received: Bytes received. byte_size=34 protocol=udp
2025-11-25T16:15:54.936186Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector::internal_events::socket: Events received. count=1 byte_size=77 mode=udp
2025-11-25T16:15:54.936325Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: mio::poll: deregistering event source from poller
2025-11-25T16:15:54.936397Z DEBUG source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector::topology::builder: Source finished normally.
2025-11-25T16:15:54.936431Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector_core::fanout: Processing control message outside of send: ControlMessage::Add(ComponentKey { id: "stdout" })
2025-11-25T16:15:54.936469Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector_core::fanout: Processing control message inside of send: None
2025-11-25T16:15:54.936493Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector_buffers::topology::channel::limited_queue: Sent item.
2025-11-25T16:15:54.936501Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector_core::fanout: Sent item to fanout.
2025-11-25T16:15:54.936504Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector_core::fanout: Processing control message inside of send: None
2025-11-25T16:15:54.936508Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector_buffers::topology::channel::limited_queue: Sent item.
2025-11-25T16:15:54.936511Z TRACE source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector_core::fanout: Sent item to fanout.
2025-11-25T16:15:54.936515Z DEBUG source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector::topology::builder: Source pump finished normally.
2025-11-25T16:15:54.936585Z DEBUG source{component_kind="source" component_id=systemd_udp_socket_activation component_type=socket}: vector::topology::builder: Source pump supervisor task finished normally.
2025-11-25T16:15:54.936651Z DEBUG sink{component_kind="sink" component_id=stdout component_type=console}: vector::utilization: Couldn't send utilization start wait message. component_id=ComponentKey { id: "stdout" } error="Closed(..)"
2025-11-25T16:15:54.936674Z DEBUG sink{component_kind="sink" component_id=stdout component_type=console}: vector::utilization: Couldn't send utilization stop wait message. component_id=ComponentKey { id: "stdout" } error="Closed(..)"
2025-11-25T16:15:54.936680Z TRACE sink{component_kind="sink" component_id=stdout component_type=console}: vector_common::internal_event::events_received: Events received. count=1 byte_size=119
2025-11-25T16:15:54.936863Z TRACE sink{component_kind="sink" component_id=stdout component_type=console}: vector_common::internal_event::events_sent: Events sent. count=1 byte_size=117
2025-11-25T16:15:54.936907Z TRACE sink{component_kind="sink" component_id=stdout component_type=console}: vector_common::internal_event::bytes_sent: Bytes sent. byte_size=139 protocol=console
2025-11-25T16:15:54.936916Z DEBUG sink{component_kind="sink" component_id=stdout component_type=console}: vector::utilization: Couldn't send utilization start wait message. component_id=ComponentKey { id: "stdout" } error="Closed(..)"
2025-11-25T16:15:54.936927Z DEBUG sink{component_kind="sink" component_id=stdout component_type=console}: vector::utilization: Couldn't send utilization stop wait message. component_id=ComponentKey { id: "stdout" } error="Closed(..)"
2025-11-25T16:15:54.936930Z TRACE sink{component_kind="sink" component_id=stdout component_type=console}: vector_common::internal_event::events_received: Events received. count=1 byte_size=125
{
  "hello": "world",
  "host": "127.0.0.1",
  "port": 52295,
  "source_type": "socket",
  "timestamp": "2025-11-25T16:15:40.511146868Z"
}
2025-11-25T16:15:54.947754Z TRACE sink{component_kind="sink" component_id=stdout component_type=console}: vector_common::internal_event::events_sent: Events sent. count=1 byte_size=123
2025-11-25T16:15:54.949495Z TRACE sink{component_kind="sink" component_id=stdout component_type=console}: vector_common::internal_event::bytes_sent: Bytes sent. byte_size=145 protocol=console
2025-11-25T16:15:54.949560Z DEBUG sink{component_kind="sink" component_id=stdout component_type=console}: vector::utilization: Couldn't send utilization start wait message. component_id=ComponentKey { id: "stdout" } error="Closed(..)"
2025-11-25T16:15:54.949741Z DEBUG sink{component_kind="sink" component_id=stdout component_type=console}: vector::utilization: Couldn't send utilization stop wait message. component_id=ComponentKey { id: "stdout" } error="Closed(..)"
2025-11-25T16:15:54.949832Z DEBUG sink{component_kind="sink" component_id=stdout component_type=console}: vector::topology::builder: Sink finished normally.
{
  "hello": "world again",
  "host": "127.0.0.1",
  "port": 53256,
  "source_type": "socket",
  "timestamp": "2025-11-25T16:15:44.149780981Z"
}

Example Data

Test with the following commands:

systemd-socket-activate --datagram --listen 127.0.0.1:12345 -E RUST_BACKTRACE=full -E VECTOR_INTERNAL_LOG_RATE_LIMIT=0 vector -vvv --config vector.yaml

echo '{"hello": "world"}' |socat STDIN UDP:127.0.0.1:12345; \
sleep 3; \
echo '{"hello": "world again"}' |socat STDIN UDP:127.0.0.1:12345; \
sleep 5; \
pkill vector; \
sleep 5; \
echo '{"hello": "send after terminate"}' |socat STDIN UDP:127.0.0.1:12345;

Additional Context

The last message send after terminate (echo '{"hello": "send after terminate"}' |socat STDIN UDP:127.0.0.1:12345;) does not appear in the output.
I guess that this message is needed to wake a thread from a blocking syscall.
I wouldn't myself consider this a bug, but the info may be helpful for you.

References

No response

[j-c-fuchs]

With the master version from commit b9ad9b3ec1ade765583e4d06fc7142f8e6b745a2, I get a panic with the following error message:

Registering a blocking socket with the tokio runtime is unsupported. If you wish to do anyways, please add `--cfg tokio_allow_from_blocking_fd` to your RUSTFLAGS. See github.com/tokio-rs/tokio/issues/7172 for details.

The same issue applies when using syslog source with systemd socket activation:

# vector.yaml
sources:
  systemd_syslog_socket_activation:
    type: syslog
    #address: 127.0.0.1:12345
    address: systemd#1
    mode: udp
    decoding:
      codec: "json"

sinks:
  stdout:
    type: console
    inputs:
      - systemd_syslog_socket_activation
    encoding:
      codec: "json"
      json:
        pretty: true

#!/usr/bin/bash

send_logs() {
  sleep 5
  echo '<13>1 2020-03-13T20:45:38.119Z dynamicwireless.name non 2426 ID931 [exampleSDID@32473 iut="3" eventSource="Application" eventID="1011"] hello world' | socat STDIN UDP:127.0.0.1:12345
  sleep 3
  echo '<13>1 2020-03-13T20:45:38.119Z dynamicwireless.name non 2426 ID931 [exampleSDID@32473 iut="3" eventSource="Application" eventID="1011"] hello world again' | socat STDIN UDP:127.0.0.1:12345
  sleep 5
  pkill vector
  sleep 5
  echo '<13>1 2020-03-13T20:45:38.119Z dynamicwireless.name non 2426 ID931 [exampleSDID@32473 iut="3" eventSource="Application" eventID="1011"] hello world after kill' | socat STDIN UDP:127.0.0.1:12345
}

send_logs &
systemd-socket-activate \
  --datagram --listen 127.0.0.1:12345 \
  -E RUST_BACKTRACE=full \
  -E VECTOR_INTERNAL_LOG_RATE_LIMIT=0 \
  ./target/debug/vector -vvv \
  --config vector.yaml

[j-c-fuchs]

I made a fix on this branch: https://github.com/j-c-fuchs/vector/tree/fix/udp-socket-activation
Maybe, I try to add integration tests for this case.

[aagor]

To summarize:

• systemd passes the opened file descriptor in blocking mode.
• vector (or tokio) expects the file descriptor in non-blocking mode.
• When vector opens the UDP socket itself, it sets the socket to non-blocking mode himself (it seems like tokio does that internally automatically).
• systemd can't know if consumers of the socket want it in non-blocking or blocking mode.
• Therefore vector should handle that case, setting the socket to non-blocking when needed.

I made a tiny Python script, that can be used as a workaround:

#!/usr/bin/env python3

import os
import sys

os.set_blocking(3, False)
os.execvp(sys.argv[1], sys.argv[1:])

This can be used as a "pre-start script". With that applied, vector behaves as I would expect.

systemd-socket-activate --datagram --listen 127.0.0.1:12345 -E VECTOR_INTERNAL_LOG_RATE_LIMIT=0 ./set_fd_3_nonblocking.py vector -vvv --config vector.yaml

[j-c-fuchs]

As far as I see, TCP sockets suffer the same problem:

# vector-tcp.yaml
sources:
  systemd_tcp_socket_activation:
    type: socket
    #address: 127.0.0.1:12345
    address: systemd#1
    mode: tcp
    decoding:
      codec: "json"

sinks:
  stdout:
    type: console
    inputs:
      - systemd_tcp_socket_activation
    encoding:
      codec: "json"
      json:
        pretty: true

#!/usr/bin/bash

send_logs() {
  sleep 5
  echo '{"hello": "world"}' | socat STDIN TCP:127.0.0.1:12345
  sleep 3
  echo '{"hello": "world again"}' | socat STDIN TCP:127.0.0.1:12345
  sleep 5
  pkill vector
  sleep 5
  echo '{"hello": "send after terminate"}' | socat STDIN TCP:127.0.0.1:12345
}

send_logs &
systemd-socket-activate \
  --listen 127.0.0.1:12345 \
  -E RUST_BACKTRACE=full \
  -E VECTOR_INTERNAL_LOG_RATE_LIMIT=0 \
  ./target/debug/vector -vvv \
  --config vector-tcp.yaml