-
Notifications
You must be signed in to change notification settings - Fork 66
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
rate-limiting: propagate back-pressure from queue as HTTP 429's
Adds a proactive handler that rejects new requests with HTTP 429's when the queue has been blocking for more than 10 consecutive seconds, allowing back- pressure to propagate in advance of filling up the connection backlog queue.
- Loading branch information
Showing
10 changed files
with
634 additions
and
9 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
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 |
|---|---|---|
| @@ -1 +1 @@ | ||
| 3.8.1 | ||
| 3.9.0 |
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
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
201 changes: 201 additions & 0 deletions
201
src/main/java/org/logstash/plugins/inputs/http/util/ExecutionObserver.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,201 @@ | ||
| package org.logstash.plugins.inputs.http.util; | ||
|
|
||
| import java.lang.invoke.MethodHandles; | ||
| import java.time.Duration; | ||
| import java.util.Optional; | ||
| import java.util.concurrent.atomic.AtomicReference; | ||
| import java.util.function.LongSupplier; | ||
|
|
||
| /** | ||
| * An {@code ExecutionObserver} observes possibly-concurrent execution, and provides information about the | ||
| * longest-running observed execution. | ||
| * | ||
| * <p> | ||
| * It is concurrency-safe and non-blocking, and uses plain memory access where practical. | ||
| * </p> | ||
| */ | ||
| public class ExecutionObserver { | ||
| private final AtomicReference<Execution> head; // newest execution | ||
| private final AtomicReference<Execution> tail; // oldest execution | ||
|
|
||
| private final LongSupplier nanosSupplier; | ||
|
|
||
| public ExecutionObserver() { | ||
| this(System::nanoTime); | ||
| } | ||
|
|
||
| ExecutionObserver(final LongSupplier nanosSupplier) { | ||
| this.nanosSupplier = nanosSupplier; | ||
| final Execution anchor = new Execution(nanosSupplier.getAsLong(), true); | ||
| this.head = new AtomicReference<>(anchor); | ||
| this.tail = new AtomicReference<>(anchor); | ||
| } | ||
|
|
||
| /** | ||
| * @see ExecutionObserver#anyExecuting(Duration) | ||
| * @return true if there are any active executions. | ||
| */ | ||
| public boolean anyExecuting() { | ||
| return this.anyExecuting(Duration.ZERO); | ||
| } | ||
|
|
||
| /** | ||
| * @param minimumDuration a threshold to exclude young executions | ||
| * @return true if any active execution has been running for at least the provided {@code Duration} | ||
| */ | ||
| public boolean anyExecuting(final Duration minimumDuration) { | ||
| final Execution tailExecution = compactTail(); | ||
| if (tailExecution.isComplete) { | ||
| return false; | ||
| } else { | ||
| return nanosSupplier.getAsLong() - tailExecution.startNanos >= minimumDuration.toNanos(); | ||
| } | ||
| } | ||
|
|
||
| // visible for test | ||
| Optional<Duration> longestExecuting() { | ||
| final Execution tailExecution = compactTail(); | ||
| if (tailExecution.isComplete) { | ||
| return Optional.empty(); | ||
| } else { | ||
| return Optional.of(Duration.ofNanos(nanosSupplier.getAsLong() - tailExecution.startNanos)); | ||
| } | ||
| } | ||
|
|
||
| // test inspections | ||
| Stats stats() { | ||
| int nodes = 0; | ||
| int executing = 0; | ||
|
|
||
| Execution candidate = this.tail.get(); | ||
| while (candidate != null) { | ||
| nodes += 1; | ||
| if (!candidate.isComplete) { | ||
| executing += 1; | ||
| } | ||
| candidate = candidate.getNextPlain(); | ||
| } | ||
| return new Stats(nodes, executing); | ||
| } | ||
|
|
||
| static class Stats { | ||
| final int nodes; | ||
| final int executing; | ||
|
|
||
| Stats(int nodes, int executing) { | ||
| this.nodes = nodes; | ||
| this.executing = executing; | ||
| } | ||
| } | ||
|
|
||
| @FunctionalInterface | ||
| public interface ExceptionalSupplier<T, E extends Throwable> { | ||
| T get() throws E; | ||
| } | ||
|
|
||
| public <T,E extends Throwable> T observeExecution(final ExceptionalSupplier<T,E> supplier) throws E { | ||
| final Execution execution = startExecution(); | ||
| try { | ||
| return supplier.get(); | ||
| } finally { | ||
| final boolean isCompact = execution.markComplete(); | ||
| if (!isCompact) { | ||
| this.compactTail(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @FunctionalInterface | ||
| public interface ExceptionalRunnable<E extends Throwable> { | ||
| void run() throws E; | ||
| } | ||
|
|
||
| public <E extends Throwable> void observeExecution(final ExceptionalRunnable<E> runnable) throws E { | ||
| observeExecution(() -> { runnable.run(); return null; }); | ||
| } | ||
|
|
||
| // visible for test | ||
| Execution startExecution() { | ||
| final Execution newHead = new Execution(nanosSupplier.getAsLong()); | ||
|
|
||
| // atomically attach the new execution as a new (detached) head | ||
| final Execution oldHead = this.head.getAndSet(newHead); | ||
| // attach our new head to the old one | ||
| oldHead.linkNext(newHead); | ||
|
|
||
| return newHead; | ||
| } | ||
|
|
||
| private Execution compactTail() { | ||
| return this.tail.updateAndGet(Execution::chaseTail); | ||
| } | ||
|
|
||
| static class Execution { | ||
| private static final java.lang.invoke.VarHandle NEXT; | ||
| static { | ||
| try { | ||
| MethodHandles.Lookup l = MethodHandles.lookup(); | ||
| NEXT = l.findVarHandle(Execution.class, "next", Execution.class); | ||
| } catch (ReflectiveOperationException e) { | ||
| throw new ExceptionInInitializerError(e); | ||
| } | ||
| } | ||
|
|
||
| private final long startNanos; | ||
|
|
||
| private volatile boolean isComplete; | ||
| private volatile Execution next; | ||
|
|
||
| Execution(long startNanos) { | ||
| this(startNanos, false); | ||
| } | ||
|
|
||
| Execution(final long startNanos, | ||
| final boolean isComplete) { | ||
| this.startNanos = startNanos; | ||
| this.isComplete = isComplete; | ||
| } | ||
|
|
||
| /** | ||
| * marks this execution as complete | ||
| * @return true if the completion resulted in a compaction | ||
| */ | ||
| boolean markComplete() { | ||
| isComplete = true; | ||
|
|
||
| // concurrency: use plain memory for reads because we can tolerate | ||
| // completed nodes remaining as the result of a race | ||
| final Execution preCompletionNext = this.getNextPlain(); | ||
| if (preCompletionNext != null) { | ||
| final Execution newNext = preCompletionNext.chaseTail(); | ||
| return (newNext != preCompletionNext) && NEXT.compareAndSet(this, preCompletionNext, newNext); | ||
| } | ||
| return false; | ||
| } | ||
|
|
||
| private void linkNext(final Execution proposedNext) { | ||
| final Execution witness = (Execution)NEXT.compareAndExchange(this, null, proposedNext); | ||
| if (witness != null && witness != proposedNext) { | ||
| throw new IllegalStateException(); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * @return the next {@code Execution} that is either not yet complete | ||
| * or is the current head, using plain memory access. | ||
| */ | ||
| private Execution chaseTail() { | ||
| Execution compactedTail = this; | ||
| Execution candidate = this.getNextPlain(); | ||
| while (candidate != null && compactedTail.isComplete) { | ||
| compactedTail = candidate; | ||
| candidate = candidate.getNextPlain(); | ||
| } | ||
| return compactedTail; | ||
| } | ||
|
|
||
| private Execution getNextPlain() { | ||
| return (Execution) NEXT.get(this); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.