Merge pull request #1747 from fluent/in_udp-add-remove_newline

in_udp: Add remove_newline parameter. fix #1741

lib/fluent/plugin/in_udp.rb

@@ -38,6 +38,8 @@ class UdpInput < Input
     config_param :body_size_limit, :size, default: nil, deprecated: "use message_length_limit instead."
     desc "The max bytes of message"
     config_param :message_length_limit, :size, default: 4096
+    desc "Remove newline from the end of incoming payload"
+    config_param :remove_newline, :bool, default: true
 
     config_param :blocking_timeout, :time, default: 0.5
 
@@ -60,7 +62,7 @@ def start
 
       log.info "listening udp socket", bind: @bind, port: @port
       server_create(:in_udp_server, @port, proto: :udp, bind: @bind, max_bytes: @message_length_limit) do |data, sock|
-        data.chomp!
+        data.chomp! if @remove_newline
         begin
           @parser.parse(data) do |time, record|
             unless time && record

test/plugin/test_in_udp.rb

@@ -149,4 +149,27 @@ def create_udp_socket(host, port)
       assert_equal expected_record, d.events[i][2]
     end
   end
+
+  test 'remove_newline' do
+    d = create_driver(BASE_CONFIG + %!
+      format none
+      remove_newline false
+    !)
+    payloads = ["test1\n", "test2\n"]
+    d.run(expect_records: 2) do
+      create_udp_socket('127.0.0.1', PORT) do |u|
+        payloads.each do |payload|
+          u.send(payload, 0)
+        end
+      end
+    end
+
+    expecteds = payloads.map { |payload| {'message' => payload} }
+    assert_equal 2, d.events.size
+    expecteds.each_with_index do |expected_record, i|
+      assert_equal "udp", d.events[i][0]
+      assert d.events[i][1].is_a?(Fluent::EventTime)
+      assert_equal expected_record, d.events[i][2]
+    end
+  end
 end