From b034153df7119caf3e26b4df96d58fd54b26e3fd Mon Sep 17 00:00:00 2001 From: Martijn van Groningen Date: Thu, 24 Oct 2019 15:33:30 +0200 Subject: [PATCH] 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. --- .../java/org/elasticsearch/grok/Grok.java | 44 ++++----- ...readWatchdog.java => MatcherWatchdog.java} | 92 ++++++++++--------- .../org/elasticsearch/grok/GrokTests.java | 2 +- ...ogTests.java => MatcherWatchdogTests.java} | 30 +++--- .../ingest/common/GrokProcessor.java | 14 +-- .../ingest/common/IngestCommonPlugin.java | 6 +- .../common/GrokProcessorFactoryTests.java | 18 ++-- .../ingest/common/GrokProcessorTests.java | 34 +++---- .../filestructurefinder/TimeoutChecker.java | 57 ++++++++---- .../TimeoutCheckerTests.java | 38 ++++---- 10 files changed, 177 insertions(+), 158 deletions(-) rename libs/grok/src/main/java/org/elasticsearch/grok/{ThreadWatchdog.java => MatcherWatchdog.java} (74%) rename libs/grok/src/test/java/org/elasticsearch/grok/{ThreadWatchdogTests.java => MatcherWatchdogTests.java} (81%) diff --git a/libs/grok/src/main/java/org/elasticsearch/grok/Grok.java b/libs/grok/src/main/java/org/elasticsearch/grok/Grok.java index a7a9a4a6f55..9c0febade2f 100644 --- a/libs/grok/src/main/java/org/elasticsearch/grok/Grok.java +++ b/libs/grok/src/main/java/org/elasticsearch/grok/Grok.java @@ -74,24 +74,24 @@ public final class Grok { private final Map patternBank; private final boolean namedCaptures; private final Regex compiledExpression; - private final ThreadWatchdog threadWatchdog; + private final MatcherWatchdog matcherWatchdog; public Grok(Map patternBank, String grokPattern) { - this(patternBank, grokPattern, true, ThreadWatchdog.noop()); + this(patternBank, grokPattern, true, MatcherWatchdog.noop()); } - public Grok(Map patternBank, String grokPattern, ThreadWatchdog threadWatchdog) { - this(patternBank, grokPattern, true, threadWatchdog); + public Grok(Map patternBank, String grokPattern, MatcherWatchdog matcherWatchdog) { + this(patternBank, grokPattern, true, matcherWatchdog); } Grok(Map patternBank, String grokPattern, boolean namedCaptures) { - this(patternBank, grokPattern, namedCaptures, ThreadWatchdog.noop()); + this(patternBank, grokPattern, namedCaptures, MatcherWatchdog.noop()); } - private Grok(Map patternBank, String grokPattern, boolean namedCaptures, ThreadWatchdog threadWatchdog) { + private Grok(Map patternBank, String grokPattern, boolean namedCaptures, MatcherWatchdog matcherWatchdog) { this.patternBank = patternBank; this.namedCaptures = namedCaptures; - this.threadWatchdog = threadWatchdog; + this.matcherWatchdog = matcherWatchdog; for (Map.Entry entry : patternBank.entrySet()) { String name = entry.getKey(); @@ -172,14 +172,12 @@ public final class Grok { 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); @@ -217,18 +215,16 @@ public final class Grok { * 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); } @@ -245,16 +241,14 @@ public final class Grok { 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; diff --git a/libs/grok/src/main/java/org/elasticsearch/grok/ThreadWatchdog.java b/libs/grok/src/main/java/org/elasticsearch/grok/MatcherWatchdog.java similarity index 74% rename from libs/grok/src/main/java/org/elasticsearch/grok/ThreadWatchdog.java rename to libs/grok/src/main/java/org/elasticsearch/grok/MatcherWatchdog.java index 602b3e97635..dcac74af1fe 100644 --- a/libs/grok/src/main/java/org/elasticsearch/grok/ThreadWatchdog.java +++ b/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,7 +29,7 @@ import java.util.function.LongSupplier; /** * 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 @@ -35,28 +37,32 @@ import java.util.function.LongSupplier; * 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 @@ -64,51 +70,51 @@ public interface ThreadWatchdog { * @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 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 scheduler; private final AtomicInteger registered = new AtomicInteger(0); private final AtomicBoolean running = new AtomicBoolean(false); - final ConcurrentHashMap registry = new ConcurrentHashMap<>(); - + final ConcurrentHashMap registry = new ConcurrentHashMap<>(); + private Default(long interval, long maxExecutionTime, LongSupplier relativeTimeSupplier, @@ -118,30 +124,30 @@ public interface ThreadWatchdog { 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 entry : registry.entrySet()) { + for (Map.Entry 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 @@ public interface ThreadWatchdog { running.set(false); } } - + } - + } diff --git a/libs/grok/src/test/java/org/elasticsearch/grok/GrokTests.java b/libs/grok/src/test/java/org/elasticsearch/grok/GrokTests.java index 10ed0495618..f25d502b6ef 100644 --- a/libs/grok/src/test/java/org/elasticsearch/grok/GrokTests.java +++ b/libs/grok/src/test/java/org/elasticsearch/grok/GrokTests.java @@ -430,7 +430,7 @@ public class GrokTests extends ESTestCase { }); 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")); diff --git a/libs/grok/src/test/java/org/elasticsearch/grok/ThreadWatchdogTests.java b/libs/grok/src/test/java/org/elasticsearch/grok/MatcherWatchdogTests.java similarity index 81% rename from libs/grok/src/test/java/org/elasticsearch/grok/ThreadWatchdogTests.java rename to libs/grok/src/test/java/org/elasticsearch/grok/MatcherWatchdogTests.java index 95a50c1fc5c..672a507733b 100644 --- a/libs/grok/src/test/java/org/elasticsearch/grok/ThreadWatchdogTests.java +++ b/libs/grok/src/test/java/org/elasticsearch/grok/MatcherWatchdogTests.java @@ -24,6 +24,7 @@ import java.util.concurrent.ScheduledExecutorService; 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; @@ -31,15 +32,16 @@ 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 class ThreadWatchdogTests extends ESTestCase { 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 class ThreadWatchdogTests extends ESTestCase { 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 class ThreadWatchdogTests extends ESTestCase { }).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 diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/GrokProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/GrokProcessor.java index 19883053d2a..c603da766a6 100644 --- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/GrokProcessor.java +++ b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/GrokProcessor.java @@ -20,7 +20,7 @@ package org.elasticsearch.ingest.common; import org.elasticsearch.grok.Grok; -import org.elasticsearch.grok.ThreadWatchdog; +import org.elasticsearch.grok.MatcherWatchdog; import org.elasticsearch.ingest.AbstractProcessor; import org.elasticsearch.ingest.ConfigurationUtils; import org.elasticsearch.ingest.IngestDocument; @@ -44,11 +44,11 @@ public final class GrokProcessor extends AbstractProcessor { private final boolean ignoreMissing; GrokProcessor(String tag, Map patternBank, List matchPatterns, String matchField, - boolean traceMatch, boolean ignoreMissing, ThreadWatchdog threadWatchdog) { + boolean traceMatch, boolean ignoreMissing, MatcherWatchdog matcherWatchdog) { super(tag); this.matchField = matchField; this.matchPatterns = matchPatterns; - this.grok = new Grok(patternBank, combinePatterns(matchPatterns, traceMatch), threadWatchdog); + this.grok = new Grok(patternBank, combinePatterns(matchPatterns, traceMatch), matcherWatchdog); this.traceMatch = traceMatch; this.ignoreMissing = ignoreMissing; } @@ -133,11 +133,11 @@ public final class GrokProcessor extends AbstractProcessor { public static final class Factory implements Processor.Factory { private final Map builtinPatterns; - private final ThreadWatchdog threadWatchdog; + private final MatcherWatchdog matcherWatchdog; - public Factory(Map builtinPatterns, ThreadWatchdog threadWatchdog) { + public Factory(Map builtinPatterns, MatcherWatchdog matcherWatchdog) { this.builtinPatterns = builtinPatterns; - this.threadWatchdog = threadWatchdog; + this.matcherWatchdog = matcherWatchdog; } @Override @@ -159,7 +159,7 @@ public final class GrokProcessor extends AbstractProcessor { try { return new GrokProcessor(processorTag, patternBank, matchPatterns, matchField, traceMatch, ignoreMissing, - threadWatchdog); + matcherWatchdog); } catch (Exception e) { throw newConfigurationException(TYPE, processorTag, "patterns", "Invalid regex pattern found in: " + matchPatterns + ". " + e.getMessage()); diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/IngestCommonPlugin.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/IngestCommonPlugin.java index b0fe00bdb34..57c720c7242 100644 --- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/IngestCommonPlugin.java +++ b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/IngestCommonPlugin.java @@ -30,7 +30,7 @@ import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.SettingsFilter; import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.grok.Grok; -import org.elasticsearch.grok.ThreadWatchdog; +import org.elasticsearch.grok.MatcherWatchdog; import org.elasticsearch.ingest.DropProcessor; import org.elasticsearch.ingest.PipelineProcessor; import org.elasticsearch.ingest.Processor; @@ -109,10 +109,10 @@ public class IngestCommonPlugin extends Plugin implements ActionPlugin, IngestPl return Arrays.asList(WATCHDOG_INTERVAL, WATCHDOG_MAX_EXECUTION_TIME); } - private static ThreadWatchdog createGrokThreadWatchdog(Processor.Parameters parameters) { + private static MatcherWatchdog createGrokThreadWatchdog(Processor.Parameters parameters) { long intervalMillis = WATCHDOG_INTERVAL.get(parameters.env.settings()).getMillis(); long maxExecutionTimeMillis = WATCHDOG_MAX_EXECUTION_TIME.get(parameters.env.settings()).getMillis(); - return ThreadWatchdog.newInstance(intervalMillis, maxExecutionTimeMillis, + return MatcherWatchdog.newInstance(intervalMillis, maxExecutionTimeMillis, parameters.relativeTimeSupplier, parameters.scheduler::apply); } diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GrokProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GrokProcessorFactoryTests.java index f35fa34eec4..c7029012da7 100644 --- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GrokProcessorFactoryTests.java +++ b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GrokProcessorFactoryTests.java @@ -20,7 +20,7 @@ package org.elasticsearch.ingest.common; import org.elasticsearch.ElasticsearchParseException; -import org.elasticsearch.grok.ThreadWatchdog; +import org.elasticsearch.grok.MatcherWatchdog; import org.elasticsearch.test.ESTestCase; import java.util.Collections; @@ -34,7 +34,7 @@ import static org.hamcrest.Matchers.notNullValue; public class GrokProcessorFactoryTests extends ESTestCase { public void testBuild() throws Exception { - GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), ThreadWatchdog.noop()); + GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), MatcherWatchdog.noop()); Map config = new HashMap<>(); config.put("field", "_field"); @@ -48,7 +48,7 @@ public class GrokProcessorFactoryTests extends ESTestCase { } public void testBuildWithIgnoreMissing() throws Exception { - GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), ThreadWatchdog.noop()); + GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), MatcherWatchdog.noop()); Map config = new HashMap<>(); config.put("field", "_field"); @@ -63,7 +63,7 @@ public class GrokProcessorFactoryTests extends ESTestCase { } public void testBuildMissingField() throws Exception { - GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), ThreadWatchdog.noop()); + GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), MatcherWatchdog.noop()); Map config = new HashMap<>(); config.put("patterns", Collections.singletonList("(?\\w+)")); ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class, () -> factory.create(null, null, config)); @@ -71,7 +71,7 @@ public class GrokProcessorFactoryTests extends ESTestCase { } public void testBuildMissingPatterns() throws Exception { - GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), ThreadWatchdog.noop()); + GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), MatcherWatchdog.noop()); Map config = new HashMap<>(); config.put("field", "foo"); ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class, () -> factory.create(null, null, config)); @@ -79,7 +79,7 @@ public class GrokProcessorFactoryTests extends ESTestCase { } public void testBuildEmptyPatternsList() throws Exception { - GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), ThreadWatchdog.noop()); + GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), MatcherWatchdog.noop()); Map config = new HashMap<>(); config.put("field", "foo"); config.put("patterns", Collections.emptyList()); @@ -88,7 +88,7 @@ public class GrokProcessorFactoryTests extends ESTestCase { } public void testCreateWithCustomPatterns() throws Exception { - GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), ThreadWatchdog.noop()); + GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), MatcherWatchdog.noop()); Map config = new HashMap<>(); config.put("field", "_field"); @@ -101,7 +101,7 @@ public class GrokProcessorFactoryTests extends ESTestCase { } public void testCreateWithInvalidPattern() throws Exception { - GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), ThreadWatchdog.noop()); + GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), MatcherWatchdog.noop()); Map config = new HashMap<>(); config.put("field", "_field"); config.put("patterns", Collections.singletonList("[")); @@ -110,7 +110,7 @@ public class GrokProcessorFactoryTests extends ESTestCase { } public void testCreateWithInvalidPatternDefinition() throws Exception { - GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), ThreadWatchdog.noop()); + GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap(), MatcherWatchdog.noop()); Map config = new HashMap<>(); config.put("field", "_field"); config.put("patterns", Collections.singletonList("%{MY_PATTERN:name}!")); diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GrokProcessorTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GrokProcessorTests.java index 68654923ae9..159115f4e38 100644 --- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GrokProcessorTests.java +++ b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GrokProcessorTests.java @@ -19,7 +19,7 @@ package org.elasticsearch.ingest.common; -import org.elasticsearch.grok.ThreadWatchdog; +import org.elasticsearch.grok.MatcherWatchdog; import org.elasticsearch.ingest.IngestDocument; import org.elasticsearch.ingest.RandomDocumentPicks; import org.elasticsearch.test.ESTestCase; @@ -40,7 +40,7 @@ public class GrokProcessorTests extends ESTestCase { IngestDocument doc = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>()); doc.setFieldValue(fieldName, "1"); GrokProcessor processor = new GrokProcessor(randomAlphaOfLength(10), Collections.singletonMap("ONE", "1"), - Collections.singletonList("%{ONE:one}"), fieldName, false, false, ThreadWatchdog.noop()); + Collections.singletonList("%{ONE:one}"), fieldName, false, false, MatcherWatchdog.noop()); processor.execute(doc); assertThat(doc.getFieldValue("one", String.class), equalTo("1")); } @@ -50,7 +50,7 @@ public class GrokProcessorTests extends ESTestCase { IngestDocument doc = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>()); doc.setFieldValue(fieldName, "23"); GrokProcessor processor = new GrokProcessor(randomAlphaOfLength(10), Collections.singletonMap("ONE", "1"), - Collections.singletonList("%{ONE:one}"), fieldName, false, false, ThreadWatchdog.noop()); + Collections.singletonList("%{ONE:one}"), fieldName, false, false, MatcherWatchdog.noop()); Exception e = expectThrows(Exception.class, () -> processor.execute(doc)); assertThat(e.getMessage(), equalTo("Provided Grok expressions do not match field value: [23]")); } @@ -61,7 +61,7 @@ public class GrokProcessorTests extends ESTestCase { doc.setFieldValue(fieldName, "23"); Exception e = expectThrows(IllegalArgumentException.class, () -> new GrokProcessor(randomAlphaOfLength(10), Collections.singletonMap("ONE", "1"), Collections.singletonList("%{NOTONE:not_one}"), fieldName, - false, false, ThreadWatchdog.noop())); + false, false, MatcherWatchdog.noop())); assertThat(e.getMessage(), equalTo("Unable to find pattern [NOTONE] in Grok's pattern dictionary")); } @@ -71,7 +71,7 @@ public class GrokProcessorTests extends ESTestCase { originalDoc.setFieldValue(fieldName, fieldName); IngestDocument doc = new IngestDocument(originalDoc); GrokProcessor processor = new GrokProcessor(randomAlphaOfLength(10), Collections.emptyMap(), - Collections.singletonList(fieldName), fieldName, false, false, ThreadWatchdog.noop()); + Collections.singletonList(fieldName), fieldName, false, false, MatcherWatchdog.noop()); processor.execute(doc); assertThat(doc, equalTo(originalDoc)); } @@ -81,7 +81,7 @@ public class GrokProcessorTests extends ESTestCase { IngestDocument doc = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>()); doc.setFieldValue(fieldName, null); GrokProcessor processor = new GrokProcessor(randomAlphaOfLength(10), Collections.singletonMap("ONE", "1"), - Collections.singletonList("%{ONE:one}"), fieldName, false, false, ThreadWatchdog.noop()); + Collections.singletonList("%{ONE:one}"), fieldName, false, false, MatcherWatchdog.noop()); Exception e = expectThrows(Exception.class, () -> processor.execute(doc)); assertThat(e.getMessage(), equalTo("field [" + fieldName + "] is null, cannot process it.")); } @@ -92,7 +92,7 @@ public class GrokProcessorTests extends ESTestCase { originalIngestDocument.setFieldValue(fieldName, null); IngestDocument ingestDocument = new IngestDocument(originalIngestDocument); GrokProcessor processor = new GrokProcessor(randomAlphaOfLength(10), Collections.singletonMap("ONE", "1"), - Collections.singletonList("%{ONE:one}"), fieldName, false, true, ThreadWatchdog.noop()); + Collections.singletonList("%{ONE:one}"), fieldName, false, true, MatcherWatchdog.noop()); processor.execute(ingestDocument); assertIngestDocument(originalIngestDocument, ingestDocument); } @@ -102,7 +102,7 @@ public class GrokProcessorTests extends ESTestCase { IngestDocument doc = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>()); doc.setFieldValue(fieldName, 1); GrokProcessor processor = new GrokProcessor(randomAlphaOfLength(10), Collections.singletonMap("ONE", "1"), - Collections.singletonList("%{ONE:one}"), fieldName, false, false, ThreadWatchdog.noop()); + Collections.singletonList("%{ONE:one}"), fieldName, false, false, MatcherWatchdog.noop()); Exception e = expectThrows(Exception.class, () -> processor.execute(doc)); assertThat(e.getMessage(), equalTo("field [" + fieldName + "] of type [java.lang.Integer] cannot be cast to [java.lang.String]")); } @@ -112,7 +112,7 @@ public class GrokProcessorTests extends ESTestCase { IngestDocument doc = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>()); doc.setFieldValue(fieldName, 1); GrokProcessor processor = new GrokProcessor(randomAlphaOfLength(10), Collections.singletonMap("ONE", "1"), - Collections.singletonList("%{ONE:one}"), fieldName, false, true, ThreadWatchdog.noop()); + Collections.singletonList("%{ONE:one}"), fieldName, false, true, MatcherWatchdog.noop()); Exception e = expectThrows(Exception.class, () -> processor.execute(doc)); assertThat(e.getMessage(), equalTo("field [" + fieldName + "] of type [java.lang.Integer] cannot be cast to [java.lang.String]")); } @@ -121,7 +121,7 @@ public class GrokProcessorTests extends ESTestCase { String fieldName = "foo.bar"; IngestDocument doc = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>()); GrokProcessor processor = new GrokProcessor(randomAlphaOfLength(10), Collections.singletonMap("ONE", "1"), - Collections.singletonList("%{ONE:one}"), fieldName, false, false, ThreadWatchdog.noop()); + Collections.singletonList("%{ONE:one}"), fieldName, false, false, MatcherWatchdog.noop()); Exception e = expectThrows(Exception.class, () -> processor.execute(doc)); assertThat(e.getMessage(), equalTo("field [foo] not present as part of path [foo.bar]")); } @@ -131,7 +131,7 @@ public class GrokProcessorTests extends ESTestCase { IngestDocument originalIngestDocument = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>()); IngestDocument ingestDocument = new IngestDocument(originalIngestDocument); GrokProcessor processor = new GrokProcessor(randomAlphaOfLength(10), Collections.singletonMap("ONE", "1"), - Collections.singletonList("%{ONE:one}"), fieldName, false, true, ThreadWatchdog.noop()); + Collections.singletonList("%{ONE:one}"), fieldName, false, true, MatcherWatchdog.noop()); processor.execute(ingestDocument); assertIngestDocument(originalIngestDocument, ingestDocument); } @@ -145,7 +145,7 @@ public class GrokProcessorTests extends ESTestCase { patternBank.put("TWO", "2"); patternBank.put("THREE", "3"); GrokProcessor processor = new GrokProcessor(randomAlphaOfLength(10), patternBank, - Arrays.asList("%{ONE:one}", "%{TWO:two}", "%{THREE:three}"), fieldName, false, false, ThreadWatchdog.noop()); + Arrays.asList("%{ONE:one}", "%{TWO:two}", "%{THREE:three}"), fieldName, false, false, MatcherWatchdog.noop()); processor.execute(doc); assertThat(doc.hasField("one"), equalTo(false)); assertThat(doc.getFieldValue("two", String.class), equalTo("2")); @@ -161,7 +161,7 @@ public class GrokProcessorTests extends ESTestCase { patternBank.put("TWO", "2"); patternBank.put("THREE", "3"); GrokProcessor processor = new GrokProcessor(randomAlphaOfLength(10), patternBank, - Arrays.asList("%{ONE:one}", "%{TWO:two}", "%{THREE:three}"), fieldName, true, false, ThreadWatchdog.noop()); + Arrays.asList("%{ONE:one}", "%{TWO:two}", "%{THREE:three}"), fieldName, true, false, MatcherWatchdog.noop()); processor.execute(doc); assertThat(doc.hasField("one"), equalTo(false)); assertThat(doc.getFieldValue("two", String.class), equalTo("2")); @@ -176,7 +176,7 @@ public class GrokProcessorTests extends ESTestCase { Map patternBank = new HashMap<>(); patternBank.put("ONE", "1"); GrokProcessor processor = new GrokProcessor(randomAlphaOfLength(10), patternBank, - Arrays.asList("%{ONE:one}"), fieldName, true, false, ThreadWatchdog.noop()); + Arrays.asList("%{ONE:one}"), fieldName, true, false, MatcherWatchdog.noop()); processor.execute(doc); assertThat(doc.hasField("one"), equalTo(true)); assertThat(doc.getFieldValue("_ingest._grok_match_index", String.class), equalTo("0")); @@ -207,7 +207,7 @@ public class GrokProcessorTests extends ESTestCase { patternBank.put("TWO", "2"); patternBank.put("THREE", "3"); GrokProcessor processor = new GrokProcessor(randomAlphaOfLength(10), patternBank, Arrays.asList("%{ONE:first}-%{TWO:second}", - "%{ONE:first}-%{THREE:second}"), fieldName, randomBoolean(), randomBoolean(), ThreadWatchdog.noop()); + "%{ONE:first}-%{THREE:second}"), fieldName, randomBoolean(), randomBoolean(), MatcherWatchdog.noop()); processor.execute(doc); assertThat(doc.getFieldValue("first", String.class), equalTo("1")); assertThat(doc.getFieldValue("second", String.class), equalTo("3")); @@ -221,7 +221,7 @@ public class GrokProcessorTests extends ESTestCase { patternBank.put("ONETWO", "1|2"); GrokProcessor processor = new GrokProcessor(randomAlphaOfLength(10), patternBank, Collections.singletonList("%{ONETWO:first}%{ONETWO:first}"), fieldName, randomBoolean(), randomBoolean(), - ThreadWatchdog.noop()); + MatcherWatchdog.noop()); processor.execute(doc); assertThat(doc.getFieldValue("first", String.class), equalTo("1")); } @@ -235,7 +235,7 @@ public class GrokProcessorTests extends ESTestCase { patternBank.put("THREE", "3"); GrokProcessor processor = new GrokProcessor(randomAlphaOfLength(10), patternBank, Collections.singletonList("%{ONETWO:first}|%{THREE:second}"), fieldName, randomBoolean(), randomBoolean(), - ThreadWatchdog.noop()); + MatcherWatchdog.noop()); processor.execute(doc); assertFalse(doc.hasField("first")); assertThat(doc.getFieldValue("second", String.class), equalTo("3")); diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/filestructurefinder/TimeoutChecker.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/filestructurefinder/TimeoutChecker.java index 97eb6146a66..a943e7b7a73 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/filestructurefinder/TimeoutChecker.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/filestructurefinder/TimeoutChecker.java @@ -6,15 +6,17 @@ package org.elasticsearch.xpack.ml.filestructurefinder; import org.elasticsearch.ElasticsearchTimeoutException; -import org.elasticsearch.common.collect.Tuple; import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.common.util.concurrent.FutureUtils; import org.elasticsearch.grok.Grok; -import org.elasticsearch.grok.ThreadWatchdog; +import org.elasticsearch.grok.MatcherWatchdog; +import org.joni.Matcher; import java.io.Closeable; +import java.util.Collection; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -38,7 +40,7 @@ import java.util.concurrent.atomic.AtomicBoolean; public class TimeoutChecker implements Closeable { private static final TimeoutCheckerWatchdog timeoutCheckerWatchdog = new TimeoutCheckerWatchdog(); - public static final ThreadWatchdog watchdog = timeoutCheckerWatchdog; + public static final MatcherWatchdog watchdog = timeoutCheckerWatchdog; private final String operation; private final TimeValue timeout; @@ -122,51 +124,68 @@ public class TimeoutChecker implements Closeable { /** * An implementation of the type of watchdog used by the {@link Grok} class to interrupt * matching operations that take too long. Rather than have a timeout per match operation - * like the {@link ThreadWatchdog.Default} implementation, the interruption is governed by + * like the {@link MatcherWatchdog.Default} implementation, the interruption is governed by * a {@link TimeoutChecker} associated with the thread doing the matching. */ - static class TimeoutCheckerWatchdog implements ThreadWatchdog { + static class TimeoutCheckerWatchdog implements MatcherWatchdog { - final ConcurrentHashMap> registry = new ConcurrentHashMap<>(); + final ConcurrentHashMap registry = new ConcurrentHashMap<>(); void add(Thread thread, TimeValue timeout) { - Tuple previousValue = registry.put(thread, new Tuple<>(new AtomicBoolean(false), timeout)); + WatchDogEntry previousValue = registry.put(thread, new WatchDogEntry(timeout)); assert previousValue == null; } @Override - public void register() { - Tuple value = registry.get(Thread.currentThread()); + public void register(Matcher matcher) { + WatchDogEntry value = registry.get(Thread.currentThread()); if (value != null) { - boolean wasFalse = value.v1().compareAndSet(false, true); + boolean wasFalse = value.registered.compareAndSet(false, true); assert wasFalse; + value.matchers.add(matcher); } } @Override public long maxExecutionTimeInMillis() { - Tuple value = registry.get(Thread.currentThread()); - return value != null ? value.v2().getMillis() : Long.MAX_VALUE; + WatchDogEntry value = registry.get(Thread.currentThread()); + return value != null ? value.timeout.getMillis() : Long.MAX_VALUE; } @Override - public void unregister() { - Tuple value = registry.get(Thread.currentThread()); + public void unregister(Matcher matcher) { + WatchDogEntry value = registry.get(Thread.currentThread()); if (value != null) { - boolean wasTrue = value.v1().compareAndSet(true, false); + boolean wasTrue = value.registered.compareAndSet(true, false); assert wasTrue; + value.matchers.remove(matcher); } } void remove(Thread thread) { - Tuple previousValue = registry.remove(thread); + WatchDogEntry previousValue = registry.remove(thread); assert previousValue != null; } void interruptLongRunningThreadIfRegistered(Thread thread) { - Tuple value = registry.get(thread); - if (value.v1().get()) { - thread.interrupt(); + WatchDogEntry value = registry.get(thread); + if (value.registered.get()) { + for (Matcher matcher : value.matchers) { + matcher.interrupt(); + } + } + } + + static class WatchDogEntry { + + final TimeValue timeout; + final AtomicBoolean registered; + final Collection matchers; + + WatchDogEntry(TimeValue timeout) { + this.timeout = timeout; + this.registered = new AtomicBoolean(false); + this.matchers = new CopyOnWriteArrayList<>(); } } } diff --git a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/filestructurefinder/TimeoutCheckerTests.java b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/filestructurefinder/TimeoutCheckerTests.java index 2770656279c..63346de64c2 100644 --- a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/filestructurefinder/TimeoutCheckerTests.java +++ b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/filestructurefinder/TimeoutCheckerTests.java @@ -9,11 +9,16 @@ import org.elasticsearch.ElasticsearchTimeoutException; import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.grok.Grok; import org.elasticsearch.threadpool.Scheduler; +import org.joni.Matcher; import org.junit.After; import org.junit.Before; import java.util.concurrent.ScheduledExecutorService; +import static org.hamcrest.Matchers.equalTo; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + public class TimeoutCheckerTests extends FileStructureTestCase { private ScheduledExecutorService scheduler; @@ -29,15 +34,12 @@ public class TimeoutCheckerTests extends FileStructureTestCase { } public void testCheckNoTimeout() { - NOOP_TIMEOUT_CHECKER.check("should never happen"); } public void testCheckTimeoutNotExceeded() throws InterruptedException { - TimeValue timeout = TimeValue.timeValueSeconds(10); try (TimeoutChecker timeoutChecker = new TimeoutChecker("timeout not exceeded test", timeout, scheduler)) { - for (int count = 0; count < 10; ++count) { timeoutChecker.check("should not timeout"); Thread.sleep(randomIntBetween(1, 10)); @@ -46,10 +48,8 @@ public class TimeoutCheckerTests extends FileStructureTestCase { } public void testCheckTimeoutExceeded() throws Exception { - TimeValue timeout = TimeValue.timeValueMillis(10); try (TimeoutChecker timeoutChecker = new TimeoutChecker("timeout exceeded test", timeout, scheduler)) { - assertBusy(() -> { ElasticsearchTimeoutException e = expectThrows(ElasticsearchTimeoutException.class, () -> timeoutChecker.check("should timeout")); @@ -59,30 +59,27 @@ public class TimeoutCheckerTests extends FileStructureTestCase { } } - public void testWatchdog() { - - assertFalse(Thread.interrupted()); - - TimeValue timeout = TimeValue.timeValueMillis(100); + public void testWatchdog() throws Exception { + TimeValue timeout = TimeValue.timeValueMillis(500); try (TimeoutChecker timeoutChecker = new TimeoutChecker("watchdog test", timeout, scheduler)) { + TimeoutChecker.TimeoutCheckerWatchdog watchdog = (TimeoutChecker.TimeoutCheckerWatchdog) TimeoutChecker.watchdog; - TimeoutChecker.watchdog.register(); + Matcher matcher = mock(Matcher.class); + TimeoutChecker.watchdog.register(matcher); + assertThat(watchdog.registry.get(Thread.currentThread()).matchers.size(), equalTo(1)); try { - expectThrows(InterruptedException.class, () -> Thread.sleep(10000)); + assertBusy(() -> { + verify(matcher).interrupt(); + }); } finally { - TimeoutChecker.watchdog.unregister(); + TimeoutChecker.watchdog.unregister(matcher); + assertThat(watchdog.registry.get(Thread.currentThread()).matchers.size(), equalTo(0)); } - } finally { - // ensure the interrupted flag is cleared to stop it making subsequent tests fail - Thread.interrupted(); } } public void testGrokCaptures() throws Exception { - - assertFalse(Thread.interrupted()); Grok grok = new Grok(Grok.getBuiltinPatterns(), "{%DATA:data}{%GREEDYDATA:greedydata}", TimeoutChecker.watchdog); - TimeValue timeout = TimeValue.timeValueMillis(1); try (TimeoutChecker timeoutChecker = new TimeoutChecker("grok captures test", timeout, scheduler)) { @@ -92,9 +89,6 @@ public class TimeoutCheckerTests extends FileStructureTestCase { assertEquals("Aborting grok captures test during [should timeout] as it has taken longer than the timeout of [" + timeout + "]", e.getMessage()); }); - } finally { - // ensure the interrupted flag is cleared to stop it making subsequent tests fail - Thread.interrupted(); } } }