LogStreamService.kt

package com.blizzard.content.filter.chatreplay.source

import com.blizzard.content.filter.chatreplay.config.CHATLOG_PREFIX

import com.blizzard.content.filter.chatreplay.config.METRICS_PREFIX

import com.blizzard.content.filter.chatreplay.config.SourceProperties

import com.blizzard.content.filter.chatreplay.config.flashbackInstant

import com.blizzard.content.filter.chatreplay.domain.ChatLog

import com.blizzard.content.filter.chatreplay.domain.hadOcurredAt

import io.micrometer.core.instrument.Counter

import io.micrometer.core.instrument.Gauge

import io.micrometer.core.instrument.MeterRegistry

import io.micrometer.core.instrument.Metrics

import io.micrometer.core.instrument.Tags

import org.jetbrains.annotations.NotNull

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 reactor.core.Disposable

import reactor.core.publisher.Flux

import reactor.core.publisher.Sinks

import reactor.core.scheduler.Schedulers

import reactor.util.retry.Retry

import java.time.Duration.ofMillis

import java.util.concurrent.ConcurrentLinkedQueue

import kotlin.random.Random.Default.nextDouble

@Service

class LogStreamService(

meterRegistry: MeterRegistry,

private val sourceProperties: SourceProperties,

@Value("\${git.commit.id}")

private val commitId: String// = "unknown"

) {

private val log = LoggerFactory.getLogger(javaClass)

private val driverLog = LoggerFactory.getLogger("${javaClass.name}.Driver")

private var logDriver: Disposable? = null

private val logStorage = ConcurrentLinkedQueue<ChatLog>()

private val driverScheduler = Schedulers.newBoundedElastic(2, 1000, "driver")

private val multiScheduler = Schedulers.newBoundedElastic(10, 1000, "multic")

private val logSink: Sinks.Many<ChatLog> = Sinks.many().multicast().directBestEffort()

private val publishCounter: Counter = meterRegistry.counter("$CHATLOG_PREFIX.logstream.published")

init {

Gauge.builder("$CHATLOG_PREFIX.fetch.queuesize", logStorage::size)

.description("Count of chatlogs collected, waiting to be processed.")

.register(meterRegistry)

// Increment it to some random value 0-100. Will probably be different from last restart

meterRegistry

.counter("$METRICS_PREFIX.application.restart", Tags.of("commitid", commitId))

.increment((nextDouble(0.0, 100.0) + 20.0).also {

log.info("Reporting restart: $it $commitId")

})

}

@EventListener

fun onApplicationEvent(@Suppress("UNUSED_PARAMETER") readyEvent: ApplicationReadyEvent) {

// See function comments

startDriver()

}

/**

* This flux interval drives sink emission

*/

private fun startDriver() {

Metrics.counter("$METRICS_PREFIX.logstream.driver.start").increment()

logDriver = Flux.interval(ofMillis(50))

.publishOn(driverScheduler)

.retryWhen(Retry.backoff(Long.MAX_VALUE, ofMillis(100))

.doBeforeRetry {

driverLog.error("Driver Retrying attempt [${it.totalRetries()}] due to ${it.failure()} : ${it.failure().message}")

Metrics.counter("$METRICS_PREFIX.logstream.driver.retry").increment()

})

.subscribe {

while (logStorage.peek()?.hadOccurred() == true) {

logSink.emitNext(logStorage.poll()) { _, _ -> true }

publishCounter.increment()

}

}

driverLog.info("ChatLog driver started.")

}

@Synchronized

fun logStream(): Flux<ChatLog> {

return logSink

.asFlux()

.retryWhen(Retry.backoff(Long.MAX_VALUE, ofMillis(100))

.doBeforeRetry {

log.error("Retrying attempt [${it.totalRetries()}] due to ${it.failure()} : ${it.failure().message}")

Metrics.counter("$CHATLOG_PREFIX.logstream.retry").increment()

})

.publishOn(multiScheduler)

}

@Synchronized

fun addAll(@NotNull incomingLogs: Collection<ChatLog>) {

// If, somehow, incomingLogs are empty

if (incomingLogs.isEmpty()) {

log.error("No chatLogs received to add. Queue size: ${logStorage.size}")

return

}

val sortedIncomingLogs = incomingLogs.sorted()

/*

In the infrequent case where incoming logs (which should already be in the queue) are

just now arriving (ie on app startup, if the first query takes significantly longer than the second),

then the arrangement will resemble this example:

last() |=========> peek() first()[0]

logStorage: 9 8 7 6 5

incomingLogs: 4 3 2 1 0 ^

Earlier logs are being inserted "after" later logs.

Time will need to pass to "9" before logs with timestamp "0" will be processed.

Unless the queue is checked and sorted first.

*/

if (logStorage.isNotEmpty() && sortedIncomingLogs.last().timestamp < logStorage.first().timestamp) {

val totalBefore = incomingLogs.size + logStorage.size

log.warn("Inverted insertion order detected in logStorage. Correcting...")

val fromLogStorage = logStorage.sorted()

logStorage.clear()

logStorage.addAll((fromLogStorage + sortedIncomingLogs).sorted())

when (logStorage.size) {

totalBefore -> log.info("Correction completed. Sizes match")

else -> log.error("Correction completed. Sizes do not match.")

}

} else {

logStorage.addAll(sortedIncomingLogs)

}

log.info("Added ${incomingLogs.size} chatLogs. New queue size: ${logStorage.size}")

}

private fun ChatLog.hadOccurred() = hadOcurredAt(sourceProperties.flashbackInstant())

}