QuestionHello, I am experimenting with the delay transform but am seeing different behavior than is outlined in the docs. As I understand the docs, events passing through this transform will be delayed some number of seconds until a condition is met, but ultimately, all logs will eventually pass through. The behavior I am seeing in testing is that events passing through this transform are dropped if they do not meet the condition. Maybe I am misunderstanding the delay function. Background: Build: vector 0.57.0 (aarch64-apple-darwin 8832452 2026-07-14 20:58:30.491174540) Vector Configdata_dir: /tmp/vector-test-data
sources:
log_file:
type: file
include:
- /Users/andrew/vector_test/sample.log
read_from: beginning
ignore_checkpoints: true
fingerprint:
strategy: device_and_inode
transforms:
parse:
type: remap
inputs:
- log_file
source: |
parsed, err = parse_regex(.message, r'^(?P<timestamp>\S+) (?P<level>\w+)\s+process=(?P<process>\S+)\s+messageId=(?P<messageId>\S+)\s+(?P<message>.*)$')
if err != null {
log("unparseable line: " + string!(.message), level: "warn")
abort
}
.timestamp = parse_timestamp!(parsed.timestamp, format: "%+")
.level = parsed.level
.process = parsed.process
.messageId = parsed.messageId
.message = parsed.message
.num = 1
thread_route:
type: route
inputs:
- parse
route:
server: '.process == "server"'
listener: '.process == "listener"'
server_delay:
type: delay
inputs:
- thread_route.server
condition:
type: "vrl"
source: '.process == "server" && contains(string!(.message), "Server up")'
delay_ms: 1000
server_reduce:
type: reduce
inputs:
- server_delay
group_by:
- messageId
ends_when:
type: "vrl"
# Flush when the last server event arrives
source: '(.process == "server" && contains(string!(.message), "Calling listener")) || (.process == "server" && contains(string!(.message), "Server up"))'
expire_after_ms: 5000
merge_strategies:
message: array
timestamp: retain
num: sum
listener_reduce:
type: reduce
inputs:
- thread_route.listener
group_by:
- messageId
ends_when:
type: "vrl"
# Flush when the last listener event arrives
# ("Payment API recovered, resuming normal traffic")
source: '.process == "listener" && contains(string!(.message), "resuming normal traffic")'
expire_after_ms: 5000
merge_strategies:
message: array
timestamp: retain
num: sum
final_reduce:
type: reduce
inputs:
- server_reduce
- listener_reduce
group_by:
- messageId
ends_when:
type: "vrl"
source: '(.process == "server" && contains(string!(.message[-1]), "Server up"))'
expire_after_ms: 7000
flush_period_ms: 5000
end_every_period_ms: 5000
merge_strategies:
message: array
timestamp: retain
num: sum
sinks:
console:
type: console
inputs:
- final_reduce
encoding:
codec: jsonVector LogsWith the above vector config and the following sample log file: Then a few seconds after the program is up, the following log item is appended to the log file: Observed output: Expected output: |
Replies: 1 comment
|
Nothing is being filtered — the condition works the opposite way round from how you have read it, and that inversion produces something that looks exactly like dropping. From Some(event) => {
let (result, event) = self.check_condition(event, true);
if result {
yield event
} else {
// ... queued with self.delayand when a queued event's timer expires: let (result, event) = self.check_condition(event, false);
if result {
yield event;
} else {
self.queue.insert(event, self.delay);
}So the condition is a release gate, not a selector for what to delay. An event matching the condition is emitted immediately, with no delay at all. An event that does not match is held — and re-checked after every delay period, forever. That last part is why it looks like filtering. Re-evaluating a condition against an unchanged event gives the same answer every time, so an event that fails a static predicate like Worth ruling out the one place events genuinely can be discarded, since it is not this: the queue is bounded, but the defaults are conservative. const fn default_queue_capacity() -> NonZeroUsize {
NonZeroUsize::new(500).expect("static non-zero number")
}with For what you actually want — hold one specific event a short while to break the race with the reduce — the mode you need is the one with no condition configured at all: fn check_condition(&self, event: Event, first: bool) -> (bool, Event) {
if let Some(condition) = self.condition.as_ref() {
condition.check(event)
} else {
// If this is the first check, we need to ensure at least one delay is
// done if no condition is configured
(!first, event)
}
}With no condition, the first check returns false so the event is queued, and the next check returns true so it is emitted. Exactly one delay period, then through. That is the plain "delay everything by Since you already have a If you would rather keep everything flowing through one delay component, a time-based condition also works — something that compares |
Nothing is being filtered — the condition works the opposite way round from how you have read it, and that inversion produces something that looks exactly like dropping.
From
src/transforms/delay.rs, when an event arrives:and when a queued event's timer expires:
So the condition is a release gate, not a selector for what to delay. An event matching the condition is emitted immediately, wi…