PERFORMANCE: Fix slow metric invocation and needless locking on timeout enforcer

Fixes #125

Author: original-brownbear

lib/logstash/filters/grok.rb

@@ -287,30 +287,29 @@ def register
           @patterns[field] << grok
         end
       end # @match.each
+      @match_counter = metric.counter(:matches)
+      @failure_counter = metric.counter(:failures)
     end # def register
 
     public
     def filter(event)
       matched = false
-      done = false
 
-      @logger.debug? and @logger.debug("Running grok filter", :event => event);
+      @logger.debug? and @logger.debug("Running grok filter", :event => event)
 
       @patterns.each do |field, groks|
-        success = match(groks, field, event)
-        
-        if success
+        if match(groks, field, event)
           matched = true
           break if @break_on_match
         end
         #break if done
       end # @patterns.each
 
       if matched
-        metric.increment(:matches)
+        @match_counter.increment(1)
         filter_matched(event)
       else
-        metric.increment(:failures)
+        @failure_counter.increment(1)
         @tag_on_failure.each {|tag| event.tag(tag)}
       end
 
@@ -345,7 +344,7 @@ def match_against_groks(groks, field, input, event)
       groks.each do |grok|
         # Convert anything else to string (number, hash, etc)
 
-        matched = @timeout_enforcer.grok_till_timeout(event, grok, field, input) { grok.execute(input) }
+        matched = @timeout_enforcer.grok_till_timeout(grok, field, input)
         if matched
           grok.capture(matched) {|field, value| handle(field, value, event)}
           break if @break_on_match

lib/logstash/filters/grok/timeout_enforcer.rb

@@ -1,5 +1,3 @@
-java_import java.util.concurrent.locks.ReentrantLock
-
 class LogStash::Filters::Grok::TimeoutEnforcer
   attr_reader :running
 
@@ -10,15 +8,14 @@ def initialize(logger, timeout_nanos)
 
     # Stores running matches with their start time, this is used to cancel long running matches
     # Is a map of Thread => start_time
-    @threads_to_start_time = {}
-    @state_lock = ReentrantLock.new
+    @threads_to_start_time = java.util.concurrent.ConcurrentHashMap.new
   end
 
-  def grok_till_timeout(event, grok, field, value)
+  def grok_till_timeout(grok, field, value)
     begin
       thread = java.lang.Thread.currentThread()
       start_thread_groking(thread)
-      yield
+      grok.execute(value)
     rescue InterruptedRegexpError => e
       raise ::LogStash::Filters::Grok::TimeoutException.new(grok, field, value)
     ensure
@@ -27,7 +24,7 @@ def grok_till_timeout(event, grok, field, value)
       # It may appear that this should go in #stop_thread_groking but that would actually
       # break functionality! If this were moved there we would clear the interrupt
       # immediately after setting it in #cancel_timed_out, hence this MUST be here
-      thread.interrupted
+      java.lang.Thread.interrupted
     end
   end
 
@@ -64,44 +61,26 @@ def stop!
 
   def start_thread_groking(thread)
     # Clear any interrupts from any previous invocations that were not caught by Joni
-    thread.interrupted
-    synchronize do
-      @threads_to_start_time[thread] = java.lang.System.nanoTime()
-    end
+    java.lang.Thread.interrupted
+    @threads_to_start_time.put(thread, java.lang.System.nanoTime)
   end
 
   def stop_thread_groking(thread)
-    synchronize do
-      @threads_to_start_time.delete(thread)
-    end
+    @threads_to_start_time.delete(thread)
   end
 
   def cancel_timed_out!
-    synchronize do
-      @threads_to_start_time.each do |thread,start_time|
-        now = java.lang.System.nanoTime # save ourselves some nanotime calls
-        elapsed = java.lang.System.nanoTime - start_time
-        if elapsed > @timeout_nanos
-          elapsed_millis = elapsed / 1000
-          thread.interrupt()
-          # Ensure that we never attempt to cancel this thread twice in the event
-          # of weird races
-          stop_thread_groking(thread)
-        end
+    now = java.lang.System.nanoTime # save ourselves some nanotime calls
+    @threads_to_start_time.entry_set.each do |entry|
+      start_time = entry.get_value
+      if start_time < now && now - start_time > @timeout_nanos
+        thread  = entry.get_key
+        thread.interrupt()
+        # Ensure that we never attempt to cancel this thread twice in the event
+        # of weird races
+        stop_thread_groking(thread)
       end
     end
   end
 
-  # We use this instead of a Mutex because JRuby mutexes are interruptible
-  # We actually don't want that behavior since we always clear the interrupt in
-  # grok_till_timeout
-  def synchronize
-    # The JRuby Mutex uses lockInterruptibly which is what we DO NOT want
-    @state_lock.lock()
-    yield
-  ensure
-    @state_lock.unlock()
-  end
-
-
 end

logstash-filter-grok.gemspec

@@ -26,5 +26,5 @@ Gem::Specification.new do |s|
   s.add_runtime_dependency 'stud', '~> 0.0.22'
   s.add_runtime_dependency 'logstash-patterns-core'
 
-  s.add_development_dependency 'logstash-devutils'
+  s.add_development_dependency 'logstash-devutils', '= 1.3.4'
 end