Skip to content

Commit 43fd83b

Browse files
nnn-gifnnn-gif
committed
feat: update logging for worker pool
1 parent 8e120d5 commit 43fd83b

1 file changed

Lines changed: 36 additions & 27 deletions

File tree

‎services/bridge/internal/worker/worker_pool.go‎

Lines changed: 36 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -118,7 +118,7 @@ func (wp *WorkerPool) Start(ctx context.Context) {
118118
// Start health monitor goroutine
119119
go wp.healthMonitor(ctx)
120120

121-
logger.Infof("Started worker pool with %d workers", wp.maxWorkers)
121+
logger.Infof("[router=%s] Started worker pool with %d workers", wp.routerID, wp.maxWorkers)
122122
}
123123

124124
// Stop stops the worker pool
@@ -130,7 +130,7 @@ func (wp *WorkerPool) Stop(ctx context.Context) {
130130
return
131131
}
132132

133-
logger.Info("Stopping worker pool")
133+
logger.Infof("[router=%s] Stopping worker pool", wp.routerID)
134134

135135
// Signal shutdown
136136
close(wp.shutdownChan)
@@ -149,9 +149,9 @@ func (wp *WorkerPool) Stop(ctx context.Context) {
149149

150150
select {
151151
case <-done:
152-
logger.Info("All workers stopped")
152+
logger.Infof("[router=%s] All workers stopped", wp.routerID)
153153
case <-ctx.Done():
154-
logger.Warn("Worker pool shutdown timed out")
154+
logger.Warnf("[router=%s] Worker pool shutdown timed out", wp.routerID)
155155
}
156156

157157
wp.running = false
@@ -191,24 +191,24 @@ func (wp *WorkerPool) healthMonitor(ctx context.Context) {
191191
wp.maxWorkers, wp.routerID, queueSize, activeCount)
192192
}
193193

194-
logger.Debugf("Worker pool health: active=%d/%d, queue=%d/%d",
195-
activeCount, wp.maxWorkers, queueSize, queueCap)
194+
logger.Debugf("[router=%s] Worker pool health: active=%d/%d, queue=%d/%d",
195+
wp.routerID, activeCount, wp.maxWorkers, queueSize, queueCap)
196196
}
197197
}
198198
}
199199

200200
// Submit submits a task to the worker pool
201201
func (wp *WorkerPool) Submit(task *WorkerTask) {
202202
if task == nil {
203-
logger.Error("Cannot submit nil task to worker pool")
203+
logger.Errorf("[router=%s] Cannot submit nil task to worker pool", wp.routerID)
204204
return
205205
}
206206

207207
wp.mu.RLock()
208208
defer wp.mu.RUnlock()
209209

210210
if !wp.running {
211-
logger.Warn("Worker pool not running, dropping task")
211+
logger.Warnf("[router=%s] Worker pool not running, dropping task", wp.routerID)
212212
return
213213
}
214214

@@ -218,7 +218,7 @@ func (wp *WorkerPool) Submit(task *WorkerTask) {
218218
wp.pendingTasks = append(wp.pendingTasks, task)
219219
wp.pendingMu.Unlock()
220220
queueSize := len(wp.taskQueue)
221-
logger.Debugf("Task %s queued (queue: %d/%d)", task.ID, queueSize, cap(wp.taskQueue))
221+
logger.Debugf("[router=%s] Task %s queued (queue: %d/%d)", wp.routerID, task.ID, queueSize, cap(wp.taskQueue))
222222
// Update queue size metric
223223
if wp.metricsCollector != nil {
224224
wp.metricsCollector.SetTaskQueueSize(int32(queueSize))
@@ -230,8 +230,8 @@ func (wp *WorkerPool) Submit(task *WorkerTask) {
230230
if task.Request != nil && task.Request.Intent != nil {
231231
symbol = task.Request.Intent.Symbol
232232
}
233-
logger.Errorf("CRITICAL: Task queue full (%d/%d), DROPPING task %s for symbol %s - consider increasing queue size or worker count",
234-
queueLen, queueCap, task.ID, symbol)
233+
logger.Errorf("[router=%s] CRITICAL: Task queue full (%d/%d), DROPPING task %s for symbol %s - consider increasing queue size or worker count",
234+
wp.routerID, queueLen, queueCap, task.ID, symbol)
235235
// Record dropped task metric
236236
if wp.metricsCollector != nil {
237237
wp.metricsCollector.IncWorkerTasksDropped()
@@ -243,17 +243,18 @@ func (wp *WorkerPool) Submit(task *WorkerTask) {
243243
func (w *Worker) start(ctx context.Context) {
244244
defer w.wg.Done()
245245

246-
logger.Debugf("Worker %d started", w.id)
246+
logger.Debugf("[router=%s][WORKER-%d] started", w.pool.routerID, w.id)
247247

248248
for {
249249
select {
250250
case <-ctx.Done():
251-
logger.Debugf("Worker %d stopped due to context cancellation", w.id)
251+
logger.Debugf("[router=%s][WORKER-%d] stopped due to context cancellation", w.pool.routerID, w.id)
252252
return
253253
case <-w.quit:
254-
logger.Debugf("Worker %d stopped due to quit signal", w.id)
254+
logger.Debugf("[router=%s][WORKER-%d] stopped due to quit signal", w.pool.routerID, w.id)
255255
return
256256
case task := <-w.taskQueue:
257+
logger.Debugf("[router=%s][WORKER-%d] picked up task %s", w.pool.routerID, w.id, task.ID)
257258
w.pool.removePending(task.ID)
258259
w.processTask(ctx, task)
259260
}
@@ -282,21 +283,23 @@ func (wp *WorkerPool) GetStats() WorkerPoolStats {
282283
func (wp *WorkerPool) removePending(taskID string) {
283284
wp.pendingMu.Lock()
284285
defer wp.pendingMu.Unlock()
286+
285287
for i, t := range wp.pendingTasks {
286288
if t.ID == taskID {
287289
wp.pendingTasks = append(wp.pendingTasks[:i], wp.pendingTasks[i+1:]...)
290+
logger.Debugf("[router=%s] Removed task %s from pending list (remaining: %d)", wp.routerID, taskID, len(wp.pendingTasks))
288291
return
289292
}
290293
}
294+
logger.Warnf("[router=%s] Task %s not found in pending list when trying to remove", wp.routerID, taskID)
291295
}
292296

293-
// ListPendingTasks returns a snapshot of all tasks currently waiting in the queue
294297
func (wp *WorkerPool) ListPendingTasks() []TaskInfo {
295298
wp.pendingMu.RLock()
296299
defer wp.pendingMu.RUnlock()
297300

298-
result := make([]TaskInfo, len(wp.pendingTasks))
299-
for i, task := range wp.pendingTasks {
301+
result := make([]TaskInfo, 0, len(wp.pendingTasks))
302+
for _, task := range wp.pendingTasks {
300303
info := TaskInfo{ID: task.ID}
301304
if task.Request != nil {
302305
info.RouterID = task.Request.RouterID
@@ -311,8 +314,14 @@ func (wp *WorkerPool) ListPendingTasks() []TaskInfo {
311314
info.ChainID = task.Request.DestinationChain.ChainID
312315
}
313316
}
314-
result[i] = info
317+
result = append(result, info)
318+
}
319+
320+
queueLen := len(wp.taskQueue)
321+
if len(result) != queueLen {
322+
logger.Warnf("[router=%s] Pending tasks list out of sync: pendingTasks=%d, taskQueue=%d", wp.routerID, len(result), queueLen)
315323
}
324+
316325
return result
317326
}
318327

@@ -337,8 +346,8 @@ func (w *Worker) processTask(ctx context.Context, task *WorkerTask) {
337346
chainID = task.Request.DestinationChain.ChainID
338347
}
339348
}
340-
logger.Infof("[WORKER-%d] Starting task: %s, router=%s, symbol=%s, chain=%d, active_workers=%d",
341-
w.id, task.ID, routerID, symbol, chainID, atomic.LoadInt32(&w.pool.activeWorkers))
349+
logger.Infof("[router=%s][WORKER-%d] Starting task: %s, symbol=%s, chain=%d, active_workers=%d",
350+
w.pool.routerID, w.id, task.ID, symbol, chainID, atomic.LoadInt32(&w.pool.activeWorkers))
342351

343352
// Create timeout context to prevent workers from blocking forever
344353
taskCtx, cancel := context.WithTimeout(ctx, w.pool.taskTimeout)
@@ -351,7 +360,7 @@ func (w *Worker) processTask(ctx context.Context, task *WorkerTask) {
351360
for retry := 0; retry < maxRetries; retry++ {
352361
if retry > 0 {
353362
retryCount++
354-
logger.Debugf("Worker %d retrying task %s (attempt %d/%d)", w.id, task.ID, retry+1, maxRetries)
363+
logger.Debugf("[router=%s][WORKER-%d] retrying task %s (attempt %d/%d)", w.pool.routerID, w.id, task.ID, retry+1, maxRetries)
355364
time.Sleep(time.Second * time.Duration(retry))
356365
}
357366

@@ -362,26 +371,26 @@ func (w *Worker) processTask(ctx context.Context, task *WorkerTask) {
362371

363372
// Don't retry on context timeout/cancellation
364373
if taskCtx.Err() != nil {
365-
logger.Warnf("[WORKER-%d] Task %s context expired, not retrying: %v", w.id, task.ID, taskCtx.Err())
374+
logger.Warnf("[router=%s][WORKER-%d] Task %s context expired, not retrying: %v", w.pool.routerID, w.id, task.ID, taskCtx.Err())
366375
err = taskCtx.Err()
367376
break
368377
}
369378

370-
logger.Errorf("Worker %d task %s failed (attempt %d/%d): %v", w.id, task.ID, retry+1, maxRetries, err)
379+
logger.Errorf("[router=%s][WORKER-%d] Task %s failed (attempt %d/%d): %v", w.pool.routerID, w.id, task.ID, retry+1, maxRetries, err)
371380
}
372381

373382
duration := time.Since(startTime)
374383

375384
if err != nil {
376-
logger.Errorf("[WORKER-%d] Task FAILED after %d retries: %s, router=%s, symbol=%s, chain=%d, duration=%v, error=%v",
377-
w.id, maxRetries, task.ID, routerID, symbol, chainID, duration, err)
385+
logger.Errorf("[router=%s][WORKER-%d] Task FAILED after %d retries: %s, symbol=%s, chain=%d, duration=%v, error=%v",
386+
w.pool.routerID, w.id, maxRetries, task.ID, symbol, chainID, duration, err)
378387
if w.metricsCollector != nil {
379388
w.metricsCollector.IncWorkerTasksFailed()
380389
w.metricsCollector.ObserveTaskProcessingDuration(duration.Seconds())
381390
}
382391
} else {
383-
logger.Infof("[WORKER-%d] Task COMPLETED: %s, router=%s, symbol=%s, chain=%d, duration=%v, retries=%d",
384-
w.id, task.ID, routerID, symbol, chainID, duration, retryCount)
392+
logger.Infof("[router=%s][WORKER-%d] Task COMPLETED: %s, symbol=%s, chain=%d, duration=%v, retries=%d",
393+
w.pool.routerID, w.id, task.ID, symbol, chainID, duration, retryCount)
385394
if w.metricsCollector != nil {
386395
w.metricsCollector.IncWorkerTasksCompleted()
387396
w.metricsCollector.ObserveTaskProcessingDuration(duration.Seconds())

0 commit comments

Comments
 (0)