Skip to content

Commit b1feec1

Browse files
committed
Emit buffered logs normally on timeout instead of throwing error
Previously, when a timeout occurred while waiting for the next multiline log, the plugin raised a `TimeoutError` via `emit_error_event`. This caused the remaining buffered logs—most notably the last line of log outputs—to be dropped or incorrectly routed. While the plugin provides a `timeout_label` option to route logs on timeout, users often use it without realizing they must explicitly configure a `<label>` block to catch and handle the rerouted logs. Without this awareness, it simply results in silent data loss. This commit changes the behavior to log an info message and emit the buffered record normally using `router.emit` when no `timeout_label` is set, ensuring no logs are lost by default. Signed-off-by: Shizuo Fujita <fujita@clear-code.com>
1 parent 602a134 commit b1feec1

3 files changed

Lines changed: 24 additions & 27 deletions

File tree

README.md

Lines changed: 12 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -124,38 +124,34 @@ Specify first line of multiline by regular expression.
124124
</filter>
125125
```
126126

127-
You can handle timeout events and remaining buffers on shutdown this plugin.
127+
If you want to intentionally separate timeout logs or handle remaining buffers on plugin shutdown, you can use the `timeout_label` option to route them to a specific label.
128128

129129
```aconf
130-
<label @ERROR>
130+
<filter docker.log>
131+
@type concat
132+
key message
133+
multiline_start_regexp /^Start/
134+
flush_interval 5
135+
timeout_label @TIMEOUT_HANDLER
136+
</filter>
137+
138+
<label @TIMEOUT_HANDLER>
131139
<match docker.log>
132140
@type file
133-
path /path/to/error.log
141+
path /path/to/timeout_and_shutdown.log
134142
</match>
135143
</label>
136144
```
137145

138-
Handle timeout log lines the same as normal logs.
146+
By default, when a timeout occurs while waiting for the next multiline log, the plugin outputs an info log and emits the buffered logs normally to the next pipeline. You don't need any special configuration to keep your last log lines.
139147

140148
```aconf
141149
<filter **>
142150
@type concat
143151
key message
144152
multiline_start_regexp /^Start/
145153
flush_interval 5
146-
timeout_label @NORMAL
147154
</filter>
148-
149-
<match **>
150-
@type relabel
151-
@label @NORMAL
152-
</match>
153-
154-
<label @NORMAL>
155-
<match **>
156-
@type stdout
157-
</match>
158-
</label>
159155
```
160156

161157
Handle single line JSON from Docker containers.

lib/fluent/plugin/filter_concat.rb

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -460,19 +460,19 @@ def flush_remaining_buffer
460460
@key => lines.join(@separator)
461461
}
462462
tag, time, record = elements.first
463-
message = "Flush remaining buffer: #{stream_identity}"
464-
handle_timeout_error(tag, time, record.merge(new_record), message)
465-
log.info(message)
463+
handle_timeout_error(tag, time, record.merge(new_record))
464+
log.info("Flush remaining buffer: #{stream_identity}")
466465
end
467466
@buffer.clear
468467
end
469468

470-
def handle_timeout_error(tag, time, record, message)
469+
def handle_timeout_error(tag, time, record)
471470
if @timeout_label
472471
event_router = event_emitter_router(@timeout_label)
473472
event_router.emit(tag, time, record)
474473
else
475-
router.emit_error_event(tag, time, record, TimeoutError.new(message))
474+
log.info "Timeout occurred while waiting for the next multiline log. Emitting the buffered log normally. tag: #{tag}"
475+
router.emit(tag, time, record)
476476
end
477477
end
478478

test/plugin/test_filter_concat.rb

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -224,7 +224,7 @@ def filter_with_time(conf, messages, wait: nil)
224224
]
225225
filtered = filter(CONFIG + "flush_interval 2s", messages, wait: 3) do |d|
226226
errored = { "container_id" => "1", "message" => "message 1\nmessage 2" }
227-
mock(d.instance.router).emit_error_event("test", anything, errored, anything)
227+
mock(d.instance.router).emit("test", anything, errored)
228228
end
229229
assert_equal([], filtered)
230230
end
@@ -307,8 +307,8 @@ def filter_with_time(conf, messages, wait: nil)
307307
errored1 = { "container_id" => "1", "message" => "start" }
308308
errored2 = { "container_id" => "2", "message" => "start" }
309309
router = d.instance.router
310-
mock(router).emit_error_event("test", anything, errored1, anything)
311-
mock(router).emit_error_event("test", anything, errored2, anything)
310+
mock(router).emit("test", anything, errored1)
311+
mock(router).emit("test", anything, errored2)
312312
end
313313
assert_equal(expected, filtered)
314314
end
@@ -456,7 +456,7 @@ def filter_with_time(conf, messages, wait: nil)
456456
]
457457
filtered = filter(config, messages, wait: 3) do |d|
458458
errored = { "container_id" => "1", "message" => "start\n message 1\n message 2" }
459-
mock(d.instance.router).emit_error_event("test", anything, errored, anything)
459+
mock(d.instance.router).emit("test", anything, errored)
460460
end
461461
assert_equal([], filtered)
462462
end
@@ -976,7 +976,7 @@ def filter_with_time(conf, messages, wait: nil)
976976
]
977977
filtered = filter_with_time(config, messages, wait: 3) do |d|
978978
errored = { "container_id" => "1", "message" => "start\n message 3\n message 4" }
979-
mock(d.instance.router).emit_error_event("test", @time, errored, anything)
979+
mock(d.instance.router).emit("test", @time, errored)
980980
end
981981
expected = [
982982
[@time, { "container_id" => "1", "message" => "start\n message 1\n message 2" }]
@@ -1000,7 +1000,7 @@ def filter_with_time(conf, messages, wait: nil)
10001000
filtered = filter_with_time(config, messages, wait: 3) do |d|
10011001
mock(d.instance).flush_timeout_buffer.at_most(0)
10021002
errored = { "container_id" => "1", "message" => "start" }
1003-
mock(d.instance.router).emit_error_event("test", @time, errored, anything)
1003+
mock(d.instance.router).emit("test", @time, errored)
10041004
end
10051005
expected = [
10061006
[@time, { "container_id" => "1", "message" => "start\n message 1\n message 2" }]
@@ -1060,6 +1060,7 @@ def filter_with_time(conf, messages, wait: nil)
10601060
"[error]: failed to flush timeout buffer error_class=StandardError error=\"timeout\"",
10611061
"[error]: failed to flush timeout buffer error_class=StandardError error=\"timeout\"",
10621062
"[error]: failed to flush timeout buffer error_class=StandardError error=\"timeout\"",
1063+
"[info]: Timeout occurred while waiting for the next multiline log. Emitting the buffered log normally. tag: test",
10631064
"[info]: Flush remaining buffer: test:default"
10641065
]
10651066
log_messages = logs.map do |line|

0 commit comments

Comments
 (0)