Make maxConnections and idleTimeout configurable

Previusly, idleTimeout has been hardcoded, maxConnections hasn't been configured. Now, these parameters are set within `networking` section and used throughout http4s backends.
snowplow · Sep 20, 2023 · 2e644b3 · 2e644b3
1 parent 66d1e79
commit 2e644b3
Show file tree

Hide file tree

Showing 8 changed files with 46 additions and 13 deletions.
diff --git a/http4s/src/main/resources/reference.conf b/http4s/src/main/resources/reference.conf
@@ -77,6 +77,11 @@
     port = 443
   }
 
+  networking {
+    maxConnections = 1024
+    idleTimeout = 610 seconds
+  }
+
   enableDefaultRedirect = false
   preTerminationPeriod = 10 seconds
 

diff --git a/http4s/src/main/scala/com.snowplowanalytics.snowplow.collector.core/Config.scala b/http4s/src/main/scala/com.snowplowanalytics.snowplow.collector.core/Config.scala
@@ -24,6 +24,7 @@ case class Config[+SinkConfig](
   streams: Config.Streams[SinkConfig],
   monitoring: Config.Monitoring,
   ssl: Config.SSL,
+  networking: Config.Networking,
   enableDefaultRedirect: Boolean,
   redirectDomains: Set[String],
   preTerminationPeriod: FiniteDuration
@@ -122,6 +123,11 @@ object Config {
     port: Int
   )
 
+  case class Networking(
+    maxConnections: Int,
+    idleTimeout: FiniteDuration
+  )
+
   implicit def decoder[SinkConfig: Decoder]: Decoder[Config[SinkConfig]] = {
     implicit val p3p         = deriveDecoder[P3P]
     implicit val crossDomain = deriveDecoder[CrossDomain]
@@ -147,6 +153,7 @@ object Config {
     implicit val metrics          = deriveDecoder[Metrics]
     implicit val monitoring       = deriveDecoder[Monitoring]
     implicit val ssl              = deriveDecoder[SSL]
+    implicit val networking       = deriveDecoder[Networking]
     deriveDecoder[Config[SinkConfig]]
   }
 

diff --git a/http4s/src/main/scala/com.snowplowanalytics.snowplow.collector.core/HttpServer.scala b/http4s/src/main/scala/com.snowplowanalytics.snowplow.collector.core/HttpServer.scala
@@ -8,8 +8,6 @@ import io.netty.handler.ssl._
 import org.typelevel.log4cats.Logger
 import org.typelevel.log4cats.slf4j.Slf4jLogger
 
-import scala.concurrent.duration.DurationLong
-
 import com.comcast.ip4s.{IpAddress, Port}
 
 import cats.implicits._
@@ -33,20 +31,22 @@ object HttpServer {
     app: HttpApp[F],
     interface: String,
     port: Int,
-    secure: Boolean
+    secure: Boolean,
+    networking: Config.Networking
   ): Resource[F, Server] =
     sys.env.get("HTTP4S_BACKEND").map(_.toUpperCase()) match {
-      case Some("BLAZE") | None => buildBlazeServer[F](app, port, secure)
-      case Some("EMBER") => buildEmberServer[F](app, interface, port, secure)
-      case Some("NETTY") => buildNettyServer[F](app, port, secure)
+      case Some("BLAZE") | None => buildBlazeServer[F](app, port, secure, networking)
+      case Some("EMBER") => buildEmberServer[F](app, interface, port, secure, networking)
+      case Some("NETTY") => buildNettyServer[F](app, port, secure, networking)
       case Some(other)   => throw new IllegalArgumentException(s"Unrecognized http4s backend $other")
     }
 
   private def buildEmberServer[F[_]: Async](
     app: HttpApp[F],
     interface: String,
     port: Int,
-    secure: Boolean
+    secure: Boolean,
+    networking: Config.Networking
   ) = {
     implicit val network = Network.forAsync[F]
     Resource.eval(Logger[F].info("Building ember server")) >>
@@ -55,34 +55,38 @@ object HttpServer {
         .withHost(IpAddress.fromString(interface).get)
         .withPort(Port.fromInt(port).get)
         .withHttpApp(app)
-        .withIdleTimeout(610.seconds)
+        .withIdleTimeout(networking.idleTimeout)
+        .withMaxConnections(networking.maxConnections)
         .cond(secure, _.withTLS(TLSContext.Builder.forAsync.fromSSLContext(SSLContext.getDefault)))
         .build
   }
 
   private def buildBlazeServer[F[_]: Async](
     app: HttpApp[F],
     port: Int,
-    secure: Boolean
+    secure: Boolean,
+    networking: Config.Networking
   ): Resource[F, Server] =
     Resource.eval(Logger[F].info("Building blaze server")) >>
       BlazeServerBuilder[F]
         .bindSocketAddress(new InetSocketAddress(port))
         .withHttpApp(app)
-        .withIdleTimeout(610.seconds)
+        .withIdleTimeout(networking.idleTimeout)
+        .withMaxConnections(networking.maxConnections)
         .cond(secure, _.withSslContext(SSLContext.getDefault))
         .resource
 
   private def buildNettyServer[F[_]: Async](
     app: HttpApp[F],
     port: Int,
-    secure: Boolean
+    secure: Boolean,
+    networking: Config.Networking
   ) =
     Resource.eval(Logger[F].info("Building netty server")) >>
       NettyServerBuilder[F]
         .bindLocal(port)
         .withHttpApp(app)
-        .withIdleTimeout(610.seconds)
+        .withIdleTimeout(networking.idleTimeout)
         .cond(
           secure,
           _.withSslContext(

diff --git a/http4s/src/main/scala/com.snowplowanalytics.snowplow.collector.core/Run.scala b/http4s/src/main/scala/com.snowplowanalytics.snowplow.collector.core/Run.scala
@@ -63,7 +63,8 @@ object Run {
         new Routes[F](config.enableDefaultRedirect, collectorService).value,
         config.interface,
         if (config.ssl.enable) config.ssl.port else config.port,
-        config.ssl.enable
+        config.ssl.enable,
+        config.networking
       )
       _ <- withGracefulShutdown(config.preTerminationPeriod)(httpServer)
     } yield ()

diff --git a/http4s/src/test/scala/com.snowplowanalytics.snowplow.collector.core/TestUtils.scala b/http4s/src/test/scala/com.snowplowanalytics.snowplow.collector.core/TestUtils.scala
@@ -100,6 +100,10 @@ object TestUtils {
       false,
       443
     ),
+    networking = Networking(
+      1024,
+      610.seconds
+    ),
     enableDefaultRedirect = false,
     redirectDomains       = Set.empty[String],
     preTerminationPeriod  = 10.seconds

diff --git a/...scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/KinesisConfigSpec.scala b/...scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/KinesisConfigSpec.scala
@@ -95,6 +95,10 @@ object KinesisConfigSpec {
     enableDefaultRedirect = false,
     redirectDomains       = Set.empty,
     preTerminationPeriod  = 10.seconds,
+    networking = Config.Networking(
+      maxConnections = 1024,
+      idleTimeout    = 610.seconds
+    ),
     streams = Config.Streams(
       good                       = "good",
       bad                        = "bad",

diff --git a/pubsub/src/test/scala/com.snowplowanalytics.snowplow.collectors.scalastream/ConfigSpec.scala b/pubsub/src/test/scala/com.snowplowanalytics.snowplow.collectors.scalastream/ConfigSpec.scala
@@ -95,6 +95,10 @@ object ConfigSpec {
     enableDefaultRedirect = false,
     redirectDomains       = Set.empty,
     preTerminationPeriod  = 10.seconds,
+    networking = Config.Networking(
+      maxConnections = 1024,
+      idleTimeout    = 610.seconds
+    ),
     streams = Config.Streams(
       good                       = "good",
       bad                        = "bad",

diff --git a/sqs/src/test/scala/com.snowplowanalytics.snowplow.collectors.scalastream/SqsConfigSpec.scala b/sqs/src/test/scala/com.snowplowanalytics.snowplow.collectors.scalastream/SqsConfigSpec.scala
@@ -106,6 +106,10 @@ object SqsConfigSpec {
     enableDefaultRedirect = false,
     redirectDomains       = Set.empty,
     preTerminationPeriod  = 10.seconds,
+    networking = Config.Networking(
+      maxConnections = 1024,
+      idleTimeout    = 610.seconds
+    ),
     streams = Config.Streams(
       good                       = "good",
       bad                        = "bad",