feat: reattempt in introducing configurable http/2 control frame limiter with observability (#3422)

* feat: reattempt in introducing configurable http/2 control frame limiter with observability

* run apiDump

* remove make Endpoint nullable

* Fix issue where multiple instance of the counter metric results in silently failed IllegalArgumentException

---------

Co-authored-by: Andrew (Paradi) Alexander <[email protected]>

Author: kevink-sq

misk/api/misk.api

@@ -2162,6 +2162,16 @@ public final class misk/web/jetty/MeasuredThreadPoolExecutor : misk/web/jetty/Me
 	public fun queueSize ()I
 }
 
+public final class misk/web/jetty/MeasuredWindowRateControl : org/eclipse/jetty/http2/parser/RateControl {
+	public synthetic fun <init> (ILmisk/metrics/v2/PeakGauge;Lio/prometheus/client/Counter;Lkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public fun onEvent (Ljava/lang/Object;)Z
+}
+
+public final class misk/web/jetty/MeasuredWindowRateControl$Factory : org/eclipse/jetty/http2/parser/RateControl$Factory {
+	public fun <init> (Lmisk/metrics/v2/Metrics;Lmisk/web/WebConfig;)V
+	public fun newRateControl (Lorg/eclipse/jetty/io/EndPoint;)Lorg/eclipse/jetty/http2/parser/RateControl;
+}
+
 public final class misk/web/jetty/ThreadPoolQueueMetrics {
 	public final fun recordQueueLatency (Ljava/time/Duration;)V
 }

misk/build.gradle.kts

@@ -15,6 +15,8 @@ dependencies {
   api(libs.jacksonDatabind)
   api(libs.jakartaInject)
   api(libs.jakartaInject)
+  api(libs.jettyHttp2Common)
+  api(libs.jettyIo)
   api(libs.jettyServer)
   api(libs.jettyServletApi)
   api(libs.jettyUtil)
@@ -44,8 +46,6 @@ dependencies {
   implementation(libs.jettyAlpnServer)
   implementation(libs.jettyHttp)
   implementation(libs.jettyHttp2)
-  implementation(libs.jettyHttp2Common)
-  implementation(libs.jettyIo)
   implementation(libs.jettyServlet)
   implementation(libs.jettyServlets)
   implementation(libs.jettyUds)
@@ -87,6 +87,7 @@ dependencies {
   testImplementation(libs.kotlinxCoroutinesCore)
   testImplementation(libs.kotlinxCoroutinesTest)
   testImplementation(libs.logbackClassic)
+  testImplementation(libs.mockitoCore)
   testImplementation(libs.okHttpMockWebServer)
   testImplementation(libs.okHttpSse)
   testImplementation(libs.openTracingMock)

misk/src/main/kotlin/misk/web/jetty/JettyService.kt

@@ -68,6 +68,7 @@ class JettyService @Inject internal constructor(
   private val connectionMetricsCollector: JettyConnectionMetricsCollector,
   private val statisticsHandler: StatisticsHandler,
   private val gzipHandler: GzipHandler,
+  private val http2RateControlFactory: MeasuredWindowRateControl.Factory
 ) : AbstractIdleService() {
   private val server = Server(threadPool)
   val healthServerUrl: HttpUrl? get() = server.healthUrl
@@ -128,6 +129,7 @@ class JettyService @Inject internal constructor(
     if (webConfig.http2) {
       val http2 = HTTP2CServerConnectionFactory(httpConfig)
       http2.customize(webConfig)
+      http2.rateControlFactory = http2RateControlFactory
       httpConnectionFactories += http2
     }
 
@@ -212,6 +214,7 @@ class JettyService @Inject internal constructor(
       if (webConfig.http2) {
         val http2 = HTTP2ServerConnectionFactory(httpsConfig)
         http2.customize(webConfig)
+        http2.rateControlFactory = http2RateControlFactory
         httpsConnectionFactories += http2
       }
 
@@ -255,7 +258,9 @@ class JettyService @Inject internal constructor(
       val udsConnFactories = mutableListOf<ConnectionFactory>()
       udsConnFactories.add(HttpConnectionFactory(httpConfig))
       if (socketConfig.h2c == true) {
-        udsConnFactories.add(HTTP2CServerConnectionFactory(httpConfig))
+        val http2 = HTTP2CServerConnectionFactory(httpConfig)
+        http2.rateControlFactory = http2RateControlFactory
+        udsConnFactories.add(http2)
       }
 
       if (isJEP380Supported(socketConfig.path)) {

misk/src/main/kotlin/misk/web/jetty/MeasuredWindowRateControl.kt

@@ -12,7 +12,12 @@
 //
 package misk.web.jetty
 
+import io.prometheus.client.Counter
+import jakarta.inject.Inject
+import jakarta.inject.Singleton
 import misk.metrics.v2.Metrics
+import misk.metrics.v2.PeakGauge
+import misk.web.WebConfig
 import org.eclipse.jetty.http2.parser.RateControl
 import org.eclipse.jetty.io.EndPoint
 import org.eclipse.jetty.util.NanoTime
@@ -24,19 +29,12 @@ import java.util.concurrent.atomic.AtomicInteger
  * Misk's RateControl implementation with observability for monitoring HTTP/2 frame rate limiting.
  * Almost the same implementation as [org.eclipse.jetty.http2.parser.WindowRateControl].
  */
-internal class MeasuredWindowRateControl internal constructor(
-  private val metrics: Metrics,
+class MeasuredWindowRateControl private constructor(
   private val maxEvents: Int,
+  private val rateEventsPeakGauge: PeakGauge,
+  private val rateLimitedEventCounter: Counter,
 ) : RateControl {
 
-  private val rateEventsPeakGauge = metrics.peakGauge(
-    "jetty_http2_rate_control_events_peak",
-    "Peak gauge of observed events per second"
-  )
-  private val rateLimitedEventCounter = metrics.counter(
-    "jetty_http2_rate_control_events_limited",
-    "Count of rate limited events"
-  )
 
   private val events = ConcurrentLinkedQueue<Long>()
   private val size = AtomicInteger()
@@ -55,20 +53,37 @@ internal class MeasuredWindowRateControl internal constructor(
 
     val count = size.incrementAndGet()
     rateEventsPeakGauge.record(count.toDouble())
-
     if (maxEvents == -1) return true
 
     val allowed = count <= maxEvents
     if (!allowed) rateLimitedEventCounter.inc()
     return allowed
   }
 
-  class Factory constructor(
-    private val metrics: Metrics,
-    private val maxEventRate: Int
+  /**
+   * Ensure the factory remains a singleton to prevent
+   * multiple instantiations of the same metric objects
+   */
+  @Singleton
+  class Factory @Inject constructor(
+    metrics: Metrics,
+    private val webConfig: WebConfig,
   ) : RateControl.Factory {
-    override fun newRateControl(endPoint: EndPoint): RateControl {
-      return MeasuredWindowRateControl(metrics, maxEventRate)
+
+    private val rateEventsPeakGauge = metrics.peakGauge(
+      "jetty_http2_rate_control_events_peak",
+      "Peak gauge of observed events per second"
+    )
+    private val rateLimitedEventCounter = metrics.counter(
+      "jetty_http2_rate_control_events_limited",
+      "Count of rate limited events"
+    )
+
+    override fun newRateControl(endPoint: EndPoint?): RateControl {
+      return MeasuredWindowRateControl(
+        webConfig.jetty_http2_max_events_per_second,
+        rateEventsPeakGauge,
+        rateLimitedEventCounter)
     }
   }
-}
+}

misk/src/test/kotlin/misk/web/jetty/MeasuredWindowRateControlTest.kt

@@ -3,19 +3,31 @@ import misk.metrics.v2.FakeMetrics
 import misk.metrics.v2.FakeMetricsModule
 import misk.testing.MiskTest
 import misk.testing.MiskTestModule
+import misk.web.WebConfig
 import misk.web.jetty.MeasuredWindowRateControl
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Test
+import org.mockito.Mockito.mock
+import org.mockito.Mockito.`when`
 
 @MiskTest(startService = false)
 class MeasuredWindowRateControlTest {
 
   @MiskTestModule val module = FakeMetricsModule()
   @Inject lateinit var metrics: FakeMetrics
 
+  fun maxEventRate(n : Int) : WebConfig {
+    val webConfig = mock(WebConfig::class.java)
+    `when`(webConfig.jetty_http2_max_events_per_second).thenReturn(n)
+    return webConfig
+  }
+
   @Test
   fun `allows events when under maxEvents limit`() {
-    val rateControl = MeasuredWindowRateControl(metrics, maxEvents = 5)
+    val rateControl = MeasuredWindowRateControl.Factory(
+      metrics,
+      maxEventRate(5)
+    ).newRateControl(null)
 
     repeat(5) {
       assertThat(rateControl.onEvent("test")).isTrue()
@@ -24,7 +36,10 @@ class MeasuredWindowRateControlTest {
 
   @Test
   fun `blocks events when over maxEvents limit`() {
-    val rateControl = MeasuredWindowRateControl(metrics, maxEvents = 2)
+    val rateControl = MeasuredWindowRateControl.Factory(
+      metrics,
+      maxEventRate(2)
+    ).newRateControl(null)
 
     assertThat(rateControl.onEvent("test1")).isTrue()
     assertThat(rateControl.onEvent("test2")).isTrue()
@@ -34,7 +49,10 @@ class MeasuredWindowRateControlTest {
 
   @Test
   fun `allows unlimited events when maxEvents is -1`() {
-    val rateControl = MeasuredWindowRateControl(metrics, maxEvents = -1)
+    val rateControl = MeasuredWindowRateControl.Factory(
+      metrics,
+      maxEventRate(-1)
+    ).newRateControl(null)
 
     repeat(100) {
       assertThat(rateControl.onEvent("test")).isTrue()
@@ -43,7 +61,10 @@ class MeasuredWindowRateControlTest {
 
   @Test
   fun `window sliding allows events after time passes`() {
-    val rateControl = MeasuredWindowRateControl(metrics, maxEvents = 2)
+    val rateControl = MeasuredWindowRateControl.Factory(
+      metrics,
+      maxEventRate(2)
+    ).newRateControl(null)
 
     assertThat(rateControl.onEvent("test1")).isTrue()
     assertThat(rateControl.onEvent("test2")).isTrue()