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