- Make multiline use grok (LOGSTASH-445)

- Add tests to cover grok patterns in multiline.

lib/logstash/filters/multiline.rb

@@ -85,6 +85,27 @@ class LogStash::Filters::Multiline < LogStash::Filters::Base
   # may have occurred between the old and new connection. To solve this use
   # case, you can use "%{@source_host}.%{@type}" instead.
   config :stream_identity , :validate => :string, :default => "%{@source}.%{@type}"
+  
+  # logstash ships by default with a bunch of patterns, so you don't
+  # necessarily need to define this yourself unless you are adding additional
+  # patterns.
+  #
+  # Pattern files are plain text with format:
+  #
+  #     NAME PATTERN
+  #
+  # For example:
+  #
+  #     NUMBER \d+
+  config :patterns_dir, :validate => :array, :default => []
+
+  # Detect if we are running from a jarfile, pick the right path.
+  @@patterns_path = Set.new
+  if __FILE__ =~ /file:\/.*\.jar!.*/
+    @@patterns_path += ["#{File.dirname(__FILE__)}/../../patterns/*"]
+  else
+    @@patterns_path += ["#{File.dirname(__FILE__)}/../../../patterns/*"]
+  end
 
   public
   def initialize(config = {})
@@ -99,12 +120,30 @@ class LogStash::Filters::Multiline < LogStash::Filters::Base
 
   public
   def register
-    begin
+    require "grok-pure" # rubygem 'jls-grok'
-      @pattern = Regexp.new(@pattern)
+
-    rescue RegexpError => e
+    @grok = Grok.new
-      @logger.fatal("Invalid pattern for multiline filter",
+
-                    :pattern => @pattern, :exception => e, :backtrace => e.backtrace)
+    @patterns_dir = @@patterns_path.to_a + @patterns_dir
+    @patterns_dir.each do |path|
+      # Can't read relative paths from jars, try to normalize away '../'
+      while path =~ /file:\/.*\.jar!.*\/\.\.\//
+        # replace /foo/bar/../baz => /foo/baz
+        path = path.gsub(/[^\/]+\/\.\.\//, "")
+      end
+
+      if File.directory?(path)
+        path = File.join(path, "*")
+      end
+
+      Dir.glob(path).each do |file|
+        @logger.info("Grok loading patterns from file", :path => file)
+        @grok.add_patterns_from_file(file)
+      end
     end
+
+    @grok.compile(@pattern)
+
     @logger.debug("Registered multiline plugin", :type => @type, :config => @config)
   end # def register
 
@@ -112,7 +151,11 @@ class LogStash::Filters::Multiline < LogStash::Filters::Base
   def filter(event)
     return unless filter?(event)
 
-    match = @pattern.match(event.message)
+    if event.message.is_a?(Array)
+      match = @grok.match(event.message.first)
+    else
+      match = @grok.match(event.message)
+    end
     key = event.sprintf(@stream_identity)
     pending = @pending[key]
 
@@ -176,17 +219,16 @@ class LogStash::Filters::Multiline < LogStash::Filters::Base
     if !event.cancelled?
       filter_matched(event)
     end
-    filter_matched(event) if !event.cancelled?
   end # def filter
 
   # Flush any pending messages. This is generally used for unit testing only.
   public
-  def flush(key)
+  def flush
-    if @pending[key]
+    events = []
-      event = @pending[key]
+    @pending.each do |key, value|
-      @pending.delete(key)
+      events << value
     end
-    return event
+    @pending.clear
+    return events
   end # def flush
-
 end # class LogStash::Filters::Date

spec/filters/multiline.rb

@@ -22,4 +22,23 @@ describe LogStash::Filters::Multiline do
       insist { subject[1].message } == "another first line"
     end
   end
+
+  describe "multiline using grok patterns" do
+    # The logstash config goes here.
+    # At this time, only filters are supported.
+    config <<-CONFIG
+    filter {
+      multiline {
+        pattern => "^%{NUMBER} %{TIME}"
+        negate => true
+        what => previous
+      }
+    }
+    CONFIG
+
+    sample [ "120913 12:04:33 first line", "second line", "third line" ] do
+      insist { subject.length } == 1
+      insist { subject[0].message } ==  "120913 12:04:33 first line\nsecond line\nthird line"
+    end
+  end
 end