Videreutvikling av bekreftelse-api

apps/bekreftelse-api/build.gradle.kts

@@ -14,6 +14,7 @@ val image: String? by project
 dependencies {
     // Project
     implementation(project(":lib:hoplite-config"))
+    implementation(project(":lib:error-handling"))
     implementation(project(":lib:kafka-streams"))
     implementation(project(":lib:kafka-key-generator-client"))
     implementation(project(":domain:bekreftelse-interne-hendelser"))

apps/bekreftelse-api/nais/nais-dev.yaml

@@ -1,16 +1,16 @@
 apiVersion: nais.io/v1alpha1
 kind: Application
 metadata:
-  name: paw-rapportering-api
+  name: paw-arbeidssoeker-bekreftelse-api
   namespace: paw
   labels:
     team: paw
 spec:
   image: {{ image }}
   port: 8080
-  env:
-    - name: KAFKA_KEY_SCOPE
-      value: "api://dev-gcp.paw.paw-kafka-key-generator/.default"
+  replicas:
+    min: 1
+    max: 1
   resources:
     limits:
       memory: 1024Mi
@@ -26,9 +26,6 @@ spec:
       claims:
         extra:
           - NAVident
-  replicas:
-    min: 1
-    max: 1
   liveness:
     path: /internal/isAlive
     initialDelay: 10
@@ -43,4 +40,18 @@ spec:
       enabled: true
       runtime: java
   kafka:
-    pool: nav-dev
+    pool: {{ kafka }}
+    streams: true
+  accessPolicy:
+    inbound:
+      rules:
+        - application: tokenx-token-generator
+          namespace: aura
+        - application: azure-token-generator
+          namespace: aura
+    outbound:
+      rules:
+        - application: paw-kafka-key-generator
+          namespace: paw
+        - application: poao-tilgang
+          namespace: poao

apps/bekreftelse-api/nais/nais-prod.yaml

@@ -1,16 +1,16 @@
 apiVersion: nais.io/v1alpha1
 kind: Application
 metadata:
-  name: paw-rapportering-api
+  name: paw-arbeidssoeker-bekreftelse-api
   namespace: paw
   labels:
     team: paw
 spec:
   image: {{ image }}
   port: 8080
-  env:
-    - name: KAFKA_KEY_SCOPE
-      value: "api://prod-gcp.paw.paw-kafka-key-generator/.default"
+  replicas:
+    min: 1
+    max: 1
   resources:
     limits:
       memory: 1024Mi
@@ -26,9 +26,6 @@ spec:
       claims:
         extra:
           - NAVident
-  replicas:
-    min: 1
-    max: 1
   liveness:
     path: /internal/isAlive
     initialDelay: 10
@@ -43,4 +40,18 @@ spec:
       enabled: true
       runtime: java
   kafka:
-    pool: nav-prod
+    pool: {{ kafka }}
+    streams: true
+  accessPolicy:
+    inbound:
+      rules:
+        - application: tokenx-token-generator
+          namespace: aura
+        - application: azure-token-generator
+          namespace: aura
+    outbound:
+      rules:
+        - application: paw-kafka-key-generator
+          namespace: paw
+        - application: poao-tilgang
+          namespace: poao

apps/bekreftelse-api/src/main/kotlin/no/nav/paw/bekreftelse/api/Application.kt

@@ -5,48 +5,41 @@ import io.ktor.server.engine.addShutdownHook
 import io.ktor.server.engine.embeddedServer
 import io.ktor.server.netty.Netty
 import io.ktor.server.routing.routing
-import no.nav.paw.config.hoplite.loadNaisOrLocalConfiguration
-import no.nav.paw.config.kafka.KAFKA_CONFIG_WITH_SCHEME_REG
-import no.nav.paw.config.kafka.KAFKA_STREAMS_CONFIG_WITH_SCHEME_REG
-import no.nav.paw.config.kafka.KafkaConfig
-import no.nav.paw.kafkakeygenerator.auth.AzureM2MConfig
-import no.nav.paw.kafkakeygenerator.client.KafkaKeyConfig
 import no.nav.paw.bekreftelse.api.config.APPLICATION_CONFIG_FILE_NAME
 import no.nav.paw.bekreftelse.api.config.ApplicationConfig
+import no.nav.paw.bekreftelse.api.config.SERVER_CONFIG_FILE_NAME
+import no.nav.paw.bekreftelse.api.config.ServerConfig
 import no.nav.paw.bekreftelse.api.plugins.configureAuthentication
 import no.nav.paw.bekreftelse.api.plugins.configureHTTP
+import no.nav.paw.bekreftelse.api.plugins.configureKafka
 import no.nav.paw.bekreftelse.api.plugins.configureLogging
 import no.nav.paw.bekreftelse.api.plugins.configureMetrics
-import no.nav.paw.bekreftelse.api.plugins.configureOtel
 import no.nav.paw.bekreftelse.api.plugins.configureSerialization
-import no.nav.paw.bekreftelse.api.routes.healthRoutes
+import no.nav.paw.bekreftelse.api.plugins.configureTracing
 import no.nav.paw.bekreftelse.api.routes.bekreftelseRoutes
+import no.nav.paw.bekreftelse.api.routes.metricsRoutes
 import no.nav.paw.bekreftelse.api.routes.swaggerRoutes
+import no.nav.paw.config.hoplite.loadNaisOrLocalConfiguration
+import no.nav.paw.health.route.healthRoutes
 import org.slf4j.LoggerFactory
 
 fun main() {
-    val logger = LoggerFactory.getLogger("rapportering-api")
-    logger.info("Starter: ${ApplicationInfo.id}")
+    val logger = LoggerFactory.getLogger("no.nav.paw.logger.application")
 
     val applicationConfig = loadNaisOrLocalConfiguration<ApplicationConfig>(APPLICATION_CONFIG_FILE_NAME)
-    val kafkaConfig = loadNaisOrLocalConfiguration<KafkaConfig>(KAFKA_CONFIG_WITH_SCHEME_REG)
-    val kafkaStreamsConfig = loadNaisOrLocalConfiguration<KafkaConfig>(KAFKA_STREAMS_CONFIG_WITH_SCHEME_REG)
-    val azureM2MConfig = loadNaisOrLocalConfiguration<AzureM2MConfig>("azure_m2m_key_config.toml")
-    val kafkaKeyConfig = loadNaisOrLocalConfiguration<KafkaKeyConfig>("kafka_key_generator_client_config.toml")
+    val serverConfig = loadNaisOrLocalConfiguration<ServerConfig>(SERVER_CONFIG_FILE_NAME)
 
-    val dependencies = createDependencies(
-        applicationConfig,
-        kafkaConfig,
-        kafkaStreamsConfig,
-        azureM2MConfig,
-        kafkaKeyConfig
-    )
+    logger.info("Starter: ${applicationConfig.appId}")
 
-    embeddedServer(Netty, port = 8080) {
-        module(applicationConfig, dependencies)
-    }.apply {
-        addShutdownHook { stop(300, 300) }
-        start(wait = true)
+    val dependencies = createDependencies(applicationConfig)
+
+    with(serverConfig) {
+        embeddedServer(Netty, port = port) {
+            module(applicationConfig, dependencies)
+        }.apply {
+            addShutdownHook { stop(gracePeriodMillis, timeoutMillis) }
+            start(wait = true)
+        }
     }
 }
 
@@ -59,19 +52,19 @@ fun Application.module(
     configureAuthentication(applicationConfig.authProviders)
     configureLogging()
     configureSerialization()
-    configureOtel()
+    configureTracing()
+    if (!applicationConfig.brukMock) { // TODO Bruker mock for utvikling
+        configureKafka(applicationConfig, listOf(dependencies.bekreftelseKafkaStreams))
+    }
 
     routing {
-        healthRoutes(dependencies.prometheusMeterRegistry, dependencies.health)
+        healthRoutes(dependencies.healthIndicatorRepository)
+        metricsRoutes(dependencies.prometheusMeterRegistry)
         swaggerRoutes()
         bekreftelseRoutes(
-            kafkaKeyClient = dependencies.kafkaKeysClient,
-            bekreftelseStateStore = dependencies.bekreftelseStateStore,
-            stateStoreName = applicationConfig.bekreftelseStateStoreName,
-            kafkaStreams = dependencies.kafkaStreams,
-            httpClient = dependencies.httpClient,
-            bekreftelseProducer = dependencies.bekreftelseProducer,
-            autorisasjonService = dependencies.autorisasjonService
+            dependencies.kafkaKeysClient,
+            dependencies.bekreftelseService,
+            dependencies.autorisasjonService
         )
     }
 }

apps/bekreftelse-api/src/main/kotlin/no/nav/paw/bekreftelse/api/Dependencies.kt

@@ -1,104 +1,43 @@
 package no.nav.paw.bekreftelse.api
 
-import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
 import io.ktor.client.HttpClient
 import io.ktor.client.plugins.contentnegotiation.ContentNegotiation
 import io.ktor.serialization.jackson.jackson
 import io.micrometer.prometheusmetrics.PrometheusConfig
 import io.micrometer.prometheusmetrics.PrometheusMeterRegistry
-import no.nav.paw.config.kafka.KafkaConfig
-import no.nav.paw.config.kafka.streams.KafkaStreamsFactory
-import no.nav.paw.kafkakeygenerator.auth.AzureM2MConfig
-import no.nav.paw.kafkakeygenerator.auth.azureAdM2MTokenClient
-import no.nav.paw.kafkakeygenerator.client.KafkaKeyConfig
-import no.nav.paw.kafkakeygenerator.client.KafkaKeysClient
-import no.nav.paw.kafkakeygenerator.client.kafkaKeysClient
 import no.nav.paw.bekreftelse.api.config.ApplicationConfig
 import no.nav.paw.bekreftelse.api.kafka.BekreftelseProducer
-import no.nav.paw.bekreftelse.api.kafka.InternState
-import no.nav.paw.bekreftelse.api.kafka.InternStateSerde
-import no.nav.paw.bekreftelse.api.kafka.appTopology
+import no.nav.paw.bekreftelse.api.kafka.buildBekreftelseTopology
+import no.nav.paw.bekreftelse.api.plugins.buildKafkaStreams
 import no.nav.paw.bekreftelse.api.services.AutorisasjonService
+import no.nav.paw.bekreftelse.api.services.BekreftelseService
+import no.nav.paw.bekreftelse.api.services.BekreftelseServiceImpl
+import no.nav.paw.bekreftelse.api.services.BekreftelseServiceMock
+import no.nav.paw.health.repository.HealthIndicatorRepository
+import no.nav.paw.kafkakeygenerator.auth.azureAdM2MTokenClient
+import no.nav.paw.kafkakeygenerator.client.KafkaKeysClient
+import no.nav.paw.kafkakeygenerator.client.kafkaKeysClient
 import no.nav.poao_tilgang.client.PoaoTilgangCachedClient
 import no.nav.poao_tilgang.client.PoaoTilgangHttpClient
-import org.apache.kafka.common.serialization.Serdes
 import org.apache.kafka.streams.KafkaStreams
-import org.apache.kafka.streams.StoreQueryParameters
-import org.apache.kafka.streams.StreamsBuilder
-import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler
-import org.apache.kafka.streams.state.QueryableStoreTypes
-import org.apache.kafka.streams.state.ReadOnlyKeyValueStore
-import org.apache.kafka.streams.state.Stores
-import org.slf4j.LoggerFactory
 
-fun createDependencies(
-    applicationConfig: ApplicationConfig,
-    kafkaConfig: KafkaConfig,
-    kafkaStreamsConfig: KafkaConfig,
-    azureM2MConfig: AzureM2MConfig,
-    kafkaKeyConfig: KafkaKeyConfig
-): Dependencies {
-    val logger = LoggerFactory.getLogger("rapportering-api")
+fun createDependencies(applicationConfig: ApplicationConfig): Dependencies {
+    val azureM2MTokenClient = azureAdM2MTokenClient(applicationConfig.naisEnv, applicationConfig.azureM2M)
 
-    val azureM2MTokenClient = azureAdM2MTokenClient(applicationConfig.naisEnv, azureM2MConfig)
-
-    val kafkaKeysClient = kafkaKeysClient(kafkaKeyConfig) {
-        azureM2MTokenClient.createMachineToMachineToken(kafkaKeyConfig.scope)
+    val kafkaKeysClient = kafkaKeysClient(applicationConfig.kafkaKeysClient) {
+        azureM2MTokenClient.createMachineToMachineToken(applicationConfig.kafkaKeysClient.scope)
     }
 
     val prometheusMeterRegistry = PrometheusMeterRegistry(PrometheusConfig.DEFAULT)
 
+    val healthIndicatorRepository = HealthIndicatorRepository()
+
     val httpClient = HttpClient {
         install(ContentNegotiation) {
             jackson()
         }
     }
 
-    val streamsConfig = KafkaStreamsFactory(applicationConfig.applicationIdSuffix, kafkaStreamsConfig)
-        .withDefaultKeySerde(Serdes.LongSerde::class)
-        .withDefaultValueSerde(SpecificAvroSerde::class)
-
-    val streamsBuilder = StreamsBuilder()
-        .addStateStore(
-            Stores.keyValueStoreBuilder(
-                Stores.persistentKeyValueStore(applicationConfig.bekreftelseStateStoreName),
-                Serdes.Long(),
-                InternStateSerde(),
-            )
-        )
-
-    val topology = streamsBuilder.appTopology(
-        prometheusRegistry = prometheusMeterRegistry,
-        bekreftelseHendelseLoggTopic = applicationConfig.bekreftelseHendelseLoggTopic,
-        bekreftelseStateStoreName = applicationConfig.bekreftelseStateStoreName,
-    )
-
-    val kafkaStreams = KafkaStreams(
-        topology,
-        streamsConfig.properties.apply {
-            put("application.server", applicationConfig.hostname)
-        }
-    )
-
-    kafkaStreams.setUncaughtExceptionHandler { throwable ->
-        logger.error("Uventet feil: ${throwable.message}", throwable)
-        StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_APPLICATION
-    }
-
-    kafkaStreams.start()
-
-    val bekreftelseStateStore: ReadOnlyKeyValueStore<Long, InternState> = kafkaStreams.store(
-        StoreQueryParameters.fromNameAndType(
-            applicationConfig.bekreftelseStateStoreName,
-            QueryableStoreTypes.keyValueStore()
-        )
-    )
-
-    val health = Health(kafkaStreams)
-
-    val bekreftelseProducer = BekreftelseProducer(kafkaConfig, applicationConfig)
-
-
     val poaoTilgangClient = PoaoTilgangCachedClient(
         PoaoTilgangHttpClient(
             applicationConfig.poaoClientConfig.url,
@@ -108,25 +47,41 @@ fun createDependencies(
 
     val autorisasjonService = AutorisasjonService(poaoTilgangClient)
 
+    val bekreftelseTopology = buildBekreftelseTopology(applicationConfig, prometheusMeterRegistry)
+    val bekreftelseKafkaStreams = buildKafkaStreams(applicationConfig, healthIndicatorRepository, bekreftelseTopology)
+
+    // TODO Bruker mock for utvikling
+    val bekreftelseService: BekreftelseService = if (applicationConfig.brukMock) {
+        healthIndicatorRepository.getLivenessIndicators().forEach { it.setHealthy() }
+        healthIndicatorRepository.getReadinessIndicators().forEach { it.setHealthy() }
+
+        BekreftelseServiceMock()
+    } else {
+        val bekreftelseProducer = BekreftelseProducer(applicationConfig)
+
+        BekreftelseServiceImpl(
+            applicationConfig,
+            httpClient,
+            bekreftelseKafkaStreams,
+            bekreftelseProducer
+        )
+    }
+
     return Dependencies(
         kafkaKeysClient,
-        httpClient,
-        kafkaStreams,
         prometheusMeterRegistry,
-        bekreftelseStateStore,
-        health,
-        bekreftelseProducer,
-        autorisasjonService
+        healthIndicatorRepository,
+        bekreftelseKafkaStreams,
+        autorisasjonService,
+        bekreftelseService
     )
 }
 
 data class Dependencies(
     val kafkaKeysClient: KafkaKeysClient,
-    val httpClient: HttpClient,
-    val kafkaStreams: KafkaStreams,
     val prometheusMeterRegistry: PrometheusMeterRegistry,
-    val bekreftelseStateStore: ReadOnlyKeyValueStore<Long, InternState>,
-    val health: Health,
-    val bekreftelseProducer: BekreftelseProducer,
-    val autorisasjonService: AutorisasjonService
+    val healthIndicatorRepository: HealthIndicatorRepository,
+    val bekreftelseKafkaStreams: KafkaStreams,
+    val autorisasjonService: AutorisasjonService,
+    val bekreftelseService: BekreftelseService
 )