Merge branch 'pr/1593'

lib/logstash/inputs/syslog.rb

@@ -56,6 +56,7 @@ class LogStash::Inputs::Syslog < LogStash::Inputs::Base
     @grok_filter = LogStash::Filters::Grok.new(
       "overwrite" => "message",
       "match" => { "message" => "<%{POSINT:priority}>%{SYSLOGLINE}" },
+      "tag_on_failure" => ["_grokparsefailure_sysloginput"],
     )
 
     @date_filter = LogStash::Filters::Date.new(
@@ -197,7 +198,7 @@ class LogStash::Inputs::Syslog < LogStash::Inputs::Base
   def syslog_relay(event)
     @grok_filter.filter(event)
 
-    if event["tags"].nil? || !event["tags"].include?("_grokparsefailure")
+    if event["tags"].nil? || !event["tags"].include?(@grok_filter.tag_on_failure)
       # Per RFC3164, priority = (facility * 8) + severity
       #                       = (facility << 3) & (severity)
       priority = event["priority"].to_i rescue 13

spec/inputs/syslog.rb

@@ -1,11 +1,13 @@
 # coding: utf-8
 require "test_utils"
 require "socket"
+require "logstash/inputs/syslog"
+require "logstash/event"
 
-describe "inputs/syslog", :socket => true do
+describe "inputs/syslog" do
   extend LogStash::RSpec
 
-  describe "properly handles priority, severity and facilities" do
+  it "should properly handle priority, severity and facilities", :socket => true do
     port = 5511
     event_count = 10
 
@@ -29,6 +31,8 @@ describe "inputs/syslog", :socket => true do
       socket.close
 
       events = event_count.times.collect { queue.pop }
+
+      insist { events.length } == event_count
       event_count.times do |i|
         insist { events[i]["priority"] } == 164
         insist { events[i]["severity"] } == 4
@@ -36,5 +40,53 @@ describe "inputs/syslog", :socket => true do
       end
     end
   end
+
+  it "should add unique tag when grok parsing fails with live syslog input", :socket => true do
+    port = 5511
+    event_count = 10
+
+    config <<-CONFIG
+      input {
+        syslog {
+          type => "blah"
+          port => #{port}
+        }
+      }
+    CONFIG
+
+    input do |pipeline, queue|
+      Thread.new { pipeline.run }
+      sleep 0.1 while !pipeline.ready?
+
+      socket = Stud.try(5.times) { TCPSocket.new("127.0.0.1", port) }
+      event_count.times do |i|
+        socket.puts("message which causes the a grok parse failure")
+      end
+      socket.close
+
+      events = event_count.times.collect { queue.pop }
+
+      insist { events.length } == event_count
+      event_count.times do |i|
+        insist { events[i]["tags"] } == ["_grokparsefailure_sysloginputplugin"]
+      end
+    end
   end
 
+  it "should add unique tag when grok parsing fails" do
+    input = LogStash::Inputs::Syslog.new({})
+    input.register
+
+    # event which is not syslog should have a new tag
+    event = LogStash::Event.new({ "message" => "hello world, this is not syslog RFC3164" })
+    input.syslog_relay(event)
+    insist { event["tags"] } ==  ["_grokparsefailure_sysloginput"]
+
+    syslog_event = LogStash::Event.new({ "message" => "<164>Oct 26 15:19:25 1.2.3.4 %ASA-4-106023: Deny udp src DRAC:10.1.2.3/43434" })
+    input.syslog_relay(syslog_event)
+    insist { syslog_event["priority"] } ==  164
+    insist { syslog_event["severity"] } ==  4
+    insist { syslog_event["tags"] } ==  nil
+  end
+
+end