in_syslog: add 'emit_unmatched_lines' option

lib/fluent/plugin/in_syslog.rb

@@ -82,6 +82,8 @@ class SyslogInput < Input
     config_param :include_source_host, :bool, default: false, deprecated: 'use "source_hostname_key" or "source_address_key" instead.'
     desc 'Specify key of source host when include_source_host is true.'
     config_param :source_host_key, :string, default: 'source_host'.freeze
+    desc 'Enable the option to emit unmatched lines.'
+    config_param :emit_unmatched_lines, :bool, default: false
 
     desc 'The field name of hostname of sender.'
     config_param :source_hostname_key, :string, default: nil
@@ -203,6 +205,9 @@ def message_handler(data, sock)
       unless @parser_parse_priority
         m = SYSLOG_REGEXP.match(data)
         unless m
+          if @emit_unmatched_lines
+            emit("#{@tag}.unmatched", Fluent::EventTime.now, {"unmatched_line" => data})
+          end
           log.warn "invalid syslog message: #{data.dump}"
           return
         end
@@ -212,6 +217,9 @@ def message_handler(data, sock)
 
       @parser.parse(text) do |time, record|
         unless time && record
+          if @emit_unmatched_lines
+            emit("#{@tag}.unmatched", Fluent::EventTime.now, {"unmatched_line" => text})
+          end
           log.warn "failed to parse message", data: data
           return
         end
@@ -229,6 +237,9 @@ def message_handler(data, sock)
         emit(tag, time, record)
       end
     rescue => e
+      if @emit_unmatched_lines
+        emit("#{@tag}.unmatched", Fluent::EventTime.now, {"unmatched_line" => text})
+      end
       log.error "invalid input", data: data, error: e
       log.error_backtrace
     end

test/plugin/test_in_syslog.rb

@@ -361,4 +361,31 @@ def create_test_case(large_message: false)
       msgs
     end
   end
+
+  def test_emit_unmatched_lines
+    d = create_driver([CONFIG, 'emit_unmatched_lines true'].join("\n"))
+    tests = [
+      # valid message
+      {'msg' => '<6>Sep 10 00:00:00 localhost logger: xxx', 'expected' => {'host'=>'localhost', 'ident'=>'logger', 'message'=>'xxx'}},
+      # missing priority
+      {'msg' => 'hello world', 'expected' => {'unmatched_line' => 'hello world'}},
+      # timestamp parsing failure
+      {'msg' => '<6>ZZZ 99 99:99:99 localhost logger: xxx', 'expected' => {'unmatched_line' => '<6>ZZZ 99 99:99:99 localhost logger: xxx'}},
+    ]
+
+    d.run(expect_emits: 3) do
+      u = UDPSocket.new
+      u.do_not_reverse_lookup = false
+      u.connect('127.0.0.1', PORT)
+      tests.each {|test|
+        u.send(test['msg'], 0)
+      }
+    end
+
+    assert_equal tests.size, d.events.size
+    tests.size.times do |i|
+      assert_equal tests[i]['expected'], d.events[i][2]
+      assert_equal 'syslog.unmatched', d.events[i][0] unless i==0
+    end
+  end
 end