pcvolkmer/etl-processor
1
0
Fork 0
mirror of https://github.com/pcvolkmer/etl-processor.git synced 2025-07-03 23:12:54 +00:00
Code Issues Releases Activity

(Near) realtime update of statistics charts

Browse Source
This commit is contained in:
Paul-Christian Volkmer
2023-07-25 20:55:32 +02:00
parent 94846deb98
commit 1a2d4ea7a2
7 changed files with 208 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/main/kotlin/dev/dnpm/etl/processor/config/AppConfiguration.kt
View File

@ -27,12 +27,16 @@ import dev.dnpm.etl.processor.pseudonym.AnonymizingGenerator
import dev.dnpm.etl.processor.pseudonym.Generator
import dev.dnpm.etl.processor.pseudonym.GpasPseudonymGenerator
import dev.dnpm.etl.processor.pseudonym.PseudonymizeService
import org.reactivestreams.Publisher
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
import org.springframework.boot.context.properties.EnableConfigurationProperties
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.kafka.core.KafkaTemplate
import reactor.core.publisher.Flux
import reactor.core.publisher.Sinks
import java.net.URI
import java.time.Duration
@Configuration
@EnableConfigurationProperties(
@ -78,5 +82,10 @@ class AppConfiguration {
return KafkaMtbFileSender(kafkaTemplate, objectMapper)
}
@Bean
fun statisticsUpdateProducer(): Sinks.Many<Any> {
return Sinks.many().multicast().directBestEffort()
}
}

7
src/main/kotlin/dev/dnpm/etl/processor/web/MtbFileController.kt
View File

@ -34,13 +34,15 @@ import org.springframework.http.ResponseEntity
import org.springframework.web.bind.annotation.PostMapping
import org.springframework.web.bind.annotation.RequestBody
import org.springframework.web.bind.annotation.RestController
import reactor.core.publisher.Sinks
@RestController
class MtbFileController(
private val pseudonymizeService: PseudonymizeService,
private val senders: List<MtbFileSender>,
private val requestRepository: RequestRepository,
private val objectMapper: ObjectMapper
private val objectMapper: ObjectMapper,
private val statisticsUpdateProducer: Sinks.Many<Any>
) {
private val logger = LoggerFactory.getLogger(MtbFileController::class.java)
@ -63,6 +65,7 @@ class MtbFileController(
report = Report("Duplikat erkannt - keine Daten weitergeleitet")
)
)
statisticsUpdateProducer.emitNext("", Sinks.EmitFailureHandler.FAIL_FAST)
return ResponseEntity.noContent().build()
}
@ -110,6 +113,8 @@ class MtbFileController(
)
)
statisticsUpdateProducer.emitNext("", Sinks.EmitFailureHandler.FAIL_FAST)
return if (requestStatus == RequestStatus.ERROR) {
ResponseEntity.unprocessableEntity().build()
} else {

5
src/main/kotlin/dev/dnpm/etl/processor/web/StatisticsController.kt
View File

@ -20,15 +20,18 @@
package dev.dnpm.etl.processor.web
import org.springframework.stereotype.Controller
import org.springframework.ui.Model
import org.springframework.web.bind.annotation.GetMapping
import org.springframework.web.bind.annotation.RequestMapping
import java.time.Instant
@Controller
@RequestMapping(path = ["/statistics"])
class StatisticsController {
@GetMapping
fun index(): String {
fun index(model: Model): String {
model.addAttribute("now", Instant.now())
return "statistics"
}

48
src/main/kotlin/dev/dnpm/etl/processor/web/StatisticsRestController.kt
View File

@ -21,9 +21,15 @@ package dev.dnpm.etl.processor.web
import dev.dnpm.etl.processor.monitoring.RequestRepository
import dev.dnpm.etl.processor.monitoring.RequestStatus
import org.reactivestreams.Publisher
import org.springframework.http.MediaType
import org.springframework.http.codec.ServerSentEvent
import org.springframework.web.bind.annotation.GetMapping
import org.springframework.web.bind.annotation.RequestMapping
import org.springframework.web.bind.annotation.RestController
import reactor.core.publisher.Flux
import reactor.core.publisher.Sinks
import reactor.kotlin.core.publisher.toFlux
import java.time.Instant
import java.time.Month
import java.time.ZoneId
@ -34,6 +40,7 @@ import java.time.temporal.TemporalUnit
@RestController
@RequestMapping(path = ["/statistics"])
class StatisticsRestController(
private val statisticsUpdateProducer: Sinks.Many<Any>,
private val requestRepository: RequestRepository
) {
@ -68,13 +75,15 @@ class StatisticsRestController(
.toMap()
Pair(
it.key.toString(),
DateNameValues(it.key.toString(), NameValues(
error = requestList[RequestStatus.ERROR] ?: 0,
warning = requestList[RequestStatus.WARNING] ?: 0,
success = requestList[RequestStatus.SUCCESS] ?: 0,
duplication = requestList[RequestStatus.DUPLICATION] ?: 0,
unknown = requestList[RequestStatus.UNKNOWN] ?: 0,
))
DateNameValues(
it.key.toString(), NameValues(
error = requestList[RequestStatus.ERROR] ?: 0,
warning = requestList[RequestStatus.WARNING] ?: 0,
success = requestList[RequestStatus.SUCCESS] ?: 0,
duplication = requestList[RequestStatus.DUPLICATION] ?: 0,
unknown = requestList[RequestStatus.UNKNOWN] ?: 0,
)
)
)
}.toMap()
@ -86,10 +95,33 @@ class StatisticsRestController(
.sortedBy { it.date }
}
@GetMapping(path = ["events"], produces = [MediaType.TEXT_EVENT_STREAM_VALUE])
fun updater(): Flux<ServerSentEvent<Any>> {
return statisticsUpdateProducer.asFlux().flatMap {
Flux.fromIterable(
listOf(
ServerSentEvent.builder<Any>()
.event("requeststates").id("none").data(this.requestStates())
.build(),
ServerSentEvent.builder<Any>()
.event("requestslastmonth").id("none").data(this.requestsLastMonth())
.build()
)
)
}
}
}
data class NameValue(val name: String, val value: Int, val color: String)
data class DateNameValues(val date: String, val nameValues: NameValues)
data class NameValues(val error: Int = 0, val warning: Int = 0, val success: Int = 0, val duplication: Int = 0, val unknown: Int = 0)
data class NameValues(
val error: Int = 0,
val warning: Int = 0,
val success: Int = 0,
val duplication: Int = 0,
val unknown: Int = 0
)