-
Notifications
You must be signed in to change notification settings - Fork 25k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[INGEST] Interrupt the current thread if evaluation grok expressions …
…take too long (#31024) This adds a thread interrupter that allows us to encapsulate calls to org.joni.Matcher#search() This method can hang forever if the regex expression is too complex. The thread interrupter in the background checks every 3 seconds whether there are threads execution the org.joni.Matcher#search() method for longer than 5 seconds and if so interrupts these threads. Joni has checks that that for every 30k iterations it checks if the current thread is interrupted and if so returns org.joni.Matcher#INTERRUPTED Closes #28731
- Loading branch information
Showing
11 changed files
with
385 additions
and
53 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
148 changes: 148 additions & 0 deletions
148
libs/grok/src/main/java/org/elasticsearch/grok/ThreadWatchdog.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,148 @@ | ||
| /* | ||
| * Licensed to Elasticsearch under one or more contributor | ||
| * license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright | ||
| * ownership. Elasticsearch licenses this file to you under | ||
| * the Apache License, Version 2.0 (the "License"); you may | ||
| * not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
| package org.elasticsearch.grok; | ||
|
|
||
| import java.util.Map; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
| import java.util.concurrent.ScheduledFuture; | ||
| import java.util.function.BiFunction; | ||
| 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 | ||
| * 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 { | ||
|
|
||
| /** | ||
| * Registers the current thread and interrupts the current thread | ||
| * if the takes too long for this thread to invoke {@link #unregister()}. | ||
| */ | ||
| void register(); | ||
|
|
||
| /** | ||
| * @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. | ||
| */ | ||
| long maxExecutionTimeInMillis(); | ||
|
|
||
| /** | ||
| * Unregisters the current thread and prevents it from being interrupted. | ||
| */ | ||
| void unregister(); | ||
|
|
||
| /** | ||
| * 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 | ||
| * 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, | ||
| long maxExecutionTime, | ||
| LongSupplier relativeTimeSupplier, | ||
| BiFunction<Long, Runnable, ScheduledFuture<?>> 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() { | ||
| return Noop.INSTANCE; | ||
| } | ||
|
|
||
| class Noop implements ThreadWatchdog { | ||
|
|
||
| private static final Noop INSTANCE = new Noop(); | ||
|
|
||
| private Noop() { | ||
| } | ||
|
|
||
| @Override | ||
| public void register() { | ||
| } | ||
|
|
||
| @Override | ||
| public long maxExecutionTimeInMillis() { | ||
| return Long.MAX_VALUE; | ||
| } | ||
|
|
||
| @Override | ||
| public void unregister() { | ||
| } | ||
| } | ||
|
|
||
| class Default implements ThreadWatchdog { | ||
|
|
||
| private final long interval; | ||
| private final long maxExecutionTime; | ||
| private final LongSupplier relativeTimeSupplier; | ||
| private final BiFunction<Long, Runnable, ScheduledFuture<?>> scheduler; | ||
| final ConcurrentHashMap<Thread, Long> registry = new ConcurrentHashMap<>(); | ||
|
|
||
| private Default(long interval, | ||
| long maxExecutionTime, | ||
| LongSupplier relativeTimeSupplier, | ||
| BiFunction<Long, Runnable, ScheduledFuture<?>> scheduler) { | ||
| this.interval = interval; | ||
| this.maxExecutionTime = maxExecutionTime; | ||
| this.relativeTimeSupplier = relativeTimeSupplier; | ||
| this.scheduler = scheduler; | ||
| scheduler.apply(interval, this::interruptLongRunningExecutions); | ||
| } | ||
|
|
||
| public void register() { | ||
| Long previousValue = registry.put(Thread.currentThread(), relativeTimeSupplier.getAsLong()); | ||
| assert previousValue == null; | ||
| } | ||
|
|
||
| @Override | ||
| public long maxExecutionTimeInMillis() { | ||
| return maxExecutionTime; | ||
| } | ||
|
|
||
| public void unregister() { | ||
| Long previousValue = registry.remove(Thread.currentThread()); | ||
| assert previousValue != null; | ||
| } | ||
|
|
||
| private void interruptLongRunningExecutions() { | ||
| final long currentRelativeTime = relativeTimeSupplier.getAsLong(); | ||
| for (Map.Entry<Thread, Long> entry : registry.entrySet()) { | ||
| if ((currentRelativeTime - entry.getValue()) > maxExecutionTime) { | ||
| entry.getKey().interrupt(); | ||
| // not removing the entry here, this happens in the unregister() method. | ||
| } | ||
| } | ||
| scheduler.apply(interval, this::interruptLongRunningExecutions); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.