feat: worker stats

Taskfile.yml

@@ -20,6 +20,8 @@ tasks:
     env:
       TEST_BY_TASK: true
     cmds:
+      - docker pull atmoz/sftp:alpine
+      - docker pull axllent/mailpit 
       - echo "Execute unit tests in Go."
       - gosec .
       - go test -tags=runOnTask -cover -v ./...

devenv.nix

@@ -336,6 +336,8 @@ tasks:
     env:
       TEST_BY_TASK: true
     cmds:
+      - docker pull atmoz/sftp:alpine
+      - docker pull axllent/mailpit 
       - echo "Execute unit tests in Go."
       - gosec .
       - go test -tags=runOnTask -cover -v ./...

queue_test.go

@@ -145,8 +145,8 @@ func TestProcessedJobs(t *testing.T) {
 		t.Fatalf("Dequeue failed: %v", err)
 	}
 
-	for _, jobInfo := range q.processedJobs {
-		if jobInfo.ID == job1.GetID() {
+	for _, jobInfo := range q.readyQueue {
+		if jobInfo.GetID() == job1.GetID() {
 			t.Fatalf("Job 1 should not be in processedJobs")
 		}
 	}

worker.go

@@ -2,6 +2,7 @@ package jobqueue
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"github.com/google/uuid"
 	"sync"
@@ -33,6 +34,29 @@ type Worker interface {
 	SetManager(manager *Manager)
 }
 
+type Statistic struct {
+	TotalThreads       int
+	ActiveThreads      int
+	JobsAssigned       int
+	JobsCompleted      int
+	FailedJobs         int
+	TotalExecutionTime time.Duration
+}
+
+func (s *Statistic) AverageExecutionTime() time.Duration {
+	if s.JobsCompleted == 0 {
+		return 0
+	}
+	return s.TotalExecutionTime / time.Duration(s.JobsCompleted)
+}
+
+func (s *Statistic) UtilizationRate() float64 {
+	if s.TotalThreads == 0 {
+		return 0
+	}
+	return float64(s.ActiveThreads) / float64(s.TotalThreads) * 100
+}
+
 // GenericWorker is a generic worker
 type GenericWorker struct {
 	ID     WorkerID
@@ -47,8 +71,10 @@ type LocalWorker struct {
 	cancelChans []chan bool
 	maxJobs     int
 	mu          sync.Mutex
+	statisticMu sync.Mutex
 	wg          sync.WaitGroup
 	manager     *Manager
+	statistic   Statistic
 }
 
 // GetID returns the ID of the worker
@@ -58,7 +84,7 @@ func (w *GenericWorker) GetID() WorkerID {
 
 // NewLocalWorker creates a new local worker
 func NewLocalWorker(maxJobs int) *LocalWorker {
-	w := &LocalWorker{maxJobs: maxJobs}
+	w := &LocalWorker{maxJobs: maxJobs, statistic: Statistic{TotalThreads: maxJobs}}
 	w.jobChannels = make([]chan GenericJob, maxJobs)
 	w.stopChans = make([]chan bool, maxJobs)
 	w.cancelChans = make([]chan bool, maxJobs)
@@ -94,6 +120,27 @@ func (w *LocalWorker) Start() error {
 	return nil
 }
 
+// UpdateStatisticExtended updates the worker's statistics with job execution details
+func (w *LocalWorker) UpdateStatisticExtended(jobDuration time.Duration, jobFailed bool) {
+	w.statisticMu.Lock()
+	defer w.statisticMu.Unlock()
+
+	if jobFailed {
+		w.statistic.FailedJobs++
+	} else {
+		w.statistic.TotalExecutionTime += jobDuration
+		w.statistic.JobsCompleted++
+	}
+
+}
+
+// GetStatistic returns the current statistics of the worker
+func (w *LocalWorker) GetStatistic() Statistic {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.statistic
+}
+
 func (w *LocalWorker) SetManager(manager *Manager) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
@@ -136,10 +183,18 @@ func (w *LocalWorker) run(jobChannel chan GenericJob, stopChan chan bool, cancel
 
 		select {
 		case job := <-jobChannel:
+
+			w.statisticMu.Lock()
+			w.statistic.JobsAssigned++
+			w.statistic.ActiveThreads++
+			w.statisticMu.Unlock()
+
 			ctx, cancel := context.WithCancel(context.Background())
 			retries := job.GetMaxRetries()
 			retryDelay := job.GetRetryDelay()
 
+			startTime := time.Now()
+
 			if retries == 0 {
 				retries = 1
 			}
@@ -159,12 +214,18 @@ func (w *LocalWorker) run(jobChannel chan GenericJob, stopChan chan bool, cancel
 				}
 
 				_, err = job.Execute(ctx)
+				jobFailed := false
+				if err != nil {
+					jobFailed = true
+				}
+
+				w.UpdateStatisticExtended(time.Since(startTime), jobFailed)
 
 				if cancel != nil {
 					cancel()
 				}
 
-				if err == nil || ctx.Err() == context.Canceled {
+				if err == nil || errors.Is(ctx.Err(), context.Canceled) {
 					break
 				}
 
@@ -186,6 +247,10 @@ func (w *LocalWorker) run(jobChannel chan GenericJob, stopChan chan bool, cancel
 				}
 			}
 
+			w.statisticMu.Lock()
+			w.statistic.ActiveThreads--
+			w.statisticMu.Unlock()
+
 		case <-stopChan:
 			stopFlag = true
 			break

worker_test.go

@@ -9,6 +9,7 @@ import (
 
 type DummyJob struct {
 	id        JobID
+	sleepTime time.Duration
 }
 
 func (j DummyJob) GetID() JobID {
@@ -48,7 +49,9 @@ func (j DummyJob) GetTimeout() time.Duration {
 }
 
 func (j DummyJob) Execute(_ context.Context) (RunGenericResult, error) {
-	time.Sleep(100 * time.Millisecond)
+	if j.sleepTime > 0 {
+		time.Sleep(j.sleepTime)
+	}
 	return nil, nil
 }
 
@@ -231,3 +234,46 @@ func TestCancelJob(t *testing.T) {
 		t.Errorf("Stop() returned error: %v", err)
 	}
 }
+
+func TestLocalWorker_Statistics(t *testing.T) {
+	maxJobs := 3
+	worker := NewLocalWorker(maxJobs)
+
+	// Start the worker
+	if err := worker.Start(); err != nil {
+		t.Fatalf("failed to start worker: %v", err)
+	}
+
+	// Assign a job with a longer execution time
+	mockJob1 := DummyJob{id: JobID("1"), sleepTime: 30 * time.Millisecond}
+	if err := worker.AssignJob(mockJob1); err != nil {
+		t.Fatalf("failed to assign job: %v", err)
+	}
+	// Assign a job with a longer execution time
+	mockJob2 := DummyJob{id: JobID("2"), sleepTime: 10 * time.Millisecond}
+	if err := worker.AssignJob(mockJob2); err != nil {
+		t.Fatalf("failed to assign job: %v", err)
+	}
+
+	// Loop to check statistics every 10ms
+	timeout := time.After(100 * time.Millisecond)
+	tick := time.Tick(1 * time.Nanosecond)
+
+	for {
+		select {
+		case <-timeout:
+			t.Fatal("Test timed out")
+		case <-tick:
+			stats := worker.GetStatistic()
+			if stats.JobsCompleted >= 2 {
+				goto TEST_SUCCESS
+			}
+		}
+	}
+
+TEST_SUCCESS:
+	// Stop the worker
+	if err := worker.Stop(); err != nil {
+		t.Fatalf("failed to stop worker: %v", err)
+	}
+}