Repository: 7mind/izumi
Pull request #1099 (2020-06-05 13:27:14)

Better `availablePorts` API, postgresSqlPortcheck API
/cc @coreyoconnor 


Patch:
diff --git a/distage/distage-core-api/src/main/scala/izumi/distage/model/providers/ProviderMagnet.scala b/distage/distage-core-api/src/main/scala/izumi/distage/model/providers/ProviderMagnet.scala
index b38607533..ef46ad676 100644
--- a/distage/distage-core-api/src/main/scala/izumi/distage/model/providers/ProviderMagnet.scala
+++ b/distage/distage-core-api/src/main/scala/izumi/distage/model/providers/ProviderMagnet.scala
@@ -103,11 +103,10 @@ final case class ProviderMagnet[+A](get: Provider) {
     zip(that).map[C](f.tupled)
   }
 
-  /** Applicative's `ap` method - can be used to chain transformations like `flatMap` */
+  /** Applicative's `ap` method - can be used to chain transformations like `flatMap`.
+    * Apply a function produced by `that` Provider to the argument produced by `this` Provider */
   def flatAp[B: Tag](that: ProviderMagnet[A => B]): ProviderMagnet[B] = {
-    implicit val rTag: Tag[A] = getRetTag
-    rTag.discard() // used for assembling Tag[A => B] below
-    map2[A => B, B](that) { case (a, f) => f(a) }
+    map2(that) { case (a, f) => f(a) }
   }
 
   /** Apply a function produced by `this` Provider to the argument produced by `that` Provider */
@@ -115,11 +114,9 @@ final case class ProviderMagnet[+A](get: Provider) {
     that.flatAp[C](this.asInstanceOf[ProviderMagnet[B => C]])
   }
 
-  /** Add `B` as an unused dependency of this constructor */
+  /** Add `B` as an unused dependency of this Provider */
   def addDependency[B: Tag]: ProviderMagnet[A] = addDependency(DIKey.get[B])
-
   def addDependency(key: DIKey): ProviderMagnet[A] = addDependencies(key :: Nil)
-
   def addDependencies(keys: Seq[DIKey]): ProviderMagnet[A] = copy[A](get = get.addUnused(keys))
 
   @inline private def getRetTag: Tag[A @uncheckedVariance] = Tag(get.ret.cls, get.ret.tag)
@@ -174,7 +171,8 @@ object ProviderMagnet {
         originalFun = () => a,
         fun = (_: Seq[Any]) => a,
         providerType = ProviderType.Function,
-      ))
+      )
+    )
   }
 
   def singleton[A <: Singleton: Tag](a: A): ProviderMagnet[A] = {
@@ -184,7 +182,8 @@ object ProviderMagnet {
         ret = SafeType.get[A],
         fun = (_: Seq[Any]) => a,
         providerType = ProviderType.Singleton,
-      ))
+      )
+    )
   }
 
   def single[A: Tag, B: Tag](f: A => B): ProviderMagnet[B] = {
@@ -223,7 +222,7 @@ object ProviderMagnet {
       name = "x$1",
       finalResultType = tpe,
       isByName = false,
-      wasGeneric = false
+      wasGeneric = false,
     )
   }
 }
diff --git a/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/DockerContainer.scala b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/DockerContainer.scala
index c76b28fb7..5c104b563 100644
--- a/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/DockerContainer.scala
+++ b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/DockerContainer.scala
@@ -55,10 +55,8 @@ object DockerContainer {
       */
     def modifyConfig(
       modify: ProviderMagnet[Docker.ContainerConfig[T] => Docker.ContainerConfig[T]]
-    )(implicit tag1: distage.Tag[ContainerResource[F, T]],
-      tag2: distage.Tag[Docker.ContainerConfig[T]],
+    )(implicit tag: distage.Tag[ContainerResource[F, T]]
     ): ProviderMagnet[ContainerResource[F, T]] = {
-      tag2.discard()
       self.zip(modify).map {
         case (that, f) =>
           import that._
diff --git a/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/examples/PostgresDocker.scala b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/examples/PostgresDocker.scala
index 724110e10..124781848 100644
--- a/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/examples/PostgresDocker.scala
+++ b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/examples/PostgresDocker.scala
@@ -14,7 +14,7 @@ object PostgresDocker extends ContainerDef {
       image = "library/postgres:12.2",
       ports = Seq(primaryPort),
       env = Map("POSTGRES_PASSWORD" -> "postgres"),
-      healthCheck = ContainerHealthCheck.postgreSqlProtocolCheck,
+      healthCheck = ContainerHealthCheck.postgreSqlProtocolCheck(primaryPort, "postgres", "postgres"),
     )
   }
 }
diff --git a/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/ContainerHealthCheck.scala b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/ContainerHealthCheck.scala
index 3df927f73..d143f1b39 100644
--- a/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/ContainerHealthCheck.scala
+++ b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/ContainerHealthCheck.scala
@@ -3,23 +3,23 @@ package izumi.distage.docker.healthcheck
 import izumi.distage.docker.Docker._
 import izumi.distage.docker.DockerContainer
 import izumi.distage.docker.healthcheck.ContainerHealthCheck.HealthCheckResult
-import izumi.distage.docker.healthcheck.ContainerHealthCheck.HealthCheckResult.AvailableOnPorts
 import izumi.fundamentals.collections.nonempty.NonEmptyList
 import izumi.logstage.api.IzLogger
 
-import scala.reflect.ClassTag
-
 trait ContainerHealthCheck[Tag] {
   def check(logger: IzLogger, container: DockerContainer[Tag]): HealthCheckResult
 
   final def ++(next: HealthCheckResult => ContainerHealthCheck[Tag]): ContainerHealthCheck[Tag] = this combine next
-  final def combine(next: HealthCheckResult => ContainerHealthCheck[Tag]): ContainerHealthCheck[Tag] = combineOn(next)
-  final def combineOn[T: ClassTag](next: T => ContainerHealthCheck[Tag]): ContainerHealthCheck[Tag] = {
+  final def combine(next: HealthCheckResult => ContainerHealthCheck[Tag]): ContainerHealthCheck[Tag] = {
+    (logger: IzLogger, container: DockerContainer[Tag]) =>
+      next(check(logger, container)).check(logger, container)
+  }
+  final def combineOnPorts(next: HealthCheckResult.AvailableOnPorts => ContainerHealthCheck[Tag]): ContainerHealthCheck[Tag] = {
     (logger: IzLogger, container: DockerContainer[Tag]) =>
       check(logger, container) match {
-        case thisCheck: T =>
-          next(thisCheck).check(logger, container) match {
-            case HealthCheckResult.Available => thisCheck
+        case thisCheckResult: HealthCheckResult.AvailableOnPorts =>
+          next(thisCheckResult).check(logger, container) match {
+            case HealthCheckResult.Available => thisCheckResult
             case other => other
           }
         case other => other
@@ -28,6 +28,14 @@ trait ContainerHealthCheck[Tag] {
 }
 
 object ContainerHealthCheck {
+  def portCheck[T]: ContainerHealthCheck[T] = new TCPContainerHealthCheck[T]
+  def httpGetCheck[T](port: DockerPort): ContainerHealthCheck[T] = portCheck.combineOnPorts(new HttpGetCheck(_, port, useHttps = false))
+  def httpsGetCheck[T](port: DockerPort): ContainerHealthCheck[T] = portCheck.combineOnPorts(new HttpGetCheck(_, port, useHttps = true))
+  def postgreSqlProtocolCheck[T](port: DockerPort, user: String, password: String): ContainerHealthCheck[T] =
+    portCheck.combineOnPorts(new PostgreSqlProtocolCheck(_, port, user, password))
+
+  def succeed[T]: ContainerHealthCheck[T] = (_, _) => HealthCheckResult.Available
+
   sealed trait HealthCheckResult
   object HealthCheckResult {
     case object Available extends HealthCheckResult
@@ -40,17 +48,21 @@ object ContainerHealthCheck {
     ) extends HealthCheckResult
   }
 
-  final case class UnavailablePorts(unavailablePorts: Map[DockerPort, List[(AvailablePort, Option[Throwable])]])
+  final case class UnavailablePorts(unavailablePorts: Map[DockerPort, List[(AvailablePort, Option[Throwable])]]) extends AnyVal {
+    override def toString: String = s"unavailable=$unavailablePorts"
+  }
+
+  final case class VerifiedContainerConnectivity(availablePorts: Map[DockerPort, NonEmptyList[AvailablePort]]) extends AnyVal {
+    def apply(port: DockerPort): NonEmptyList[AvailablePort] = availablePorts(port)
+    def first(port: DockerPort): AvailablePort = availablePorts(port).head
+
+    def get(port: DockerPort): Option[NonEmptyList[AvailablePort]] = availablePorts.get(port)
+    def firstOption(port: DockerPort): Option[AvailablePort] = availablePorts.get(port).map(_.head)
 
-  final case class VerifiedContainerConnectivity(availablePorts: Map[DockerPort, NonEmptyList[AvailablePort]]) {
     override def toString: String = s"accessible=$availablePorts"
   }
   object VerifiedContainerConnectivity {
     def empty: VerifiedContainerConnectivity = VerifiedContainerConnectivity(Map.empty)
   }
 
-  def ignore[T]: ContainerHealthCheck[T] = (_, _) => HealthCheckResult.Available
-  def portCheck[T]: ContainerHealthCheck[T] = new TCPContainerHealthCheck[T]
-  def postgreSqlProtocolCheck[T]: ContainerHealthCheck[T] = portCheck.combineOn[AvailableOnPorts](new PostgreSqlProtocolCheck(_))
-  def httpGetCheck[T](port: DockerPort): ContainerHealthCheck[T] = portCheck.combineOn[AvailableOnPorts](new HttpGetCheck(_, port))
 }
diff --git a/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/HttpGetCheck.scala b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/HttpGetCheck.scala
index d11d34f63..00c770048 100644
--- a/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/HttpGetCheck.scala
+++ b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/HttpGetCheck.scala
@@ -10,12 +10,11 @@ import izumi.logstage.api.IzLogger
 final class HttpGetCheck[Tag](
   portStatus: HealthCheckResult.AvailableOnPorts,
   port: DockerPort,
-  useHttps: Boolean = false
+  useHttps: Boolean,
 ) extends ContainerHealthCheck[Tag] {
   override def check(logger: IzLogger, container: DockerContainer[Tag]): ContainerHealthCheck.HealthCheckResult = {
-    portStatus.availablePorts.availablePorts.get(port) match {
-      case Some(value) if portStatus.requiredPortsAccessible =>
-        val availablePort = value.head
+    portStatus.availablePorts.firstOption(port) match {
+      case Some(availablePort) if portStatus.requiredPortsAccessible =>
         val protocol = if (useHttps) "https" else "http"
         val url = new URL(s"$protocol://${availablePort.hostV4}:${availablePort.port}")
         logger.info(s"Checking docker port $port via $url for $container. Will try to establish HTTP connection.")
@@ -28,12 +27,12 @@ final class HttpGetCheck[Tag](
             logger.info(s"HTTP connection was successfully established with $port.")
             ContainerHealthCheck.HealthCheckResult.Available
           } else {
-            logger.debug(s"Cannot establish HTTP connection with $port. Wrong protocol.")
+            logger.info(s"Cannot establish HTTP connection with $port. Wrong protocol.")
             ContainerHealthCheck.HealthCheckResult.Unavailable
           }
         } catch {
           case t: Throwable =>
-            logger.debug(s"Cannot establish HTTP connection with $port due to ${t.getMessage -> "failure"}")
+            logger.warn(s"Cannot establish HTTP connection with $port due to ${t.getMessage -> "failure"}")
             ContainerHealthCheck.HealthCheckResult.Unavailable
         }
       case _ => ContainerHealthCheck.HealthCheckResult.Unavailable
diff --git a/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/PostgreSqlProtocolCheck.scala b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/PostgreSqlProtocolCheck.scala
index 75794ca26..ed70929b9 100644
--- a/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/PostgreSqlProtocolCheck.scala
+++ b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/PostgreSqlProtocolCheck.scala
@@ -11,17 +11,16 @@ import izumi.logstage.api.IzLogger
 
 final class PostgreSqlProtocolCheck[Tag](
   portStatus: HealthCheckResult.AvailableOnPorts,
-  port: DockerPort = DockerPort.TCP(5432),
-  userName: String = "postgres",
-  databaseName: String = "postgres",
+  port: DockerPort,
+  userName: String,
+  databaseName: String,
 ) extends ContainerHealthCheck[Tag] {
   override def check(logger: IzLogger, container: DockerContainer[Tag]): ContainerHealthCheck.HealthCheckResult = {
-    portStatus.availablePorts.availablePorts.get(port) match {
-      case Some(value) if portStatus.requiredPortsAccessible =>
+    portStatus.availablePorts.firstOption(port) match {
+      case Some(availablePort) if portStatus.requiredPortsAccessible =>
         val startupMessage = genStartupMessage()
         val socket = new Socket()
         try {
-          val availablePort = value.head
           socket.connect(new InetSocketAddress(availablePort.hostV4, availablePort.port), container.containerConfig.portProbeTimeout.toMillis.toInt)
           logger.info(s"Checking PostgreSQL protocol on $port for $container. ${startupMessage.toIterable.toHex -> "Startup message"}.")
           val out = socket.getOutputStream
@@ -34,12 +33,12 @@ final class PostgreSqlProtocolCheck[Tag](
             logger.info(s"PostgreSQL protocol on $port is available.")
             ContainerHealthCheck.HealthCheckResult.Available
           } else {
-            logger.debug(s"PostgreSQL protocol on $port unavailable due to unknown message type: $messageType.")
+            logger.info(s"PostgreSQL protocol on $port unavailable due to unknown message type: $messageType.")
             ContainerHealthCheck.HealthCheckResult.Unavailable
           }
         } catch {
           case t: Throwable =>
-            logger.debug(s"PostgreSQL protocol on $port unavailable due to unexpected exception. ${t.getMessage -> "Failure"}")
+            logger.warn(s"PostgreSQL protocol on $port unavailable due to unexpected exception. ${t.getMessage -> "Failure"}")
             ContainerHealthCheck.HealthCheckResult.Unavailable
         } finally {
           socket.close()
diff --git a/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/TCPContainerHealthCheck.scala b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/TCPContainerHealthCheck.scala
index 22e6858be..5678aab31 100644
--- a/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/TCPContainerHealthCheck.scala
+++ b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/healthcheck/TCPContainerHealthCheck.scala
@@ -11,10 +11,14 @@ import izumi.fundamentals.platform.integration.{PortCheck, ResourceCheck}
 import izumi.fundamentals.platform.strings.IzString._
 import izumi.logstage.api.IzLogger
 
-
 class TCPContainerHealthCheck[Tag] extends ContainerHealthCheckBase[Tag] {
 
-  override protected def perform(logger: IzLogger, container: DockerContainer[Tag], tcpPorts: Map[DockerPort.TCPBase, NonEmptyList[ServicePort]], udpPorts: Map[DockerPort.UDPBase, NonEmptyList[ServicePort]]): HealthCheckResult.AvailableOnPorts = {
+  override protected def perform(
+    logger: IzLogger,
+    container: DockerContainer[Tag],
+    tcpPorts: Map[DockerPort.TCPBase, NonEmptyList[ServicePort]],
+    udpPorts: Map[DockerPort.UDPBase, NonEmptyList[ServicePort]],
+  ): HealthCheckResult.AvailableOnPorts = {
     val check = new PortCheck(container.containerConfig.portProbeTimeout)
 
     val dockerHostCandidates = findDockerHostCandidates(container, tcpPorts)
@@ -26,7 +30,6 @@ class TCPContainerHealthCheck[Tag] extends ContainerHealthCheckBase[Tag] {
 
     logger.debug(s"going to check ports on $container: ${allCandidates.map { case PortCandidate(k, v) => s"if $k is available at $v" }.niceList() -> "port mappings"}")
 
-
     val checks = allCandidates.map {
       case PortCandidate(dp, ap) =>
         check.checkPort(ap.hostV4, ap.port, s"open port ${ap.hostV4}:${ap.port} for ${container.id}") match {
@@ -48,7 +51,8 @@ class TCPContainerHealthCheck[Tag] extends ContainerHealthCheckBase[Tag] {
           case (dp, ap) =>
             (dp: DockerPort, ap.toList)
         }
-        .toMap)
+        .toMap
+    )
 
     val available = VerifiedContainerConnectivity(
       good
@@ -66,8 +70,7 @@ class TCPContainerHealthCheck[Tag] extends ContainerHealthCheckBase[Tag] {
       available,
       errors,
       udpPorts.keySet.map(p => p: DockerPort),
-      tcpPortsGood(container, available)
+      tcpPortsGood(container, available),
     )
   }
 }
-


Test patch:
diff --git a/distage/distage-framework-docker/src/test/scala/izumi/distage/testkit/docker/fixtures/DockerPlugin.scala b/distage/distage-framework-docker/src/test/scala/izumi/distage/testkit/docker/fixtures/DockerPlugin.scala
index 80126c3bd..dd104fe50 100644
--- a/distage/distage-framework-docker/src/test/scala/izumi/distage/testkit/docker/fixtures/DockerPlugin.scala
+++ b/distage/distage-framework-docker/src/test/scala/izumi/distage/testkit/docker/fixtures/DockerPlugin.scala
@@ -41,17 +41,17 @@ object DockerPlugin extends PluginDef {
 
   make[AvailablePort].named("mq").tagged(Env.Test).from {
     cs: ElasticMQDocker.Container =>
-      cs.availablePorts.availablePorts(ElasticMQDocker.primaryPort).head
+      cs.availablePorts.first(ElasticMQDocker.primaryPort)
   }
 
   make[AvailablePort].named("cs").tagged(Env.Test).from {
     cs: CassandraDocker.Container =>
-      cs.availablePorts.availablePorts(CassandraDocker.primaryPort).head
+      cs.availablePorts.first(CassandraDocker.primaryPort)
   }
 
   make[AvailablePort].named("kafka").tagged(Env.Test).from {
     kafka: KafkaDocker.Container =>
-      kafka.availablePorts.availablePorts(KafkaDocker.primaryPort).head
+      kafka.availablePorts.first(KafkaDocker.primaryPort)
   }
 
   // this container will start once `DynamoContainer` is up and running
@@ -62,11 +62,11 @@ object DockerPlugin extends PluginDef {
   // these lines are for test scope
   make[AvailablePort].named("pg").tagged(Env.Test).from {
     pg: PostgresDocker.Container =>
-      pg.availablePorts.availablePorts(PostgresDocker.primaryPort).head
+      pg.availablePorts.first(PostgresDocker.primaryPort)
   }
   make[AvailablePort].named("ddb").tagged(Env.Test).from {
     dn: DynamoDocker.Container =>
-      dn.availablePorts.availablePorts(DynamoDocker.primaryPort).head
+      dn.availablePorts.first(DynamoDocker.primaryPort)
   }
 
   // and this one is for production