Skip to content

Commit 6466e27

Browse files
JanCizmarclaude
JanCizmar
and
claude
committed
feat: add batch job timing instrumentation for e2e performance testing
Add AOP-based timing instrumentation (gated behind tolgee.internal.controller-enabled) with internal API endpoints for profiling batch job bottlenecks. Add per-instance and aggregated timing report output to the e2e batch job performance script. - Add BatchJobTimerProvider interface and BatchJobOperationTimer impl - Add BatchJobTimingAspect for AOP-based state provider timing - Add timing endpoints to InternalBatchJobController - Add NO_OP and NO_OP_EXCLUSIVE job types for benchmarking - Add multi-job and exclusive mode support to perf test script - Add project lock caching with timing instrumentation - Add timing wrappers to BatchJobConcurrentLauncher Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1 parent 007f2f6 commit 6466e27

12 files changed

Lines changed: 869 additions & 80 deletions

File tree

‎backend/data/src/main/kotlin/io/tolgee/batch/BatchJobChunkExecutionQueue.kt‎

Lines changed: 52 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import io.tolgee.Metrics
55
import io.tolgee.batch.data.BatchJobChunkExecutionDto
66
import io.tolgee.batch.data.ExecutionQueueItem
77
import io.tolgee.batch.data.QueueEventType
8+
import io.tolgee.batch.timing.BatchJobTimerProvider
89
import io.tolgee.batch.events.JobQueueItemsEvent
910
import io.tolgee.component.UsingRedisProvider
1011
import io.tolgee.configuration.tolgee.BatchProperties
@@ -36,6 +37,7 @@ class BatchJobChunkExecutionQueue(
3637
@Lazy
3738
private val redisTemplate: StringRedisTemplate,
3839
private val metrics: Metrics,
40+
private val timer: BatchJobTimerProvider?,
3941
) : Logging,
4042
InitializingBean {
4143
companion object {
@@ -57,6 +59,11 @@ class BatchJobChunkExecutionQueue(
5759
private var lastServedJobId: Long? = null
5860
}
5961

62+
private fun <T> timed(
63+
name: String,
64+
block: () -> T,
65+
): T = timer?.measure(name, block) ?: block()
66+
6067
private fun incrementCharacterCount(character: JobCharacter) {
6168
jobCharacterCounts.computeIfAbsent(character) { AtomicInteger(0) }.incrementAndGet()
6269
}
@@ -75,9 +82,11 @@ class BatchJobChunkExecutionQueue(
7582
QueueEventType.REMOVE -> {
7683
// Remove and decrement atomically per item to prevent double-decrement
7784
// if poll() removes an item between removeAll and forEach
78-
event.items.forEach { item ->
79-
if (queue.remove(item)) {
80-
decrementCharacterCount(item.jobCharacter)
85+
timed("QUEUE_INTERNAL.removeConsuming") {
86+
event.items.forEach { item ->
87+
if (queue.remove(item)) {
88+
decrementCharacterCount(item.jobCharacter)
89+
}
8190
}
8291
}
8392
}
@@ -117,7 +126,10 @@ class BatchJobChunkExecutionQueue(
117126
}
118127

119128
fun addExecutionsToLocalQueue(data: List<BatchJobChunkExecutionDto>) {
120-
val ids = queue.map { it.chunkExecutionId }.toSet()
129+
val ids =
130+
timed("QUEUE_INTERNAL.buildIdSet") {
131+
queue.map { it.chunkExecutionId }.toSet()
132+
}
121133
var count = 0
122134
data.forEach {
123135
if (!ids.contains(it.id)) {
@@ -133,7 +145,10 @@ class BatchJobChunkExecutionQueue(
133145

134146
fun addItemsToLocalQueue(data: List<ExecutionQueueItem>) {
135147
// Use Set for O(1) lookup instead of O(n) queue.contains()
136-
val existingIds = queue.mapTo(HashSet()) { it.chunkExecutionId }
148+
val existingIds =
149+
timed("QUEUE_INTERNAL.buildIdSet") {
150+
queue.mapTo(HashSet()) { it.chunkExecutionId }
151+
}
137152
val toAdd = mutableListOf<ExecutionQueueItem>()
138153
var filteredOutCount = 0
139154

@@ -188,12 +203,14 @@ class BatchJobChunkExecutionQueue(
188203

189204
fun removeJobExecutions(jobId: Long) {
190205
logger.debug("Removing job $jobId from queue, queue size: ${queue.size}")
191-
val iterator = queue.iterator()
192-
while (iterator.hasNext()) {
193-
val item = iterator.next()
194-
if (item.jobId == jobId) {
195-
iterator.remove()
196-
decrementCharacterCount(item.jobCharacter)
206+
timed("QUEUE_INTERNAL.removeJobExecutions") {
207+
val iterator = queue.iterator()
208+
while (iterator.hasNext()) {
209+
val item = iterator.next()
210+
if (item.jobId == jobId) {
211+
iterator.remove()
212+
decrementCharacterCount(item.jobCharacter)
213+
}
197214
}
198215
}
199216
logger.debug("Removed job $jobId from queue, queue size: ${queue.size}")
@@ -237,8 +254,11 @@ class BatchJobChunkExecutionQueue(
237254
return null
238255
}
239256

240-
// Get distinct job IDs in queue order
241-
val jobIds = queue.mapTo(LinkedHashSet()) { it.jobId }.toList()
257+
// Get distinct job IDs in queue order — O(n) scan of entire queue
258+
val jobIds =
259+
timed("QUEUE_INTERNAL.collectJobIds") {
260+
queue.mapTo(LinkedHashSet()) { it.jobId }.toList()
261+
}
242262
if (jobIds.isEmpty()) {
243263
return null
244264
}
@@ -258,13 +278,22 @@ class BatchJobChunkExecutionQueue(
258278
val jobIndex = (startIndex + i) % jobIds.size
259279
val targetJobId = jobIds[jobIndex]
260280

261-
// Find first item for this job and try to remove it
262-
val item = queue.firstOrNull { it.jobId == targetJobId }
263-
if (item != null && queue.remove(item)) {
264-
// Successfully removed - update state and return
265-
decrementCharacterCount(item.jobCharacter)
266-
lastServedJobId = targetJobId
267-
return item
281+
// Find first item for this job — O(n) scan to find matching item
282+
val item =
283+
timed("QUEUE_INTERNAL.findFirstForJob") {
284+
queue.firstOrNull { it.jobId == targetJobId }
285+
}
286+
if (item != null) {
287+
// Remove item — O(n) scan for ConcurrentLinkedQueue.remove()
288+
val removed =
289+
timed("QUEUE_INTERNAL.removeItem") {
290+
queue.remove(item)
291+
}
292+
if (removed) {
293+
decrementCharacterCount(item.jobCharacter)
294+
lastServedJobId = targetJobId
295+
return item
296+
}
268297
}
269298
// Item was null or already removed by another thread, try next job
270299
}
@@ -298,7 +327,9 @@ class BatchJobChunkExecutionQueue(
298327
}
299328

300329
fun getQueuedJobItems(jobId: Long): List<ExecutionQueueItem> {
301-
return queue.filter { it.jobId == jobId }
330+
return timed("QUEUE_INTERNAL.getQueuedJobItems") {
331+
queue.filter { it.jobId == jobId }
332+
}
302333
}
303334

304335
fun getAllQueueItems(): List<ExecutionQueueItem> {

‎backend/data/src/main/kotlin/io/tolgee/batch/BatchJobConcurrentLauncher.kt‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package io.tolgee.batch
33
import io.sentry.Sentry
44
import io.tolgee.batch.data.BatchJobDto
55
import io.tolgee.batch.data.ExecutionQueueItem
6+
import io.tolgee.batch.timing.BatchJobTimerProvider
67
import io.tolgee.component.CurrentDateProvider
78
import io.tolgee.configuration.tolgee.BatchProperties
89
import io.tolgee.fixtures.waitFor
@@ -31,6 +32,7 @@ class BatchJobConcurrentLauncher(
3132
private val progressManager: ProgressManager,
3233
private val batchJobActionService: BatchJobActionService,
3334
private val tracingContext: TolgeeTracingContext,
35+
private val timerProvider: BatchJobTimerProvider?,
3436
) : Logging {
3537
companion object {
3638
const val MIN_TIME_BETWEEN_OPERATIONS = 100
@@ -203,7 +205,11 @@ class BatchJobConcurrentLauncher(
203205
/**
204206
* Only single job can run in project at the same time
205207
*/
206-
if (!batchJobProjectLockingManager.canLockJobForProject(executionItem.jobId)) {
208+
val canLock =
209+
timed("LAUNCHER.projectLockCheck") {
210+
batchJobProjectLockingManager.canLockJobForProject(executionItem.jobId)
211+
}
212+
if (!canLock) {
207213
logger.debug(
208214
"⚠️ Cannot run execution ${executionItem.chunkExecutionId}. " +
209215
"Other job from the project is currently running, skipping",
@@ -299,6 +305,13 @@ class BatchJobConcurrentLauncher(
299305
return runningCount < allowedCharacterCounts
300306
}
301307

308+
private fun <T> timed(
309+
operationName: String,
310+
block: () -> T,
311+
): T {
312+
return timerProvider?.measure(operationName, block) ?: block()
313+
}
314+
302315
private fun ExecutionQueueItem.trySetRunningState(batchJobDto: BatchJobDto): Boolean {
303316
// Check maxPerJobConcurrency before trying to set running state
304317
val maxPerJobConcurrency = batchJobDto.maxPerJobConcurrency

‎backend/data/src/main/kotlin/io/tolgee/batch/BatchJobProjectLockingManager.kt‎

Lines changed: 55 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package io.tolgee.batch
22

33
import io.tolgee.batch.data.BatchJobDto
4+
import io.tolgee.batch.timing.BatchJobTimerProvider
45
import io.tolgee.component.UsingRedisProvider
56
import io.tolgee.util.Logging
67
import io.tolgee.util.logger
@@ -22,27 +23,43 @@ class BatchJobProjectLockingManager(
2223
@Lazy
2324
private val redissonClient: RedissonClient,
2425
private val usingRedisProvider: UsingRedisProvider,
26+
private val timerProvider: BatchJobTimerProvider?,
2527
) : Logging {
2628
companion object {
2729
private val localProjectLocks by lazy {
2830
ConcurrentHashMap<Long, Long?>()
2931
}
32+
33+
/**
34+
* Local cache of Redis lock state per project, to avoid redundant Redis round-trips.
35+
* Key: projectId, Value: (lockedJobId, timestampMs).
36+
* Entries expire after [LOCK_CACHE_TTL_MS] and are eagerly invalidated on unlock.
37+
*/
38+
private val redisLockCache = ConcurrentHashMap<Long, LockCacheEntry>()
39+
private const val LOCK_CACHE_TTL_MS = 1000L
3040
}
3141

42+
private data class LockCacheEntry(
43+
val lockedJobId: Long,
44+
val timestamp: Long,
45+
)
46+
3247
fun canLockJobForProject(batchJobId: Long): Boolean {
33-
val jobDto = batchJobService.getJobDto(batchJobId)
34-
if (!jobDto.type.exclusive) {
35-
return true
48+
return timed("PROJECT_LOCK.canLock") {
49+
val jobDto = batchJobService.getJobDto(batchJobId)
50+
if (!jobDto.type.exclusive) {
51+
return@timed true
52+
}
53+
tryLockJobForProject(jobDto)
3654
}
37-
return tryLockJobForProject(jobDto)
3855
}
3956

4057
private fun tryLockJobForProject(jobDto: BatchJobDto): Boolean {
4158
logger.debug("Trying to lock job ${jobDto.id} for project ${jobDto.projectId}")
4259
return if (usingRedisProvider.areWeUsingRedis) {
43-
tryLockWithRedisson(jobDto)
60+
timed("PROJECT_LOCK.tryLockRedis") { tryLockWithRedisson(jobDto) }
4461
} else {
45-
tryLockLocal(jobDto)
62+
timed("PROJECT_LOCK.tryLockLocal") { tryLockLocal(jobDto) }
4663
}
4764
}
4865

@@ -51,14 +68,18 @@ class BatchJobProjectLockingManager(
5168
jobId: Long,
5269
) {
5370
projectId ?: return
54-
getMap().compute(projectId) { _, lockedJobId ->
55-
logger.debug("Unlocking job: $jobId for project $projectId")
56-
if (lockedJobId == jobId) {
71+
timed("PROJECT_LOCK.unlock") {
72+
// Eagerly invalidate the local cache so other jobs can acquire the lock immediately
73+
redisLockCache.remove(projectId)
74+
getMap().compute(projectId) { _, lockedJobId ->
5775
logger.debug("Unlocking job: $jobId for project $projectId")
58-
return@compute 0L
76+
if (lockedJobId == jobId) {
77+
logger.debug("Unlocking job: $jobId for project $projectId")
78+
return@compute 0L
79+
}
80+
logger.debug("Job: $jobId for project $projectId is not locked")
81+
return@compute lockedJobId
5982
}
60-
logger.debug("Job: $jobId for project $projectId is not locked")
61-
return@compute lockedJobId
6283
}
6384
}
6485

@@ -71,10 +92,24 @@ class BatchJobProjectLockingManager(
7192

7293
private fun tryLockWithRedisson(batchJobDto: BatchJobDto): Boolean {
7394
val projectId = batchJobDto.projectId ?: return true
95+
96+
// Check local cache first to avoid Redis round-trip for the common "still locked" case
97+
val cached = redisLockCache[projectId]
98+
if (cached != null &&
99+
cached.lockedJobId != batchJobDto.id &&
100+
cached.lockedJobId != 0L &&
101+
System.currentTimeMillis() - cached.timestamp < LOCK_CACHE_TTL_MS
102+
) {
103+
timed("PROJECT_LOCK.cacheHit") {}
104+
return false
105+
}
106+
74107
val computed =
75108
getRedissonProjectLocks().compute(projectId) { _, value ->
76109
computeFnBody(batchJobDto, value)
77110
}
111+
// Update local cache with the result
112+
redisLockCache[projectId] = LockCacheEntry(computed ?: 0L, System.currentTimeMillis())
78113
return computed == batchJobDto.id
79114
}
80115

@@ -116,7 +151,7 @@ class BatchJobProjectLockingManager(
116151
if (currentValue == null) {
117152
logger.debug("Getting initial locked state from DB state")
118153
// we have to find out from database if there is any running job for the project
119-
val initial = getInitialJobId(projectId)
154+
val initial = timed("PROJECT_LOCK.getInitialJobId") { getInitialJobId(projectId) }
120155
logger.debug("Initial locked job $initial for project ${toLock.projectId}")
121156
if (initial == null) {
122157
logger.debug("No job found, locking ${toLock.id}")
@@ -171,4 +206,11 @@ class BatchJobProjectLockingManager(
171206
fun getLockedJobIds(): Set<Long> {
172207
return getMap().values.filterNotNull().toSet()
173208
}
209+
210+
private fun <T> timed(
211+
operationName: String,
212+
block: () -> T,
213+
): T {
214+
return timerProvider?.measure(operationName, block) ?: block()
215+
}
174216
}

‎backend/data/src/main/kotlin/io/tolgee/batch/data/BatchJobType.kt‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -114,4 +114,9 @@ enum class BatchJobType(
114114
processor = NoOpChunkProcessor::class,
115115
exclusive = false,
116116
),
117+
NO_OP_EXCLUSIVE(
118+
activityType = null,
119+
maxRetries = 0,
120+
processor = NoOpChunkProcessor::class,
121+
),
117122
}

‎backend/data/src/main/kotlin/io/tolgee/batch/processors/NoOpChunkProcessor.kt‎

Lines changed: 13 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -7,25 +7,32 @@ import io.tolgee.batch.request.NoOpRequest
77
import org.springframework.stereotype.Component
88
import kotlin.coroutines.CoroutineContext
99

10+
data class NoOpParams(
11+
val chunkProcessingDelayMs: Long = 0,
12+
)
13+
1014
@Component
1115
class NoOpChunkProcessor(
1216
private val progressManager: ProgressManager,
13-
) : ChunkProcessor<NoOpRequest, Any?, Long> {
17+
) : ChunkProcessor<NoOpRequest, NoOpParams, Long> {
1418
override fun process(
1519
job: BatchJobDto,
1620
chunk: List<Long>,
1721
coroutineContext: CoroutineContext,
1822
) {
19-
// Report progress for the whole chunk at once
23+
val params = getParams(job)
24+
if (params.chunkProcessingDelayMs > 0) {
25+
Thread.sleep(params.chunkProcessingDelayMs)
26+
}
2027
progressManager.reportSingleChunkProgress(job.id, chunk.size)
2128
}
2229

23-
override fun getParamsType(): Class<Any?>? {
24-
return null
30+
override fun getParamsType(): Class<NoOpParams> {
31+
return NoOpParams::class.java
2532
}
2633

27-
override fun getParams(data: NoOpRequest): Any? {
28-
return null
34+
override fun getParams(data: NoOpRequest): NoOpParams {
35+
return NoOpParams(chunkProcessingDelayMs = data.chunkProcessingDelayMs)
2936
}
3037

3138
override fun getTargetItemType(): Class<Long> {

‎backend/data/src/main/kotlin/io/tolgee/batch/request/NoOpRequest.kt‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,4 +5,12 @@ import jakarta.validation.constraints.NotEmpty
55
class NoOpRequest {
66
@NotEmpty
77
var itemIds: List<Long> = listOf()
8+
var chunkProcessingDelayMs: Long = 0
9+
}
10+
11+
class NoOpMultiRequest {
12+
var totalItems: Int = 10000
13+
var numberOfJobs: Int = 1
14+
var chunkProcessingDelayMs: Long = 0
15+
var numberOfProjects: Int = 0
816
}

‎backend/data/src/main/kotlin/io/tolgee/batch/timing/BatchJobTimerProvider.kt‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
package io.tolgee.batch.timing
2+
3+
/**
4+
* Interface for inline timing of internal operations.
5+
* Implemented by BatchJobOperationTimer in the development module.
6+
* When not available (production), operations run without timing overhead.
7+
*/
8+
interface BatchJobTimerProvider {
9+
fun <T> measure(
10+
operationName: String,
11+
block: () -> T,
12+
): T
13+
}

‎backend/development/build.gradle‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@ dependencies {
6666
implementation 'org.springframework.boot:spring-boot-starter-web'
6767
implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
6868
implementation 'org.springframework.boot:spring-boot-starter-hateoas'
69+
implementation 'org.springframework.boot:spring-boot-starter-aop'
6970
implementation "org.springframework.boot:spring-boot-configuration-processor"
7071
implementation project(":data")
7172
implementation project(":api")

0 commit comments

Comments
 (0)