From 0a931651e242e8bac795973e5a1bfc38a668fec3 Mon Sep 17 00:00:00 2001
From: bernii <bkobos@gmail.com>
Date: Tue, 4 Apr 2017 17:13:07 +0200
Subject: [PATCH] join_separator support for the output log

---
 README.rdoc                                |  3 +++
 lib/fluent/plugin/exception_detector.rb    |  5 +++--
 lib/fluent/plugin/out_detect_exceptions.rb |  5 ++++-
 test/plugin/test_exception_detector.rb     | 13 +++++++++++++
 4 files changed, 23 insertions(+), 3 deletions(-)

diff --git a/README.rdoc b/README.rdoc
index 0a4008c..f2c6ae2 100644
--- a/README.rdoc
+++ b/README.rdoc
@@ -95,6 +95,9 @@ The plugin supports the following parameters:
           This parameter is only applicable to structured (JSON) log streams.
           Default: ''.
 
+[join_separator] String used when joining multiple lines of the stack trace.
+                 Default: ''.
+
 Example configuration:
 
     <match **>
diff --git a/lib/fluent/plugin/exception_detector.rb b/lib/fluent/plugin/exception_detector.rb
index 0e9ba1f..496f0a4 100644
--- a/lib/fluent/plugin/exception_detector.rb
+++ b/lib/fluent/plugin/exception_detector.rb
@@ -194,10 +194,11 @@ class TraceAccumulator
     # 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,
-                   &emit_callback)
+                   join_separator: '', &emit_callback)
       @exception_detector = Fluent::ExceptionDetector.new(*languages)
       @max_lines = max_lines
       @max_bytes = max_bytes
+      @join_separator = join_separator
       @message_field = message_field
       @messages = []
       @buffer_start_time = Time.now
@@ -230,7 +231,7 @@ def flush
       when 1
         @emit.call(@first_timestamp, @first_record)
       else
-        combined_message = @messages.join
+        combined_message = @messages.join(@join_separator)
         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
index 3e4fe42..601c999 100644
--- 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 'String used when joining multiple lines of the stack trace'
+    config_param :join_separator, :string, default: ''
 
     Fluent::Plugin.register_output('detect_exceptions', self)
 
@@ -93,7 +95,8 @@ def process_record(tag, time_sec, record)
           @accumulators[log_id] =
             Fluent::TraceAccumulator.new(@message, @languages,
                                          max_lines: @max_lines,
-                                         max_bytes: @max_bytes) do |t, r|
+                                         max_bytes: @max_bytes,
+                                         join_separator: @join_separator) do |t, r|
               router.emit(out_tag, t, r)
             end
         end
diff --git a/test/plugin/test_exception_detector.rb b/test/plugin/test_exception_detector.rb
index 6642ea6..7575199 100644
--- a/test/plugin/test_exception_detector.rb
+++ b/test/plugin/test_exception_detector.rb
@@ -404,4 +404,17 @@ def test_low_max_bytes_limit
     # Check that the trace is flushed after the first part.
     assert_equal([JAVA_EXC_PART1] + JAVA_EXC_PART2.lines, out)
   end
+
+  def test_join_separator
+    # Custom join separator is used to join accumulated output
+    join_separator = ';'
+    out = []
+    buffer = Fluent::TraceAccumulator.new(nil,
+                                          [:all],
+                                          join_separator: join_separator) do |_, m|
+      out << m
+    end
+    feed_lines(buffer, JAVA_EXC)
+    assert_equal(JAVA_EXC.lines.length - 1, out[0].count(join_separator))
+  end
 end