in_tcp: do not put the whole received data into logs

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

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: "<extract>\ntag_key tag\n</extract>\n" }, keep: true)
+    test "log the whole message when it is short" do |data|
+      d = create_driver(base_config + %!
+        <parse>
+          @type json
+        </parse>
+        #{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 + %!
+        <parse>
+          @type json
+        </parse>
+        #{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 + %!
+        <parse>
+          @type json
+        </parse>
+        #{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