added structured logging

Signed-off-by: redpinecube <tara.chakkithara@icloud.com>

Author: redpinecube

src/api.go

@@ -2,8 +2,9 @@ package main
 
 import (
 	"context"
+	"errors"
 	"fmt"
-	"log"
+	"log/slog"
 	"net/http"
 	"os"
 	"os/signal"
@@ -20,44 +21,50 @@ type App struct {
 	supervisor  *Supervisor
 	httpServer  *http.Server
 	wg          sync.WaitGroup
+	log         *slog.Logger
 }
 
-func NewApp(redisAddr, gpuType string) *App {
+func NewApp(redisAddr, gpuType string, log *slog.Logger) *App {
 	client := redis.NewClient(&redis.Options{Addr: redisAddr})
-	scheduler := NewScheduler(redisAddr)
+	scheduler := NewScheduler(redisAddr, log)
 
 	consumerID := fmt.Sprintf("worker_%d", os.Getpid())
-	supervisor := NewSupervisor(redisAddr, consumerID, gpuType)
+	supervisor := NewSupervisor(redisAddr, consumerID, gpuType, log)
 
 	mux := http.NewServeMux()
 	a := &App{
 		redisClient: client,
 		scheduler:   scheduler,
 		supervisor:  supervisor,
 		httpServer:  &http.Server{Addr: ":3000", Handler: mux},
+		log:         log,
 	}
 
 	mux.HandleFunc("/auth/login", a.login)
 	mux.HandleFunc("/auth/refresh", a.refresh)
 	mux.HandleFunc("/jobs", a.enqueueJob)
 	mux.HandleFunc("/jobs/status", a.getJobStatus)
 
+	a.log.Info("new app initialized", "redis_address", redisAddr,
+		"gpu_type", gpuType, "http_address", a.httpServer.Addr)
+
 	return a
 }
 
 func (a *App) Start() error {
 	// Connect to redis
 	if err := a.redisClient.Ping(context.Background()).Err(); err != nil {
-		return fmt.Errorf("redis ping failed: %w", err)
+		a.log.Error("redis ping failed", "err", err)
+		return err
 	}
-	
+
 	// Launch HTTP server
 	a.wg.Add(1)
 	go func() {
 		defer a.wg.Done()
-		log.Println("HTTP server listening on", a.httpServer.Addr)
+		slog.Info("http server started", "address", a.httpServer.Addr)
 		if err := a.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
-			log.Fatalf("HTTP server error: %v", err)
+			a.log.Error("HTTP server error", "err", err)
 		}
 	}()
 
@@ -66,7 +73,7 @@ func (a *App) Start() error {
 
 func (a *App) Shutdown(ctx context.Context) error {
 	if err := a.httpServer.Shutdown(ctx); err != nil {
-		log.Printf("error shutting down HTTP server: %v", err)
+		a.log.Error("error shutting down HTTP server", "err", err)
 	}
 
 	// Wait for ListenAndServe goroutine to finish
@@ -75,44 +82,61 @@ func (a *App) Shutdown(ctx context.Context) error {
 	a.supervisor.Stop()
 
 	if err := a.scheduler.Close(); err != nil {
-		log.Printf("error closing scheduler: %v", err)
+		a.log.Error("error closing scheduler", "err", err)
+
+	} else {
+		a.log.Info("scheduler closed successfully")
 	}
 
 	if err := a.redisClient.Close(); err != nil {
-		log.Printf("error closing redis client: %v", err)
+		a.log.Error("error closing redis client", "err", err)
+	} else {
+		a.log.Info("redis client closed successfully")
 	}
 
+	a.log.Info("shutdown completed")
+
 	return nil
 }
 
 func main() {
-	app := NewApp("localhost:6379", "AMD")
+	log := slog.New(slog.NewJSONHandler(os.Stdout, nil))
+	app := NewApp("localhost:6379", "AMD", log)
 
 	if err := app.Start(); err != nil {
-		log.Fatalf("failed to start app: %v", err)
+		log.Error("failed to start app", "err", err)
+		os.Exit(1)
 	}
 
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
 	<-ctx.Done()
-	log.Println("shutdown signal received")
+	log.Info("shutdown signal received")
 
 	shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
 	defer cancel()
 	if err := app.Shutdown(shutdownCtx); err != nil {
-		log.Fatalf("shutdown error: %v", err)
+		log.Error("shutdown error", "err", err)
 	}
 
-	log.Println("all services stopped cleanly")
+	log.Info("all services stopped cleanly")
 }
 
 func (a *App) login(w http.ResponseWriter, r *http.Request) {
 	ctx := r.Context()
+	a.log.Info("login handler accessed", "remote_address", r.RemoteAddr)
 	val, err := a.redisClient.Get(ctx, "some:key").Result()
-	if err != nil || err != redis.Nil {
+	if errors.Is(err, redis.Nil) {
+		a.log.Info("redis key not found")
+		http.Error(w, "redis key not found", http.StatusNotFound)
+		return
+	}
+	if err != nil {
+		a.log.Error("redis error on login", "err", err)
 		http.Error(w, "redis error", http.StatusInternalServerError)
 		return
 	}
+	a.log.Info("login success", "remote_address", r.RemoteAddr)
 	fmt.Fprintf(w, "login page; redis says: %q\n", val)
 }
 
@@ -121,14 +145,17 @@ func (a *App) refresh(w http.ResponseWriter, r *http.Request) {
 }
 
 func (a *App) enqueueJob(w http.ResponseWriter, r *http.Request) {
+	a.log.Info("enqueueJob handler accessed", "remote_address", r.RemoteAddr)
 	payload := map[string]interface{}{
 		"task_id": 123,
 		"data":    "test_data_123",
 	}
 	if err := a.scheduler.Enqueue("jobType", payload); err != nil {
+		a.log.Error("enqueue failed", "err", err, "payload", payload)
 		http.Error(w, "enqueue failed", http.StatusInternalServerError)
 		return
 	}
+	a.log.Info("job enqueued", "payload", payload)
 	w.WriteHeader(http.StatusAccepted)
 	fmt.Fprint(w, "enqueued")
 }

src/int_test.go

@@ -3,9 +3,10 @@ package main
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"os"
-	"sync"
 	"os/signal"
+	"sync"
 	"syscall"
 	"testing"
 
@@ -14,18 +15,19 @@ import (
 
 func TestIntegration(t *testing.T) {
 	redisAddr := "localhost:6379"
+	log := slog.New(slog.NewJSONHandler(os.Stdout, nil))
 
 	client := redis.NewClient(&redis.Options{Addr: redisAddr})
 	defer client.Close()
 	if err := client.Ping(context.Background()).Err(); err != nil {
 		t.Errorf("Failed to connect to Redis: %v", err)
 	}
 
-	scheduler := NewScheduler(redisAddr)
+	scheduler := NewScheduler(redisAddr, log)
 	defer scheduler.Close()
 
 	consumerID := fmt.Sprintf("worker_%d", os.Getpid())
-	supervisor := NewSupervisor(redisAddr, consumerID, "AMD")
+	supervisor := NewSupervisor(redisAddr, consumerID, "AMD", log)
 
 	if err := supervisor.Start(); err != nil {
 		t.Errorf("Failed to start supervisor: %v", err)
@@ -36,7 +38,7 @@ func TestIntegration(t *testing.T) {
 
 	// test jobs
 	var wg sync.WaitGroup
-    wg.Add(1)
+	wg.Add(1)
 	go func() {
 		defer wg.Done()
 		jobTypes := []string{"a", "b", "c"}
@@ -52,7 +54,7 @@ func TestIntegration(t *testing.T) {
 			}
 		}
 	}()
-	
+
 	wg.Wait()
 	supervisor.Stop()
 }

src/scheduler.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"log"
+	"log/slog"
 	"time"
 
 	"github.com/redis/go-redis/v9"
@@ -13,16 +13,18 @@ import (
 type Scheduler struct {
 	client *redis.Client
 	ctx    context.Context
+	log    *slog.Logger
 }
 
-func NewScheduler(redisAddr string) *Scheduler {
+func NewScheduler(redisAddr string, log *slog.Logger) *Scheduler {
 	client := redis.NewClient(&redis.Options{
 		Addr: redisAddr,
 	})
 
 	return &Scheduler{
 		client: client,
 		ctx:    context.Background(),
+		log:    log,
 	}
 }
 
@@ -52,7 +54,7 @@ func (s *Scheduler) Enqueue(jobType string, payload map[string]interface{}) erro
 		return fmt.Errorf("failed to enqueue job: %w", result.Err())
 	}
 
-	log.Printf("Enqueued job %s of type %s", job.ID, job.Type)
+	s.log.Info("enqueued job", "job_id", job.ID, "job_type", job.Type)
 	return nil
 }
 

src/supervisor.go

@@ -3,8 +3,9 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
-	"log"
+	"log/slog"
 	"sync"
 	"time"
 
@@ -18,9 +19,10 @@ type Supervisor struct {
 	consumerID  string
 	gpuType     string
 	wg          sync.WaitGroup
+	log         *slog.Logger
 }
 
-func NewSupervisor(redisAddr, consumerID, gpuType string) *Supervisor {
+func NewSupervisor(redisAddr, consumerID, gpuType string, log *slog.Logger) *Supervisor {
 	client := redis.NewClient(&redis.Options{
 		Addr: redisAddr,
 	})
@@ -33,6 +35,7 @@ func NewSupervisor(redisAddr, consumerID, gpuType string) *Supervisor {
 		cancel:      cancel,
 		consumerID:  consumerID,
 		gpuType:     gpuType,
+		log:         log,
 	}
 }
 
@@ -46,7 +49,7 @@ func (s *Supervisor) Start() error {
 	s.wg.Add(1)
 	go s.processJobs()
 
-	log.Printf("Supervisor %s started with GPU: %s", s.consumerID, s.gpuType)
+	s.log.Info("supervisor started", "consumer_id", s.consumerID, "gpu_type", s.gpuType)
 	return nil
 }
 
@@ -63,6 +66,8 @@ func (s *Supervisor) createConsumerGroup() error {
 
 func (s *Supervisor) processJobs() {
 	defer s.wg.Done()
+	s.log.Info("job processor started", "consumer_id", s.consumerID)
+	defer s.log.Info("job processor stopped", "consumer_id", s.consumerID)
 
 	for {
 		select {
@@ -79,8 +84,8 @@ func (s *Supervisor) processJobs() {
 			})
 
 			if result.Err() != nil {
-				if result.Err() != redis.Nil {
-					log.Printf("Error reading from stream: %v", result.Err())
+				if !errors.Is(result.Err(), redis.Nil) {
+					s.log.Error("error reading from stream", "error", result.Err())
 				}
 				continue
 			}
@@ -98,36 +103,36 @@ func (s *Supervisor) processJobs() {
 func (s *Supervisor) handleMessage(message redis.XMessage) {
 	jobData, ok := message.Values["data"].(string)
 	if !ok {
-		log.Printf("Invalid job data in message %s", message.ID)
+		s.log.Error("invalid job data in message", "message_id", message.ID)
 		s.ackMessage(message.ID)
 		return
 	}
 
 	var job Job
 	if err := json.Unmarshal([]byte(jobData), &job); err != nil {
-		log.Printf("Failed to unmarshal job data: %v", err)
+		s.log.Error("failed to unmarshal job data", "error", err, "message_id", message.ID)
 		s.ackMessage(message.ID)
 		return
 	}
 
 	// certain jobs require a specific GPU
 	if !s.canHandleJob(job) {
-		log.Printf("Job %s requires GPU type %s, but supervisor has %s - skipping",
-			job.ID, job.RequiredGPU, s.gpuType)
+		s.log.Info("skipping job due to GPU mismatch",
+			"job_id", job.ID, "required_gpu", job.RequiredGPU, "supervisor_gpu", s.gpuType)
 		// let another supervisor can pick it up
 		return
 	}
 
-	log.Printf("Processing job %s:%s", job.ID, job.Type)
+	s.log.Info("processing job", "job_id", job.ID, "job_type", job.Type)
 
 	// Simulate job processing
 	success := s.processJob(job)
 
 	if success {
 		s.ackMessage(message.ID)
-		log.Printf("Job %s completed successfully", job.ID)
+		s.log.Info("job completed successfully", "job_id", job.ID)
 	} else {
-		log.Printf("Job %s failed", job.ID)
+		s.log.Error("job failed", "job_id", job.ID)
 		s.ackMessage(message.ID) // TODO: change this once we have docker support
 	}
 }
@@ -151,12 +156,12 @@ func (s *Supervisor) processJob(job Job) bool {
 func (s *Supervisor) ackMessage(messageID string) {
 	result := s.redisClient.XAck(s.ctx, StreamName, ConsumerGroup, messageID)
 	if result.Err() != nil {
-		log.Printf("Failed to ack message %s: %v", messageID, result.Err())
+		s.log.Error("failed to ack message", "message_id", messageID, "error", result.Err())
 	}
 }
 
 func (s *Supervisor) Stop() {
-	log.Printf("Stopping supervisor %s", s.consumerID)
+	s.log.Info("stopping supervisor", "consumer_id", s.consumerID)
 	s.cancel()
 	s.wg.Wait()
 	s.redisClient.Close()