WIP

Author: migesok

build.sbt

@@ -87,7 +87,7 @@ lazy val k8sDnsNameResolverIt = project.in(file("k8s-dns-name-resolver-it"))
       // to manipulate its DNS settings
       Cmd("USER", "root"),
       // TODO: WIP do we really need bind-tools?
-      ExecCmd("RUN", "apk", "add", "--no-cache", "bash", "lsof", "coredns", "bind-tools"),
+      ExecCmd("RUN", "apk", "add", "--no-cache", "bash", "lsof", "coredns"),
     ),
     dockerExposedPorts := Seq(9000), // Should match the test app GRPC server port.
     // The int test here needs the test app docker container staged before running the code.

k8s-dns-name-resolver-it/src/main/scala/com/evolution/jgrpc/tools/k8sdns/it/TestClient.scala

@@ -1,5 +1,6 @@
 package com.evolution.jgrpc.tools.k8sdns.it
 
+import com.evolution.jgrpc.tools.k8sdns.K8sDnsNameResolverProvider
 import com.evolution.jgrpc.tools.k8sdns.it.ItAppShared._
 import io.grpc.netty.NettyChannelBuilder
 import k8sdns.it.test_svc.TestSvcGrpc.TestSvcBlockingStub
@@ -68,43 +69,57 @@ private[it] final class TestClient {
   }
 
   private def testCaseDiscoverNewPod(fixture: Fixture): Unit = {
-    fixture.coreDns.ensureStarted()
+    fixture.coreDns.ensureStarted(serviceIps = Set(fixture.srv1Ip))
 
     withRoundRobinLbClient { client =>
-      require(callServerNTimes(client, 2).toSet == Set(1))
+      callHost2TimesAssertServerIds(client, expectedServerIds = Set(1))
 
-      fixture.coreDns.setServiceIps(Vector(fixture.srv1Ip, fixture.srv2Ip))
+      fixture.coreDns.setServiceIps(Set(fixture.srv1Ip, fixture.srv2Ip))
 
-      Thread.sleep(15000L) // TODO: wait time?
+      sleepUntilClientGetsDnsUpdate()
 
-      require(callServerNTimes(client, 2).toSet == Set(1, 2))
+      callHost2TimesAssertServerIds(client, expectedServerIds = Set(1, 2))
     }
   }
 
   private def testCaseDnsFailureRecover(fixture: Fixture): Unit = {
-    fixture.coreDns.ensureStarted()
+    fixture.coreDns.ensureStarted(serviceIps = Set(fixture.srv1Ip))
 
     withRoundRobinLbClient { client =>
-      require(callServerNTimes(client, 2).toSet == Set(1))
+      callHost2TimesAssertServerIds(client, expectedServerIds = Set(1))
 
       fixture.coreDns.ensureStopped()
 
-      fixture.coreDns.setServiceIps(Vector(fixture.srv1Ip, fixture.srv2Ip))
+      sleepUntilClientGetsDnsUpdate()
 
-      Thread.sleep(15000L) // TODO: wait time?
+      fixture.coreDns.ensureStarted(serviceIps = Set(fixture.srv1Ip, fixture.srv2Ip))
 
-      fixture.coreDns.ensureStarted()
+      sleepUntilClientGetsDnsUpdate()
 
-      Thread.sleep(15000L) // TODO: wait time?
-
-      require(callServerNTimes(client, 2).toSet == Set(1, 2))
+      callHost2TimesAssertServerIds(client, expectedServerIds = Set(1, 2))
     }
   }
 
-  private def callServerNTimes(client: TestSvcBlockingStub, times: Int): Vector[Int] = {
-    0.until(times).map { _ =>
+  private def sleepUntilClientGetsDnsUpdate(): Unit = {
+    val sleepIntervalSeconds =
+      coreDnsHostsReloadIntervalSeconds +
+        K8sDnsNameResolverProvider.DEFAULT_REFRESH_INTERVAL_SECONDS +
+        2 // adding 2 seconds on top just in case
+
+    println(s"Sleeping until GRPC client gets DNS update: $sleepIntervalSeconds seconds")
+    Thread.sleep(sleepIntervalSeconds.toLong * 1000)
+  }
+
+  private def callHost2TimesAssertServerIds(
+    client: TestSvcBlockingStub,
+    expectedServerIds: Set[Int],
+  ): Unit = {
+    val actualServerIds = 0.until(2).map { _ =>
       client.getId(GetIdRequest()).id
-    }.toVector
+    }.toSet
+    if (actualServerIds != expectedServerIds) {
+      sys.error(s"GRPC client observed server IDs $actualServerIds, expected $expectedServerIds")
+    }
   }
 
   private def withRoundRobinLbClient[T](body: TestSvcBlockingStub => T): T = {
@@ -132,11 +147,12 @@ private object TestClient {
   private val coreDnsHealthEndpointPort = 8080
   private val coreDnsReadyTimeout = 10.seconds
   private val coreDnsReadyCheckAttemptDelay = 2.seconds
+  private val coreDnsHostsReloadIntervalSeconds = 2
 
   private final class Fixture {
     val srv1Ip: String = InetAddress.getByName(TestAppSvcNames.Server1).getHostAddress
     val srv2Ip: String = InetAddress.getByName(TestAppSvcNames.Server2).getHostAddress
-    val coreDns: CoreDns = new CoreDns(defaultServiceIps = Vector(srv1Ip))
+    val coreDns: CoreDns = new CoreDns
     setCoreDnsAsPrimaryDns()
   }
 
@@ -177,11 +193,11 @@ private object TestClient {
     }
   }
 
-  private final class CoreDns(defaultServiceIps: Vector[String]) {
+  private final class CoreDns {
     private val healthCheckHttpClient = HttpClient.newHttpClient()
 
     writeCoreFile()
-    setServiceIps(defaultServiceIps)
+    writeHostsFile(ips = Set.empty)
 
     private var processOpt: Option[CoreDns.StartedProcess] = None
 
@@ -190,10 +206,9 @@ private object TestClient {
         Paths.get(coreDnsCoreFilePath),
         s"""$svcHostname {
            |    hosts $coreDnsHostsFilePath {
-           |        ttl 5
-           |        reload 5s
+           |        ttl $coreDnsHostsReloadIntervalSeconds
+           |        reload ${ coreDnsHostsReloadIntervalSeconds }s
            |    }
-           |    reload 5s 1s
            |    errors         # show errors
            |    log            # enable query logs
            |    health :$coreDnsHealthEndpointPort   # enable healthcheck HTTP endpoint
@@ -240,7 +255,8 @@ private object TestClient {
       }
     }
 
-    def ensureStarted(): Unit = {
+    def ensureStarted(serviceIps: Set[String]): Unit = {
+      setServiceIps(serviceIps)
       processOpt match {
         case None =>
           startNewProcess()
@@ -256,8 +272,8 @@ private object TestClient {
         case Some(process) if process.sysProcess.isAlive() =>
           process.sysProcess.destroy() // TODO: does it wait for the process to stop?
         case _ =>
-          processOpt = None
       }
+      processOpt = None
     }
 
     private def startNewProcess(): Unit = {
@@ -284,19 +300,25 @@ private object TestClient {
       }
     }
 
-    def setServiceIps(ips: Vector[String]): Unit = {
+    private def writeHostsFile(ips: Set[String]): Unit = {
       // overwriting hosts file atomically so CoreDNS couldn't observe broken file content
 
       val tmpHostsFile = Files.createTempFile("hosts", "txt")
       Files.write(
         tmpHostsFile,
-        ips.map(ip => s"$ip $svcHostname").asJava,
+        ips.toVector.sorted.map(ip => s"$ip $svcHostname").asJava,
         StandardOpenOption.CREATE,
         StandardOpenOption.TRUNCATE_EXISTING,
       )
 
       Files.move(tmpHostsFile, Paths.get(coreDnsHostsFilePath), StandardCopyOption.ATOMIC_MOVE)
     }
+
+    def setServiceIps(ips: Set[String]): Unit = {
+      writeHostsFile(ips)
+      // force CoreDNS to reread configs just to be sure
+      processOpt.foreach(p => sys.process.Process(s"kill -HUP ${ p.pid }").!!)
+    }
   }
 
   private object CoreDns {