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