-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: consolidate event processing and optimize application performance
The event processing system has been updated to improve readability and efficiency: - Switched from Reactor to Kotlin Flow for event handling to improve readability. - Events are now saved in batch, reducing concurrency and improving performance. - Removed reconciliation logic with a more aggressive in memory cache approach. - Replaced Caffeine with a simple LinkedHashMap cache to support LRU eviction policy. - Added configuration parameters to adjust the size of the cache, buffer and batch chunk size. These properties can also be modified with environment variables. - Removed actuator: benchmarks are now log, reducing memory consumption and eliminating the need for starting a server. - Updated dependencies to the latest compatible versions. - Replaced dynamic mapping in OpenSearch with a handmade mapping for document fields. This change reduces storage requirements, and eliminates unnecessary fields, making queries more logical and efficient. Co-authored-by: Gaëtan Muller <[email protected]>
- Loading branch information
Showing
41 changed files
with
961 additions
and
498 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
30 changes: 30 additions & 0 deletions
30
src/main/kotlin/ch/srgssr/pillarbox/monitoring/benchmark/BenchmarkScheduledLogger.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| package ch.srgssr.pillarbox.monitoring.benchmark | ||
|
|
||
| import ch.srgssr.pillarbox.monitoring.log.info | ||
| import ch.srgssr.pillarbox.monitoring.log.logger | ||
| import org.springframework.scheduling.annotation.Scheduled | ||
| import org.springframework.stereotype.Component | ||
| import java.util.concurrent.TimeUnit | ||
|
|
||
| /** | ||
| * A scheduled logger that periodically logs the average execution times | ||
| * and stats regarding the number of processed events. | ||
| */ | ||
| @Component | ||
| class BenchmarkScheduledLogger { | ||
| private companion object { | ||
| /** | ||
| * Logger instance for logging within this component. | ||
| */ | ||
| private val logger = logger() | ||
| } | ||
|
|
||
| /** | ||
| * The scheduled logging function, executes every minute. | ||
| */ | ||
| @Scheduled(fixedRate = 1, timeUnit = TimeUnit.MINUTES) | ||
| fun logBenchmarkAverages() { | ||
| logger.info { "Benchmark averages: ${TimeTracker.averages}" } | ||
| logger.info { "Latest stats per minute: ${StatsTracker.getAndResetAll()}" } | ||
| } | ||
| } |
41 changes: 41 additions & 0 deletions
41
src/main/kotlin/ch/srgssr/pillarbox/monitoring/benchmark/MovingAverageCalculator.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| package ch.srgssr.pillarbox.monitoring.benchmark | ||
|
|
||
| import java.util.concurrent.atomic.AtomicLong | ||
| import java.util.concurrent.atomic.LongAdder | ||
|
|
||
| /** | ||
| * A thread-safe class for calculating the moving average of a stream of long values. | ||
| * | ||
| * This implementation accumulates values until the average is calculated, at which point | ||
| * the internal state (sum and count) is reset. | ||
| */ | ||
| internal class MovingAverageCalculator { | ||
| private val sum = LongAdder() | ||
| private val count = AtomicLong() | ||
|
|
||
| /** | ||
| * Adds a new value to the moving average calculation. | ||
| * | ||
| * @param n The value to be added to the moving average calculation. | ||
| */ | ||
| fun add(n: Long) { | ||
| sum.add(n) | ||
| count.incrementAndGet() | ||
| } | ||
|
|
||
| /** | ||
| * Calculates the average of the accumulated values and resets the internal state. | ||
| * | ||
| * This method atomically retrieves and resets both the accumulated sum and count. | ||
| * If no values have been added, the method returns `Double.NaN`. | ||
| * | ||
| * @return The average of the accumulated values, or `NaN` if no values were added. | ||
| */ | ||
| val average: Double | ||
| get() = | ||
| synchronized(this) { | ||
| val totalCount = count.getAndSet(0) | ||
| val total = sum.sumThenReset().toDouble() | ||
| if (totalCount == 0L) Double.NaN else total / totalCount | ||
| } | ||
| } |
56 changes: 56 additions & 0 deletions
56
src/main/kotlin/ch/srgssr/pillarbox/monitoring/benchmark/StatsTracker.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| package ch.srgssr.pillarbox.monitoring.benchmark | ||
|
|
||
| import java.util.concurrent.ConcurrentHashMap | ||
| import java.util.concurrent.atomic.AtomicLong | ||
|
|
||
| /** | ||
| * Utility object for tracking and aggregating statistics. | ||
| * | ||
| * This class uses a [ConcurrentHashMap] to store statistics counters identified by unique keys. | ||
| * It provides methods to increment, retrieve, and reset these counters. | ||
| */ | ||
| object StatsTracker { | ||
| private val stats = ConcurrentHashMap<String, AtomicLong>() | ||
|
|
||
| /** | ||
| * Increments the count for a specific key by the specified delta. | ||
| * If the key does not exist, it is initialized to 0 before incrementing. | ||
| * | ||
| * @param key The unique identifier for the statistic. | ||
| * @param delta The amount to increment the counter by. Defaults to 1. | ||
| */ | ||
| fun increment( | ||
| key: String, | ||
| delta: Long = 1, | ||
| ) { | ||
| stats.computeIfAbsent(key) { AtomicLong(0) }.addAndGet(delta) | ||
| } | ||
|
|
||
| /** | ||
| * Increments the count for a specific key by the specified delta. | ||
| * | ||
| * @param key The unique identifier for the statistic. | ||
| * @param delta The amount to increment the counter by, as an [Int]. | ||
| */ | ||
| fun increment( | ||
| key: String, | ||
| delta: Int, | ||
| ) { | ||
| increment(key, delta.toLong()) | ||
| } | ||
|
|
||
| /** | ||
| * Retrieves the current count for a specific key. | ||
| * | ||
| * @param key The unique identifier for the statistic. | ||
| * @return The current count for the key, or 0 if the key does not exist. | ||
| */ | ||
| operator fun get(key: String): Long = stats[key]?.get() ?: 0L | ||
|
|
||
| /** | ||
| * Retrieves all current statistics as a map and resets their counters to 0. | ||
| * | ||
| * @return A map containing the keys and their corresponding counts before reset. | ||
| */ | ||
| fun getAndResetAll(): Map<String, Long> = stats.mapValues { it.value.getAndSet(0) } | ||
| } |
61 changes: 61 additions & 0 deletions
61
src/main/kotlin/ch/srgssr/pillarbox/monitoring/benchmark/TimeTracker.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| package ch.srgssr.pillarbox.monitoring.benchmark | ||
|
|
||
| import ch.srgssr.pillarbox.monitoring.log.logger | ||
| import ch.srgssr.pillarbox.monitoring.log.trace | ||
| import java.util.concurrent.ConcurrentHashMap | ||
| import kotlin.time.measureTimedValue | ||
|
|
||
| /** | ||
| * Utility object for tracking and logging execution times of code blocks with support for moving averages. | ||
| */ | ||
| object TimeTracker { | ||
| private val logger = logger() | ||
|
|
||
| private val movingAverages = ConcurrentHashMap<String, MovingAverageCalculator>() | ||
|
|
||
| /** | ||
| * Tracks the execution time of a code block. | ||
| * | ||
| * @param T The return type of the code block. | ||
| * @param signature A unique identifier (e.g., method name) for tracking and logging purposes. | ||
| * @param block The suspending code block to measure. | ||
| * | ||
| * @return The result of the code block. | ||
| */ | ||
| suspend fun <T> track( | ||
| signature: String, | ||
| block: suspend () -> T, | ||
| ): T { | ||
| val (result, executionTime) = measureTimedValue { block() } | ||
| val calculator = | ||
| movingAverages.computeIfAbsent(signature) { | ||
| MovingAverageCalculator() | ||
| } | ||
|
|
||
| calculator.add(executionTime.inWholeMilliseconds) | ||
| logger.trace { "$signature took $executionTime" } | ||
|
|
||
| return result | ||
| } | ||
|
|
||
| /** | ||
| * Provides the average execution times for all monitored methods. | ||
| * | ||
| * @return A map where the keys are method signatures and the values are the average execution times. | ||
| */ | ||
| val averages get() = movingAverages.mapValues { it.value.average } | ||
| } | ||
|
|
||
| /** | ||
| * Convenience function to track execution time of a code block using [TimeTracker]. | ||
| * | ||
| * @param T The return type of the code block. | ||
| * @param signature A unique identifier (e.g., method name) for tracking and logging purposes. | ||
| * @param block The suspending code block to measure. | ||
| * | ||
| * @return The result of the code block. | ||
| */ | ||
| suspend fun <T> timed( | ||
| signature: String, | ||
| block: suspend () -> T, | ||
| ): T = TimeTracker.track(signature, block) |
Oops, something went wrong.