Add dirty warmup

This adds a simplistic, dirty warmup requests to health endpoint
snowplow · Jun 28, 2024 · 6150942 · 6150942
1 parent 7f1b06a
commit 6150942
Show file tree

Hide file tree

Showing 9 changed files with 67 additions and 27 deletions.
diff --git a/core/src/main/resources/reference.conf b/core/src/main/resources/reference.conf
@@ -111,6 +111,11 @@
 
   experimental {
     backend = blaze
+    warmup {
+      enable = false
+      numRequests = 2000
+      maxCycles = 3
+    }
   }
 
   enableDefaultRedirect = false

diff --git a/core/src/main/scala/com.snowplowanalytics.snowplow.collector.core/Config.scala b/core/src/main/scala/com.snowplowanalytics.snowplow.collector.core/Config.scala
@@ -176,7 +176,7 @@ object Config {
     case class Debug(http: Http)
   }
 
-  case class Experimental(backend: Experimental.Backend)
+  case class Experimental(backend: Experimental.Backend, warmup: Experimental.Warmup)
   object Experimental {
     sealed trait Backend
     object Backend {
@@ -185,6 +185,7 @@ object Config {
       case object Netty extends Backend
       case object Armeria extends Backend
     }
+    case class Warmup(enable: Boolean, numRequests: Int, maxCycles: Long)
   }
 
   implicit def decoder[SinkConfig: Decoder]: Decoder[Config[SinkConfig]] = {
@@ -206,23 +207,24 @@ object Config {
         case Left(err) => Left(err)
       }
     }
-    implicit val cookie           = deriveDecoder[Cookie]
-    implicit val doNotTrackCookie = deriveDecoder[DoNotTrackCookie]
-    implicit val cookieBounce     = deriveDecoder[CookieBounce]
-    implicit val redirectMacro    = deriveDecoder[RedirectMacro]
-    implicit val rootResponse     = deriveDecoder[RootResponse]
-    implicit val cors             = deriveDecoder[CORS]
-    implicit val statsd           = deriveDecoder[Statsd]
-    implicit val metrics          = deriveDecoder[Metrics]
-    implicit val monitoring       = deriveDecoder[Monitoring]
-    implicit val ssl              = deriveDecoder[SSL]
-    implicit val hsts             = deriveDecoder[HSTS]
-    implicit val telemetry        = deriveDecoder[Telemetry]
-    implicit val networking       = deriveDecoder[Networking]
-    implicit val http             = deriveDecoder[Debug.Http]
-    implicit val debug            = deriveDecoder[Debug.Debug]
-    implicit val sinkConfig       = newDecoder[SinkConfig].or(legacyDecoder[SinkConfig])
-    implicit val streams          = deriveDecoder[Streams[SinkConfig]]
+    implicit val cookie                               = deriveDecoder[Cookie]
+    implicit val doNotTrackCookie                     = deriveDecoder[DoNotTrackCookie]
+    implicit val cookieBounce                         = deriveDecoder[CookieBounce]
+    implicit val redirectMacro                        = deriveDecoder[RedirectMacro]
+    implicit val rootResponse                         = deriveDecoder[RootResponse]
+    implicit val cors                                 = deriveDecoder[CORS]
+    implicit val statsd                               = deriveDecoder[Statsd]
+    implicit val metrics                              = deriveDecoder[Metrics]
+    implicit val monitoring                           = deriveDecoder[Monitoring]
+    implicit val ssl                                  = deriveDecoder[SSL]
+    implicit val hsts                                 = deriveDecoder[HSTS]
+    implicit val telemetry                            = deriveDecoder[Telemetry]
+    implicit val networking                           = deriveDecoder[Networking]
+    implicit val http                                 = deriveDecoder[Debug.Http]
+    implicit val debug                                = deriveDecoder[Debug.Debug]
+    implicit val sinkConfig                           = newDecoder[SinkConfig].or(legacyDecoder[SinkConfig])
+    implicit val streams                              = deriveDecoder[Streams[SinkConfig]]
+    implicit val warmup: Decoder[Experimental.Warmup] = deriveDecoder[Experimental.Warmup]
     implicit val backend: Decoder[Experimental.Backend] = Decoder[String].emap {
       case s if s.toLowerCase() == "blaze"   => Right(Experimental.Backend.Blaze)
       case s if s.toLowerCase() == "ember"   => Right(Experimental.Backend.Ember)

diff --git a/core/src/main/scala/com.snowplowanalytics.snowplow.collector.core/Run.scala b/core/src/main/scala/com.snowplowanalytics.snowplow.collector.core/Run.scala
@@ -15,14 +15,15 @@ import java.nio.file.Path
 import org.typelevel.log4cats.Logger
 import org.typelevel.log4cats.slf4j.Slf4jLogger
 
-import scala.concurrent.duration.FiniteDuration
+import scala.concurrent.duration._
 
 import cats.implicits._
 import cats.data.EitherT
 
 import cats.effect.{Async, ExitCode, Sync}
 import cats.effect.kernel.Resource
 
+import fs2.Stream
 import fs2.io.net.Network
 
 import org.http4s.blaze.client.BlazeClientBuilder
@@ -117,11 +118,37 @@ object Run {
 
     resources.use { httpClient =>
       val appId = java.util.UUID.randomUUID.toString
-      Telemetry
+      val warmup = if (config.experimental.warmup.enable) {
+        Stream
+          .emits(1 to config.experimental.warmup.numRequests)
+          .evalMap(_ =>
+            httpClient
+              .expect[String](
+                s"""${if (config.ssl.enable) "https" else "http"}://${config.interface}:${if (config.ssl.enable)
+                  config.ssl.port
+                else config.port}/health"""
+              )
+              .recover { case _ => "error" }
+          )
+          .fold((0, 0))((acc, cur) => if (cur == "error") (acc._1 + 1, acc._2) else (acc._1, acc._2 + 1))
+          .evalTap(status =>
+            Logger[F].info(s"Warmup pass complete. Successful requests: ${status._2}, failures: ${status._1}")
+          )
+          .repeatN(config.experimental.warmup.maxCycles)
+          .fold((0, 0))((acc, cur) => (acc._1 + acc._2, cur._1 + cur._2))
+          .onComplete(Stream.eval(Logger[F].info("Warmup completed.")))
+          .compile
+          .drain
+
+      } else {
+        Sync[F].unit
+      }
+      warmup *> Telemetry
         .run(config.telemetry, httpClient, appInfo, appId, telemetryInfo(config.streams))
         .compile
         .drain
         .flatMap(_ => Async[F].never[ExitCode])
+
     }
   }
 

diff --git a/core/src/test/scala/com.snowplowanalytics.snowplow.collector.core/TestUtils.scala b/core/src/test/scala/com.snowplowanalytics.snowplow.collector.core/TestUtils.scala
@@ -144,7 +144,8 @@ object TestUtils {
       instanceId      = None,
       autoGeneratedId = None
     ),
-    license      = License(accept       = true),
-    experimental = Experimental(backend = Config.Experimental.Backend.Blaze)
+    license = License(accept = true),
+    experimental =
+      Experimental(backend = Config.Experimental.Backend.Blaze, warmup = Config.Experimental.Warmup(false, 2000, 3))
   )
 }
diff --git a/...rc/test/scala/com.snowplowanalytics.snowplow.collectors.scalastream/KafkaConfigSpec.scala b/...rc/test/scala/com.snowplowanalytics.snowplow.collectors.scalastream/KafkaConfigSpec.scala
@@ -180,6 +180,7 @@ object KafkaConfigSpec {
     debug = Config
       .Debug
       .Debug(Config.Debug.Http(enable = false, logHeaders = true, logBody = false, redactHeaders = List.empty)),
-    experimental = Config.Experimental(backend = Config.Experimental.Backend.Blaze)
+    experimental = Config
+      .Experimental(backend = Config.Experimental.Backend.Blaze, warmup = Config.Experimental.Warmup(false, 2000, 3))
   )
 }
diff --git a/...scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/KinesisConfigSpec.scala b/...scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/KinesisConfigSpec.scala
@@ -196,7 +196,8 @@ object KinesisConfigSpec {
     debug = Config
       .Debug
       .Debug(Config.Debug.Http(enable = false, logHeaders = true, logBody = false, redactHeaders = List.empty)),
-    experimental = Config.Experimental(backend = Config.Experimental.Backend.Blaze)
+    experimental = Config
+      .Experimental(backend = Config.Experimental.Backend.Blaze, warmup = Config.Experimental.Warmup(false, 2000, 3))
   )
 
 }
diff --git a/nsq/src/test/scala/com.snowplowanalytics.snowplow.collectors.scalastream/NsqConfigSpec.scala b/nsq/src/test/scala/com.snowplowanalytics.snowplow.collectors.scalastream/NsqConfigSpec.scala
@@ -167,6 +167,7 @@ object NsqConfigSpec {
     debug = Config
       .Debug
       .Debug(Config.Debug.Http(enable = false, logHeaders = true, logBody = false, redactHeaders = List.empty)),
-    experimental = Config.Experimental(backend = Config.Experimental.Backend.Blaze)
+    experimental = Config
+      .Experimental(backend = Config.Experimental.Backend.Blaze, warmup = Config.Experimental.Warmup(false, 2000, 3))
   )
 }
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
@@ -187,7 +187,8 @@ object ConfigSpec {
     debug = Config
       .Debug
       .Debug(Config.Debug.Http(enable = false, logHeaders = true, logBody = false, redactHeaders = List.empty)),
-    experimental = Config.Experimental(backend = Config.Experimental.Backend.Blaze)
+    experimental = Config
+      .Experimental(backend = Config.Experimental.Backend.Blaze, warmup = Config.Experimental.Warmup(false, 2000, 3))
   )
 
 }
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
@@ -176,7 +176,8 @@ object SqsConfigSpec {
     debug = Config
       .Debug
       .Debug(Config.Debug.Http(enable = false, logHeaders = true, logBody = false, redactHeaders = List.empty)),
-    experimental = Config.Experimental(backend = Config.Experimental.Backend.Blaze)
+    experimental = Config
+      .Experimental(backend = Config.Experimental.Backend.Blaze, warmup = Config.Experimental.Warmup(false, 2000, 3))
   )
 
 }