From c625ccd4899cb5083f7b73d47cc0dba6a0427814 Mon Sep 17 00:00:00 2001
From: Brian Candler <b.candler@pobox.com>
Date: Mon, 15 Jul 2019 14:35:05 +0000
Subject: [PATCH] in_syslog: add 'emit_unmatched_lines' option

Fixes #2497

Signed-off-by: Brian Candler <b.candler@pobox.com>
---
 lib/fluent/plugin/in_syslog.rb | 11 +++++++++++
 test/plugin/test_in_syslog.rb  | 27 +++++++++++++++++++++++++++
 2 files changed, 38 insertions(+)

diff --git a/lib/fluent/plugin/in_syslog.rb b/lib/fluent/plugin/in_syslog.rb
index 2cd410d827..3d60fde3fd 100644
--- a/lib/fluent/plugin/in_syslog.rb
+++ b/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
diff --git a/test/plugin/test_in_syslog.rb b/test/plugin/test_in_syslog.rb
index f75fb87b64..5e02b0e2cc 100755
--- a/test/plugin/test_in_syslog.rb
+++ b/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