Change grok watch dog to be Matcher based instead of thread based. (#…

…48346)

There is a watchdog in order to avoid long running (and expensive)
grok expressions. Currently the watchdog is thread based, threads
that run grok expressions are registered and after completion unregister.
If these threads stay registered for too long then the watch dog interrupts
these threads. Joni (the library that powers grok expressions) has a
mechanism that checks whether the current thread is interrupted and
if so abort the pattern matching.

Newer versions have an additional method to abort long running pattern
matching inside joni. Instead of checking the thread's interrupted flag,
joni now also checks a volatile field that can be set via a `Matcher`
instance. This is more efficient method for aborting long running matches.
(joni checks each 30k iterations whether interrupted flag is set vs.
just checking a volatile field)

Recently we upgraded to a recent joni version (#47374), and this PR
is a followup of that PR.

This change should also fix #43673, since it appears when unit tests
are ran the a test runner thread's interrupted flag may already have
been set, due to some thread reuse.

libs/grok/src/main/java/org/elasticsearch/grok/Grok.java

@@ -73,24 +73,24 @@ public final class Grok {
     private final Map<String, String> patternBank;
     private final boolean namedCaptures;
     private final Regex compiledExpression;
-    private final ThreadWatchdog threadWatchdog;
+    private final MatcherWatchdog matcherWatchdog;
 
     public Grok(Map<String, String> patternBank, String grokPattern) {
-        this(patternBank, grokPattern, true, ThreadWatchdog.noop());
+        this(patternBank, grokPattern, true, MatcherWatchdog.noop());
     }
 
-    public Grok(Map<String, String> patternBank, String grokPattern, ThreadWatchdog threadWatchdog) {
-        this(patternBank, grokPattern, true, threadWatchdog);
+    public Grok(Map<String, String> patternBank, String grokPattern, MatcherWatchdog matcherWatchdog) {
+        this(patternBank, grokPattern, true, matcherWatchdog);
     }
 
     Grok(Map<String, String> patternBank, String grokPattern, boolean namedCaptures) {
-        this(patternBank, grokPattern, namedCaptures, ThreadWatchdog.noop());
+        this(patternBank, grokPattern, namedCaptures, MatcherWatchdog.noop());
     }
 
-    private Grok(Map<String, String> patternBank, String grokPattern, boolean namedCaptures, ThreadWatchdog threadWatchdog) {
+    private Grok(Map<String, String> patternBank, String grokPattern, boolean namedCaptures, MatcherWatchdog matcherWatchdog) {
         this.patternBank = patternBank;
         this.namedCaptures = namedCaptures;
-        this.threadWatchdog = threadWatchdog;
+        this.matcherWatchdog = matcherWatchdog;
 
         for (Map.Entry<String, String> entry : patternBank.entrySet()) {
             String name = entry.getKey();
@@ -168,14 +168,12 @@ public String toRegex(String grokPattern) {
 
         int result;
         try {
-            threadWatchdog.register();
-            result = matcher.searchInterruptible(0, grokPatternBytes.length, Option.NONE);
-        } catch (InterruptedException e) {
-            result = Matcher.INTERRUPTED;
+            matcherWatchdog.register(matcher);
+            result = matcher.search(0, grokPatternBytes.length, Option.NONE);
         } finally {
-            threadWatchdog.unregister();
+            matcherWatchdog.unregister(matcher);
         }
-        if (result != -1) {
+        if (result >= 0) {
             Region region = matcher.getEagerRegion();
             String namedPatternRef = groupMatch(NAME_GROUP, region, grokPattern);
             String subName = groupMatch(SUBNAME_GROUP, region, grokPattern);
@@ -213,18 +211,16 @@ public String toRegex(String grokPattern) {
      * Checks whether a specific text matches the defined grok expression.
      *
      * @param text the string to match
-     * @return true if grok expression matches text, false otherwise.
+     * @return true if grok expression matches text or there is a timeout, false otherwise.
      */
     public boolean match(String text) {
         Matcher matcher = compiledExpression.matcher(text.getBytes(StandardCharsets.UTF_8));
         int result;
         try {
-            threadWatchdog.register();
-            result = matcher.searchInterruptible(0, text.length(), Option.DEFAULT);
-        } catch (InterruptedException e) {
-            result = Matcher.INTERRUPTED;
+            matcherWatchdog.register(matcher);
+            result = matcher.search(0, text.length(), Option.DEFAULT);
         } finally {
-            threadWatchdog.unregister();
+            matcherWatchdog.unregister(matcher);
         }
         return (result != -1);
     }
@@ -241,16 +237,14 @@ public Map<String, Object> captures(String text) {
         Matcher matcher = compiledExpression.matcher(textAsBytes);
         int result;
         try {
-            threadWatchdog.register();
-            result = matcher.searchInterruptible(0, textAsBytes.length, Option.DEFAULT);
-        } catch (InterruptedException e) {
-            result = Matcher.INTERRUPTED;
+            matcherWatchdog.register(matcher);
+            result = matcher.search(0, textAsBytes.length, Option.DEFAULT);
         } finally {
-            threadWatchdog.unregister();
+            matcherWatchdog.unregister(matcher);
         }
         if (result == Matcher.INTERRUPTED) {
             throw new RuntimeException("grok pattern matching was interrupted after [" +
-                threadWatchdog.maxExecutionTimeInMillis() + "] ms");
+                matcherWatchdog.maxExecutionTimeInMillis() + "] ms");
         } else if (result == Matcher.FAILED) {
             // TODO: I think we should throw an error here?
             return null;

libs/grok/src/main/java/org/elasticsearch/grok/ThreadWatchdog.java → libs/grok/src/main/java/org/elasticsearch/grok/MatcherWatchdog.java

@@ -18,6 +18,8 @@
  */
 package org.elasticsearch.grok;
 
+import org.joni.Matcher;
+
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -27,88 +29,92 @@
 
 /**
  * Protects against long running operations that happen between the register and unregister invocations.
- * Threads that invoke {@link #register()}, but take too long to invoke the {@link #unregister()} method
+ * Threads that invoke {@link #register(Matcher)}, but take too long to invoke the {@link #unregister(Matcher)} method
  * will be interrupted.
  *
  * This is needed for Joni's {@link org.joni.Matcher#search(int, int, int)} method, because
  * it can end up spinning endlessly if the regular expression is too complex. Joni has checks
  * that for every 30k iterations it checks if the current thread is interrupted and if so
  * returns {@link org.joni.Matcher#INTERRUPTED}.
  */
-public interface ThreadWatchdog {
-    
+public interface MatcherWatchdog {
+
     /**
-     * Registers the current thread and interrupts the current thread
-     * if the takes too long for this thread to invoke {@link #unregister()}.
+     * Registers the current matcher and interrupts the this matcher
+     * if the takes too long for this thread to invoke {@link #unregister(Matcher)}.
+     *
+     * @param matcher The matcher to register
      */
-    void register();
-    
+    void register(Matcher matcher);
+
     /**
-     * @return The maximum allowed time in milliseconds for a thread to invoke {@link #unregister()}
-     *         after {@link #register()} has been invoked before this ThreadWatchDog starts to interrupting that thread.
+     * @return The maximum allowed time in milliseconds for a thread to invoke {@link #unregister(Matcher)}
+     *         after {@link #register(Matcher)} has been invoked before this ThreadWatchDog starts to interrupting that thread.
      */
     long maxExecutionTimeInMillis();
-    
+
     /**
-     * Unregisters the current thread and prevents it from being interrupted.
+     * Unregisters the current matcher and prevents it from being interrupted.
+     *
+     * @param matcher The matcher to unregister
      */
-    void unregister();
-    
+    void unregister(Matcher matcher);
+
     /**
-     * Returns an implementation that checks for each fixed interval if there are threads that have invoked {@link #register()}
-     * and not {@link #unregister()} and have been in this state for longer than the specified max execution interval and
+     * Returns an implementation that checks for each fixed interval if there are threads that have invoked {@link #register(Matcher)}
+     * and not {@link #unregister(Matcher)} and have been in this state for longer than the specified max execution interval and
      * then interrupts these threads.
      *
      * @param interval              The fixed interval to check if there are threads to interrupt
      * @param maxExecutionTime      The time a thread has the execute an operation.
      * @param relativeTimeSupplier  A supplier that returns relative time
      * @param scheduler             A scheduler that is able to execute a command for each fixed interval
      */
-    static ThreadWatchdog newInstance(long interval,
+    static MatcherWatchdog newInstance(long interval,
                                       long maxExecutionTime,
                                       LongSupplier relativeTimeSupplier,
                                       BiConsumer<Long, Runnable> scheduler) {
         return new Default(interval, maxExecutionTime, relativeTimeSupplier, scheduler);
     }
-    
+
     /**
      * @return A noop implementation that does not interrupt threads and is useful for testing and pre-defined grok expressions.
      */
-    static ThreadWatchdog noop() {
+    static MatcherWatchdog noop() {
         return Noop.INSTANCE;
     }
-    
-    class Noop implements ThreadWatchdog {
-    
+
+    class Noop implements MatcherWatchdog {
+
         private static final Noop INSTANCE = new Noop();
-        
+
         private Noop() {
         }
-    
+
         @Override
-        public void register() {
+        public void register(Matcher matcher) {
         }
-    
+
         @Override
         public long maxExecutionTimeInMillis() {
             return Long.MAX_VALUE;
         }
-        
+
         @Override
-        public void unregister() {
+        public void unregister(Matcher matcher) {
         }
     }
-    
-    class Default implements ThreadWatchdog {
-        
+
+    class Default implements MatcherWatchdog {
+
         private final long interval;
         private final long maxExecutionTime;
         private final LongSupplier relativeTimeSupplier;
         private final BiConsumer<Long, Runnable> scheduler;
         private final AtomicInteger registered = new AtomicInteger(0);
         private final AtomicBoolean running = new AtomicBoolean(false);
-        final ConcurrentHashMap<Thread, Long> registry = new ConcurrentHashMap<>();
-        
+        final ConcurrentHashMap<Matcher, Long> registry = new ConcurrentHashMap<>();
+
         private Default(long interval,
                         long maxExecutionTime,
                         LongSupplier relativeTimeSupplier,
@@ -118,30 +124,30 @@ private Default(long interval,
             this.relativeTimeSupplier = relativeTimeSupplier;
             this.scheduler = scheduler;
         }
-        
-        public void register() {
+
+        public void register(Matcher matcher) {
             registered.getAndIncrement();
-            Long previousValue = registry.put(Thread.currentThread(), relativeTimeSupplier.getAsLong());
+            Long previousValue = registry.put(matcher, relativeTimeSupplier.getAsLong());
             if (running.compareAndSet(false, true) == true) {
                 scheduler.accept(interval, this::interruptLongRunningExecutions);
             }
             assert previousValue == null;
         }
-    
+
         @Override
         public long maxExecutionTimeInMillis() {
             return maxExecutionTime;
         }
-    
-        public void unregister() {
-            Long previousValue = registry.remove(Thread.currentThread());
+
+        public void unregister(Matcher matcher) {
+            Long previousValue = registry.remove(matcher);
             registered.decrementAndGet();
             assert previousValue != null;
         }
-        
+
         private void interruptLongRunningExecutions() {
             final long currentRelativeTime = relativeTimeSupplier.getAsLong();
-            for (Map.Entry<Thread, Long> entry : registry.entrySet()) {
+            for (Map.Entry<Matcher, Long> entry : registry.entrySet()) {
                 if ((currentRelativeTime - entry.getValue()) > maxExecutionTime) {
                     entry.getKey().interrupt();
                     // not removing the entry here, this happens in the unregister() method.
@@ -153,7 +159,7 @@ private void interruptLongRunningExecutions() {
                 running.set(false);
             }
         }
-        
+
     }
-    
+
 }

libs/grok/src/test/java/org/elasticsearch/grok/GrokTests.java

@@ -430,7 +430,7 @@ public void testExponentialExpressions() {
             });
             t.start();
         };
-        Grok grok = new Grok(basePatterns, grokPattern, ThreadWatchdog.newInstance(10, 200, System::currentTimeMillis, scheduler));
+        Grok grok = new Grok(basePatterns, grokPattern, MatcherWatchdog.newInstance(10, 200, System::currentTimeMillis, scheduler));
         Exception e = expectThrows(RuntimeException.class, () -> grok.captures(logLine));
         run.set(false);
         assertThat(e.getMessage(), equalTo("grok pattern matching was interrupted after [200] ms"));

libs/grok/src/test/java/org/elasticsearch/grok/ThreadWatchdogTests.java → libs/grok/src/test/java/org/elasticsearch/grok/MatcherWatchdogTests.java

@@ -24,22 +24,24 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.elasticsearch.test.ESTestCase;
+import org.joni.Matcher;
 import org.mockito.Mockito;
 
 import static org.hamcrest.Matchers.is;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.verifyZeroInteractions;
 
-public class ThreadWatchdogTests extends ESTestCase {
+public class MatcherWatchdogTests extends ESTestCase {
 
     public void testInterrupt() throws Exception {
         AtomicBoolean run = new AtomicBoolean(true); // to avoid a lingering thread when test has completed
-        ThreadWatchdog watchdog = ThreadWatchdog.newInstance(10, 100, System::currentTimeMillis, (delay, command) -> {
+        MatcherWatchdog watchdog = MatcherWatchdog.newInstance(10, 100, System::currentTimeMillis, (delay, command) -> {
             try {
                 Thread.sleep(delay);
             } catch (InterruptedException e) {
@@ -53,17 +55,17 @@ public void testInterrupt() throws Exception {
             thread.start();
         });
 
-        Map<?, ?> registry = ((ThreadWatchdog.Default) watchdog).registry;
+        Map<?, ?> registry = ((MatcherWatchdog.Default) watchdog).registry;
         assertThat(registry.size(), is(0));
         // need to call #register() method on a different thread, assertBusy() fails if current thread gets interrupted
         AtomicBoolean interrupted = new AtomicBoolean(false);
         Thread thread = new Thread(() -> {
-            Thread currentThread = Thread.currentThread();
-            watchdog.register();
-            while (currentThread.isInterrupted() == false) {}
+            Matcher matcher = mock(Matcher.class);
+            watchdog.register(matcher);
+            verify(matcher, timeout(9999).times(1)).interrupt();
             interrupted.set(true);
             while (run.get()) {} // wait here so that the size of the registry can be asserted
-            watchdog.unregister();
+            watchdog.unregister(matcher);
         });
         thread.start();
         assertBusy(() -> {
@@ -79,7 +81,7 @@ public void testInterrupt() throws Exception {
     public void testIdleIfNothingRegistered() throws Exception {
         long interval = 1L;
         ScheduledExecutorService threadPool = mock(ScheduledExecutorService.class);
-        ThreadWatchdog watchdog = ThreadWatchdog.newInstance(interval, Long.MAX_VALUE, System::currentTimeMillis,
+        MatcherWatchdog watchdog = MatcherWatchdog.newInstance(interval, Long.MAX_VALUE, System::currentTimeMillis,
             (delay, command) -> threadPool.schedule(command, delay, TimeUnit.MILLISECONDS));
         // Periodic action is not scheduled because no thread is registered
         verifyZeroInteractions(threadPool);
@@ -91,16 +93,20 @@ public void testIdleIfNothingRegistered() throws Exception {
         }).when(threadPool).schedule(
             any(Runnable.class), eq(interval), eq(TimeUnit.MILLISECONDS)
         );
-        watchdog.register();
+        Matcher matcher = mock(Matcher.class);
+        watchdog.register(matcher);
         // Registering the first thread should have caused the command to get scheduled again
         Runnable command = commandFuture.get(1L, TimeUnit.MILLISECONDS);
         Mockito.reset(threadPool);
-        watchdog.unregister();
+        watchdog.unregister(matcher);
         command.run();
         // Periodic action is not scheduled again because no thread is registered
         verifyZeroInteractions(threadPool);
-        watchdog.register();
-        Thread otherThread = new Thread(watchdog::register);
+        watchdog.register(matcher);
+        Thread otherThread = new Thread(() -> {
+            Matcher otherMatcher = mock(Matcher.class);
+            watchdog.register(otherMatcher);
+        });
         try {
             verify(threadPool).schedule(any(Runnable.class), eq(interval), eq(TimeUnit.MILLISECONDS));
             // Registering a second thread does not cause the command to get scheduled twice