package com.blizzard.content.filter.chatreplay.sink
import com.blizzard.content.filter.chatreplay.config.CHATLOG_PREFIX
import com.blizzard.content.filter.chatreplay.config.MODE
import com.blizzard.content.filter.chatreplay.config.TIME
import com.blizzard.content.filter.chatreplay.config.TRANSFORM
import com.blizzard.content.filter.chatreplay.domain.ChatLog
import com.blizzard.content.filter.chatreplay.metrics.timerTags
import com.blizzard.content.filter.chatreplay.metrics.toSpec
import com.blizzard.content.filter.chatreplay.sink.task.ChatLogTask
import com.blizzard.content.filter.chatreplay.sink.task.PredicateTask
import com.blizzard.content.filter.chatreplay.sink.task.ProcessTask
import com.blizzard.content.filter.chatreplay.sink.task.ReductionTask
import com.blizzard.content.filter.chatreplay.sink.task.TransformTask
import com.blizzard.content.filter.chatreplay.source.LogStreamService
import io.micrometer.core.instrument.MeterRegistry
import io.micrometer.core.instrument.Metrics
import io.micrometer.core.instrument.Tag
import io.micrometer.core.instrument.Tags
import io.micrometer.core.instrument.Timer
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.launch
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Value
import org.springframework.boot.context.event.ApplicationReadyEvent
import org.springframework.context.event.EventListener
import org.springframework.stereotype.Service
import java.util.concurrent.TimeUnit
import kotlin.system.measureNanoTime
/**
* Some work on ChatLogs must be done individually, per ChatLog
*
* This class centralizes the dispatching of those individual ChatLog tasks into Kotlin coroutines for better concurrent performance.
*
* For work that is done on the stream of ChatLogs from LogStreamService, see other subscribers.
*
* @see ChatLogTask
*/
@Service
class TaskBroker(
private val logStreamService: LogStreamService,
private val chatLogTasks: List<ChatLogTask>,
meterRegistry: MeterRegistry,
) {
private val log = LoggerFactory.getLogger(javaClass)
init {
log.info("Brokering ChatLogs for Tasks:")
chatLogTasks.forEach {
log.info("...Task: ${it.javaClass.simpleName}")
}
}
@Value("\${chat-replay.task-broker.enable-brokering}")
private var brokeringEnabled: Boolean = true
private val brokerTags = Tags.of(Tag.of(MODE, "taskbroker"))
/**
* When this field is created using `Metrics.counter()` instead, the metrics are slightly off in EndToEndIT and TaskBrokerMetricsIT tests.
* When created with `registry.counter()`, it matches perfectly.
* /shrug
*/
private val brokeredTotalCounter = meterRegistry.counter("$CHATLOG_PREFIX.taskbroker.brokered", brokerTags)
@EventListener(condition = "@environment.getProperty('chat-replay.subscribers.task-broker.enable')")
fun onApplicationEvent(readyEvent: ApplicationReadyEvent) {
log.info("ApplicationReady: ${javaClass.simpleName}")
CoroutineScope(Dispatchers.Default).launch {
log.info("${LogStreamService::class.java.simpleName} is ready")
brokerChatLogs()
}
}
private fun brokerChatLogs() {
logStreamService.logStream()
.doOnSubscribe {
if (brokeringEnabled) log.info("Now brokering chat logs")
}
.subscribe { chatLog ->
brokeredTotalCounter.increment()
if (brokeringEnabled) {
forEachChatLog(chatLog)
}
}
}
// https://micrometer.io/docs/concepts#_global_registry
private fun forEachChatLog(chatLog: ChatLog) {
if (log.isDebugEnabled) log.debug("Brokering: ${chatLog.id}")
chatLogTasks.asSequence().forEach { task ->
task.taskScope().launch {
with(TaskResult()) {
taskDuration = measureNanoTime {
when (task) {
is ProcessTask -> {
val processed: String = task.process(chatLog)
timerSpec = task.toSpec(/*Too much cardinality to put the processed value here*/)
}
is PredicateTask -> {
val predicate: Boolean = task.process(chatLog)
timerSpec = task.toSpec("detection", "$predicate")
}
is ReductionTask -> {
val reduced: Long = task.process(chatLog)
timerSpec = task.toSpec("reduction", "$reduced")
}
is TransformTask -> {
task.process(chatLog)
.wasModified.toString().let {
Metrics.counter(TRANSFORM, brokerTags.and(task.timerTags()).and("transformed", it)).increment()
timerSpec = task.toSpec("transformed", it)
}
}
}
}
this
}.apply {
taskTimerPlus(brokerTags).record(taskDuration, TimeUnit.NANOSECONDS)
}
}
}
}
inner class TaskResult {
var taskDuration: Long = 1
var timerSpec: Pair<String, Tags> = Pair("GenericTaskResultName", Tags.empty())
fun name() = "${timerSpec.first}$TIME"
fun taskTimerPlus(otherTags: Tags): Timer = Metrics.timer(name(), timerSpec.second.and(otherTags))
}
}