in_tail.rb in fluentd-0.12.32

- old
+ new

@@ -46,10 +46,12 @@
     config_param :refresh_interval, :time, default: 60
     desc 'The number of reading lines at each IO.'
     config_param :read_lines_limit, :integer, default: 1000
     desc 'The interval of flushing the buffer for multiline format'
     config_param :multiline_flush_interval, :time, default: nil
+    desc 'Enable the option to emit unmatched lines.'
+    config_param :emit_unmatched_lines, :bool, default: false
     desc 'Enable the additional watch timer.'
     config_param :enable_watch_timer, :bool, default: true
     desc 'The encoding after conversion of the input.'
     config_param :encoding, :string, default: nil
     desc 'The encoding of the input.'
@@ -313,10 +315,15 @@
         @parser.parse(line) { |time, record|
           if time && record
             record[@path_key] ||= tail_watcher.path unless @path_key.nil?
             es.add(time, record)
           else
+            if @emit_unmatched_lines
+              record = {'unmatched_line' => line}
+              record[@path_key] ||= tail_watcher.path unless @path_key.nil?
+              es.add(::Fluent::Engine.now, record)
+            end
             log.warn "pattern not match: #{line.inspect}"
           end
         }
       rescue => e
         log.warn line.dump, error: e.to_s
@@ -343,9 +350,12 @@
               convert_line_to_event(lb, es, tail_watcher)
             end
             lb = line
           else
             if lb.nil?
+              if @emit_unmatched_lines
+                convert_line_to_event(line, es, tail_watcher)
+              end
               log.warn "got incomplete line before first line from #{tail_watcher.path}: #{line.inspect}"
             else
               lb << line
             end
           end