From f13629203653ffd9f75d7470f492f444d7dc5e51 Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Thu, 5 Nov 2026 04:04:07 +0000 Subject: [PATCH] in_tcp: do not put the whole received data into logs Before: when a message does not match the parser, the whole message is written into one warn log, so a 32MiB message makes a 32MiB log line. After: when a message longer than 16KiB does not match the parser, only the first 16KiB and the size are logged. 16KiB is larger than normal log lines, such as lines split by Docker or Kubernetes, so the warn log does not change for them. Co-Authored-By: Claude Signed-off-by: Kentaro Hayashi --- lib/fluent/plugin/in_tcp.rb | 14 ++++++-- test/plugin/test_in_tcp.rb | 66 +++++++++++++++++++++++++++++++++++++ 2 files changed, 78 insertions(+), 2 deletions(-) diff --git a/lib/fluent/plugin/in_tcp.rb b/lib/fluent/plugin/in_tcp.rb index 99e066c6e4..6c979215b7 100644 --- a/lib/fluent/plugin/in_tcp.rb +++ b/lib/fluent/plugin/in_tcp.rb @@ -22,6 +22,8 @@ class TcpInput < Input helpers :server, :parser, :extract, :compat_parameters + MAX_LOGGED_MESSAGE_LENGTH = 16 * 1024 + desc 'Tag of output events.' config_param :tag, :string desc 'The port to listen to.' @@ -137,7 +139,7 @@ def start @parser.parse(msg) do |time, record| unless time && record - log.on_warn { log.warn "pattern not matched", message: msg } + log.on_warn { warn_pattern_not_matched(msg) } next end @@ -187,7 +189,7 @@ def start @parser.parse(msg) do |time, record| unless time && record - log.on_warn { log.warn "pattern not matched", message: msg } + log.on_warn { warn_pattern_not_matched(msg) } next end @@ -214,6 +216,14 @@ def start private + def warn_pattern_not_matched(msg) + if msg.length > MAX_LOGGED_MESSAGE_LENGTH + log.warn "pattern not matched", message: msg[...MAX_LOGGED_MESSAGE_LENGTH], size: msg.bytesize + else + log.warn "pattern not matched", message: msg + end + end + def check_client(conn) if @nodes remote_addr = conn.remote_addr diff --git a/test/plugin/test_in_tcp.rb b/test/plugin/test_in_tcp.rb index a27e00bba3..8b0ca5a25c 100755 --- a/test/plugin/test_in_tcp.rb +++ b/test/plugin/test_in_tcp.rb @@ -325,4 +325,70 @@ def create_tcp_socket(host, port, &block) ) end end + + sub_test_case "pattern not matched" do + data("batch_emit", { extract: "" }, keep: true) + data("single_emit", { extract: "\ntag_key tag\n\n" }, keep: true) + test "log the whole message when it is short" do |data| + d = create_driver(base_config + %! + + @type json + + #{data[:extract]} + !) + d.run(expect_records: 1, timeout: 10) do + create_tcp_socket('127.0.0.1', @port) do |sock| + sock.send("hello\n", 0) + sock.send('{"k":"v"}' + "\n", 0) + end + end + + logs = d.logs.collect do |log| + log.gsub(/\A\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} [-+]\d{4} /, "") + end + assert_equal(['[warn]: pattern not matched message="hello"' + "\n"], logs[1..]) + end + + test "log the whole message when its length is the limit" do |data| + max = Fluent::Plugin::TcpInput::MAX_LOGGED_MESSAGE_LENGTH + d = create_driver(base_config + %! + + @type json + + #{data[:extract]} + !) + d.run(expect_records: 1, timeout: 10) do + create_tcp_socket('127.0.0.1', @port) do |sock| + sock.send("a" * max + "\n", 0) + sock.send('{"k":"v"}' + "\n", 0) + end + end + + logs = d.logs.collect do |log| + log.gsub(/\A\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} [-+]\d{4} /, "") + end + assert_equal(["[warn]: pattern not matched message=\"#{"a" * max}\"\n"], logs[1..]) + end + + test "log only the head of the message when it is long" do |data| + max = Fluent::Plugin::TcpInput::MAX_LOGGED_MESSAGE_LENGTH + d = create_driver(base_config + %! + + @type json + + #{data[:extract]} + !) + d.run(expect_records: 1, timeout: 10) do + create_tcp_socket('127.0.0.1', @port) do |sock| + sock.send("a" * (max + 1) + "\n", 0) + sock.send('{"k":"v"}' + "\n", 0) + end + end + + logs = d.logs.collect do |log| + log.gsub(/\A\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} [-+]\d{4} /, "") + end + assert_equal(["[warn]: pattern not matched message=\"#{"a" * max}\" size=#{max + 1}\n"], logs[1..]) + end + end end