Add debug logging and timeout configurations

Previously it was only possible to enable trace logs for requests and responses. These can contain PII and should not be enabled. After this change it is possible to explicitly configure which parts of request we want to log. For now, body is left on or off, but we should provide a better mechanism to filter out PIIs on demand.
snowplow · Mar 26, 2024 · 63aea42 · 63aea42
1 parent e3a6d0b
commit 63aea42
Show file tree

Hide file tree

Showing 10 changed files with 65 additions and 12 deletions.
diff --git a/core/src/main/resources/reference.conf b/core/src/main/resources/reference.conf
@@ -98,6 +98,14 @@
     bodyReadTimeout = 500 millis
   }
 
+  debug {
+    http {
+      enable = false
+      logHeaders = true
+      logBody = false
+      redactHeaders = []
+    }
+  }
   enableDefaultRedirect = false
   preTerminationPeriod = 10 seconds
 

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
@@ -41,7 +41,8 @@ case class Config[+SinkConfig](
   enableDefaultRedirect: Boolean,
   redirectDomains: Set[String],
   preTerminationPeriod: FiniteDuration,
-  license: Config.License
+  license: Config.License,
+  debug: Config.Debug.Debug
 )
 
 object Config {
@@ -167,6 +168,11 @@ object Config {
     accept: Boolean
   )
 
+  object Debug {
+    case class Http(enable: Boolean, logHeaders: Boolean, logBody: Boolean, redactHeaders: List[String])
+    case class Debug(http: Http)
+  }
+
   implicit def decoder[SinkConfig: Decoder]: Decoder[Config[SinkConfig]] = {
     implicit val license: Decoder[License] = {
       val truthy = Set("true", "yes", "on", "1")
@@ -199,6 +205,8 @@ object Config {
     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]]
 

diff --git a/core/src/main/scala/com.snowplowanalytics.snowplow.collector.core/HttpServer.scala b/core/src/main/scala/com.snowplowanalytics.snowplow.collector.core/HttpServer.scala
@@ -19,7 +19,8 @@ import org.http4s.{HttpApp, HttpRoutes}
 import org.http4s.blaze.server.BlazeServerBuilder
 import org.http4s.headers.`Strict-Transport-Security`
 import org.http4s.server.Server
-import org.http4s.server.middleware.{HSTS, Metrics}
+import org.http4s.server.middleware.{HSTS, Logger => LoggerMiddleware, Metrics, Timeout}
+import org.typelevel.ci.CIString
 import org.typelevel.log4cats.Logger
 import org.typelevel.log4cats.slf4j.Slf4jLogger
 
@@ -36,11 +37,12 @@ object HttpServer {
     secure: Boolean,
     hsts: Config.HSTS,
     networking: Config.Networking,
-    metricsConfig: Config.Metrics
+    metricsConfig: Config.Metrics,
+    debugHttp: Config.Debug.Http
   ): Resource[F, Server] =
     for {
       withMetricsMiddleware <- createMetricsMiddleware(routes, metricsConfig)
-      server                <- buildBlazeServer[F](withMetricsMiddleware, port, secure, hsts, networking)
+      server                <- buildBlazeServer[F](withMetricsMiddleware, port, secure, hsts, networking, debugHttp)
     } yield server
 
   private def createMetricsMiddleware[F[_]: Async](
@@ -67,17 +69,33 @@ object HttpServer {
       HSTS(routes, `Strict-Transport-Security`.unsafeFromDuration(hsts.maxAge))
     else routes
 
+  private def loggerMiddleware[F[_]: Async](routes: HttpApp[F], config: Config.Debug.Http): HttpApp[F] =
+    if (config.enable) {
+      LoggerMiddleware.httpApp[F](
+        logHeaders        = config.logHeaders,
+        logBody           = config.logBody,
+        redactHeadersWhen = config.redactHeaders.map(CIString(_)).contains(_),
+        logAction         = Some((msg: String) => Logger[F].debug(msg))
+      )(routes)
+    } else routes
+
+  private def timeoutMiddleware[F[_]: Async](routes: HttpApp[F], networking: Config.Networking): HttpApp[F] =
+    Timeout.httpApp[F](timeout = networking.responseHeaderTimeout)(routes)
+
   private def buildBlazeServer[F[_]: Async](
     routes: HttpRoutes[F],
     port: Int,
     secure: Boolean,
     hsts: Config.HSTS,
-    networking: Config.Networking
+    networking: Config.Networking,
+    debugHttp: Config.Debug.Http
   ): Resource[F, Server] =
     Resource.eval(Logger[F].info("Building blaze server")) >>
       BlazeServerBuilder[F]
         .bindSocketAddress(new InetSocketAddress(port))
-        .withHttpApp(hstsMiddleware(hsts, routes.orNotFound))
+        .withHttpApp(
+          loggerMiddleware(timeoutMiddleware(hstsMiddleware(hsts, routes.orNotFound), networking), debugHttp)
+        )
         .withIdleTimeout(networking.idleTimeout)
         .withMaxConnections(networking.maxConnections)
         .withResponseHeaderTimeout(networking.responseHeaderTimeout)

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
@@ -105,7 +105,8 @@ object Run {
         config.ssl.enable,
         config.hsts,
         config.networking,
-        config.monitoring.metrics
+        config.monitoring.metrics,
+        config.debug.http
       )
       _          <- withGracefulShutdown(config.preTerminationPeriod)(httpServer)
       httpClient <- BlazeClientBuilder[F].resource

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
@@ -123,6 +123,9 @@ object TestUtils {
       2.second,
       500.millis
     ),
+    debug = Debug.Debug(
+      http = Debug.Http(enable = false, logHeaders = true, logBody = false, redactHeaders = List.empty)
+    ),
     enableDefaultRedirect = false,
     redirectDomains       = Set.empty[String],
     preTerminationPeriod  = 10.seconds,

diff --git a/...rc/test/scala/com.snowplowanalytics.snowplow.collectors.scalastream/KafkaConfigSpec.scala b/...rc/test/scala/com.snowplowanalytics.snowplow.collectors.scalastream/KafkaConfigSpec.scala
@@ -174,6 +174,9 @@ object KafkaConfigSpec {
       responseHeaderTimeout = 2.seconds,
       bodyReadTimeout       = 500.millis
     ),
-    license = Config.License(accept = true)
+    license = Config.License(accept = true),
+    debug = Config
+      .Debug
+      .Debug(Config.Debug.Http(enable = false, logHeaders = true, logBody = false, redactHeaders = List.empty))
   )
 }
diff --git a/...scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/KinesisConfigSpec.scala b/...scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/KinesisConfigSpec.scala
@@ -190,7 +190,10 @@ object KinesisConfigSpec {
       instanceId      = None,
       autoGeneratedId = None
     ),
-    license = Config.License(accept = true)
+    license = Config.License(accept = true),
+    debug = Config
+      .Debug
+      .Debug(Config.Debug.Http(enable = false, logHeaders = true, logBody = false, redactHeaders = List.empty))
   )
 
 }
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
@@ -161,6 +161,9 @@ object NsqConfigSpec {
       responseHeaderTimeout = 2.seconds,
       bodyReadTimeout       = 500.millis
     ),
-    license = Config.License(accept = true)
+    license = Config.License(accept = true),
+    debug = Config
+      .Debug
+      .Debug(Config.Debug.Http(enable = false, logHeaders = true, logBody = false, redactHeaders = List.empty))
   )
 }
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
@@ -181,7 +181,10 @@ object ConfigSpec {
       instanceId      = None,
       autoGeneratedId = None
     ),
-    license = Config.License(accept = true)
+    license = Config.License(accept = true),
+    debug = Config
+      .Debug
+      .Debug(Config.Debug.Http(enable = false, logHeaders = true, logBody = false, redactHeaders = List.empty))
   )
 
 }
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
@@ -170,7 +170,10 @@ object SqsConfigSpec {
       instanceId      = None,
       autoGeneratedId = None
     ),
-    license = Config.License(accept = true)
+    license = Config.License(accept = true),
+    debug = Config
+      .Debug
+      .Debug(Config.Debug.Http(enable = false, logHeaders = true, logBody = false, redactHeaders = List.empty))
   )
 
 }