From 85dcf10c6b858cdc7b233772c6719f3916479524 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Piotr=20Poniedzia=C5=82ek?= Date: Fri, 16 Feb 2024 13:11:33 +0100 Subject: [PATCH] Simplify building HTTP/S resources --- .../Configuration.scala | 6 -- .../MicroHttpServer.scala | 66 ++++++++----------- .../Run.scala | 32 +-------- 3 files changed, 31 insertions(+), 73 deletions(-) diff --git a/src/main/scala/com.snowplowanalytics.snowplow.micro/Configuration.scala b/src/main/scala/com.snowplowanalytics.snowplow.micro/Configuration.scala index ccaba4c..d09566f 100644 --- a/src/main/scala/com.snowplowanalytics.snowplow.micro/Configuration.scala +++ b/src/main/scala/com.snowplowanalytics.snowplow.micro/Configuration.scala @@ -81,7 +81,6 @@ object Configuration { namespaced(ConfigFactory.load(namespaced(config.withFallback(namespaced(ConfigFactory.parseResources("collector-micro.conf")))))) loadConfig[CollectorConfig[SinkConfig]](path, resolveOrder) - .map(adjustSslConfig) } private def loadIgluResources(path: Option[Path]): EitherT[IO, String, IgluResources] = { @@ -113,11 +112,6 @@ object Configuration { client <- EitherT.liftF(IgluCirceClient.fromResolver[IO](completeResolver, resolverConfig.cacheSize)) } yield IgluResources(resolver, client) - private def adjustSslConfig(config: CollectorConfig[SinkConfig]): CollectorConfig[SinkConfig] = { - val envVarPresent = sys.env.contains(EnvironmentVariables.sslCertificatePassword) - config.copy(ssl = config.ssl.copy(enable = envVarPresent)) - } - private def loadEnrichmentsAsSDD(enrichmentsDirectory: Path, igluClient: IgluCirceClient[IO], fileType: String): EitherT[IO, String, List[EnrichmentConf]] = { diff --git a/src/main/scala/com.snowplowanalytics.snowplow.micro/MicroHttpServer.scala b/src/main/scala/com.snowplowanalytics.snowplow.micro/MicroHttpServer.scala index ed9727f..6f24cf1 100644 --- a/src/main/scala/com.snowplowanalytics.snowplow.micro/MicroHttpServer.scala +++ b/src/main/scala/com.snowplowanalytics.snowplow.micro/MicroHttpServer.scala @@ -1,11 +1,11 @@ package com.snowplowanalytics.snowplow.micro import cats.effect.{IO, Resource} -import cats.implicits._ import com.avast.datadog4s.api.Tag import com.avast.datadog4s.extension.http4s.DatadogMetricsOps import com.avast.datadog4s.{StatsDMetricFactory, StatsDMetricFactoryConfig} import com.snowplowanalytics.snowplow.collector.core.{Config => CollectorConfig} +import com.snowplowanalytics.snowplow.micro.Configuration.MicroConfig import org.http4s.blaze.server.BlazeServerBuilder import org.http4s.headers.`Strict-Transport-Security` import org.http4s.server.Server @@ -18,28 +18,41 @@ import java.net.InetSocketAddress import javax.net.ssl.SSLContext -/** The same http server builder as in collector but with option to customize SSLContext */ +/** Similar to HTTP server builder as in collector but with option to customize SSLContext */ object MicroHttpServer { implicit private def logger: Logger[IO] = Slf4jLogger.getLogger[IO] def build(routes: HttpRoutes[IO], - port: Int, - secure: Boolean, - customSslContext: Option[SSLContext], - hsts: CollectorConfig.HSTS, - networking: CollectorConfig.Networking, - metricsCollectorConfig: CollectorConfig.Metrics - ): Resource[IO, Server] = + config: MicroConfig, + sslContext: Option[SSLContext]): Resource[IO, Unit] = for { - withMetricsMiddleware <- createMetricsMiddleware(routes, metricsCollectorConfig) - server <- buildBlazeServer(withMetricsMiddleware, port, secure, customSslContext, hsts, networking) - } yield server + withMetricsMiddleware <- createMetricsMiddleware(routes, config.collector.monitoring.metrics) + _ <- Resource.eval(Logger[IO].info("Building blaze server")) + _ <- buildHTTPServer(withMetricsMiddleware, config) + _ <- sslContext.map(definedSSL => buildHTTPSServer(withMetricsMiddleware, config, definedSSL)).getOrElse(Resource.unit[IO]) + } yield () - private def createMetricsMiddleware( - routes: HttpRoutes[IO], - metricsCollectorConfig: CollectorConfig.Metrics - ): Resource[IO, HttpRoutes[IO]] = + private def buildHTTPServer(routes: HttpRoutes[IO], config: MicroConfig): Resource[IO, Server] = + builder(routes, config) + .bindSocketAddress(new InetSocketAddress(config.collector.port)) + .resource + + private def buildHTTPSServer(routes: HttpRoutes[IO], config: MicroConfig, sslContext: SSLContext): Resource[IO, Server] = + builder(routes, config) + .bindSocketAddress(new InetSocketAddress(config.collector.ssl.port)) + .withSslContext(sslContext) + .resource + + private def builder(routes: HttpRoutes[IO], config: MicroConfig): BlazeServerBuilder[IO] = { + BlazeServerBuilder[IO] + .withHttpApp(hstsMiddleware(config.collector.hsts, routes.orNotFound)) + .withIdleTimeout(config.collector.networking.idleTimeout) + .withMaxConnections(config.collector.networking.maxConnections) + } + + private def createMetricsMiddleware(routes: HttpRoutes[IO], + metricsCollectorConfig: CollectorConfig.Metrics): Resource[IO, HttpRoutes[IO]] = if (metricsCollectorConfig.statsd.enabled) { val metricsFactory = StatsDMetricFactory.make[IO](createStatsdCollectorConfig(metricsCollectorConfig)) metricsFactory.evalMap(DatadogMetricsOps.builder[IO](_).useDistributionBasedTimers().build()).map { metricsOps => @@ -59,25 +72,4 @@ object MicroHttpServer { if (hsts.enable) HSTS(routes, `Strict-Transport-Security`.unsafeFromDuration(hsts.maxAge)) else routes - - private def buildBlazeServer(routes: HttpRoutes[IO], - port: Int, - secure: Boolean, - customSslContext: Option[SSLContext], - hsts: CollectorConfig.HSTS, - networking: CollectorConfig.Networking - ): Resource[IO, Server] = - Resource.eval(Logger[IO].info("Building blaze server")) >> - BlazeServerBuilder[IO] - .bindSocketAddress(new InetSocketAddress(port)) - .withHttpApp(hstsMiddleware(hsts, routes.orNotFound)) - .withIdleTimeout(networking.idleTimeout) - .withMaxConnections(networking.maxConnections) - .cond(secure, _.withSslContext(customSslContext.getOrElse(SSLContext.getDefault))) - .resource - - implicit class ConditionalAction[A](item: A) { - def cond(cond: Boolean, action: A => A): A = - if (cond) action(item) else item - } } diff --git a/src/main/scala/com.snowplowanalytics.snowplow.micro/Run.scala b/src/main/scala/com.snowplowanalytics.snowplow.micro/Run.scala index b42120d..c6ac1e2 100644 --- a/src/main/scala/com.snowplowanalytics.snowplow.micro/Run.scala +++ b/src/main/scala/com.snowplowanalytics.snowplow.micro/Run.scala @@ -22,7 +22,6 @@ import com.snowplowanalytics.snowplow.enrich.common.enrichments.EnrichmentRegist import com.snowplowanalytics.snowplow.enrich.common.enrichments.registry.{Enrichment, EnrichmentConf} import com.snowplowanalytics.snowplow.enrich.common.utils.{HttpClient, ShiftExecution} import com.snowplowanalytics.snowplow.micro.Configuration.MicroConfig -import org.http4s.HttpRoutes import org.http4s.ember.client.EmberClientBuilder import org.typelevel.log4cats.Logger import org.typelevel.log4cats.slf4j.Slf4jLogger @@ -53,7 +52,7 @@ object Run { private def buildEnvironment(config: MicroConfig): Resource[IO, Unit] = { for { - customSslContext <- Resource.eval(setupSSLContext()) + sslContext <- Resource.eval(setupSSLContext()) enrichmentRegistry <- buildEnrichmentRegistry(config.enrichmentsConfig) badProcessor = Processor(BuildInfo.name, BuildInfo.version) adapterRegistry = MicroAdapterRegistry.create() @@ -73,16 +72,7 @@ object Run { miniRoutes = new Routing(config.iglu.resolver)(lookup).value allRoutes = miniRoutes <+> collectorRoutes - _ <- MicroHttpServer.build( - allRoutes, - config.collector.port, - secure = false, - customSslContext = None, - config.collector.hsts, - config.collector.networking, - config.collector.monitoring.metrics - ) - _ <- runHttpsServerIfEnabled(config, customSslContext, allRoutes) + _ <- MicroHttpServer.build(allRoutes, config, sslContext) } yield () } @@ -138,24 +128,6 @@ object Run { } } - private def runHttpsServerIfEnabled(config: MicroConfig, - customSslContext: Option[SSLContext], - routes: HttpRoutes[IO]): Resource[IO, Unit] = { - if (config.collector.ssl.enable) { - MicroHttpServer.build( - routes, - config.collector.ssl.port, - secure = true, - customSslContext, - config.collector.hsts, - config.collector.networking, - config.collector.monitoring.metrics - ).void - } else { - Resource.unit - } - } - private def handleAppErrors(appOutput: EitherT[IO, String, ExitCode]): IO[ExitCode] = { appOutput .leftSemiflatMap { error =>