in_syslog: Support octet counting frame. fix #1679

lib/fluent/plugin/in_syslog.rb

@@ -75,6 +75,8 @@ class SyslogInput < Input
     config_param :tag, :string
     desc 'The transport protocol used to receive logs.(udp, tcp)'
     config_param :protocol_type, :enum, list: [:tcp, :udp], default: :udp
+    desc 'The message frame type.(normal, octet_count)'
+    config_param :frame_type, :enum, list: [:normal, :octet_count], default: :normal
 
     desc 'If true, add source host to event record.'
     config_param :include_source_host, :bool, default: false, deprecated: 'use "source_hostname_key" or "source_address_key" instead.'
@@ -152,18 +154,29 @@ def start_udp_server
     end
 
     def start_tcp_server
-      # syslog family add "\n" to each message and this seems only way to split messages in tcp stream
-      delimiter = "\n"
+      octet_count_frame = @frame_type == :octet_count
+
+      # syslog family adds "\n" to each message when transport is TCP and traditional frame
+      delimiter = octet_count_frame ? " " : "\n"
       delimiter_size = delimiter.size
       server_create_connection(:in_syslog_tcp_server, @port, bind: @bind, resolve_name: @resolve_hostname) do |conn|
         conn.data do |data|
           buffer = conn.buffer
           buffer << data
           pos = 0
-          while idx = buffer.index(delimiter, pos)
-            msg = buffer[pos...idx]
-            pos = idx + delimiter_size
-            message_handler(msg, conn)
+          if octet_count_frame
+            while idx = buffer.index(delimiter, pos)
+              num = Integer(buffer[pos..idx])
+              pos = idx + num
+              msg = buffer[idx + 1...pos]
+              message_handler(msg, conn)
+            end
+          else
+            while idx = buffer.index(delimiter, pos)
+              msg = buffer[pos...idx]
+              pos = idx + delimiter_size
+              message_handler(msg, conn)
+            end
           end
           buffer.slice!(0, pos) if pos > 0
         end

test/plugin/test_in_syslog.rb

@@ -286,4 +286,50 @@ def compare_test_result(events, tests, options = {})
       assert_equal(options[:facility], events[i][2]['facility']) if options[:facility]
     }
   end
+
+  sub_test_case 'octet counting frame' do
+    def test_msg_size_with_tcp
+      d = create_driver([CONFIG, 'protocol_type tcp', 'frame_type octet_count'].join("\n"))
+      tests = create_test_case
+
+      d.run(expect_emits: 2) do
+        tests.each {|test|
+          TCPSocket.open('127.0.0.1', PORT) do |s|
+            s.send(test['msg'], 0)
+          end
+        }
+      end
+
+      assert(d.events.size > 0)
+      compare_test_result(d.events, tests)
+    end
+
+    def test_msg_size_with_same_tcp_connection
+      d = create_driver([CONFIG, 'protocol_type tcp', 'frame_type octet_count'].join("\n"))
+      tests = create_test_case
+
+      d.run(expect_emits: 2) do
+        TCPSocket.open('127.0.0.1', PORT) do |s|
+          tests.each {|test|
+            s.send(test['msg'], 0)
+          }
+        end
+      end
+
+      assert(d.events.size > 0)
+      compare_test_result(d.events, tests)
+    end
+
+    def create_test_case(large_message: false)
+      msgs = [
+        {'msg' => '<6>Sep 10 00:00:00 localhost logger: ' + 'x' * 100, 'expected' => 'x' * 100},
+        {'msg' => '<6>Sep 10 00:00:00 localhost logger: ' + 'x' * 1024, 'expected' => 'x' * 1024},
+      ]
+      msgs.each { |msg|
+        m = msg['msg']
+        msg['msg'] = "#{m.size + 1} #{m}"
+      }
+      msgs
+    end
+  end
 end