улучшить observability и документацию outbox

- обновить README с текущим Kafka, DB, outbox и scaling flow - расширить логи listener, ingestion, publisher и outbox worker - не логировать raw payload и outbox payload - закрыть test controller явным property - добавить Micrometer counters для ingestion и outbox outcomes - сохранить listener flow, payload contract и delivery semantics без изменений
2026-05-25 15:20:25 +03:00
parent 4073897efc
commit fa357a78b4
10 changed files with 306 additions and 65 deletions
@@ -2,6 +2,7 @@ package space.nstart.pcp.routepassportconsumer.controller

 import io.swagger.v3.oas.annotations.Operation
 import io.swagger.v3.oas.annotations.tags.Tag
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.core.io.ClassPathResource
 import org.springframework.kafka.core.KafkaTemplate
 import org.springframework.web.bind.annotation.PostMapping
@@ -14,6 +15,11 @@ import java.nio.charset.StandardCharsets
@RestController
@RequestMapping("/api/route-processing")
@Tag(name = "Route Processing", description = "Тестовые endpoint'ы сервиса обработки маршрутов")
+@ConditionalOnProperty(
+    name = ["app.route-processing.test-controller.enabled"],
+    havingValue = "true",
+    matchIfMissing = false,
+)
 class RouteProcessingTestController(
    private val kafkaTemplate: KafkaTemplate<String, String>,
    private val kafkaTopicsProperties: KafkaTopicsProperties,
@@ -8,6 +8,7 @@ import org.springframework.stereotype.Component
 import space.nstart.pcp.routepassportconsumer.config.KafkaTopicsProperties
 import space.nstart.pcp.routepassportconsumer.service.RoutePassportIngestionService
 import space.nstart.pcp.routepassportconsumer.service.RoutePassportJsonParserService
+import space.nstart.pcp.routepassportconsumer.service.RoutePassportMetrics
 import space.nstart.pcp.routepassportconsumer.service.RoutePassportSourceRecordMetadata

 /**
@@ -22,6 +23,9 @@ class RoutePassportMessageListener(
    /** Сервис транзакционного сохранения паспорта маршрута и outbox-событий. */
    private val routePassportIngestionService: RoutePassportIngestionService,

+    /** Метрики ingestion flow. */
+    private val routePassportMetrics: RoutePassportMetrics,
+
    /** Набор Kafka topic'ов, который приходит из config server и валидируется при старте. */
    val kafkaTopicsProperties: KafkaTopicsProperties,
 ) {
@@ -59,6 +63,7 @@ class RoutePassportMessageListener(
            )

            val parsedMessage = routePassportJsonParserService.parse(rawPayload)
+            val routePassportDto = parsedMessage.routePassportDto
            routePassportIngestionService.ingest(
                parsedMessage = parsedMessage,
                rawPayload = rawPayload,
@@ -67,25 +72,29 @@ class RoutePassportMessageListener(

            // Transactional ingest commits before returning; only then the Kafka offset is acknowledged.
            acknowledgment.acknowledge()
+            routePassportMetrics.incrementIngestSuccess()

            log.info(
-                "Route passport ingested: sourceTopic={}, partition={}, offset={}, key={}, payloadSize={}, routeId={}, traceId={}, correlationId={}",
+                "Route passport ingested: sourceTopic={}, partition={}, offset={}, key={}, payloadSize={}, routeId={}, routeNameFull={}, traceId={}, correlationId={}",
                sourceRecord.topic,
                sourceRecord.partition,
                sourceRecord.offset,
                sourceRecord.key,
                rawPayload.length,
-                parsedMessage.routePassportDto.routeId,
+                routePassportDto.routeId,
+                routePassportDto.routeNameFull,
                parsedMessage.metadata.traceId,
                parsedMessage.metadata.correlationId,
            )
        } catch (exception: Exception) {
+            routePassportMetrics.incrementIngestFailure()
            log.error(
-                "Failed to process Kafka record: sourceTopic={}, partition={}, offset={}, key={}, error={}",
+                "Failed to process Kafka record: sourceTopic={}, partition={}, offset={}, key={}, exceptionClass={}, exceptionMessage={}",
                sourceRecord.topic,
                sourceRecord.partition,
                sourceRecord.offset,
                sourceRecord.key,
+                exception::class.java.name,
                exception.message,
                exception,
            )
@@ -1,5 +1,6 @@
 package space.nstart.pcp.routepassportconsumer.service

+import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Service
 import org.springframework.transaction.annotation.Transactional
 import space.nstart.pcp.pcp_types_lib.dto.routes.ParsedRoutePassportMessage
@@ -27,6 +28,7 @@ class RoutePassportIngestionService(
    private val objectMapper: ObjectMapper,
    private val kafkaTopicsProperties: KafkaTopicsProperties,
 ) {
+    private val log = LoggerFactory.getLogger(this::class.java)

    @Transactional
    fun ingest(
@@ -36,8 +38,10 @@ class RoutePassportIngestionService(
    ) {
        val now = LocalDateTime.now()
        val routePassportDto = parsedMessage.routePassportDto
+        val existingRoutePassport = routePassportRepository.findById(routePassportDto.routeId)
+        val routeCreated = existingRoutePassport.isEmpty

-        val routePassportEntity = routePassportRepository.findById(routePassportDto.routeId)
+        val routePassportEntity = existingRoutePassport
            .map { existingEntity ->
                existingEntity.applyRoutePassport(
                    routePassportDto = routePassportDto,
@@ -58,17 +62,33 @@ class RoutePassportIngestionService(

        routePassportRepository.save(routePassportEntity)

-        createOutboxIfAbsent(
-            eventType = PcpKafkaEvent.ModeStatusChangedEvent,
-            topic = kafkaTopicsProperties.processedRoute,
-            parsedMessage = parsedMessage,
-            now = now,
+        val outboxResults = listOf(
+            createOutboxIfAbsent(
+                eventType = PcpKafkaEvent.ModeStatusChangedEvent,
+                topic = kafkaTopicsProperties.processedRoute,
+                parsedMessage = parsedMessage,
+                now = now,
+            ),
+            createOutboxIfAbsent(
+                eventType = PcpKafkaEvent.RouteGeoRefEvent,
+                topic = kafkaTopicsProperties.routeGeoreference,
+                parsedMessage = parsedMessage,
+                now = now,
+            ),
        )
-        createOutboxIfAbsent(
-            eventType = PcpKafkaEvent.RouteGeoRefEvent,
-            topic = kafkaTopicsProperties.routeGeoreference,
-            parsedMessage = parsedMessage,
-            now = now,
+
+        log.info(
+            "Route passport persisted: routeId={}, routeNameFull={}, traceId={}, correlationId={}, sourceTopic={}, sourcePartition={}, sourceOffset={}, routeCreated={}, outboxEventsCreated={}, outboxEventsExisting={}",
+            routePassportDto.routeId,
+            routePassportDto.routeNameFull,
+            parsedMessage.metadata.traceId,
+            parsedMessage.metadata.correlationId,
+            sourceRecord.topic,
+            sourceRecord.partition,
+            sourceRecord.offset,
+            routeCreated,
+            outboxResults.filter { it.created }.map { "${it.eventType}:${it.topic}" },
+            outboxResults.filterNot { it.created }.map { "${it.eventType}:${it.topic}" },
        )
    }

@@ -77,34 +97,57 @@ class RoutePassportIngestionService(
        topic: String,
        parsedMessage: ParsedRoutePassportMessage,
        now: LocalDateTime,
-    ) {
+    ): OutboxEventResult {
        val routeId = parsedMessage.routePassportDto.routeId
+        val routeNameFull = parsedMessage.routePassportDto.routeNameFull
        val eventTypeName = eventType.name

        if (routePassportOutboxRepository.existsByRouteIdAndEventType(routeId, eventTypeName)) {
-            return
+            log.debug(
+                "Route passport outbox event already exists: routeId={}, routeNameFull={}, traceId={}, correlationId={}, eventType={}, topic={}",
+                routeId,
+                routeNameFull,
+                parsedMessage.metadata.traceId,
+                parsedMessage.metadata.correlationId,
+                eventTypeName,
+                topic,
+            )
+            return OutboxEventResult(eventType = eventTypeName, topic = topic, created = false)
        }

-        routePassportOutboxRepository.save(
-            RoutePassportOutboxEntity(
-                id = UUID.randomUUID(),
-                routeId = routeId,
-                eventType = eventTypeName,
-                topic = topic,
-                messageKey = routeId.toString(),
-                payload = buildOutboxPayload(eventType, parsedMessage),
-                headers = null,
-                status = RoutePassportOutboxStatus.PENDING,
-                attempts = 0,
-                maxAttempts = 10,
-                nextAttemptAt = now,
-                lockedUntil = null,
-                lastError = null,
-                createdAt = now,
-                updatedAt = now,
-                publishedAt = null,
-            ),
+        val outbox = RoutePassportOutboxEntity(
+            id = UUID.randomUUID(),
+            routeId = routeId,
+            eventType = eventTypeName,
+            topic = topic,
+            messageKey = routeId.toString(),
+            payload = buildOutboxPayload(eventType, parsedMessage),
+            headers = null,
+            status = RoutePassportOutboxStatus.PENDING,
+            attempts = 0,
+            maxAttempts = 10,
+            nextAttemptAt = now,
+            lockedUntil = null,
+            lastError = null,
+            createdAt = now,
+            updatedAt = now,
+            publishedAt = null,
        )
+        routePassportOutboxRepository.save(outbox)
+
+        log.debug(
+            "Route passport outbox event created: outboxId={}, routeId={}, routeNameFull={}, traceId={}, correlationId={}, eventType={}, topic={}, status={}",
+            outbox.id,
+            routeId,
+            routeNameFull,
+            parsedMessage.metadata.traceId,
+            parsedMessage.metadata.correlationId,
+            eventTypeName,
+            topic,
+            outbox.status,
+        )
+
+        return OutboxEventResult(eventType = eventTypeName, topic = topic, created = true)
    }

    private fun buildOutboxPayload(
@@ -197,4 +240,10 @@ class RoutePassportIngestionService(
            createdAt = now,
            updatedAt = now,
        )
+
+    private data class OutboxEventResult(
+        val eventType: String,
+        val topic: String,
+        val created: Boolean,
+    )
 }
@@ -9,6 +9,7 @@ import tools.jackson.databind.ObjectMapper
 import java.nio.charset.StandardCharsets
 import java.util.concurrent.ExecutionException
 import java.util.concurrent.TimeUnit
+import kotlin.time.Duration.Companion.nanoseconds

 /**
 * Сервис физической публикации одной готовой outbox-записи в Kafka.
@@ -30,19 +31,23 @@ class RoutePassportKafkaPublisherService(
        val payload = objectMapper.writeValueAsString(outbox.payload)
        val record = ProducerRecord(outbox.topic, outbox.messageKey, payload)
        record.headers().add(TYPE_HEADER, outbox.eventType.toByteArray(StandardCharsets.UTF_8))
+        val startedAt = System.nanoTime()

        try {
            val result = kafkaTemplate.send(record)
                .get(DEFAULT_PUBLISH_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+            val sendDurationMs = (System.nanoTime() - startedAt).nanoseconds.inWholeMilliseconds

            log.info(
-                "Kafka outbox message published: outboxId={}, routeId={}, eventType={}, topic={}, partition={}, offset={}",
+                "Kafka outbox message published: outboxId={}, routeId={}, eventType={}, topic={}, messageKey={}, partition={}, offset={}, sendDurationMs={}",
                outbox.id,
                outbox.routeId,
                outbox.eventType,
                outbox.topic,
+                outbox.messageKey,
                result.recordMetadata.partition(),
                result.recordMetadata.offset(),
+                sendDurationMs,
            )
        } catch (exception: InterruptedException) {
            Thread.currentThread().interrupt()
@@ -64,11 +69,14 @@ class RoutePassportKafkaPublisherService(

    private fun logPublishFailure(outbox: RoutePassportOutboxEntity, exception: Throwable) {
        log.error(
-            "Failed to publish kafka outbox message: outboxId={}, routeId={}, eventType={}, topic={}",
+            "Failed to publish kafka outbox message: outboxId={}, routeId={}, eventType={}, topic={}, messageKey={}, exceptionClass={}, exceptionMessage={}",
            outbox.id,
            outbox.routeId,
            outbox.eventType,
            outbox.topic,
+            outbox.messageKey,
+            exception::class.java.name,
+            exception.message,
            exception,
        )
    }
@@ -0,0 +1,42 @@
+package space.nstart.pcp.routepassportconsumer.service
+
+import io.micrometer.core.instrument.Counter
+import io.micrometer.core.instrument.MeterRegistry
+import org.springframework.stereotype.Component
+
+/**
+ * Centralizes route passport counters so operational metrics names stay consistent.
+ */
+@Component
+class RoutePassportMetrics(
+    meterRegistry: MeterRegistry,
+) {
+    private val ingestSuccessCounter = meterRegistry.counter("route_passport_ingest_success_total")
+    private val ingestFailureCounter = meterRegistry.counter("route_passport_ingest_failure_total")
+    private val outboxPublishedCounter = meterRegistry.counter("route_passport_outbox_published_total")
+    private val outboxRetryCounter = meterRegistry.counter("route_passport_outbox_retry_total")
+    private val outboxFailedCounter = meterRegistry.counter("route_passport_outbox_failed_total")
+
+    fun incrementIngestSuccess() {
+        ingestSuccessCounter.increment()
+    }
+
+    fun incrementIngestFailure() {
+        ingestFailureCounter.increment()
+    }
+
+    fun incrementOutboxPublished() {
+        outboxPublishedCounter.increment()
+    }
+
+    fun incrementOutboxRetry() {
+        outboxRetryCounter.increment()
+    }
+
+    fun incrementOutboxFailed() {
+        outboxFailedCounter.increment()
+    }
+
+    private fun MeterRegistry.counter(name: String): Counter =
+        Counter.builder(name).register(this)
+}
@@ -1,5 +1,6 @@
 package space.nstart.pcp.routepassportconsumer.service

+import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Service
 import org.springframework.transaction.annotation.Transactional
 import space.nstart.pcp.routepassportconsumer.config.RoutePassportOutboxProperties
@@ -17,7 +18,9 @@ import java.util.UUID
 class RoutePassportOutboxService(
    private val routePassportOutboxRepository: RoutePassportOutboxRepository,
    private val outboxProperties: RoutePassportOutboxProperties,
+    private val routePassportMetrics: RoutePassportMetrics,
 ) {
+    private val log = LoggerFactory.getLogger(this::class.java)

    /**
     * Claims due pending rows in one DB transaction using SELECT FOR UPDATE SKIP LOCKED.
@@ -25,6 +28,7 @@ class RoutePassportOutboxService(
    @Transactional
    fun claimDueBatch(): List<RoutePassportOutboxEntity> {
        if (!outboxProperties.enabled) {
+            log.debug("Route passport outbox claim skipped because worker is disabled")
            return emptyList()
        }

@@ -37,7 +41,28 @@ class RoutePassportOutboxService(
            outbox.updatedAt = now
        }

-        return routePassportOutboxRepository.saveAll(claimedRows).toList()
+        val savedRows = routePassportOutboxRepository.saveAll(claimedRows).toList()
+
+        log.debug(
+            "Route passport outbox rows claimed: claimedCount={}, batchSize={}, lockedUntil={}",
+            savedRows.size,
+            outboxProperties.batchSize,
+            lockedUntil,
+        )
+        savedRows.forEach { outbox ->
+            log.debug(
+                "Route passport outbox row claimed: outboxId={}, routeId={}, eventType={}, status={}, attempts={}, nextAttemptAt={}, lockedUntil={}",
+                outbox.id,
+                outbox.routeId,
+                outbox.eventType,
+                outbox.status,
+                outbox.attempts,
+                outbox.nextAttemptAt,
+                outbox.lockedUntil,
+            )
+        }
+
+        return savedRows
    }

    /**
@@ -55,6 +80,18 @@ class RoutePassportOutboxService(
        outbox.updatedAt = now

        routePassportOutboxRepository.save(outbox)
+        routePassportMetrics.incrementOutboxPublished()
+
+        log.info(
+            "Route passport outbox status changed: outboxId={}, routeId={}, eventType={}, transition=PENDING->PUBLISHED, attempts={}, nextAttemptAt={}, lockedUntil={}, publishedAt={}",
+            outbox.id,
+            outbox.routeId,
+            outbox.eventType,
+            outbox.attempts,
+            outbox.nextAttemptAt,
+            outbox.lockedUntil,
+            outbox.publishedAt,
+        )
    }

    /**
@@ -78,6 +115,25 @@ class RoutePassportOutboxService(
        }

        routePassportOutboxRepository.save(outbox)
+        if (outbox.status == RoutePassportOutboxStatus.FAILED) {
+            routePassportMetrics.incrementOutboxFailed()
+        } else {
+            routePassportMetrics.incrementOutboxRetry()
+        }
+
+        log.warn(
+            "Route passport outbox status changed after publish failure: outboxId={}, routeId={}, eventType={}, transition=LOCKED->{}, attempts={}, maxAttempts={}, nextAttemptAt={}, lockedUntil={}, exceptionClass={}, exceptionMessage={}",
+            outbox.id,
+            outbox.routeId,
+            outbox.eventType,
+            outbox.status,
+            outbox.attempts,
+            outbox.maxAttempts,
+            outbox.nextAttemptAt,
+            outbox.lockedUntil,
+            exception::class.java.name,
+            exception.message,
+        )
    }

    private fun findOutbox(outboxId: UUID): RoutePassportOutboxEntity =
@@ -20,11 +20,14 @@ class RoutePassportOutboxWorker(

    @Scheduled(fixedDelayString = "\${app.route-passport.outbox.fixed-delay-ms:1000}")
    fun publishDueOutboxRows() {
+        log.debug("Route passport outbox worker tick started: enabled={}", outboxProperties.enabled)
        if (!outboxProperties.enabled) {
+            log.debug("Route passport outbox worker tick skipped because worker is disabled")
            return
        }

        val outboxRows = routePassportOutboxService.claimDueBatch()
+        log.debug("Route passport outbox worker batch claimed: batchSize={}", outboxRows.size)
        outboxRows.forEach { outbox ->
            logClaimed(outbox)
        }
@@ -46,6 +49,7 @@ class RoutePassportOutboxWorker(
                )
            }
        }
+        log.debug("Route passport outbox worker tick finished: batchSize={}", outboxRows.size)
    }

    private fun publishOne(outbox: RoutePassportOutboxEntity) {
@@ -54,7 +58,7 @@ class RoutePassportOutboxWorker(
            routePassportOutboxService.markPublished(outbox.id)

            log.info(
-                "Kafka outbox row marked published: outboxId={}, routeId={}, eventType={}, topic={}, attempts={}, status={}",
+                "Kafka outbox row publish succeeded: outboxId={}, routeId={}, eventType={}, topic={}, attempts={}, status={}",
                outbox.id,
                outbox.routeId,
                outbox.eventType,
@@ -1,5 +1,6 @@
 package space.nstart.pcp.routepassportconsumer.listener

+import io.micrometer.core.instrument.simple.SimpleMeterRegistry
 import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.junit.jupiter.api.Assertions.assertThrows
 import org.junit.jupiter.api.Test
@@ -18,6 +19,7 @@ import space.nstart.pcp.pcp_types_lib.dto.routes.RoutePassportMessageMetadata
 import space.nstart.pcp.routepassportconsumer.config.KafkaTopicsProperties
 import space.nstart.pcp.routepassportconsumer.service.RoutePassportIngestionService
 import space.nstart.pcp.routepassportconsumer.service.RoutePassportJsonParserService
+import space.nstart.pcp.routepassportconsumer.service.RoutePassportMetrics
 import space.nstart.pcp.routepassportconsumer.service.RoutePassportSourceRecordMetadata
 import java.math.BigDecimal
 import java.time.LocalDateTime
@@ -33,6 +35,7 @@ class RoutePassportMessageListenerTest {
        val listener = RoutePassportMessageListener(
            routePassportJsonParserService = parserService,
            routePassportIngestionService = ingestionService,
+            routePassportMetrics = routePassportMetrics(),
            kafkaTopicsProperties = kafkaTopicsProperties(),
        )
        val rawPayload = """{"id":"route-1"}"""
@@ -61,6 +64,7 @@ class RoutePassportMessageListenerTest {
        val listener = RoutePassportMessageListener(
            routePassportJsonParserService = parserService,
            routePassportIngestionService = ingestionService,
+            routePassportMetrics = routePassportMetrics(),
            kafkaTopicsProperties = kafkaTopicsProperties(),
        )
        val rawPayload = """{"id":"route-1"}"""
@@ -86,6 +90,7 @@ class RoutePassportMessageListenerTest {
        val listener = RoutePassportMessageListener(
            routePassportJsonParserService = parserService,
            routePassportIngestionService = ingestionService,
+            routePassportMetrics = routePassportMetrics(),
            kafkaTopicsProperties = kafkaTopicsProperties(),
        )
        val rawPayload = """{"id":"route-1"}"""
@@ -121,6 +126,9 @@ class RoutePassportMessageListenerTest {
            routeGeoreference = "pcp.route.georeference.v1",
        )

+    private fun routePassportMetrics(): RoutePassportMetrics =
+        RoutePassportMetrics(SimpleMeterRegistry())
+
    private fun consumerRecord(rawPayload: String): ConsumerRecord<String, String> =
        ConsumerRecord(
            INPUT_TOPIC,
@@ -1,5 +1,6 @@
 package space.nstart.pcp.routepassportconsumer.service

+import io.micrometer.core.instrument.simple.SimpleMeterRegistry
 import org.junit.jupiter.api.Test
 import org.mockito.ArgumentMatchers.any
 import org.mockito.ArgumentMatchers.anyInt
@@ -170,6 +171,7 @@ class RoutePassportOutboxServiceTest {
        RoutePassportOutboxService(
            routePassportOutboxRepository = repository,
            outboxProperties = properties,
+            routePassportMetrics = RoutePassportMetrics(SimpleMeterRegistry()),
        )

    private fun properties(