Lagt til mer logging og metrics

Author: naviktthomas

apps/min-side-varsler/nais/nais-dev.yaml

@@ -19,7 +19,7 @@ spec:
       memory: 256Mi
   env:
     - name: KAFKA_BEKREFTELSE_STREAM_SUFFIX
-      value: bekreftelse-beta-v1
+      value: bekreftelse-beta-v2
     - name: KAFKA_VARSEL_HENDELSE_STREAM_SUFFIX
       value: varsel-hendelser-beta-v1
     - name: KAFKA_PAW_ARBEIDSOKERPERIODE_TOPIC

apps/min-side-varsler/src/main/kotlin/no/nav/paw/arbeidssoekerregisteret/context/ApplicationContext.kt

@@ -61,7 +61,12 @@ data class ApplicationContext(
 
             val varselMeldingBygger = VarselMeldingBygger(serverConfig.runtimeEnvironment, minSideVarselConfig)
 
-            val varselService = VarselService(periodeRepository, varselRepository, varselMeldingBygger)
+            val varselService = VarselService(
+                meterRegistry = prometheusMeterRegistry,
+                periodeRepository = periodeRepository,
+                varselRepository = varselRepository,
+                varselMeldingBygger = varselMeldingBygger
+            )
 
             val bekreftelseKafkaStreams = buildBekreftelseKafkaStreams(
                 serverConfig = serverConfig,

apps/min-side-varsler/src/main/kotlin/no/nav/paw/arbeidssoekerregisteret/repository/PeriodeRepository.kt

@@ -20,6 +20,11 @@ class PeriodeRepository {
         PeriodeTable.selectAll().count()
     }
 
+    fun findAll(): List<PeriodeRow> = transaction {
+        PeriodeTable.selectAll()
+            .map { it.asPeriodeRow() }
+    }
+
     fun findByPeriodeId(periodeId: UUID): PeriodeRow? = transaction {
         PeriodeTable.selectAll()
             .where { PeriodeTable.periodeId eq periodeId }

apps/min-side-varsler/src/main/kotlin/no/nav/paw/arbeidssoekerregisteret/repository/VarselRepository.kt

@@ -20,6 +20,11 @@ class VarselRepository {
         VarselTable.selectAll().count()
     }
 
+    fun findAll(): List<VarselRow> = transaction {
+        VarselTable.selectAll()
+            .map { it.asVarselRow() }
+    }
+
     fun findByVarselId(varselId: UUID): VarselRow? = transaction {
         VarselTable.selectAll()
             .where { VarselTable.varselId eq varselId }

apps/min-side-varsler/src/main/kotlin/no/nav/paw/arbeidssoekerregisteret/service/VarselService.kt

@@ -1,5 +1,6 @@
 package no.nav.paw.arbeidssoekerregisteret.service
 
+import io.micrometer.core.instrument.MeterRegistry
 import no.nav.paw.arbeidssoekerregisteret.model.OppgaveMelding
 import no.nav.paw.arbeidssoekerregisteret.model.VarselHendelse
 import no.nav.paw.arbeidssoekerregisteret.model.VarselKilde
@@ -11,6 +12,8 @@ import no.nav.paw.arbeidssoekerregisteret.model.asUpdatePeriodeRow
 import no.nav.paw.arbeidssoekerregisteret.model.asUpdateVarselRow
 import no.nav.paw.arbeidssoekerregisteret.repository.PeriodeRepository
 import no.nav.paw.arbeidssoekerregisteret.repository.VarselRepository
+import no.nav.paw.arbeidssoekerregisteret.utils.bekreftelseHendelseCounter
+import no.nav.paw.arbeidssoekerregisteret.utils.periodeCounter
 import no.nav.paw.arbeidssokerregisteret.api.v1.Periode
 import no.nav.paw.bekreftelse.internehendelser.BekreftelseHendelse
 import no.nav.paw.bekreftelse.internehendelser.BekreftelseMeldingMottatt
@@ -21,6 +24,7 @@ import org.jetbrains.exposed.sql.transactions.transaction
 import java.util.*
 
 class VarselService(
+    private val meterRegistry: MeterRegistry,
     private val periodeRepository: PeriodeRepository,
     private val varselRepository: VarselRepository,
     private val varselMeldingBygger: VarselMeldingBygger
@@ -30,11 +34,13 @@ class VarselService(
     fun mottaPeriode(periode: Periode) = transaction {
         val periodeRow = periodeRepository.findByPeriodeId(periode.id)
         if (periodeRow != null) {
-            logger.warn("Oppdaterer innslag for periode")
+            logger.warn("Oppdaterer innslag for periode {}", periode.id)
+            meterRegistry.periodeCounter("update", periode)
             val updatePeriodeRow = periode.asUpdatePeriodeRow()
             periodeRepository.update(updatePeriodeRow)
         } else {
-            logger.debug("Oppretter innslag for periode")
+            logger.debug("Oppretter innslag for periode {}", periode.id)
+            meterRegistry.periodeCounter("insert", periode)
             val insertPeriodeRow = periode.asInsertPeriodeRow()
             periodeRepository.insert(insertPeriodeRow)
         }
@@ -48,15 +54,18 @@ class VarselService(
                     val varselRow = varselRepository.findByVarselId(hendelse.bekreftelseId)
                     if (varselRow != null) {
                         logger.warn(
-                            "Ignorerer bekreftelse-hendelse av type {} siden varsel allerede finnes",
+                            "Ignorerer hendelse {} siden varsel allerede finnes",
                             hendelse.hendelseType
                         )
+                        meterRegistry.bekreftelseHendelseCounter("ignore", hendelse)
                         emptyList()
                     } else {
                         logger.debug(
-                            "Bestiller og oppretter varsel basert på bekreftelse-hendelse av type {}",
-                            hendelse.hendelseType
+                            "Bestiller og oppretter varsel for hendelse {} og periode {}",
+                            hendelse.hendelseType,
+                            hendelse.periodeId
                         )
+                        meterRegistry.bekreftelseHendelseCounter("insert", hendelse)
                         val insertVarselRow = hendelse.asInsertVarselRow(
                             varselKilde = VarselKilde.BEKREFTELSE_TILGJENGELIG,
                             varselType = VarselType.OPPGAVE,
@@ -72,7 +81,12 @@ class VarselService(
                         )
                     }
                 } else {
-                    logger.warn("Fant ingen aktiv periode for bekreftelse-hendelse av type {}", hendelse.hendelseType)
+                    logger.warn(
+                        "Fant ingen aktiv periode for hendelse {} og periode {}",
+                        hendelse.hendelseType,
+                        hendelse.periodeId
+                    )
+                    meterRegistry.bekreftelseHendelseCounter("fail", hendelse)
                     emptyList()
                 }
             }
@@ -81,32 +95,56 @@ class VarselService(
                 val varselRow = varselRepository.findByVarselId(hendelse.bekreftelseId)
                 if (varselRow != null) {
                     logger.debug(
-                        "Avslutter og sletter varsel for varsel på grunn av bekreftelse-hendelse av type {}",
+                        "Avslutter og sletter varsel for varsel for hendelse {}",
                         hendelse.hendelseType
                     )
+                    meterRegistry.bekreftelseHendelseCounter("delete", hendelse)
                     varselRepository.deleteByVarselId(hendelse.bekreftelseId)
-
                     listOf(varselMeldingBygger.avsluttOppgave(hendelse.bekreftelseId))
                 } else {
-                    logger.warn("Fant ingen varsel for bekreftelse-hendelse av type {}", hendelse.hendelseType)
+                    logger.warn("Fant ingen varsel for hendelse {}", hendelse.hendelseType)
+                    meterRegistry.bekreftelseHendelseCounter("fail", hendelse)
                     emptyList()
                 }
             }
 
             is PeriodeAvsluttet -> {
                 logger.debug(
-                    "Avlutter og sletter alle varsler på grunn av bekreftelse-hendelse av type {}",
-                    hendelse.hendelseType
+                    "Avlutter og sletter alle varsler for hendelse {} og periode {}",
+                    hendelse.hendelseType,
+                    hendelse.periodeId
                 )
+                val periodeRow = periodeRepository.findByPeriodeId(hendelse.periodeId)
+                if (periodeRow == null) {
+                    logger.warn(
+                        "Fant ingen periode for hendelse {} og periode {}",
+                        hendelse.hendelseType,
+                        hendelse.periodeId
+                    )
+                    meterRegistry.bekreftelseHendelseCounter("fail", hendelse)
+                } else {
+                    meterRegistry.bekreftelseHendelseCounter("delete", hendelse)
+                    periodeRepository.deleteByPeriodeId(hendelse.periodeId)
+                }
+
                 val varselRows = varselRepository.findByPeriodeId(hendelse.periodeId)
-                varselRepository.deleteByPeriodeId(hendelse.periodeId)
-                periodeRepository.deleteByPeriodeId(hendelse.periodeId)
+                if (varselRows.isEmpty()) {
+                    logger.warn(
+                        "Fant ingen varsler for hendelse {} og periode {}",
+                        hendelse.hendelseType,
+                        hendelse.periodeId
+                    )
+                    emptyList()
+                } else {
+                    varselRepository.deleteByPeriodeId(hendelse.periodeId)
+                    varselRows.map { varselMeldingBygger.avsluttOppgave(it.varselId) }
+                }
 
-                varselRows.map { varselMeldingBygger.avsluttOppgave(it.varselId) }
             }
 
             else -> {
-                logger.debug("Ignorerer bekreftelse-hendelse av type {}", hendelse.hendelseType)
+                logger.debug("Ignorerer hendelse {}", hendelse.hendelseType)
+                meterRegistry.bekreftelseHendelseCounter("ignore", hendelse)
                 emptyList()
             }
         }
@@ -116,19 +154,26 @@ class VarselService(
         val varselRow = varselRepository.findByVarselId(UUID.fromString(hendelse.varselId))
         if (varselRow != null) {
             if (varselRow.hendelseTimestamp.isBefore(hendelse.tidspunkt)) {
-                logger.debug("Oppdaterer varsel basert på varsel-hendelse med status {}", hendelse.status)
+                logger.debug(
+                    "Oppdaterer varsel for hendelse {} med type {} og status {}",
+                    VarselHendelse::class.java.simpleName,
+                    hendelse.varseltype,
+                    hendelse.status
+                )
                 val updateVarselRow = hendelse.asUpdateVarselRow()
                 varselRepository.update(updateVarselRow)
             } else {
                 logger.warn(
-                    "Ignorerer varsel-hendelse med status {} siden lagret varsel {} er nyere enn hendelse {}",
+                    "Ignorerer hendelse {} med type {} og status {} siden lagret varsel {} er nyere enn hendelse {}",
+                    VarselHendelse::class.java.simpleName,
+                    hendelse.varseltype,
                     hendelse.status,
                     varselRow.hendelseTimestamp,
                     hendelse.tidspunkt
                 )
             }
         } else {
-            logger.warn("Fant ikke lagret varsel for varsel-hendelse")
+            logger.warn("Fant ikke lagret varsel for hendelse {}", VarselHendelse::class.java.simpleName)
         }
     }
 }

apps/min-side-varsler/src/main/kotlin/no/nav/paw/arbeidssoekerregisteret/topology/Topology.kt

@@ -30,13 +30,13 @@ fun StreamsBuilder.bekreftelseKafkaTopology(
 ): StreamsBuilder {
     with(kafkaTopicsConfig) {
         stream<Long, Periode>(periodeTopic)
-            .peek { _, periode -> meterRegistry.periodeCounter(periode) }
+            .peek { _, periode -> meterRegistry.periodeCounter("read", periode) }
             .foreach { _, periode ->
                 varselService.mottaPeriode(periode)
             }
 
         stream(bekreftelseHendelseTopic, Consumed.with(Serdes.Long(), BekreftelseHendelseSerde()))
-            .peek { _, hendelse -> meterRegistry.bekreftelseHendelseCounter(hendelse) }
+            .peek { _, hendelse -> meterRegistry.bekreftelseHendelseCounter("read", hendelse) }
             .flatMapValues { _, hendelse ->
                 varselService.mottaBekreftelseHendelse(hendelse)
             }

apps/min-side-varsler/src/main/kotlin/no/nav/paw/arbeidssoekerregisteret/utils/Metrics.kt

@@ -17,30 +17,34 @@ import no.nav.paw.config.env.namespaceOrDefaultForLocal
 private const val METRIC_PREFIX = "paw_min_side_varsler"
 
 fun MeterRegistry.periodeCounter(
+    action: String,
     periode: Periode
 ) {
     counter(
         "${METRIC_PREFIX}_antall_operasjoner",
         Tags.of(
+            Tag.of("type", "periode"),
             Tag.of("source", "kafka"),
             Tag.of("target", "database"),
-            Tag.of("action", "read"),
+            Tag.of("action", action),
             Tag.of("event.topic", "paw.arbeidssokerperioder-v1"),
-            Tag.of("event.name", "periode"),
-            Tag.of("event.status", if (periode.avsluttet == null) "aapen" else "lukket")
+            Tag.of("event.name", periode::class.java.simpleName),
+            Tag.of("event.type", if (periode.avsluttet == null) "periode.startet" else "periode.avsluttet")
         )
     ).increment()
 }
 
 fun MeterRegistry.bekreftelseHendelseCounter(
+    action: String,
     hendelse: BekreftelseHendelse
 ) {
     counter(
         "${METRIC_PREFIX}_antall_operasjoner",
         Tags.of(
+            Tag.of("type", "bekreftelse"),
             Tag.of("source", "kafka"),
             Tag.of("target", "database"),
-            Tag.of("action", "read"),
+            Tag.of("action", action),
             Tag.of("event.topic", "paw.arbeidssoker-bekreftelse-hendelseslogg-v1"),
             Tag.of("event.name", hendelse::class.java.simpleName),
             Tag.of("event.type", hendelse.hendelseType)
@@ -55,6 +59,7 @@ fun MeterRegistry.varselCounter(
     counter(
         "${METRIC_PREFIX}_antall_operasjoner",
         Tags.of(
+            Tag.of("type", "varsel"),
             Tag.of("source", "kafka"),
             Tag.of("target", "kafka"),
             Tag.of("action", "write"),
@@ -75,6 +80,7 @@ fun MeterRegistry.varselHendelseCounter(
     counter(
         "${METRIC_PREFIX}_antall_operasjoner",
         Tags.of(
+            Tag.of("type", "varsel.hendelse"),
             Tag.of("source", "kafka"),
             Tag.of("target", "database"),
             Tag.of("action", "read"),

apps/min-side-varsler/src/test/kotlin/no/nav/paw/arbeidssoekerregisteret/context/TestContext.kt

@@ -8,18 +8,18 @@ import io.confluent.kafka.serializers.KafkaAvroSerializerConfig
 import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
 import io.micrometer.prometheusmetrics.PrometheusConfig
 import io.micrometer.prometheusmetrics.PrometheusMeterRegistry
-import no.nav.paw.arbeidssoekerregisteret.utils.VarselHendelseJsonSerde
-import no.nav.paw.arbeidssoekerregisteret.topology.bekreftelseKafkaTopology
-import no.nav.paw.arbeidssoekerregisteret.topology.varselHendelserKafkaTopology
-import no.nav.paw.arbeidssoekerregisteret.model.VarselMeldingBygger
 import no.nav.paw.arbeidssoekerregisteret.config.KafkaTopologyConfig
 import no.nav.paw.arbeidssoekerregisteret.config.MIN_SIDE_VARSEL_CONFIG
 import no.nav.paw.arbeidssoekerregisteret.config.MinSideVarselConfig
 import no.nav.paw.arbeidssoekerregisteret.model.VarselHendelse
+import no.nav.paw.arbeidssoekerregisteret.model.VarselMeldingBygger
 import no.nav.paw.arbeidssoekerregisteret.repository.PeriodeRepository
 import no.nav.paw.arbeidssoekerregisteret.repository.VarselRepository
 import no.nav.paw.arbeidssoekerregisteret.service.VarselService
 import no.nav.paw.arbeidssoekerregisteret.testdata.KafkaKeyContext
+import no.nav.paw.arbeidssoekerregisteret.topology.bekreftelseKafkaTopology
+import no.nav.paw.arbeidssoekerregisteret.topology.varselHendelserKafkaTopology
+import no.nav.paw.arbeidssoekerregisteret.utils.VarselHendelseJsonSerde
 import no.nav.paw.arbeidssokerregisteret.api.v1.Periode
 import no.nav.paw.bekreftelse.internehendelser.BekreftelseHendelse
 import no.nav.paw.bekreftelse.internehendelser.BekreftelseHendelseSerde
@@ -52,6 +52,8 @@ data class TestContext(
     val periodeRepository: PeriodeRepository,
     val varselRepository: VarselRepository,
     val varselService: VarselService,
+    val bekreftelseTopologyTestDriver: TopologyTestDriver,
+    val varselTopologyTestDriver: TopologyTestDriver,
     val periodeTopic: TestInputTopic<Long, Periode>,
     val bekreftelseHendelseTopic: TestInputTopic<Long, BekreftelseHendelse>,
     val tmsVarselTopic: TestOutputTopic<String, String>,
@@ -91,7 +93,12 @@ data class TestContext(
             val prometheusMeterRegistry = PrometheusMeterRegistry(PrometheusConfig.DEFAULT)
             val periodeRepository = PeriodeRepository()
             val varselRepository = VarselRepository()
-            val varselService = VarselService(periodeRepository, varselRepository, varselMeldingBygger)
+            val varselService = VarselService(
+                meterRegistry = prometheusMeterRegistry,
+                periodeRepository = periodeRepository,
+                varselRepository = varselRepository,
+                varselMeldingBygger = varselMeldingBygger
+            )
             val bekreftelseTopology = StreamsBuilder()
                 .bekreftelseKafkaTopology(
                     runtimeEnvironment = runtimeEnvironment,
@@ -143,6 +150,8 @@ data class TestContext(
                 periodeRepository = periodeRepository,
                 varselRepository = varselRepository,
                 varselService = varselService,
+                bekreftelseTopologyTestDriver = bekreftelseTopologyTestDriver,
+                varselTopologyTestDriver = varselTopologyTestDriver,
                 periodeTopic = periodeInputTopic,
                 bekreftelseHendelseTopic = bekreftelseHendelseTopic,
                 tmsVarselTopic = tmsVarselTopic,