chore: debug logs

Author: cisse21

router/batchrouter/handle.go

@@ -168,6 +168,7 @@ func (brt *Handle) mainLoop(ctx context.Context) {
 			return
 		case <-time.After(mainLoopSleep):
 			for _, partition := range brt.activePartitions(ctx) {
+				brt.logger.Infof("Pinging worker for partition: %s", partition)
 				pool.PingWorker(partition)
 			}
 			mainLoopSleep = brt.mainLoopFreq.Load()

router/batchrouter/worker.go

@@ -3,7 +3,9 @@ package batchrouter
 import (
 	"context"
 	"fmt"
+	"runtime"
 	"slices"
+	"strconv"
 	"strings"
 	"time"
 
@@ -32,6 +34,17 @@ type worker struct {
 	brt       *Handle
 }
 
+func goid() int {
+	var buf [64]byte
+	n := runtime.Stack(buf[:], false)
+	idField := strings.Fields(strings.TrimPrefix(string(buf[:n]), "goroutine"))[0]
+	id, err := strconv.Atoi(idField)
+	if err != nil {
+		panic(fmt.Sprintf("cannot get goroutine id: %v", err))
+	}
+	return id
+}
+
 // Work retrieves jobs from batch router for the worker's partition and processes them,
 // grouped by destination and in parallel.
 // The function returns when processing completes and the return value is true if at least 1 job was processed,
@@ -40,6 +53,7 @@ func (w *worker) Work() bool {
 	brt := w.brt
 	workerJobs := brt.getWorkerJobs(w.partition)
 	brt.logger.Infof("Worker %s has %v jobs to process", w.partition, len(workerJobs))
+	brt.logger.Infof("Goroutine ID: %d", goid())
 	if len(workerJobs) == 0 {
 		return false
 	}

utils/workerpool/internal_worker.go

@@ -2,6 +2,10 @@ package workerpool
 
 import (
 	"context"
+	"fmt"
+	"runtime"
+	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -62,10 +66,10 @@ func (w *internalWorker) start() {
 
 			w.setIdleSince(time.Time{})
 			if w.delegate.Work() {
-				w.logger.Infof("worker %q produced work", w.partition)
+				w.logger.Infof("worker %q produced work %d", w.partition, goid())
 				exponentialSleep.Reset()
 			} else {
-				w.logger.Infof("worker %q didn't produce any work", w.partition)
+				w.logger.Infof("worker %q didn't produce any work %d", w.partition, goid())
 				if err := misc.SleepCtx(w.lifecycle.ctx, exponentialSleep.Next(w.delegate.SleepDurations())); err != nil {
 					w.logger.Debugf("worker %q sleep interrupted: %v", w.partition, err)
 					return
@@ -84,6 +88,17 @@ func (w *internalWorker) setIdleSince(t time.Time) {
 	}
 }
 
+func goid() int {
+	var buf [64]byte
+	n := runtime.Stack(buf[:], false)
+	idField := strings.Fields(strings.TrimPrefix(string(buf[:n]), "goroutine"))[0]
+	id, err := strconv.Atoi(idField)
+	if err != nil {
+		panic(fmt.Sprintf("cannot get goroutine id: %v", err))
+	}
+	return id
+}
+
 // Ping triggers the worker to pick more jobs
 func (w *internalWorker) Ping() {
 	w.logger.Debugf("worker %q pinged", w.partition)
@@ -94,9 +109,9 @@ func (w *internalWorker) Ping() {
 	}
 	select {
 	case w.ping <- struct{}{}:
-		w.logger.Infof("worker %q pinged", w.partition)
+		w.logger.Infof("worker %q pinged %d", w.partition, goid())
 	default:
-		w.logger.Infof("worker %q pinged but channel was full", w.partition)
+		w.logger.Infof("worker %q pinged but channel was full %d", w.partition, goid())
 	}
 }