feat(vector source, vector sink): Add vector source fetch mode and sink serve mode#25808
feat(vector source, vector sink): Add vector source fetch mode and sink serve mode#25808Voldemat wants to merge 8 commits into
Conversation
github-actions
Bot
added
domain: sources
|
All contributors have signed the CLA ✍️ ✅ |
This comment has been minimized.
This comment has been minimized.
![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: 6c3bbdb2df
ℹ️ 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".
| }), | ||
| ))); | ||
| while let Some(batch) = futures::StreamExt::next(&mut batched_stream).await { | ||
| let _ = self.emitter.send(batch.events); |
There was a problem hiding this comment.
When no fetch client is currently subscribed, broadcast::Sender::send returns an error containing the unsent batch; this _ = drops that error and the events. Because the finalizers were already removed from the events while batching and are never marked failed, upstream acknowledgements can still complete as delivered even though the serve-mode sink discarded the batch during startup gaps or client disconnects.
Useful? React with 👍 / 👎.
| break; | ||
| } | ||
| Err(_) => { | ||
| error!("Error reading from stream"); | ||
| break; |
There was a problem hiding this comment.
If the serve side restarts, closes the RPC, or the broadcast stream ends after an error, these breaks let the source future fall through to Ok(()), so the component stops pulling and never reconnects until Vector is restarted. The retry policy only wraps the initial PullEvents call, not subsequent stream termination, so fetch mode loses all future events after any transient stream closure.
Useful? React with 👍 / 👎.
| .map_err(|error| { | ||
| error!(message = "Sink serve future failed.", %error); | ||
| }); | ||
| tokio::spawn(source); |
There was a problem hiding this comment.
run_grpc_server_with_routes performs the socket bind when this spawned future is polled, so errors such as an address already in use happen after build has returned; the map_err above only logs and the healthcheck returned for the sink is always Ok(()). In that serve-mode scenario the component starts and consumes input, but no PullEvents listener is running for fetch clients.
Useful? React with 👍 / 👎.
| } | ||
| VectorMode::Fetch => { | ||
| let client = new_client(&tls_settings, &cx.proxy)?; | ||
| let uri = with_default_scheme(&format!("{}", self.address), tls_settings.is_tls())?; |
There was a problem hiding this comment.
In fetch mode this uses the already-deserialized SocketAddr, so configurations that need to connect to a DNS name such as a Kubernetes Service (address: vector-source:6000) are rejected before this code can run. That makes the new client mode usable only with literal IP addresses, unlike the vector sink's existing client-side address handling.
Useful? React with 👍 / 👎.
| let events = response | ||
| .events | ||
| .into_iter() | ||
| .map(Event::from) | ||
| .collect::<Vec<_>>(); |
There was a problem hiding this comment.
The receive path adds the standard vector source metadata before forwarding events, and outputs() advertises that metadata in the schema, but the fetch path converts the streamed wrappers directly and sends them unchanged. Fetched log events therefore miss source_type = "vector" and the vector ingest timestamp unless an upstream component happened to set them, breaking pipelines that rely on the vector source schema matching runtime events.
Useful? React with 👍 / 👎.
| Ok(Box::pin(source)) | ||
| } | ||
| VectorMode::Fetch => { | ||
| let client = new_client(&tls_settings, &cx.proxy)?; |
There was a problem hiding this comment.
In fetch mode this tls_settings value was created at the start of build with for_server = true for the receive listener, but here it is reused for an outbound client connector. A normal TLS client config with only CA verification fails as MissingRequiredIdentity, and a config with a client cert inherits the server defaults that disable certificate and hostname verification, so TLS fetch connections are either unusable or under-verified.
Useful? React with 👍 / 👎.
| let mut proto_client = proto_vector::Client::new(HyperSvc { | ||
| uri, | ||
| client: hyper_client, | ||
| }); |
There was a problem hiding this comment.
This client receives PullEventsResponse batches, but it keeps tonic's default inbound message limit instead of matching the existing vector receive server's unlimited decoding behavior. Large log events or a 1000-event batch can exceed the default 4 MiB response limit, causing stream.message() to error and the fetch source to stop reading from the serve sink.
Useful? React with 👍 / 👎.
| let mut batched_stream = Box::pin(input.batched(self.batch_settings.as_reducer_config( | ||
| |event: &crate::event::Event| event.size_of(), | ||
| BatchReduce::new(|batch: &mut EventBatch, mut event: crate::event::Event| { | ||
| batch.finalizers.merge(event.take_finalizers()); |
There was a problem hiding this comment.
When serve-mode sink acknowledgements are enabled and a fetch client is connected, this removes the original finalizers before converting events to protobuf, but those finalizers are then dropped at the end of the loop without waiting for the fetch-side pipeline or its sinks to accept the batch. That makes upstream sources see the batch as delivered as soon as it is broadcast locally, so any later fetch disconnection or downstream failure cannot be reported back through end-to-end acknowledgements.
Useful? React with 👍 / 👎.
| let emitter = self.emitter.subscribe(); | ||
| let mut stream = tokio_stream::wrappers::BroadcastStream::new(emitter); |
There was a problem hiding this comment.
Using a bounded broadcast channel means a fetch client that falls more than 10,000 batches behind gets a Lagged error after older batches have already been overwritten, and the stream is closed rather than slowing the serve sink. In high-throughput or slow-downstream scenarios this loses events even while the sink continues accepting input, which is not the normal lossless/backpressure behavior expected from Vector sinks.
Useful? React with 👍 / 👎.
| timeout: std::time::Duration::from_secs(3), | ||
| rate_limit_duration: std::time::Duration::from_secs(3), | ||
| rate_limit_num: 100, | ||
| retry_attempts: 10, |
There was a problem hiding this comment.
If the serve endpoint is unavailable at startup for longer than this bounded retry window, the initial PullEvents call returns an error and the fetch source future exits instead of waiting for the remote Vector to come back. That leaves deployments that start the fetching side before the serving side permanently disconnected until Vector itself is restarted.
Useful? React with 👍 / 👎.
|
I have read the CLA Document and I hereby sign the CLA |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 2933c747ab
ℹ️ 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".
| .map_err(|_| -> crate::Error { | ||
| "failed to parse address into SocketAddr".into() | ||
| })?, | ||
| tls, |
There was a problem hiding this comment.
When mode = "serve" and TLS is enabled, this passes the tls value that was built once at the top of build with client semantics (for_server = false) into the gRPC listener. Client-side defaults make verify_certificate true, so the server acceptor requires fetch clients to present a certificate by default; a normal TLS fetch source that only verifies the serve sink's certificate will fail the handshake. Build TLS settings with server semantics inside the Serve branch before binding.
Useful? React with 👍 / 👎.
| .ok_or_else(|| -> crate::Error { | ||
| "address must be defined if mode is serve".into() | ||
| })? | ||
| .parse::<std::net::SocketAddr>() |
There was a problem hiding this comment.
In mode = "serve", this reuses the vector sink's existing address option, which is documented and validated as a URI/hostname-friendly endpoint for push mode, but then parses it as a literal SocketAddr. A config that is valid for the same sink today, such as address = "http://127.0.0.1:6000" or address = "localhost:6000", fails before the listener is started, so users cannot switch those sink configs to serve mode without changing the address format unexpectedly.
Useful? React with 👍 / 👎.
| let events = response | ||
| .events | ||
| .into_iter() | ||
| .map(Event::from) | ||
| .collect::<Vec<_>>(); |
There was a problem hiding this comment.
When mode = "fetch", this path forwards the pulled batch without emitting the EventsReceived counters that the receive-mode vector source emits before sending downstream. As a result, fetched events still appear as sent by the source, but component_received_events_total and component_received_event_bytes_total stay underreported for the vector source, breaking the standard source telemetry expected by component validation and dashboards.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: edffbcbddd
ℹ️ 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".
| service, | ||
| }) | ||
| VectorMode::Serve => { | ||
| super::serve::config_to_serve_sink(self).await |
There was a problem hiding this comment.
In mode = "serve" this branch starts a gRPC listener, but this SinkConfig impl still inherits the default empty resources() list. Vector's config validation and topology reload logic collect SinkConfig::resources() for conflict detection, so a serve sink can be configured on the same TCP port as a source or another serve sink without being rejected or reload-ordered before the old listener is released; the conflict then turns into a runtime bind failure/race instead of a configuration error.
Useful? React with 👍 / 👎.
| }); | ||
|
|
||
| if let Some(encoding) = compression.as_tonic_encoding() { | ||
| proto_client = proto_client.send_compressed(encoding); |
There was a problem hiding this comment.
When mode = "fetch" and compression is configured, this only applies tonic compression to the outbound PullEventsRequest, which carries no event data. The large payloads are the server-streamed PullEventsResponse batches, and the serve side builds proto::Server::new(service) without response compression, so users who enable compression for the new pull path still send all fetched events uncompressed.
Useful? React with 👍 / 👎.
| tokio::spawn(source); | ||
| Ok(( | ||
| crate::sinks::VectorSink::from_event_streamsink(super::serve::ServeSink { | ||
| emitter, | ||
| batch_settings: config.batch.into_batcher_settings()?, |
There was a problem hiding this comment.
In serve mode, a config with invalid batch settings such as batch.max_events = 0 reaches this spawn before config.batch.into_batcher_settings()? is evaluated. If that later conversion returns an error, build fails but the spawned gRPC listener has no retained handle or cancellation path, leaving the port bound and serving even though the sink was rejected.
Useful? React with 👍 / 👎.
| let mut stream = tokio_stream::wrappers::BroadcastStream::new(emitter); | ||
| Ok(tonic::Response::new(Box::pin(async_stream::stream! { | ||
| while let Some(result) = tokio_stream::StreamExt::next(&mut stream).await { | ||
| yield Ok(proto::PullEventsResponse { |
There was a problem hiding this comment.
When mode = "serve" and a fetch client is connected, this is the path that actually delivers batches to the downstream Vector, but it yields the response directly instead of going through the sink driver or emitting EventsSent. The serve sink therefore leaves component_sent_events_total and its byte counts at zero even while clients are successfully pulling events, breaking the standard sink telemetry dashboards and compliance checks.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: e2d21c0c08
ℹ️ 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".
| ) -> crate::Result<(crate::sinks::VectorSink, crate::sinks::Healthcheck)> { | ||
| let tls = vector_lib::tls::MaybeTlsSettings::from_config(config.tls.as_ref(), false)?; | ||
| let (emitter, _) = tokio::sync::broadcast::channel(10000); | ||
| let (endpoint, protocol) = crate::sinks::util::uri::protocol_endpoint( |
There was a problem hiding this comment.
When mode = "serve" is used, protocol_endpoint returns (protocol, endpoint), but this destructures the first value into endpoint; for address = "http://127.0.0.1:6000", endpoint becomes "http", and the bind setup below then resolves that with port 0 instead of listening on the configured 127.0.0.1:6000. Fresh evidence beyond the earlier address-format comment is this reversed tuple assignment, which prevents fetch sources from reaching the serve sink even with an otherwise valid address.
Useful? React with 👍 / 👎.
| let events_received = register!(vector_lib::internal_event::EventsReceived); | ||
| events_received.emit(vector_lib::internal_event::CountByteSize(count, byte_size)); | ||
|
|
||
| if pipeline.clone().send_batch(events).await.is_err() { |
There was a problem hiding this comment.
When a vector source runs with mode = "fetch" and a downstream sink/global setting enables acknowledgements, VectorConfig::can_acknowledge() still marks this source as ack-capable, but this path forwards pulled events directly with send_batch and never applies a BatchNotifier or waits for the downstream batch status like receive mode does. That lets ack-enabled topologies run without warning even though rejected downstream batches cannot be reported back through the pull path.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 8cc342b03c
ℹ️ 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".
| protocol: protocol.leak(), | ||
| endpoint: endpoint.leak(), | ||
| }; | ||
| let proto_server = proto::Server::new(service).max_decoding_message_size(usize::MAX); |
There was a problem hiding this comment.
In mode = "serve", this listener only needs to accept an empty PullEventsRequest, but setting the gRPC server's decoding limit to usize::MAX applies to every inbound RPC on this service, including PushEvents. A client can send a very large PushEventsRequest to the serve port and force Vector to decode it before the handler returns Unimplemented, bypassing the decompressed-size cap used by the normal vector source and creating an avoidable memory-DoS risk.
Useful? React with 👍 / 👎.
| let emitter = self.emitter.subscribe(); | ||
| let mut stream = tokio_stream::wrappers::BroadcastStream::new(emitter); |
There was a problem hiding this comment.
When more than one fetch source connects to the same serve sink, each RPC creates its own broadcast subscription, so every active client receives every batch. During common overlap cases such as rolling restarts or an accidentally scaled fetch deployment, the vector sink no longer has its normal deliver-once behavior and downstream receives duplicated events rather than load-balancing or rejecting the extra consumer.
Useful? React with 👍 / 👎.
|
@vectordotdev/vector can you please take a look? |
1 participant
Summary
Idea is summarised in issue #25183
Vector configuration
Source vector process:
Sink vector process:
How did you test this PR?
I ran two vector processes with provided configurations and successfully received events in terminal from sink vector process. I was testing it on Macbook Pro M1 Pro MacOS 26.5.1.
Change Type
Is this a breaking change?
Does this PR include user facing changes?
no-changeloglabel to this PR.References
Notes
@vectordotdev/vectorto reach out to us regarding this PR.pre-pushhook, please see this template.make fmtmake check-clippy(if there are failures it's possible some of them can be fixed withmake clippy-fix)make testgit merge origin masterandgit push.Cargo.lock), pleaserun
make build-licensesto regenerate the license inventory and commit the changes (if any). More details on the dd-rust-license-tool.