fix: bound HTTPServer request resources (#2333)

## Summary

- use a fixed-size default executor with a bounded queue and
non-blocking rejection
- close rejected authenticated request bodies instead of draining an
unbounded body
- close rejected exchanges after sending HTTP 403

This is the focused replacement for the #2284 portion of #2297.

Fixes #2284

## Ongoing discussion

None currently. Earlier review feedback about preserving default
concurrency and closing rejected exchanges is incorporated here.

## Validation

- `mise run lint:fix`
- `mise run build`
- `./mvnw test -pl prometheus-metrics-exporter-httpserver
-Dcoverage.skip=true -Dcheckstyle.skip=true`

---------

Signed-off-by: Gregor Zeitlinger <gregor.zeitlinger@grafana.com>

Author: zeitlinger

prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/BlockingRejectedExecutionHandler.java

@@ -12,14 +12,13 @@
 import io.prometheus.metrics.model.registry.PrometheusRegistry;
 import java.io.Closeable;
 import java.io.IOException;
-import java.io.InputStream;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.security.PrivilegedActionException;
 import java.security.PrivilegedExceptionAction;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
@@ -39,6 +38,10 @@
 @StableApi
 public class HTTPServer implements Closeable {
 
+  private static final int DEFAULT_MIN_THREADS = 10;
+  private static final int DEFAULT_MAX_THREADS = 10;
+  private static final int DEFAULT_QUEUE_SIZE = 100;
+
   static {
     if (!System.getProperties().containsKey("sun.net.httpserver.maxReqTime")) {
       System.setProperty("sun.net.httpserver.maxReqTime", "60");
@@ -154,22 +157,14 @@ public void handle(HttpExchange exchange) throws IOException {
             }
           }
         } else {
-          drainInputAndClose(exchange);
+          exchange.getRequestBody().close();
           exchange.sendResponseHeaders(403, -1);
+          exchange.close();
         }
       }
     };
   }
 
-  private void drainInputAndClose(HttpExchange httpExchange) throws IOException {
-    InputStream inputStream = httpExchange.getRequestBody();
-    byte[] b = new byte[4096];
-    while (inputStream.read(b) != -1) {
-      // nop
-    }
-    inputStream.close();
-  }
-
   /** Stop the HTTP server. Same as {@link #close()}. */
   public void stop() {
     close();
@@ -354,13 +349,12 @@ private ExecutorService makeExecutorService() {
         return executorService;
       } else {
         return new ThreadPoolExecutor(
-            1,
-            10,
+            DEFAULT_MIN_THREADS,
+            DEFAULT_MAX_THREADS,
             120,
             TimeUnit.SECONDS,
-            new SynchronousQueue<>(true),
-            NamedDaemonThreadFactory.defaultThreadFactory(true),
-            new BlockingRejectedExecutionHandler());
+            new ArrayBlockingQueue<>(DEFAULT_QUEUE_SIZE),
+            NamedDaemonThreadFactory.defaultThreadFactory(true));
       }
     }
 

prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/HTTPServer.java

@@ -25,7 +25,7 @@
 import java.security.Principal;
 import java.util.List;
 import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.ThreadPoolExecutor;
 import javax.net.ssl.SSLContext;
 import javax.security.auth.Subject;
 import org.junit.jupiter.api.BeforeEach;
@@ -46,7 +46,7 @@ void setUp() {
   }
 
   @Test
-  public void testSubjectDoAs() throws Exception {
+  void testSubjectDoAs() throws Exception {
     final String user = "joe";
     final Subject subject = new Subject();
     subject.getPrincipals().add(() -> user);
@@ -160,55 +160,35 @@ void metricsCustomRootPath() throws Exception {
   }
 
   @Test
-  void registryThrows() throws Exception {
-    HTTPServer server = HTTPServer.builder().port(0).registry(throwingRegistry()).buildAndStart();
-    run(
-        server,
-        "/metrics",
-        500,
-        "Configure an HTTP error reporter for details.",
-        "IllegalStateException",
-        "test");
-  }
-
-  @Test
-  void registryExceptionIsPassedToConfiguredReporter() throws Exception {
-    AtomicReference<Throwable> reportedError = new AtomicReference<>();
-    HTTPServer server =
-        HTTPServer.builder()
-            .port(0)
-            .registry(throwingRegistry())
-            .errorHandlingPolicy(
-                HttpErrorHandlingPolicy.builder().errorReporter(reportedError::set).build())
-            .buildAndStart();
-
-    run(
-        server,
-        "/metrics",
-        500,
-        "Configure an HTTP error reporter for details.",
-        "IllegalStateException",
-        "test");
-
-    assertThat(reportedError.get()).isInstanceOf(IllegalStateException.class).hasMessage("test");
+  void defaultExecutorHasBoundedQueueAndNonBlockingRejection() throws Exception {
+    HTTPServer server = HTTPServer.builder().port(0).buildAndStart();
+    try {
+      assertThat(server.executorService).isInstanceOf(ThreadPoolExecutor.class);
+      ThreadPoolExecutor executor = (ThreadPoolExecutor) server.executorService;
+      assertThat(executor.getCorePoolSize()).isEqualTo(10);
+      assertThat(executor.getMaximumPoolSize()).isEqualTo(10);
+      assertThat(executor.getQueue().remainingCapacity()).isEqualTo(100);
+      assertThat(executor.getRejectedExecutionHandler())
+          .isInstanceOf(ThreadPoolExecutor.AbortPolicy.class);
+    } finally {
+      server.stop();
+    }
   }
 
   @Test
-  void registryExceptionCanUseUnsafeDebugResponse() throws Exception {
+  void registryThrows() throws Exception {
     HTTPServer server =
         HTTPServer.builder()
             .port(0)
-            .registry(throwingRegistry())
-            .errorHandlingPolicy(
-                HttpErrorHandlingPolicy.builder().unsafeDebugResponse(true).build())
+            .registry(
+                new PrometheusRegistry() {
+                  @Override
+                  public MetricSnapshots scrape(PrometheusScrapeRequest scrapeRequest) {
+                    throw new IllegalStateException("test");
+                  }
+                })
             .buildAndStart();
-
-    run(
-        server,
-        "/metrics",
-        500,
-        "IllegalStateException: test",
-        "Configure an HTTP error reporter for details.");
+    run(server, "/metrics", 500, "An internal error occurred while scraping metrics");
   }
 
   @Test
@@ -274,25 +254,6 @@ void healthDisabled() throws Exception {
   private static void run(
       HTTPServer server, String path, int expectedStatusCode, String expectedBody)
       throws Exception {
-    run(server, path, expectedStatusCode, expectedBody, new String[0]);
-  }
-
-  private static PrometheusRegistry throwingRegistry() {
-    return new PrometheusRegistry() {
-      @Override
-      public MetricSnapshots scrape(PrometheusScrapeRequest scrapeRequest) {
-        throw new IllegalStateException("test");
-      }
-    };
-  }
-
-  private static void run(
-      HTTPServer server,
-      String path,
-      int expectedStatusCode,
-      String expectedBody,
-      String... unexpectedBody)
-      throws Exception {
     // we cannot use try-with-resources or even client.close(), or the test will fail with Java 17
     @SuppressWarnings("resource")
     final HttpClient client = HttpClient.newBuilder().build();
@@ -304,9 +265,6 @@ private static void run(
           client.send(request, HttpResponse.BodyHandlers.ofString());
       assertThat(response.statusCode()).isEqualTo(expectedStatusCode);
       assertThat(response.body()).contains(expectedBody);
-      if (unexpectedBody.length > 0) {
-        assertThat(response.body()).doesNotContain(unexpectedBody);
-      }
     } finally {
       server.stop();
     }