Throttle using overload handling from ThreadedRequestHandler

5 files changed, 66 insertions, 81 deletions

diff --git a/vespaclient-container-plugin/src/main/java/com/yahoo/vespa/http/server/ClientFeederV3.java b/vespaclient-container-plugin/src/main/java/com/yahoo/vespa/http/server/ClientFeederV3.java
index c63593b3af6..7959945e8a0 100644
--- a/vespaclient-container-plugin/src/main/java/com/yahoo/vespa/http/server/ClientFeederV3.java
+++ b/vespaclient-container-plugin/src/main/java/com/yahoo/vespa/http/server/ClientFeederV3.java
@@ -49,7 +49,6 @@ class ClientFeederV3 {
     private final String clientId;
     private final ReplyHandler feedReplyHandler;
     private final Metric metric;
-    private final HttpThrottlePolicy httpThrottlePolicy;
     private Instant prevOpsPerSecTime = Instant.now();
     private double operationsForOpsPerSec = 0d;
     private final Object monitor = new Object();
@@ -63,13 +62,11 @@ class ClientFeederV3 {
             DocumentTypeManager docTypeManager,
             String clientId,
             Metric metric,
-            ReplyHandler feedReplyHandler,
-            HttpThrottlePolicy httpThrottlePolicy) {
+            ReplyHandler feedReplyHandler) {
         this.sourceSession = sourceSession;
         this.clientId = clientId;
         this.feedReplyHandler = feedReplyHandler;
         this.metric = metric;
-        this.httpThrottlePolicy = httpThrottlePolicy;
         this.streamReaderV3 = new StreamReaderV3(feedReaderFactory, docTypeManager);
         this.hostName = HostName.getLocalhost();
     }
@@ -105,10 +102,6 @@ class ClientFeederV3 {
         ongoingRequests.incrementAndGet();
         try {
             FeederSettings feederSettings = new FeederSettings(request);
-            if (httpThrottlePolicy.shouldThrottle()) {
-                return new ErrorHttpResponse(getOverloadReturnCode(request), "Gateway overloaded");
-            }
-
             InputStream inputStream = StreamReaderV3.unzipStreamIfNeeded(request);
             BlockingQueue<OperationStatus> replies = new LinkedBlockingQueue<>();
             try {
@@ -133,11 +126,6 @@ class ClientFeederV3 {
         }
     }
 
-    private int getOverloadReturnCode(HttpRequest request) {
-        if (request.getHeader(Headers.SILENTUPGRADE) != null ) return 299;
-        return 429;
-    }
-
     private Optional<DocumentOperationMessageV3> pullMessageFromRequest(FeederSettings settings,
                                                                         InputStream requestInputStream,
                                                                         BlockingQueue<OperationStatus> repliesFromOldMessages) {
@@ -294,26 +282,4 @@ class ClientFeederV3 {
             }
         }
     }
-
-    /*
-     * The gateway handle overload from clients in different ways.
-     *
-     * If the backend is overloaded, but not the gateway, it will fill the backend, messagebus throttler
-     * will start to block new documents and finally all threadsAvailableForFeeding will be blocking.
-     * However, as more threads are added, the gateway will not block on messagebus but return
-     * transitive errors on the documents that can not be processed. These errors will cause the client(s) to
-     * back off a bit.
-     *
-     * However, we can also have the case that the gateway becomes the bottleneck (e.g. CPU). In this case
-     * we need to stop processing of new messages as early as possible and reject the request. This
-     * will cause the client(s) to back off for a while. We want some slack before we enter this mode.
-     * If we can simply transitively fail each document, it is nicer. Therefor we allow some threads to be
-     * busy processing requests with transitive errors before entering this mode. Since we already
-     * have flooded the backend, have several threads hanging and waiting for capacity, the number should
-     * not be very large. Too much slack can lead to too many threads handling feed and impacting query traffic.
-     */
-    interface HttpThrottlePolicy {
-        boolean shouldThrottle();
-    }
-
 }
diff --git a/vespaclient-container-plugin/src/main/java/com/yahoo/vespa/http/server/FeedHandler.java b/vespaclient-container-plugin/src/main/java/com/yahoo/vespa/http/server/FeedHandler.java
index fc26034a311..2ba1f0e98cc 100644
--- a/vespaclient-container-plugin/src/main/java/com/yahoo/vespa/http/server/FeedHandler.java
+++ b/vespaclient-container-plugin/src/main/java/com/yahoo/vespa/http/server/FeedHandler.java
@@ -11,6 +11,9 @@ import com.yahoo.container.logging.AccessLog;
 import com.yahoo.document.config.DocumentmanagerConfig;
 import com.yahoo.documentapi.metrics.DocumentApiMetrics;
 import com.yahoo.jdisc.Metric;
+import com.yahoo.jdisc.Request;
+import com.yahoo.jdisc.handler.ResponseDispatch;
+import com.yahoo.jdisc.handler.ResponseHandler;
 import com.yahoo.messagebus.ReplyHandler;
 import com.yahoo.metrics.simple.MetricReceiver;
 import com.yahoo.vespa.http.client.core.Headers;
@@ -49,7 +52,7 @@ public class FeedHandler extends LoggingRequestHandler {
                        MetricReceiver metricReceiver) {
         super(threadpool.executor(), accessLog, metric);
         metricsHelper = new DocumentApiMetrics(metricReceiver, "vespa.http.server");
-        feedHandlerV3 = new FeedHandlerV3(threadpool, metric, accessLog, documentManagerConfig, sessionCache, metricsHelper);
+        feedHandlerV3 = new FeedHandlerV3(threadpool.executor(), metric, accessLog, documentManagerConfig, sessionCache, metricsHelper);
         feedReplyHandler = new FeedReplyReader(metric, metricsHelper);
     }
 
@@ -115,6 +118,12 @@ public class FeedHandler extends LoggingRequestHandler {
         return feedHandlerV3.handle(request);
     }
 
+    @Override
+    protected void writeErrorResponseOnOverload(Request request, ResponseHandler responseHandler) {
+        int responseCode = request.headers().getFirst(Headers.SILENTUPGRADE) != null ? 299 : 429;
+        ResponseDispatch.newInstance(responseCode).dispatch(responseHandler);
+    }
+
     private static Optional<String> findClientVersion(HttpRequest request) {
         String versionHeader = request.getHeader(Headers.CLIENT_VERSION);
         if (versionHeader != null) {
diff --git a/vespaclient-container-plugin/src/main/java/com/yahoo/vespa/http/server/FeedHandlerV3.java b/vespaclient-container-plugin/src/main/java/com/yahoo/vespa/http/server/FeedHandlerV3.java
index 0f5a808c507..be432834d1b 100644
--- a/vespaclient-container-plugin/src/main/java/com/yahoo/vespa/http/server/FeedHandlerV3.java
+++ b/vespaclient-container-plugin/src/main/java/com/yahoo/vespa/http/server/FeedHandlerV3.java
@@ -2,7 +2,6 @@
 package com.yahoo.vespa.http.server;
 
 import com.yahoo.concurrent.ThreadFactoryFactory;
-import com.yahoo.container.handler.threadpool.ContainerThreadPool;
 import com.yahoo.container.jdisc.HttpRequest;
 import com.yahoo.container.jdisc.HttpResponse;
 import com.yahoo.container.jdisc.LoggingRequestHandler;
@@ -43,34 +42,16 @@ public class FeedHandlerV3 extends LoggingRequestHandler {
     private final SessionCache sessionCache;
     protected final ReplyHandler feedReplyHandler;
     private final Metric metric;
-    private final ClientFeederV3.HttpThrottlePolicy httpThrottlePolicy;
     private final Object monitor = new Object();
     private static final Logger log = Logger.getLogger(FeedHandlerV3.class.getName());
 
-    public FeedHandlerV3(ContainerThreadPool threadpool,
+    public FeedHandlerV3(Executor executor,
                          Metric metric,
                          AccessLog accessLog,
                          DocumentmanagerConfig documentManagerConfig,
                          SessionCache sessionCache,
                          DocumentApiMetrics metricsHelper) {
-        this(threadpool.executor(),
-                () -> threadpool.queuedTasks() > 0,
-                metric,
-                accessLog,
-                documentManagerConfig,
-                sessionCache,
-                metricsHelper);
-    }
-
-    FeedHandlerV3(Executor executor,
-                  ClientFeederV3.HttpThrottlePolicy httpThrottlePolicy,
-                  Metric metric,
-                  AccessLog accessLog,
-                  DocumentmanagerConfig documentManagerConfig,
-                  SessionCache sessionCache,
-                  DocumentApiMetrics metricsHelper) {
         super(executor, accessLog, metric);
-        this.httpThrottlePolicy = httpThrottlePolicy;
         docTypeManager = new DocumentTypeManager(documentManagerConfig);
         this.sessionCache = sessionCache;
         feedReplyHandler = new FeedReplyReader(metric, metricsHelper);
@@ -98,8 +79,7 @@ public class FeedHandlerV3 extends LoggingRequestHandler {
                                                               docTypeManager,
                                                               clientId,
                                                               metric,
-                                                              feedReplyHandler,
-                                                              httpThrottlePolicy));
+                                                              feedReplyHandler));
             }
             clientFeederV3 = clientFeederByClientId.get(clientId);
         }
diff --git a/vespaclient-container-plugin/src/test/java/com/yahoo/vespa/http/server/FeedHandlerTest.java b/vespaclient-container-plugin/src/test/java/com/yahoo/vespa/http/server/FeedHandlerTest.java
new file mode 100644
index 00000000000..1fdd0764417
--- /dev/null
+++ b/vespaclient-container-plugin/src/test/java/com/yahoo/vespa/http/server/FeedHandlerTest.java
@@ -0,0 +1,46 @@
+package com.yahoo.vespa.http.server;// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+import com.yahoo.container.handler.threadpool.ContainerThreadPool;
+import com.yahoo.container.jdisc.RequestHandlerTestDriver;
+import com.yahoo.container.logging.AccessLog;
+import com.yahoo.document.config.DocumentmanagerConfig;
+import com.yahoo.jdisc.handler.OverloadException;
+import com.yahoo.metrics.simple.MetricReceiver;
+import org.junit.Test;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
+
+import static com.yahoo.vespa.http.server.FeedHandlerV3Test.createRequest;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+/**
+ * @author bjorncs
+ */
+public class FeedHandlerTest {
+
+    @Test
+    public void response_has_status_code_429_when_throttling() {
+        FeedHandler handler = new FeedHandler(
+                new RejectingContainerThreadpool(),
+                new CollectingMetric(),
+                AccessLog.voidAccessLog(),
+                new DocumentmanagerConfig(new DocumentmanagerConfig.Builder().enablecompression(true)),
+                null /* session cache */,
+                MetricReceiver.nullImplementation);
+        var responseHandler = new RequestHandlerTestDriver.MockResponseHandler();
+        try {
+            handler.handleRequest(createRequest(100).getJDiscRequest(), responseHandler);
+            fail();
+        } catch (OverloadException e) {}
+        assertEquals(429, responseHandler.getStatus());
+    }
+
+    private static class RejectingContainerThreadpool implements ContainerThreadPool {
+        private final Executor executor = ignored -> { throw new RejectedExecutionException(); };
+
+        @Override public Executor executor() { return executor;  }
+    }
+
+}
\ No newline at end of file
diff --git a/vespaclient-container-plugin/src/test/java/com/yahoo/vespa/http/server/FeedHandlerV3Test.java b/vespaclient-container-plugin/src/test/java/com/yahoo/vespa/http/server/FeedHandlerV3Test.java
index 4b76abf785c..bda2d8ea861 100644
--- a/vespaclient-container-plugin/src/test/java/com/yahoo/vespa/http/server/FeedHandlerV3Test.java
+++ b/vespaclient-container-plugin/src/test/java/com/yahoo/vespa/http/server/FeedHandlerV3Test.java
@@ -22,7 +22,6 @@ import com.yahoo.vespa.http.client.config.FeedParams;
 import com.yahoo.vespa.http.client.core.ErrorCode;
 import com.yahoo.vespa.http.client.core.Headers;
 import com.yahoo.vespa.http.client.core.OperationStatus;
-import com.yahoo.vespa.http.server.ClientFeederV3.HttpThrottlePolicy;
 import org.junit.Test;
 import org.mockito.stubbing.Answer;
 
@@ -34,7 +33,6 @@ import java.util.concurrent.Executors;
 
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.startsWith;
-import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
@@ -42,13 +40,11 @@ import static org.mockito.Mockito.when;
 
 public class FeedHandlerV3Test {
     final CollectingMetric metric = new CollectingMetric();
-
-    private static final HttpThrottlePolicy NON_THROTTLE = () -> false;
-    private static final HttpThrottlePolicy THROTTLE_ALWAYS = () -> true;
+    private final Executor simpleThreadpool = Executors.newCachedThreadPool();
 
     @Test
     public void feedOneDocument() throws Exception {
-        final FeedHandlerV3 feedHandlerV3 = setupFeederHandler(NON_THROTTLE);
+        final FeedHandlerV3 feedHandlerV3 = setupFeederHandler(simpleThreadpool);
         HttpResponse httpResponse = feedHandlerV3.handle(createRequest(1));
         ByteArrayOutputStream outStream = new ByteArrayOutputStream();
         httpResponse.render(outStream);
@@ -58,7 +54,7 @@ public class FeedHandlerV3Test {
 
     @Test
     public void feedOneBrokenDocument() throws Exception {
-        final FeedHandlerV3 feedHandlerV3 = setupFeederHandler(NON_THROTTLE);
+        final FeedHandlerV3 feedHandlerV3 = setupFeederHandler(simpleThreadpool);
         HttpResponse httpResponse = feedHandlerV3.handle(createBrokenRequest());
         ByteArrayOutputStream outStream = new ByteArrayOutputStream();
         httpResponse.render(outStream);
@@ -69,7 +65,7 @@ public class FeedHandlerV3Test {
 
     @Test
     public void feedManyDocument() throws Exception {
-        final FeedHandlerV3 feedHandlerV3 = setupFeederHandler(NON_THROTTLE);
+        final FeedHandlerV3 feedHandlerV3 = setupFeederHandler(simpleThreadpool);
         HttpResponse httpResponse = feedHandlerV3.handle(createRequest(100));
         ByteArrayOutputStream outStream = new ByteArrayOutputStream();
         httpResponse.render(outStream);
@@ -78,15 +74,6 @@ public class FeedHandlerV3Test {
         assertThat(Splitter.on("\n").splitToList(result).size(), is(101));
     }
 
-    @Test
-    public void response_has_status_code_429_when_throttling() throws Exception {
-        final FeedHandlerV3 feedHandlerV3 = setupFeederHandler(THROTTLE_ALWAYS);
-        HttpResponse httpResponse = feedHandlerV3.handle(createRequest(100));
-        ByteArrayOutputStream outStream = new ByteArrayOutputStream();
-        httpResponse.render(outStream);
-        assertEquals(httpResponse.getStatus(), 429);
-    }
-
     private static DocumentTypeManager createDoctypeManager() {
         DocumentTypeManager docTypeManager = new DocumentTypeManager();
         DocumentType documentType = new DocumentType("testdocument");
@@ -96,7 +83,7 @@ public class FeedHandlerV3Test {
         return docTypeManager;
     }
 
-    private static HttpRequest createRequest(int numberOfDocs) {
+    static HttpRequest createRequest(int numberOfDocs) {
         StringBuilder wireData = new StringBuilder();
         for (int x = 0; x < numberOfDocs; x++) {
             String docData = "[{\"put\": \"id:testdocument:testdocument::c\", \"fields\": { \"title\": \"fooKey\", \"body\": \"value\"}}]";
@@ -112,7 +99,7 @@ public class FeedHandlerV3Test {
         return createRequestWithPayload(wireData);
     }
 
-    private static HttpRequest createRequestWithPayload(String payload) {
+    static HttpRequest createRequestWithPayload(String payload) {
         InputStream inputStream = new ByteArrayInputStream(payload.getBytes());
         HttpRequest request = HttpRequest.createTestRequest("http://dummyhostname:19020/reserved-for-internal-use/feedapi",
                 com.yahoo.jdisc.http.HttpRequest.Method.POST, inputStream);
@@ -126,13 +113,10 @@ public class FeedHandlerV3Test {
         return request;
     }
 
-    private FeedHandlerV3 setupFeederHandler(HttpThrottlePolicy policy) throws Exception {
-        Executor threadPool = Executors.newCachedThreadPool();
-
+    private FeedHandlerV3 setupFeederHandler(Executor threadPool) {
         DocumentmanagerConfig docMan = new DocumentmanagerConfig(new DocumentmanagerConfig.Builder().enablecompression(true));
         FeedHandlerV3 feedHandlerV3 = new FeedHandlerV3(
                 threadPool,
-                policy,
                 metric,
                 AccessLog.voidAccessLog(),
                 docMan,