La til tester for felles helsesjekk-lib

navikt · Sep 12, 2024 · 86b8acb · 86b8acb
1 parent 34bc1e7
commit 86b8acb
Show file tree

Hide file tree

Showing 10 changed files with 383 additions and 103 deletions.
diff --git a/lib/error-handling/build.gradle.kts b/lib/error-handling/build.gradle.kts
@@ -9,6 +9,14 @@ dependencies {
 
     // Test
     testImplementation(testLibs.bundles.withUnitTesting)
+    testImplementation(ktorServer.testJvm)
+    testImplementation(ktorServer.contentNegotiation)
+    testImplementation(ktorServer.statusPages)
+    testImplementation(ktor.serializationJackson)
+    testImplementation(ktorClient.contentNegotiation)
+    testImplementation(ktorServer.core)
+    testImplementation(orgApacheKafka.kafkaStreams)
+    testImplementation(loggingLibs.logbackClassic)
 }
 
 tasks.withType<Test>().configureEach {

diff --git a/...error-handling/src/main/kotlin/no/nav/paw/health/listener/KafkaHealthIndicatorListener.kt b/...error-handling/src/main/kotlin/no/nav/paw/health/listener/KafkaHealthIndicatorListener.kt
diff --git a/lib/error-handling/src/main/kotlin/no/nav/paw/health/listener/KafkaStreamsStatusListener.kt b/lib/error-handling/src/main/kotlin/no/nav/paw/health/listener/KafkaStreamsStatusListener.kt
@@ -0,0 +1,59 @@
+package no.nav.paw.health.listener
+
+import no.nav.paw.health.model.LivenessHealthIndicator
+import no.nav.paw.health.model.ReadinessHealthIndicator
+import org.apache.kafka.streams.KafkaStreams
+import org.slf4j.LoggerFactory
+
+private val logger = LoggerFactory.getLogger("paw.application.health.kafka")
+
+fun KafkaStreams.withHealthIndicatorStateListener(
+    livenessIndicator: LivenessHealthIndicator,
+    readinessIndicator: ReadinessHealthIndicator
+) = KafkaStreams.StateListener { newState, previousState ->
+    when (newState) {
+        KafkaStreams.State.CREATED -> {
+            readinessIndicator.setUnhealthy()
+            livenessIndicator.setHealthy()
+        }
+
+        KafkaStreams.State.RUNNING -> {
+            readinessIndicator.setHealthy()
+            livenessIndicator.setHealthy()
+        }
+
+        KafkaStreams.State.REBALANCING -> {
+            readinessIndicator.setHealthy()
+            livenessIndicator.setHealthy()
+        }
+
+        KafkaStreams.State.PENDING_ERROR -> {
+            readinessIndicator.setUnhealthy()
+            livenessIndicator.setHealthy()
+        }
+
+        KafkaStreams.State.ERROR -> {
+            readinessIndicator.setUnhealthy()
+            livenessIndicator.setUnhealthy()
+        }
+
+        KafkaStreams.State.PENDING_SHUTDOWN -> {
+            readinessIndicator.setUnhealthy()
+            livenessIndicator.setUnhealthy()
+        }
+
+        KafkaStreams.State.NOT_RUNNING -> {
+            readinessIndicator.setUnhealthy()
+            livenessIndicator.setUnhealthy()
+        }
+
+        else -> {
+            readinessIndicator.setUnknown()
+            livenessIndicator.setUnknown()
+        }
+    }
+
+    logger.debug("Kafka Streams state endret seg ${previousState.name} -> ${newState.name}")
+    logger.info("Kafka Streams liveness er ${livenessIndicator.getStatus().value}")
+    logger.info("Kafka Streams readiness er ${readinessIndicator.getStatus().value}")
+}
diff --git a/lib/error-handling/src/main/kotlin/no/nav/paw/health/model/Health.kt b/lib/error-handling/src/main/kotlin/no/nav/paw/health/model/Health.kt
@@ -8,30 +8,50 @@ enum class HealthStatus(val value: String) {
     UNHEALTHY("UNHEALTHY"),
 }
 
-interface HealthIndicator {
-    fun setUnknown()
-    fun setHealthy()
-    fun setUnhealthy()
-    fun getStatus(): HealthStatus
-}
-
-class StandardHealthIndicator(initialStatus: HealthStatus) : HealthIndicator {
+open class HealthIndicator(initialStatus: HealthStatus) {
 
     private val status = AtomicReference(initialStatus)
 
-    override fun setUnknown() {
+    fun setUnknown() {
         status.set(HealthStatus.UNKNOWN)
     }
 
-    override fun setHealthy() {
+    fun setHealthy() {
         status.set(HealthStatus.HEALTHY)
     }
 
-    override fun setUnhealthy() {
+    fun setUnhealthy() {
         status.set(HealthStatus.UNHEALTHY)
     }
 
-    override fun getStatus(): HealthStatus {
+    fun getStatus(): HealthStatus {
         return status.get()
     }
-}
+
+    override fun equals(other: Any?): Boolean {
+        if (this === other) return true
+        other as HealthIndicator
+        return status == other.status
+    }
+
+    override fun hashCode(): Int {
+        return status.hashCode()
+    }
+}
+
+typealias HealthIndicatorList = MutableList<HealthIndicator>
+
+fun HealthIndicatorList.getAggregatedStatus(): HealthStatus {
+    return if (this.isEmpty()) {
+        HealthStatus.UNKNOWN
+    } else if (this.all { it.getStatus() == HealthStatus.HEALTHY }) {
+        HealthStatus.HEALTHY
+    } else if (this.any { it.getStatus() == HealthStatus.UNHEALTHY }) {
+        HealthStatus.UNHEALTHY
+    } else {
+        HealthStatus.UNKNOWN
+    }
+}
+
+class ReadinessHealthIndicator(initialStatus: HealthStatus = HealthStatus.UNKNOWN) : HealthIndicator(initialStatus)
+class LivenessHealthIndicator(initialStatus: HealthStatus = HealthStatus.HEALTHY) : HealthIndicator(initialStatus)
diff --git a/lib/error-handling/src/main/kotlin/no/nav/paw/health/repository/HealthIndicatorRepository.kt b/lib/error-handling/src/main/kotlin/no/nav/paw/health/repository/HealthIndicatorRepository.kt
@@ -0,0 +1,29 @@
+package no.nav.paw.health.repository
+
+import no.nav.paw.health.model.HealthIndicatorList
+import no.nav.paw.health.model.LivenessHealthIndicator
+import no.nav.paw.health.model.ReadinessHealthIndicator
+
+class HealthIndicatorRepository {
+
+    private val readinessIndicators: HealthIndicatorList = mutableListOf()
+    private val livenessIndicators: HealthIndicatorList = mutableListOf()
+
+    fun addReadinessIndicator(healthIndicator: ReadinessHealthIndicator): ReadinessHealthIndicator {
+        readinessIndicators.add(healthIndicator)
+        return healthIndicator
+    }
+
+    fun addLivenessIndicator(healthIndicator: LivenessHealthIndicator): LivenessHealthIndicator {
+        livenessIndicators.add(healthIndicator)
+        return healthIndicator
+    }
+
+    fun getReadinessIndicators(): HealthIndicatorList {
+        return readinessIndicators
+    }
+
+    fun getLivenessIndicators(): HealthIndicatorList {
+        return livenessIndicators
+    }
+}
diff --git a/lib/error-handling/src/main/kotlin/no/nav/paw/health/route/HealthRoutes.kt b/lib/error-handling/src/main/kotlin/no/nav/paw/health/route/HealthRoutes.kt
@@ -7,14 +7,16 @@ import io.ktor.server.response.respondText
 import io.ktor.server.routing.Route
 import io.ktor.server.routing.get
 import no.nav.paw.health.model.HealthStatus
-import no.nav.paw.health.service.HealthIndicatorService
+import no.nav.paw.health.model.getAggregatedStatus
+import no.nav.paw.health.repository.HealthIndicatorRepository
 
 fun Route.healthRoutes(
-    healthIndicatorService: HealthIndicatorService,
+    healthIndicatorRepository: HealthIndicatorRepository,
 ) {
 
     get("/internal/isAlive") {
-        when (val status = healthIndicatorService.getLivenessStatus()) {
+        val livenessIndicators = healthIndicatorRepository.getLivenessIndicators()
+        when (val status = livenessIndicators.getAggregatedStatus()) {
             HealthStatus.HEALTHY -> call.respondText(
                 ContentType.Text.Plain,
                 HttpStatusCode.OK
@@ -28,7 +30,8 @@ fun Route.healthRoutes(
     }
 
     get("/internal/isReady") {
-        when (val status = healthIndicatorService.getReadinessStatus()) {
+        val readinessIndicators = healthIndicatorRepository.getReadinessIndicators()
+        when (val status = readinessIndicators.getAggregatedStatus()) {
             HealthStatus.HEALTHY -> call.respondText(
                 ContentType.Text.Plain,
                 HttpStatusCode.OK

diff --git a/lib/error-handling/src/main/kotlin/no/nav/paw/health/service/HealthIndicatorService.kt b/lib/error-handling/src/main/kotlin/no/nav/paw/health/service/HealthIndicatorService.kt
diff --git a/...ror-handling/src/test/kotlin/no/nav/paw/health/listener/KafkaStreamsStatusListenerTest.kt b/...ror-handling/src/test/kotlin/no/nav/paw/health/listener/KafkaStreamsStatusListenerTest.kt
@@ -0,0 +1,62 @@
+package no.nav.paw.health.listener
+
+import io.kotest.core.spec.style.FreeSpec
+import io.kotest.matchers.shouldBe
+import io.mockk.mockkClass
+import no.nav.paw.health.model.HealthStatus
+import no.nav.paw.health.model.LivenessHealthIndicator
+import no.nav.paw.health.model.ReadinessHealthIndicator
+import no.nav.paw.health.model.getAggregatedStatus
+import no.nav.paw.health.repository.HealthIndicatorRepository
+import org.apache.kafka.streams.KafkaStreams
+
+class KafkaStreamsStatusListenerTest : FreeSpec({
+
+    "Kafka Streams Status Listener skal returnere korrekt helsesjekk-status" {
+        val kafkaStreams = mockkClass(KafkaStreams::class)
+        val healthIndicatorRepository = HealthIndicatorRepository()
+
+        val liveness = healthIndicatorRepository.addLivenessIndicator(LivenessHealthIndicator())
+        val readiness = healthIndicatorRepository.addReadinessIndicator(ReadinessHealthIndicator())
+
+        val listener = kafkaStreams.withHealthIndicatorStateListener(liveness, readiness)
+
+        healthIndicatorRepository.getReadinessIndicators().getAggregatedStatus() shouldBe HealthStatus.UNKNOWN
+        healthIndicatorRepository.getLivenessIndicators().getAggregatedStatus() shouldBe HealthStatus.HEALTHY
+
+        listener.onChange(KafkaStreams.State.CREATED, KafkaStreams.State.CREATED)
+
+        healthIndicatorRepository.getReadinessIndicators().getAggregatedStatus() shouldBe HealthStatus.UNHEALTHY
+        healthIndicatorRepository.getLivenessIndicators().getAggregatedStatus() shouldBe HealthStatus.HEALTHY
+
+        listener.onChange(KafkaStreams.State.RUNNING, KafkaStreams.State.RUNNING)
+
+        healthIndicatorRepository.getReadinessIndicators().getAggregatedStatus() shouldBe HealthStatus.HEALTHY
+        healthIndicatorRepository.getLivenessIndicators().getAggregatedStatus() shouldBe HealthStatus.HEALTHY
+
+        listener.onChange(KafkaStreams.State.REBALANCING, KafkaStreams.State.REBALANCING)
+
+        healthIndicatorRepository.getReadinessIndicators().getAggregatedStatus() shouldBe HealthStatus.HEALTHY
+        healthIndicatorRepository.getLivenessIndicators().getAggregatedStatus() shouldBe HealthStatus.HEALTHY
+
+        listener.onChange(KafkaStreams.State.PENDING_ERROR, KafkaStreams.State.PENDING_ERROR)
+
+        healthIndicatorRepository.getReadinessIndicators().getAggregatedStatus() shouldBe HealthStatus.UNHEALTHY
+        healthIndicatorRepository.getLivenessIndicators().getAggregatedStatus() shouldBe HealthStatus.HEALTHY
+
+        listener.onChange(KafkaStreams.State.ERROR, KafkaStreams.State.ERROR)
+
+        healthIndicatorRepository.getReadinessIndicators().getAggregatedStatus() shouldBe HealthStatus.UNHEALTHY
+        healthIndicatorRepository.getLivenessIndicators().getAggregatedStatus() shouldBe HealthStatus.UNHEALTHY
+
+        listener.onChange(KafkaStreams.State.PENDING_SHUTDOWN, KafkaStreams.State.PENDING_SHUTDOWN)
+
+        healthIndicatorRepository.getReadinessIndicators().getAggregatedStatus() shouldBe HealthStatus.UNHEALTHY
+        healthIndicatorRepository.getLivenessIndicators().getAggregatedStatus() shouldBe HealthStatus.UNHEALTHY
+
+        listener.onChange(KafkaStreams.State.NOT_RUNNING, KafkaStreams.State.NOT_RUNNING)
+
+        healthIndicatorRepository.getReadinessIndicators().getAggregatedStatus() shouldBe HealthStatus.UNHEALTHY
+        healthIndicatorRepository.getLivenessIndicators().getAggregatedStatus() shouldBe HealthStatus.UNHEALTHY
+    }
+})