Provide CRLF newline style choice on formatter plugins. Fix #3151

lib/fluent/plugin/formatter.rb

@@ -46,5 +46,29 @@ def format(tag, time, record)
         @proc.call(tag, time, record)
       end
     end
+
+    module Newline
+      module Mixin
+        include Fluent::Configurable
+
+        DEFAULT_NEWLINE = if Fluent.windows?
+                            :crlf
+                          else
+                            :lf
+                          end
+
+        config_param :newline, :enum, list: [:lf, :crlf], default: DEFAULT_NEWLINE
+
+        def configure(conf)
+          super
+          @newline = case newline
+                     when :lf
+                       "\n"
+                     when :crlf
+                       "\r\n"
+                     end
+        end
+      end
+    end
   end
 end

lib/fluent/plugin/formatter_hash.rb

@@ -19,13 +19,15 @@
 module Fluent
   module Plugin
     class HashFormatter < Formatter
+      include Fluent::Plugin::Newline::Mixin
+
       Plugin.register_formatter('hash', self)
 
       config_param :add_newline, :bool, default: true
 
       def format(tag, time, record)
         line = record.to_s
-        line << "\n".freeze if @add_newline
+        line << @newline.freeze if @add_newline
         line
       end
     end

lib/fluent/plugin/formatter_json.rb

@@ -20,6 +20,8 @@
 module Fluent
   module Plugin
     class JSONFormatter < Formatter
+      include Fluent::Plugin::Newline::Mixin
+
       Plugin.register_formatter('json', self)
 
       config_param :json_parser, :string, default: 'oj'
@@ -44,7 +46,7 @@ def configure(conf)
       end
 
       def format(tag, time, record)
-        "#{@dump_proc.call(record)}\n"
+        "#{@dump_proc.call(record)}#{@newline}"
       end
 
       def format_without_nl(tag, time, record)

lib/fluent/plugin/formatter_ltsv.rb

@@ -19,6 +19,8 @@
 module Fluent
   module Plugin
     class LabeledTSVFormatter < Formatter
+      include Fluent::Plugin::Newline::Mixin
+
       Plugin.register_formatter('ltsv', self)
 
       # http://ltsv.org/
@@ -34,7 +36,7 @@ def format(tag, time, record)
           formatted << @delimiter if formatted.length.nonzero?
           formatted << "#{label}#{@label_delimiter}#{value}"
         end
-        formatted << "\n".freeze if @add_newline
+        formatted << @newline.freeze if @add_newline
         formatted
       end
     end

lib/fluent/plugin/formatter_out_file.rb

@@ -21,6 +21,8 @@
 module Fluent
   module Plugin
     class OutFileFormatter < Formatter
+      include Fluent::Plugin::Newline::Mixin
+
       Plugin.register_formatter('out_file', self)
 
       config_param :output_time, :bool, default: true
@@ -44,7 +46,7 @@ def format(tag, time, record)
         header = ''
         header << "#{@timef.format(time)}#{@delimiter}" if @output_time
         header << "#{tag}#{@delimiter}" if @output_tag
-        "#{header}#{Yajl.dump(record)}\n"
+        "#{header}#{Yajl.dump(record)}#{@newline}"
       end
     end
   end

lib/fluent/plugin/formatter_single_value.rb

@@ -19,14 +19,16 @@
 module Fluent
   module Plugin
     class SingleValueFormatter < Formatter
+      include Fluent::Plugin::Newline::Mixin
+
       Plugin.register_formatter('single_value', self)
 
       config_param :message_key, :string, default: 'message'
       config_param :add_newline, :bool, default: true
 
       def format(tag, time, record)
         text = record[@message_key].to_s.dup
-        text << "\n" if @add_newline
+        text << @newline.freeze if @add_newline
         text
       end
     end

lib/fluent/plugin/formatter_tsv.rb

@@ -19,6 +19,8 @@
 module Fluent
   module Plugin
     class TSVFormatter < Formatter
+      include Fluent::Plugin::Newline::Mixin
+
       Plugin.register_formatter('tsv', self)
 
       desc 'Field names included in each lines'
@@ -30,7 +32,7 @@ class TSVFormatter < Formatter
 
       def format(tag, time, record)
         formatted = @keys.map{|k| record[k].to_s }.join(@delimiter)
-        formatted << "\n".freeze if @add_newline
+        formatted << @newline.freeze if @add_newline
         formatted
       end
     end

test/command/test_binlog_reader.rb

@@ -82,6 +82,14 @@ def timezone(timezone = 'UTC')
 end
 
 class TestHead < TestBaseCommand
+  setup do
+    @default_newline = if Fluent.windows?
+                         "\r\n"
+                       else
+                         "\n"
+                       end
+  end
+
   sub_test_case 'initialize' do
     data(
       'file is not passed' => %w(),
@@ -138,7 +146,7 @@ class TestHead < TestBaseCommand
         create_message_packed_file(@file_name, [event_time(@t).to_i] * 6, [@record] * 6)
         head = BinlogReaderCommand::Head.new(argv)
         out = capture_stdout { head.call }
-        assert_equal "2011-01-02T13:14:15+00:00\t#{TMP_DIR}/#{@file_name}\t#{Yajl.dump(@record)}\n" * 5, out
+        assert_equal "2011-01-02T13:14:15+00:00\t#{TMP_DIR}/#{@file_name}\t#{Yajl.dump(@record)}#{@default_newline}" * 5, out
       end
     end
 
@@ -149,7 +157,7 @@ class TestHead < TestBaseCommand
         create_message_packed_file(@file_name, [event_time(@t).to_i] * 6, [@record] * 6)
         head = BinlogReaderCommand::Head.new(argv)
         out = capture_stdout { head.call }
-        assert_equal "2011-01-02T13:14:15+00:00\t#{TMP_DIR}/#{@file_name}\t#{Yajl.dump(@record)}\n", out
+        assert_equal "2011-01-02T13:14:15+00:00\t#{TMP_DIR}/#{@file_name}\t#{Yajl.dump(@record)}#{@default_newline}", out
       end
     end
 
@@ -169,7 +177,7 @@ class TestHead < TestBaseCommand
         create_message_packed_file(@file_name, [event_time(@t).to_i], [@record])
         head = BinlogReaderCommand::Head.new(argv)
         out = capture_stdout { head.call }
-        assert_equal "#{Yajl.dump(@record)}\n", out
+        assert_equal "#{Yajl.dump(@record)}#{@default_newline}", out
       end
     end
 
@@ -198,6 +206,14 @@ class TestHead < TestBaseCommand
 end
 
 class TestCat < TestBaseCommand
+  setup do
+    @default_newline = if Fluent.windows?
+                         "\r\n"
+                       else
+                         "\n"
+                       end
+  end
+
   sub_test_case 'initialize' do
     data(
       'file is not passed' => [],
@@ -254,7 +270,7 @@ class TestCat < TestBaseCommand
         create_message_packed_file(@file_name, [event_time(@t).to_i] * 6, [@record] * 6)
         head = BinlogReaderCommand::Cat.new(argv)
         out = capture_stdout { head.call }
-        assert_equal "2011-01-02T13:14:15+00:00\t#{TMP_DIR}/#{@file_name}\t#{Yajl.dump(@record)}\n" * 6, out
+        assert_equal "2011-01-02T13:14:15+00:00\t#{TMP_DIR}/#{@file_name}\t#{Yajl.dump(@record)}#{@default_newline}" * 6, out
       end
     end
 
@@ -265,7 +281,7 @@ class TestCat < TestBaseCommand
         create_message_packed_file(@file_name, [event_time(@t).to_i] * 6, [@record] * 6)
         head = BinlogReaderCommand::Cat.new(argv)
         out = capture_stdout { head.call }
-        assert_equal "2011-01-02T13:14:15+00:00\t#{TMP_DIR}/#{@file_name}\t#{Yajl.dump(@record)}\n", out
+        assert_equal "2011-01-02T13:14:15+00:00\t#{TMP_DIR}/#{@file_name}\t#{Yajl.dump(@record)}#{@default_newline}", out
       end
     end
 
@@ -276,7 +292,7 @@ class TestCat < TestBaseCommand
         create_message_packed_file(@file_name, [event_time(@t).to_i], [@record])
         head = BinlogReaderCommand::Cat.new(argv)
         out = capture_stdout { head.call }
-        assert_equal "#{Yajl.dump(@record)}\n", out
+        assert_equal "#{Yajl.dump(@record)}#{@default_newline}", out
       end
     end
 

test/plugin/test_filter_stdout.rb

@@ -12,6 +12,11 @@ def setup
     @old_tz = ENV["TZ"]
     ENV["TZ"] = "UTC"
     Timecop.freeze
+    @default_newline = if Fluent.windows?
+                         "\r\n"
+                       else
+                         "\n"
+                       end
   end
 
   def teardown
@@ -106,7 +111,7 @@ def test_include_time_key
     def test_format_json
       d = create_driver(CONFIG + config_element("", "", { "format" => "json" }))
       out = capture_log(d) { filter(d, event_time, {'test' => 'test'}) }
-      assert_equal "{\"test\":\"test\"}\n", out
+      assert_equal "{\"test\":\"test\"}#{@default_newline}", out
     end
   end
 

test/plugin/test_formatter_hash.rb

@@ -19,11 +19,14 @@ def record
     {'message' => 'awesome', 'greeting' => 'hello'}
   end
 
-  def test_format
-    d = create_driver({})
+  data("newline (LF)" => ["lf", "\n"],
+       "newline (CRLF)" => ["crlf", "\r\n"])
+  def test_format(data)
+    newline_conf, newline = data
+    d = create_driver({"newline" => newline_conf})
     formatted = d.instance.format(tag, @time, record)
 
-    assert_equal(%Q!{"message"=>"awesome", "greeting"=>"hello"}\n!, formatted.encode(Encoding::UTF_8))
+    assert_equal(%Q!{"message"=>"awesome", "greeting"=>"hello"}#{newline}!, formatted.encode(Encoding::UTF_8))
   end
 
   def test_format_without_newline

test/plugin/test_formatter_json.rb

@@ -7,6 +7,11 @@ class JsonFormatterTest < ::Test::Unit::TestCase
 
   def setup
     @time = event_time
+    @default_newline = if Fluent.windows?
+                         "\r\n"
+                       else
+                         "\n"
+                       end
   end
 
   def create_driver(conf = "")
@@ -25,12 +30,17 @@ def symbolic_record
     {:message => :awesome}
   end
 
-  data('oj' => 'oj', 'yajl' => 'yajl')
+  data('oj with LF' => ['oj', "lf", "\n"],
+       'oj with CRLF' => ['oj', "crlf", "\r\n"],
+       'yajl with LF' => ['yajl', "lf", "\n"],
+       'yajl with CRLF' => ['yajl', "crlf", "\r\n"]
+      )
   def test_format(data)
-    d = create_driver('json_parser' => data)
+    parser, newline_conf, newline = data
+    d = create_driver('json_parser' => parser, 'newline' => newline_conf)
     formatted = d.instance.format(tag, @time, record)
 
-    assert_equal("#{JSON.generate(record)}\n", formatted)
+    assert_equal("#{JSON.generate(record)}#{newline}", formatted)
   end
 
   data('oj' => 'oj', 'yajl' => 'yajl')
@@ -46,6 +56,6 @@ def test_format_with_symbolic_record(data)
     d = create_driver('json_parser' => data)
     formatted = d.instance.format(tag, @time, symbolic_record)
 
-    assert_equal("#{JSON.generate(record)}\n", formatted)
+    assert_equal("#{JSON.generate(record)}#{@default_newline}", formatted)
   end
 end

test/plugin/test_formatter_ltsv.rb

@@ -36,11 +36,14 @@ def test_config_params
     assert_equal  false, d.instance.add_newline
   end
 
-  def test_format
-    d = create_driver({})
+  data("newline (LF)" => ["lf", "\n"],
+       "newline (CRLF)" => ["crlf", "\r\n"])
+  def test_format(data)
+    newline_conf, newline = data
+    d = create_driver({"newline" => newline_conf})
     formatted = d.instance.format(tag, @time, record)
 
-    assert_equal("message:awesome\tgreeting:hello\n", formatted)
+    assert_equal("message:awesome\tgreeting:hello#{newline}", formatted)
   end
 
   def test_format_without_newline
@@ -50,13 +53,18 @@ def test_format_without_newline
     assert_equal("message:awesome\tgreeting:hello", formatted)
   end
 
-  def test_format_with_customized_delimiters
+  data("newline (LF)" => ["lf", "\n"],
+       "newline (CRLF)" => ["crlf", "\r\n"])
+  def test_format_with_customized_delimiters(data)
+    newline_conf, newline = data
+
     d = create_driver(
       'delimiter'       => ',',
       'label_delimiter' => '=',
+      'newline' => newline_conf,
     )
     formatted = d.instance.format(tag, @time, record)
 
-    assert_equal("message=awesome,greeting=hello\n", formatted)
+    assert_equal("message=awesome,greeting=hello#{newline}", formatted)
   end
 end