diff --git a/bom/application/pom.xml b/bom/application/pom.xml index bd22de0b477a2..72f73bb1cc576 100644 --- a/bom/application/pom.xml +++ b/bom/application/pom.xml @@ -1886,6 +1886,16 @@ quarkus-smallrye-openapi-common-deployment ${project.version} + + io.quarkus + quarkus-load-shedding + ${project.version} + + + io.quarkus + quarkus-load-shedding-deployment + ${project.version} + io.quarkus quarkus-vertx diff --git a/extensions/load-shedding/deployment/pom.xml b/extensions/load-shedding/deployment/pom.xml new file mode 100644 index 0000000000000..1b874dc9a8316 --- /dev/null +++ b/extensions/load-shedding/deployment/pom.xml @@ -0,0 +1,52 @@ + + + 4.0.0 + + + io.quarkus + quarkus-load-shedding-parent + 999-SNAPSHOT + + + quarkus-load-shedding-deployment + + Quarkus - Load Shedding - Deployment + + + + io.quarkus + quarkus-load-shedding + + + + io.quarkus + quarkus-vertx-http-deployment + + + + io.quarkus + quarkus-junit5-internal + test + + + + + + + + maven-compiler-plugin + + + + io.quarkus + quarkus-extension-processor + ${project.version} + + + + + + + diff --git a/extensions/load-shedding/deployment/src/main/java/io/quarkus/load/shedding/deployment/LoadSheddingProcessor.java b/extensions/load-shedding/deployment/src/main/java/io/quarkus/load/shedding/deployment/LoadSheddingProcessor.java new file mode 100644 index 0000000000000..669bf6c4c252f --- /dev/null +++ b/extensions/load-shedding/deployment/src/main/java/io/quarkus/load/shedding/deployment/LoadSheddingProcessor.java @@ -0,0 +1,34 @@ +package io.quarkus.load.shedding.deployment; + +import java.util.ArrayList; +import java.util.List; + +import io.quarkus.arc.deployment.AdditionalBeanBuildItem; +import io.quarkus.deployment.annotations.BuildStep; +import io.quarkus.deployment.builditem.FeatureBuildItem; +import io.quarkus.load.shedding.runtime.HttpLoadShedding; +import io.quarkus.load.shedding.runtime.HttpRequestClassifier; +import io.quarkus.load.shedding.runtime.ManagementRequestPrioritizer; +import io.quarkus.load.shedding.runtime.OverloadDetector; +import io.quarkus.load.shedding.runtime.PriorityLoadShedding; + +public class LoadSheddingProcessor { + private static final String FEATURE = "load-shedding"; + + @BuildStep + FeatureBuildItem feature() { + return new FeatureBuildItem(FEATURE); + } + + @BuildStep + AdditionalBeanBuildItem beans() { + List beans = new ArrayList<>(); + beans.add(OverloadDetector.class.getName()); + beans.add(HttpLoadShedding.class.getName()); + beans.add(PriorityLoadShedding.class.getName()); + beans.add(ManagementRequestPrioritizer.class.getName()); + beans.add(HttpRequestClassifier.class.getName()); + + return AdditionalBeanBuildItem.builder().addBeanClasses(beans).build(); + } +} diff --git a/extensions/load-shedding/pom.xml b/extensions/load-shedding/pom.xml new file mode 100644 index 0000000000000..1116e835d64b1 --- /dev/null +++ b/extensions/load-shedding/pom.xml @@ -0,0 +1,24 @@ + + + 4.0.0 + + + quarkus-extensions-parent + io.quarkus + 999-SNAPSHOT + ../pom.xml + + + quarkus-load-shedding-parent + pom + + Quarkus - Load Shedding + + + deployment + runtime + + + diff --git a/extensions/load-shedding/runtime/pom.xml b/extensions/load-shedding/runtime/pom.xml new file mode 100644 index 0000000000000..f2ecfe9143594 --- /dev/null +++ b/extensions/load-shedding/runtime/pom.xml @@ -0,0 +1,44 @@ + + + 4.0.0 + + + io.quarkus + quarkus-load-shedding-parent + 999-SNAPSHOT + + + quarkus-load-shedding + + Quarkus - Load Shedding - Runtime + + + + io.quarkus + quarkus-vertx-http + + + + + + + io.quarkus + quarkus-extension-maven-plugin + + + maven-compiler-plugin + + + + io.quarkus + quarkus-extension-processor + ${project.version} + + + + + + + diff --git a/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/RequestClassifier.java b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/RequestClassifier.java new file mode 100644 index 0000000000000..48c9ad2823b0d --- /dev/null +++ b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/RequestClassifier.java @@ -0,0 +1,10 @@ +package io.quarkus.load.shedding; + +public interface RequestClassifier { + int MIN_COHORT = 1; + int MAX_COHORT = 128; + + boolean appliesTo(Object request); + + int cohort(R request); +} diff --git a/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/RequestPrioritizer.java b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/RequestPrioritizer.java new file mode 100644 index 0000000000000..602b8b91c1678 --- /dev/null +++ b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/RequestPrioritizer.java @@ -0,0 +1,7 @@ +package io.quarkus.load.shedding; + +public interface RequestPrioritizer { + boolean appliesTo(Object request); + + RequestPriority priority(R request); +} diff --git a/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/RequestPriority.java b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/RequestPriority.java new file mode 100644 index 0000000000000..9417214da3558 --- /dev/null +++ b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/RequestPriority.java @@ -0,0 +1,20 @@ +package io.quarkus.load.shedding; + +public enum RequestPriority { + CRITICAL(0), + IMPORTANT(1), + NORMAL(2), + BACKGROUND(3), + DEGRADED(4), + ; + + private final int value; + + RequestPriority(int value) { + this.value = value; + } + + public int value() { + return value; + } +} diff --git a/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/HttpLoadShedding.java b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/HttpLoadShedding.java new file mode 100644 index 0000000000000..8f92fe8850c41 --- /dev/null +++ b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/HttpLoadShedding.java @@ -0,0 +1,49 @@ +package io.quarkus.load.shedding.runtime; + +import jakarta.annotation.Priority; +import jakarta.enterprise.event.Observes; +import jakarta.inject.Singleton; + +import io.netty.handler.codec.http.HttpHeaderNames; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.vertx.core.AsyncResult; +import io.vertx.core.Handler; +import io.vertx.core.http.HttpServerResponse; +import io.vertx.ext.web.Router; + +@Singleton +public class HttpLoadShedding { + public void init(@Observes @Priority(-1_000_000_000) Router router, OverloadDetector detector, + PriorityLoadShedding priorityLoadShedding, LoadSheddingRuntimeConfig config) { + + if (!config.enabled()) { + return; + } + + router.route().order(-1_000_000_000).handler(ctx -> { + if (detector.isOverloaded() && priorityLoadShedding.shedLoad(ctx.request())) { + HttpServerResponse response = ctx.response(); + response.setStatusCode(HttpResponseStatus.SERVICE_UNAVAILABLE.code()); + response.headers().add(HttpHeaderNames.CONNECTION, "close"); + response.endHandler(new Handler() { + @Override + public void handle(Void ignored) { + ctx.request().connection().close(); + } + }); + response.end(); + } else { + detector.requestBegin(); + long start = System.nanoTime(); + ctx.addEndHandler(new Handler>() { + @Override + public void handle(AsyncResult ignored) { + long end = System.nanoTime(); + detector.requestEnd((end - start) / 1_000); + } + }); + ctx.next(); + } + }); + } +} diff --git a/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/HttpRequestClassifier.java b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/HttpRequestClassifier.java new file mode 100644 index 0000000000000..a0214a7337bd1 --- /dev/null +++ b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/HttpRequestClassifier.java @@ -0,0 +1,26 @@ +package io.quarkus.load.shedding.runtime; + +import jakarta.inject.Singleton; + +import io.quarkus.load.shedding.RequestClassifier; +import io.vertx.core.http.HttpServerRequest; + +@Singleton +public class HttpRequestClassifier implements RequestClassifier { + @Override + public boolean appliesTo(Object request) { + return request instanceof HttpServerRequest; + } + + @Override + public int cohort(HttpServerRequest request) { + long hour = System.currentTimeMillis() >> 23; // roughly 2.5 hours + String host = request.remoteAddress().hostAddress(); // TODO proxying, load balancing, etc.? + if (host == null) { + host = ""; + } + long hash = hour + host.hashCode(); + return (int) ((hash >> 56) & 0xFF + (hash >> 48) & 0xFF + (hash >> 40) & 0xFF + (hash >> 32) & 0xFF + + (hash >> 24) & 0xFF + (hash >> 16) & 0xFF + (hash >> 8) & 0xFF + (hash) & 0xFF); + } +} diff --git a/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/LoadSheddingRuntimeConfig.java b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/LoadSheddingRuntimeConfig.java new file mode 100644 index 0000000000000..69f28ffbcca61 --- /dev/null +++ b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/LoadSheddingRuntimeConfig.java @@ -0,0 +1,62 @@ +package io.quarkus.load.shedding.runtime; + +import io.quarkus.runtime.annotations.ConfigGroup; +import io.quarkus.runtime.annotations.ConfigPhase; +import io.quarkus.runtime.annotations.ConfigRoot; +import io.smallrye.config.ConfigMapping; +import io.smallrye.config.WithDefault; + +@ConfigMapping(prefix = "quarkus.load-shedding") +@ConfigRoot(phase = ConfigPhase.RUN_TIME) +public interface LoadSheddingRuntimeConfig { + /** + * Whether load shedding should be enabled. + * Currently, this only applies to incoming HTTP requests. + */ + @WithDefault("true") + boolean enabled(); + + /** + * The maximum number of concurrent requests allowed. + */ + @WithDefault("1000") + int maxLimit(); + + /** + * The {@code alpha} factor of the Vegas overload detection algorithm. + */ + @WithDefault("3") + int alphaFactor(); + + /** + * The {@code beta} factor of the Vegas overload detection algorithm. + */ + @WithDefault("6") + int betaFactor(); + + /** + * The probe factor of the Vegas overload detection algorithm. + */ + @WithDefault("30.0") + double probeFactor(); + + /** + * The initial limit of concurrent requests allowed. + */ + @WithDefault("100") + int initialLimit(); + + /** + * Configuration of priority load shedding. + */ + PriorityLoadShedding priority(); + + @ConfigGroup + interface PriorityLoadShedding { + /** + * Whether priority load shedding should be enabled. + */ + @WithDefault("true") + boolean enabled(); + } +} diff --git a/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/ManagementRequestPrioritizer.java b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/ManagementRequestPrioritizer.java new file mode 100644 index 0000000000000..b811ce2cff5fa --- /dev/null +++ b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/ManagementRequestPrioritizer.java @@ -0,0 +1,42 @@ +package io.quarkus.load.shedding.runtime; + +import jakarta.inject.Inject; +import jakarta.inject.Singleton; + +import io.quarkus.load.shedding.RequestPrioritizer; +import io.quarkus.load.shedding.RequestPriority; +import io.quarkus.vertx.http.runtime.HttpBuildTimeConfig; +import io.quarkus.vertx.http.runtime.management.ManagementInterfaceBuildTimeConfig; +import io.vertx.core.http.HttpServerRequest; + +@Singleton +public class ManagementRequestPrioritizer implements RequestPrioritizer { + private final String managementPath; + + @Inject + public ManagementRequestPrioritizer(HttpBuildTimeConfig httpConfig, + ManagementInterfaceBuildTimeConfig managementInterfaceConfig) { + if (managementInterfaceConfig.enabled) { + managementPath = null; + return; + } + if (httpConfig.nonApplicationRootPath.startsWith("/")) { + managementPath = httpConfig.nonApplicationRootPath; + return; + } + managementPath = httpConfig.rootPath + httpConfig.nonApplicationRootPath; + } + + @Override + public boolean appliesTo(Object request) { + if (managementPath != null && request instanceof HttpServerRequest httpRequest) { + return httpRequest.path().startsWith(managementPath); + } + return false; + } + + @Override + public RequestPriority priority(HttpServerRequest request) { + return RequestPriority.CRITICAL; + } +} diff --git a/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/OverloadDetector.java b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/OverloadDetector.java new file mode 100644 index 0000000000000..300409e04536e --- /dev/null +++ b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/OverloadDetector.java @@ -0,0 +1,95 @@ +package io.quarkus.load.shedding.runtime; + +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicInteger; + +import jakarta.inject.Inject; +import jakarta.inject.Singleton; + +/** + * An overload detector based on TCP Vegas, as implemented by + * Netflix Concurrency Limits. + */ +@Singleton +public class OverloadDetector { + private final int maxLimit; + private final int alphaFactor; + private final int betaFactor; + private final double probeFactor; + + private final AtomicInteger currentRequests = new AtomicInteger(); + private volatile long currentLimit; + + private long lowestRequestTime = Long.MAX_VALUE; + private double probeCount = 0.0; + private double probeJitter; + + @Inject + public OverloadDetector(LoadSheddingRuntimeConfig config) { + maxLimit = config.maxLimit(); + alphaFactor = config.alphaFactor(); + betaFactor = config.betaFactor(); + probeFactor = config.probeFactor(); + currentLimit = config.initialLimit(); + resetProbeJitter(); + } + + public boolean isOverloaded() { + return currentRequests.get() >= currentLimit; + } + + public void requestBegin() { + currentRequests.incrementAndGet(); + } + + public void requestEnd(long timeInMicros) { + int current = currentRequests.getAndDecrement(); + + update(timeInMicros, current); + } + + private synchronized void update(long requestTime, int currentRequests) { + probeCount++; + if (probeFactor * probeJitter * currentLimit <= probeCount) { + resetProbeJitter(); + probeCount = 0.0; + lowestRequestTime = requestTime; + return; + } + + if (requestTime < lowestRequestTime) { + lowestRequestTime = requestTime; + return; + } + + long currentLimit = this.currentLimit; + + if (2L * currentRequests < currentLimit) { + return; + } + + int queueSize = (int) Math.ceil(currentLimit * (1.0 - (double) lowestRequestTime / (double) requestTime)); + + int currentLimitLog10 = 1 + (int) Math.log10(currentLimit); + int alpha = alphaFactor * currentLimitLog10; + int beta = betaFactor * currentLimitLog10; + + long newLimit; + if (queueSize <= currentLimitLog10) { + newLimit = currentLimit + beta; + } else if (queueSize < alpha) { + newLimit = currentLimit + currentLimitLog10; + } else if (queueSize > beta) { + newLimit = currentLimit - currentLimitLog10; + } else { + return; + } + + newLimit = Math.max(1, Math.min(maxLimit, newLimit)); + this.currentLimit = newLimit; + } + + private void resetProbeJitter() { + probeJitter = ThreadLocalRandom.current().nextDouble(0.5, 1); + } +} diff --git a/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/PriorityLoadShedding.java b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/PriorityLoadShedding.java new file mode 100644 index 0000000000000..64f65ea56dcca --- /dev/null +++ b/extensions/load-shedding/runtime/src/main/java/io/quarkus/load/shedding/runtime/PriorityLoadShedding.java @@ -0,0 +1,90 @@ +package io.quarkus.load.shedding.runtime; + +import java.lang.management.ManagementFactory; +import java.util.List; + +import jakarta.inject.Inject; +import jakarta.inject.Singleton; + +import org.jboss.logging.Logger; + +import com.sun.management.OperatingSystemMXBean; + +import io.quarkus.arc.All; +import io.quarkus.load.shedding.RequestClassifier; +import io.quarkus.load.shedding.RequestPrioritizer; +import io.quarkus.load.shedding.RequestPriority; + +@Singleton +public class PriorityLoadShedding { + @Inject + @All + List> requestPrioritizers; + + @Inject + @All + List> requestClassifiers; + + private final boolean enabled; + + private final int max; + + private final OperatingSystemMXBean os; + + private double lastCpuLoad; + + private long lastCpuLoadTime; + + @Inject + PriorityLoadShedding(LoadSheddingRuntimeConfig config) { + enabled = config.priority().enabled(); + max = RequestPriority.values().length * RequestClassifier.MAX_COHORT; + os = (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean(); + } + + // when this is called, we know we're overloaded + public boolean shedLoad(Object request) { + if (!enabled) { + return true; + } + + synchronized (this) { + long now = System.currentTimeMillis(); + if (now - lastCpuLoadTime > 1_000) { + lastCpuLoad = os.getCpuLoad(); + lastCpuLoadTime = now; + } + } + double load = lastCpuLoad; + if (load < 0) { + return true; + } + + RequestPriority priority = RequestPriority.NORMAL; + for (RequestPrioritizer requestPrioritizer : requestPrioritizers) { + if (requestPrioritizer.appliesTo(request)) { + priority = requestPrioritizer.priority(request); + break; + } + } + + int cohort = 1; + for (RequestClassifier requestClassifier : requestClassifiers) { + if (requestClassifier.appliesTo(request)) { + cohort = requestClassifier.cohort(request); + break; + } + } + if (cohort == Integer.MIN_VALUE) { + cohort = 1; + } else if (cohort < 0) { + cohort = -cohort % RequestClassifier.MAX_COHORT; + } else if (cohort == 0) { + cohort = 1; + } else if (cohort > RequestClassifier.MAX_COHORT) { + cohort = cohort % RequestClassifier.MAX_COHORT + 1; + } + + return priority.value() * RequestClassifier.MAX_COHORT + cohort > max * (1.0 - load * load * load); + } +} diff --git a/extensions/load-shedding/runtime/src/main/resources/META-INF/quarkus-extension.yaml b/extensions/load-shedding/runtime/src/main/resources/META-INF/quarkus-extension.yaml new file mode 100644 index 0000000000000..9631c75bb9938 --- /dev/null +++ b/extensions/load-shedding/runtime/src/main/resources/META-INF/quarkus-extension.yaml @@ -0,0 +1,13 @@ +--- +artifact: ${project.groupId}:${project.artifactId}:${project.version} +name: "Load Shedding" +metadata: + keywords: + - "fault-tolerance" + - "load-shedding" + guide: "https://quarkus.io/guides/load-shedding" + categories: + - "cloud" + status: "experimental" + config: + - "quarkus.load-shedding." diff --git a/extensions/pom.xml b/extensions/pom.xml index 254514c22fcdb..f6182bedf8b99 100644 --- a/extensions/pom.xml +++ b/extensions/pom.xml @@ -53,6 +53,7 @@ opentelemetry info observability-devservices + load-shedding resteasy-classic diff --git a/extensions/vertx-http/runtime/src/main/java/io/quarkus/vertx/http/runtime/RouteConstants.java b/extensions/vertx-http/runtime/src/main/java/io/quarkus/vertx/http/runtime/RouteConstants.java index 6d00a3afa9b07..e1030bacc24ac 100644 --- a/extensions/vertx-http/runtime/src/main/java/io/quarkus/vertx/http/runtime/RouteConstants.java +++ b/extensions/vertx-http/runtime/src/main/java/io/quarkus/vertx/http/runtime/RouteConstants.java @@ -1,7 +1,7 @@ package io.quarkus.vertx.http.runtime; /** - * Route order value constants used in Quarkus, update {@code reactive-routes.adoc} when changing this class. + * Route order value constants used in Quarkus, update {@code http-reference.adoc} when changing this class. */ @SuppressWarnings("JavadocDeclaration") public final class RouteConstants {