Wait for service availability at the ingress before running tests

settings.gradle.kts

@@ -31,7 +31,7 @@ dependencyResolutionManagement {
  version("protobuf", "3.24.3")
  version("grpc", "1.58.0")
  version("grpckt", "1.4.0")
- version("log4j", "2.19.0")
+ version("log4j", "2.22.0")
  version("jackson", "2.15.2")
 
  version("junit-jupiter", "5.10.0")

test-utils/build.gradle.kts

@@ -37,6 +37,8 @@ dependencies {
  implementation(libs.jackson.databind)
  implementation(libs.jackson.yaml)
 
+ implementation(libs.awaitility)
+
  testImplementation(libs.junit.all)
  testImplementation(libs.assertj)
 }

test-utils/src/main/kotlin/dev/restate/e2e/utils/JsonUtils.kt

@@ -0,0 +1,43 @@
+// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate e2e tests,
+// which are released under the MIT license.
+//
+// You can find a copy of the license in file LICENSE in the root
+// directory of this repository or package, or at
+// https://github.com/restatedev/e2e/blob/main/LICENSE
+
+package dev.restate.e2e.utils
+
+import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.databind.ObjectMapper
+import java.net.URI
+import java.net.http.HttpClient
+import java.net.http.HttpRequest
+import java.net.http.HttpResponse
+import java.nio.charset.StandardCharsets
+
+object JsonUtils {
+ private val objMapper = ObjectMapper()
+ private val httpClient = HttpClient.newHttpClient()
+
+ fun jacksonBodyHandler(): HttpResponse.BodyHandler<JsonNode> {
+ return HttpResponse.BodyHandler {
+ HttpResponse.BodySubscribers.mapping(
+ HttpResponse.BodySubscribers.ofString(StandardCharsets.UTF_8), objMapper::readTree)
+ }
+ }
+
+ fun jacksonBodyPublisher(value: Any): HttpRequest.BodyPublisher {
+ return HttpRequest.BodyPublishers.ofString(objMapper.writeValueAsString(value))
+ }
+
+ fun postJsonRequest(uri: String, reqBody: Any): HttpResponse<JsonNode> {
+ val req =
+ HttpRequest.newBuilder(URI.create(uri))
+ .headers("Content-Type", "application/json")
+ .POST(jacksonBodyPublisher(reqBody))
+ .build()
+ return httpClient.send(req, jacksonBodyHandler())
+ }
+}

test-utils/src/main/kotlin/dev/restate/e2e/utils/RestateDeployer.kt

@@ -24,6 +24,7 @@ import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder
 import java.io.File
 import java.lang.reflect.Method
 import java.net.URI
+import java.net.URL
 import java.nio.file.Path
 import java.time.LocalDateTime
 import java.time.format.DateTimeFormatter
@@ -87,6 +88,8 @@ private constructor(
 
  private const val RESTATE_URI_ENV = "RESTATE_URI"
 
+ const val HEALTH_CHECK_SERVICE = "grpc.health.v1.Health/Check"
+
  private val logger = LogManager.getLogger(RestateDeployer::class.java)
 
  @JvmStatic
@@ -250,6 +253,8 @@ private constructor(
  .setPort(getContainerPort(RESTATE_RUNTIME, RUNTIME_META_ENDPOINT_PORT)))
  serviceContainers.values.forEach { (spec, _) -> discoverDeployment(client, spec) }
 
+ waitForServicesBeingAvailable()
+
  // Log environment
  writeEnvironmentReport(testReportDir)
  }
@@ -382,13 +387,28 @@ private constructor(
  RUNTIME_META_ENDPOINT_PORT,
  )
  proxyContainer.waitHttp(
- Wait.forHttp("/grpc.health.v1.Health/Check"),
+ Wait.forHttp("/${HEALTH_CHECK_SERVICE}"),
  RESTATE_RUNTIME,
  RUNTIME_GRPC_INGRESS_ENDPOINT_PORT,
  )
  logger.debug("Runtime META and Ingress healthy")
  }
 
+ private fun waitForServicesBeingAvailable() {
+ val adminUrl =
+ URL(
+ "http",
+ "127.0.0.1",
+ getContainerPort(RESTATE_RUNTIME, RUNTIME_GRPC_INGRESS_ENDPOINT_PORT),
+ "")
+
+ for ((spec, _) in serviceContainers.values) {
+ if (!spec.skipRegistration) {
+ waitForServicesBeingAvailable(spec.services, adminUrl)
+ }
+ }
+ }
+
  fun discoverDeployment(client: DeploymentApi, spec: ServiceSpec) {
  val url = spec.getEndpointUrl()
  if (spec.skipRegistration) {

test-utils/src/main/kotlin/dev/restate/e2e/utils/ServiceSpec.kt

@@ -19,6 +19,7 @@ data class ServiceSpec(
  internal val containerImage: String,
  val hostName: String,
  internal val envs: Map<String, String>,
+ val services: Set<String>,
  internal val port: Int,
  internal val skipRegistration: Boolean,
  internal val dependencies: List<Startable>,
@@ -44,6 +45,7 @@ data class ServiceSpec(
  private var port: Int = 8080,
  private var skipRegistration: Boolean = false,
  private var dependencies: MutableList<Startable> = mutableListOf(),
+ private var services: MutableSet<String> = mutableSetOf(),
  ) {
  fun withHostName(hostName: String) = apply { this.hostName = hostName }
 
@@ -54,11 +56,14 @@ data class ServiceSpec(
 
  fun withEnvs(envs: Map<String, String>) = apply { this.envs.putAll(envs) }
 
+ fun withServices(vararg services: String) = apply { this.services.addAll(services) }
+
  fun skipRegistration() = apply { this.skipRegistration = true }
 
  fun dependsOn(container: Startable) = apply { this.dependencies.add(container) }
 
- fun build() = ServiceSpec(containerImage, hostName, envs, port, skipRegistration, dependencies)
+ fun build() =
+ ServiceSpec(containerImage, hostName, envs, services, port, skipRegistration, dependencies)
  }
 
  fun toBuilder(): Builder {
@@ -67,13 +72,14 @@ data class ServiceSpec(
  hostName,
  envs = envs.toMutableMap(),
  port,
- dependencies = dependencies.toMutableList())
+ dependencies = dependencies.toMutableList(),
+ services = services.toMutableSet())
  }
 
  internal fun toContainer(): GenericContainer<*> {
  return GenericContainer(DockerImageName.parse(containerImage))
  .withEnv("PORT", port.toString())
- .withEnv(envs)
+ .withEnv(envs + ("SERVICES" to services.joinToString(",")))
  .dependsOn(dependencies)
  }
 

test-utils/src/main/kotlin/dev/restate/e2e/utils/utils.kt

@@ -1,4 +1,4 @@
-// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH
+// Copyright (c) 2024 - Restate Software, Inc., Restate GmbH
 //
 // This file is part of the Restate e2e tests,
 // which are released under the MIT license.
@@ -10,6 +10,10 @@
 package dev.restate.e2e.utils
 
 import com.github.dockerjava.api.command.InspectContainerResponse
+import java.net.URL
+import org.awaitility.kotlin.await
+import org.awaitility.kotlin.matches
+import org.awaitility.kotlin.untilCallTo
 import org.testcontainers.containers.wait.strategy.WaitStrategyTarget
 
 internal open class NotCachedContainerInfo(private val delegate: WaitStrategyTarget) :
@@ -29,3 +33,23 @@ internal class WaitOnSpecificPortsTarget(
  return ports.toMutableList()
  }
 }
+
+fun waitForServicesBeingAvailable(services: Collection<String>, ingressURL: URL) {
+ val healthCheckURL =
+ URL(
+ ingressURL.protocol,
+ ingressURL.host,
+ ingressURL.port,
+ "/${RestateDeployer.HEALTH_CHECK_SERVICE}")
+
+ for (service in services) {
+ val body = mapOf("service" to service)
+
+ await
+ .untilCallTo { JsonUtils.postJsonRequest(healthCheckURL.toString(), body) }
+ .matches { response ->
+ response!!.statusCode() == 200 &&
+ response!!.body().get("status").asText().equals("SERVING")
+ }
+ }
+}

tests/src/test/kotlin/dev/restate/e2e/Containers.kt

@@ -41,7 +41,7 @@ object Containers {
  fun javaServicesContainer(hostName: String, vararg services: String): ServiceSpec.Builder {
  assert(services.isNotEmpty())
  return ServiceSpec.builder("restatedev/e2e-java-services")
- .withEnv("SERVICES", services.joinToString(","))
+ .withServices(*services)
  .withHostName(hostName)
  }
 
@@ -82,7 +82,7 @@ object Containers {
  fun nodeServicesContainer(hostName: String, vararg services: String): ServiceSpec.Builder {
  assert(services.isNotEmpty())
  return ServiceSpec.builder("restatedev/e2e-node-services")
- .withEnv("SERVICES", services.joinToString(","))
+ .withServices(*services)
  .withEnv("RESTATE_DEBUG_LOGGING", "JOURNAL")
  .withHostName(hostName)
  }

tests/src/test/kotlin/dev/restate/e2e/KafkaIngressTest.kt

@@ -185,11 +185,11 @@ class NodeHandlerAPIKafkaIngressTest {
  HttpRequest.newBuilder(
  URI.create(
  "${httpEndpointURL}${Containers.HANDLER_API_COUNTER_SERVICE_NAME}/get"))
- .POST(Utils.jacksonBodyPublisher(mapOf("key" to counter)))
+ .POST(JsonUtils.jacksonBodyPublisher(mapOf("key" to counter)))
  .headers("Content-Type", "application/json")
  .build()
 
- val response = client.send(req, Utils.jacksonBodyHandler())
+ val response = client.send(req, JsonUtils.jacksonBodyHandler())
 
  assertThat(response.statusCode()).isEqualTo(200)
  assertThat(response.headers().firstValue("content-type"))

tests/src/test/kotlin/dev/restate/e2e/Utils.kt

@@ -10,46 +10,18 @@
 package dev.restate.e2e
 
 import com.fasterxml.jackson.databind.JsonNode
-import com.fasterxml.jackson.databind.ObjectMapper
-import java.net.URI
-import java.net.http.HttpClient
-import java.net.http.HttpRequest
-import java.net.http.HttpResponse
-import java.nio.charset.StandardCharsets
+import dev.restate.e2e.utils.JsonUtils
 import org.assertj.core.api.Assertions.assertThat
 
 object Utils {
 
- private val objMapper = ObjectMapper()
- private val httpClient = HttpClient.newHttpClient()
-
- fun jacksonBodyHandler(): HttpResponse.BodyHandler<JsonNode> {
- return HttpResponse.BodyHandler {
- HttpResponse.BodySubscribers.mapping(
- HttpResponse.BodySubscribers.ofString(StandardCharsets.UTF_8), objMapper::readTree)
- }
- }
-
- fun jacksonBodyPublisher(value: Any): HttpRequest.BodyPublisher {
- return HttpRequest.BodyPublishers.ofString(objMapper.writeValueAsString(value))
- }
-
- fun postJsonRequest(uri: String, reqBody: Any): HttpResponse<JsonNode> {
- val req =
- HttpRequest.newBuilder(URI.create(uri))
- .headers("Content-Type", "application/json")
- .POST(jacksonBodyPublisher(reqBody))
- .build()
- return httpClient.send(req, jacksonBodyHandler())
- }
-
  fun doJsonRequestToService(
  restateEndpoint: String,
  service: String,
  method: String,
  reqBody: Any
  ): JsonNode {
- val res = postJsonRequest("${restateEndpoint}${service}/${method}", reqBody)
+ val res = JsonUtils.postJsonRequest("${restateEndpoint}${service}/${method}", reqBody)
  assertThat(res.statusCode()).isEqualTo(200)
  assertThat(res.headers().firstValue("content-type"))
  .get()

tests/src/test/kotlin/dev/restate/e2e/node/EmbeddedHandlerApiTest.kt

@@ -15,7 +15,6 @@ import dev.restate.e2e.Containers.EMBEDDED_HANDLER_SERVER_HOSTNAME
 import dev.restate.e2e.Containers.EMBEDDED_HANDLER_SERVER_PORT
 import dev.restate.e2e.Containers.HANDLER_API_COUNTER_SERVICE_NAME
 import dev.restate.e2e.Containers.nodeServicesContainer
-import dev.restate.e2e.Utils.postJsonRequest
 import dev.restate.e2e.utils.*
 import java.net.URL
 import java.util.UUID
@@ -55,7 +54,7 @@ class EmbeddedHandlerApiTest {
 
  for (i in 0..2) {
  val response =
- postJsonRequest(
+ JsonUtils.postJsonRequest(
  "http://localhost:${embeddedHandlerServerPort}/increment_counter_test",
  mapOf("id" to operationUuid, "input" to counterUuid))
  assertThat(response.statusCode()).isEqualTo(200)
@@ -64,7 +63,7 @@ class EmbeddedHandlerApiTest {
  }
 
  val response =
- postJsonRequest(
+ JsonUtils.postJsonRequest(
  "${httpEndpointURL}$HANDLER_API_COUNTER_SERVICE_NAME/get", mapOf("key" to counterUuid))
  assertThat(response.statusCode()).isEqualTo(200)
  assertThat(response.body().get("response").get("counter").asInt()).isEqualTo(1)
@@ -100,12 +99,11 @@ class EmbeddedHandlerApiTest {
  @InjectContainerPort(
  hostName = EMBEDDED_HANDLER_SERVER_HOSTNAME, port = EMBEDDED_HANDLER_SERVER_PORT)
  embeddedHandlerServerPort: Int,
- @InjectGrpcIngressURL httpEndpointURL: URL
  ) {
  val operationUuid = UUID.randomUUID().toString()
 
  val response =
- postJsonRequest(
+ JsonUtils.postJsonRequest(
  "http://localhost:${embeddedHandlerServerPort}/side_effect_and_awakeable",
  mapOf("id" to operationUuid, "itemsNumber" to 10))
  assertThat(response.statusCode()).isEqualTo(200)
@@ -122,12 +120,11 @@ class EmbeddedHandlerApiTest {
  @InjectContainerPort(
  hostName = EMBEDDED_HANDLER_SERVER_HOSTNAME, port = EMBEDDED_HANDLER_SERVER_PORT)
  embeddedHandlerServerPort: Int,
- @InjectGrpcIngressURL httpEndpointURL: URL
  ) {
  val operationUuid = UUID.randomUUID().toString()
 
  val response =
- postJsonRequest(
+ JsonUtils.postJsonRequest(
  "http://localhost:${embeddedHandlerServerPort}/consecutive_side_effects",
  mapOf("id" to operationUuid))
  assertThat(response.statusCode()).isEqualTo(200)
@@ -144,7 +141,7 @@ class EmbeddedHandlerApiTest {
 
  for (i in 0..2) {
  val response =
- postJsonRequest(
+ JsonUtils.postJsonRequest(
  "http://localhost:${embeddedHandlerServerPort}/${path}",
  mapOf("id" to operationUuid, "input" to counterUuid))
  assertThat(response.statusCode()).isEqualTo(200)
@@ -153,7 +150,7 @@ class EmbeddedHandlerApiTest {
  await untilAsserted
  {
  val response =
- postJsonRequest(
+ JsonUtils.postJsonRequest(
  "${httpEndpointURL}$HANDLER_API_COUNTER_SERVICE_NAME/get",
  mapOf("key" to counterUuid))
  assertThat(response.statusCode()).isEqualTo(200)