Merge 98a6a2f into 341b634

GoogleCloudPlatform · Jan 26, 2018 · 9eeb92b · 9eeb92b
2 parents 341b634 + 98a6a2f
commit 9eeb92b
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 3 deletions.
diff --git a/lib/fluent/plugin/exception_detector.rb b/lib/fluent/plugin/exception_detector.rb
@@ -198,7 +198,7 @@ class TraceAccumulator
     # for some pre-defined common field names of text logs.
     # The named parameters max_lines and max_bytes limit the maximum amount
     # of data to be buffered. The default value 0 indicates 'no limit'.
-    def initialize(message_field, languages, max_lines: 0, max_bytes: 0,
+    def initialize(newline_separator, message_field, languages, max_lines: 0, max_bytes: 0,
                    &emit_callback)
       @exception_detector = Fluent::ExceptionDetector.new(*languages)
       @max_lines = max_lines
@@ -210,6 +210,7 @@ def initialize(message_field, languages, max_lines: 0, max_bytes: 0,
       @first_record = nil
       @first_timestamp = nil
       @emit = emit_callback
+      @newline_separator = newline_separator
     end
 
     def push(time_sec, record)
@@ -235,7 +236,11 @@ def flush
       when 1
         @emit.call(@first_timestamp, @first_record)
       else
-        combined_message = @messages.join
+        if @newline_separator
+          combined_message = @messages.join("\r\n")
+        else
+          combined_message = @messages.join()
+        end
         if @message_field.nil?
           output_record = combined_message
         else

diff --git a/lib/fluent/plugin/out_detect_exceptions.rb b/lib/fluent/plugin/out_detect_exceptions.rb
@@ -36,6 +36,8 @@ class DetectExceptionsOutput < Output
     config_param :max_bytes, :integer, default: 0
     desc 'Separate log streams by this field in the input JSON data.'
     config_param :stream, :string, default: ''
+    desc 'Separate lines by new line.'
+    config_param :newline_separator, :bool, default: false
 
     Fluent::Plugin.register_output('detect_exceptions', self)
 
@@ -91,7 +93,7 @@ def process_record(tag, time_sec, record)
         unless @accumulators.key?(log_id)
           out_tag = tag.sub(/^#{Regexp.escape(@remove_tag_prefix)}\./, '')
           @accumulators[log_id] =
-            Fluent::TraceAccumulator.new(@message, @languages,
+            Fluent::TraceAccumulator.new(@newline_separator, @message, @languages,
                                          max_lines: @max_lines,
                                          max_bytes: @max_bytes) do |t, r|
               router.emit(out_tag, t, r)