chore(observability): make kafka message handling distinct spans (#228)

Author: karl-run

build.gradle.kts

@@ -22,7 +22,7 @@ val kotlinVersion = "1.9.22"
 val testContainerVersion = "1.19.6"
 val ktfmtVersion = "0.44"
 val snappyJavaVersion = "1.1.10.5"
-
+val opentelemetryVersion = "2.3.0"
 
 plugins {
     id("application")
@@ -78,6 +78,7 @@ dependencies {
 
     implementation("ch.qos.logback:logback-classic:$logbackVersion")
     implementation("net.logstash.logback:logstash-logback-encoder:$logstashEncoderVersion")
+    implementation("io.opentelemetry.instrumentation:opentelemetry-instrumentation-annotations:$opentelemetryVersion")
 
     testImplementation("org.jetbrains.kotlin:kotlin-test:$kotlinVersion")
     testImplementation("org.amshove.kluent:kluent:$kluentVersion")

src/main/kotlin/no/nav/syfo/syfosmvarsel/Bootstrap.kt

@@ -5,6 +5,8 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule
 import com.fasterxml.jackson.module.kotlin.readValue
 import com.fasterxml.jackson.module.kotlin.registerKotlinModule
+import io.opentelemetry.api.trace.Span
+import io.opentelemetry.instrumentation.annotations.WithSpan
 import java.time.Duration
 import java.util.UUID
 import kotlinx.coroutines.CoroutineScope
@@ -28,6 +30,7 @@ import no.nav.syfo.syfosmvarsel.util.KafkaFactory.Companion.getBrukernotifikasjo
 import no.nav.syfo.syfosmvarsel.util.KafkaFactory.Companion.getKafkaStatusConsumerAiven
 import no.nav.syfo.syfosmvarsel.util.KafkaFactory.Companion.getNyKafkaAivenConsumer
 import no.nav.syfo.util.util.Unbounded
+import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.apache.kafka.clients.consumer.KafkaConsumer
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
@@ -94,7 +97,7 @@ fun createListener(
             log.error(
                 "En uhåndtert feil oppstod, applikasjonen restarter {}",
                 fields(e.loggingMeta),
-                e.cause
+                e.cause,
             )
         } finally {
             applicationState.alive = false
@@ -118,15 +121,15 @@ fun launchListeners(
             nyKafkaConsumerAiven,
             nySykmeldingService,
             avvistSykmeldingService,
-            environment
+            environment,
         )
     }
 
     createListener(applicationState) {
         blockingApplicationLogicStatusendringAiven(
             applicationState,
             statusendringService,
-            kafkaStatusConsumerAiven
+            kafkaStatusConsumerAiven,
         )
     }
 }
@@ -143,33 +146,46 @@ suspend fun blockingApplicationLogicNySykmelding(
             .poll(Duration.ofMillis(1000))
             .filterNot { it.value() == null }
             .forEach {
-                val receivedSykmelding: ReceivedSykmelding = objectMapper.readValue(it.value())
-
-                val loggingMeta =
-                    LoggingMeta(
-                        mottakId = receivedSykmelding.navLogId,
-                        orgNr = receivedSykmelding.legekontorOrgNr,
-                        msgId = receivedSykmelding.msgId,
-                        sykmeldingId = receivedSykmelding.sykmelding.id,
-                    )
-                wrapExceptions(loggingMeta) {
-                    when (it.topic()) {
-                        environment.avvistSykmeldingTopicAiven ->
-                            avvistSykmeldingService.opprettVarselForAvvisteSykmeldinger(
-                                receivedSykmelding,
-                                loggingMeta
-                            )
-                        else ->
-                            nySykmeldingService.opprettVarselForNySykmelding(
-                                receivedSykmelding,
-                                loggingMeta
-                            )
-                    }
-                }
+                handleNySykmelding(it, environment, avvistSykmeldingService, nySykmeldingService)
             }
     }
 }
 
+@WithSpan
+private suspend fun handleNySykmelding(
+    message: ConsumerRecord<String, String>,
+    environment: Environment,
+    avvistSykmeldingService: AvvistSykmeldingService,
+    nySykmeldingService: NySykmeldingService
+) {
+    val receivedSykmelding: ReceivedSykmelding = objectMapper.readValue(message.value())
+
+    val currentSpan = Span.current()
+    currentSpan.setAttribute("sykmeldingId", receivedSykmelding.sykmelding.id)
+
+    val loggingMeta =
+        LoggingMeta(
+            mottakId = receivedSykmelding.navLogId,
+            orgNr = receivedSykmelding.legekontorOrgNr,
+            msgId = receivedSykmelding.msgId,
+            sykmeldingId = receivedSykmelding.sykmelding.id,
+        )
+    wrapExceptions(loggingMeta) {
+        when (message.topic()) {
+            environment.avvistSykmeldingTopicAiven ->
+                avvistSykmeldingService.opprettVarselForAvvisteSykmeldinger(
+                    receivedSykmelding,
+                    loggingMeta,
+                )
+            else ->
+                nySykmeldingService.opprettVarselForNySykmelding(
+                    receivedSykmelding,
+                    loggingMeta,
+                )
+        }
+    }
+}
+
 fun blockingApplicationLogicStatusendringAiven(
     applicationState: ApplicationState,
     statusendringService: StatusendringService,
@@ -180,21 +196,31 @@ fun blockingApplicationLogicStatusendringAiven(
             .poll(Duration.ofMillis(1000))
             .filter { it.value() != null }
             .filter { it.key().erUuid() }
-            .forEach {
-                val sykmeldingStatusKafkaMessageDTO: SykmeldingStatusKafkaMessageDTO = it.value()
-                try {
-                    log.info(
-                        "Mottatt statusmelding fra aiven ${sykmeldingStatusKafkaMessageDTO.kafkaMetadata.sykmeldingId}"
-                    )
-                    statusendringService.handterStatusendring(sykmeldingStatusKafkaMessageDTO)
-                } catch (e: Exception) {
-                    log.error(
-                        "Noe gikk galt ved behandling av statusendring fra aiven for sykmelding med id {}",
-                        sykmeldingStatusKafkaMessageDTO.kafkaMetadata.sykmeldingId
-                    )
-                    throw e
-                }
-            }
+            .forEach { handleStatusEndring(it, statusendringService) }
+    }
+}
+
+@WithSpan
+private fun handleStatusEndring(
+    it: ConsumerRecord<String, SykmeldingStatusKafkaMessageDTO>,
+    statusendringService: StatusendringService
+) {
+    val sykmeldingStatusKafkaMessageDTO: SykmeldingStatusKafkaMessageDTO = it.value()
+
+    val currentSpan = Span.current()
+    currentSpan.setAttribute("sykmeldingId", sykmeldingStatusKafkaMessageDTO.event.sykmeldingId)
+
+    try {
+        log.info(
+            "Mottatt statusmelding fra aiven ${sykmeldingStatusKafkaMessageDTO.kafkaMetadata.sykmeldingId}",
+        )
+        statusendringService.handterStatusendring(sykmeldingStatusKafkaMessageDTO)
+    } catch (e: Exception) {
+        log.error(
+            "Noe gikk galt ved behandling av statusendring fra aiven for sykmelding med id {}",
+            sykmeldingStatusKafkaMessageDTO.kafkaMetadata.sykmeldingId,
+        )
+        throw e
     }
 }