Merge pull request #33 from Naganathan05/main

Migration to Redis List from RabbitMQ

Author: Ashrockzzz2003

README.md

@@ -9,6 +9,13 @@ Go implementation of the Runner Controller.
 1. GoLang - Install GoLang.
 2. MinIO - Start a MinIO server instance.
 3. CockroachDB - Start a CockroachDB instance.
+4. Redis - Start a Redis instance.
+
+(or) Use Docker Compose to start the services.
+
+```sh
+docker compose up -d
+```
 
 ### Installation
 
@@ -26,11 +33,11 @@ export DATABASE_URL=<cockroachdb_url>
 export MINIO_ENDPOINT=<minio_endpoint>
 export MINIO_ACCESS_KEY_ID=<minio_access_key>
 export MINIO_SECRET_KEY=<minio_secret_key>
-export RABBITMQ_URL=<rabbitmq_url>
 export FRONTEND_URL=<frontend_url>
 export HTTP_PORT=<http_port>
 export AUTH_GRPC_ADDRESS=<auth_grpc_address>
 export REDIS_URL=<redis_url>
+export REDIS_QUEUE_NAME=<redis_queue_name>
 ```
 
 3. Run the following command to start the server.

docker-compose.yml

@@ -12,8 +12,8 @@ services:
       MINIO_ENDPOINT : host.docker.internal:9000
       MINIO_ACCESS_KEY_ID : minioadmin
       MINIO_SECRET_KEY : minioadmin
-      RABBITMQ_URL : amqp://user:[email protected]:5672/
       FRONTEND_URL : http://localhost:3000
       HTTP_PORT : 5002
       AUTH_GRPC_ADDRESS : host.docker.internal:5001
-      REDIS_URL: redis://host.docker.internal:6379/0
+      REDIS_URL: redis://host.docker.internal:6379/0
+      REDIS_QUEUE_NAME: task_queue

go.mod

@@ -7,7 +7,6 @@ toolchain go1.24.2
 require (
 	github.com/jackc/pgx/v5 v5.7.4
 	github.com/minio/minio-go/v7 v7.0.91
-	github.com/rabbitmq/amqp091-go v1.10.0
 	github.com/redis/go-redis/v9 v9.8.0
 	github.com/rs/cors v1.11.1
 	google.golang.org/grpc v1.72.0

go.sum

@@ -46,8 +46,6 @@ github.com/minio/minio-go/v7 v7.0.91 h1:tWLZnEfo3OZl5PoXQwcwTAPNNrjyWwOh6cbZitW5
 github.com/minio/minio-go/v7 v7.0.91/go.mod h1:uvMUcGrpgeSAAI6+sD3818508nUyMULw94j2Nxku/Go=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/rabbitmq/amqp091-go v1.10.0 h1:STpn5XsHlHGcecLmMFCtg7mqq0RnD+zFr4uzukfVhBw=
-github.com/rabbitmq/amqp091-go v1.10.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
 github.com/redis/go-redis/v9 v9.8.0 h1:q3nRvjrlge/6UD7eTu/DSg2uYiU2mCL0G/uzBWqhicI=
 github.com/redis/go-redis/v9 v9.8.0/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw=
 github.com/rs/cors v1.11.1 h1:eU3gRzXLRK57F5rKMGMZURNdIG4EoAmX8k94r9wXWHA=
@@ -71,8 +69,6 @@ go.opentelemetry.io/otel/sdk/metric v1.34.0 h1:5CeK9ujjbFVL5c1PhLuStg1wxA7vQv7ce
 go.opentelemetry.io/otel/sdk/metric v1.34.0/go.mod h1:jQ/r8Ze28zRKoNRdkjCZxfs6YvBTG1+YIqyFVFYec5w=
 go.opentelemetry.io/otel/trace v1.34.0 h1:+ouXS2V8Rd4hp4580a8q23bg0azF2nI8cqLYnC8mh/k=
 go.opentelemetry.io/otel/trace v1.34.0/go.mod h1:Svm7lSjQD7kG7KJ/MUHPVXSDGz2OX4h0M2jHBhmSfRE=
-go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
-go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE=
 golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc=
 golang.org/x/net v0.39.0 h1:ZCu7HMWDxpXpaiKdhzIfaltL9Lp31x/3fCP11bc6/fY=

main.go

@@ -35,7 +35,7 @@ func main() {
 
 	var logger = util.NewLogger()
 
-	redisClient, err := sse.GetRedisClient(*logger)
+	err := util.InitRedisClient(*logger)
 	if err != nil {
 		logger.Error(fmt.Sprintf("Failed to initialize Redis client: %v. Exiting.", err))
 		os.Exit(1)
@@ -58,7 +58,7 @@ func main() {
 	mux.HandleFunc(routes.SHARE_RUN, controller.ShareRun)
 	mux.HandleFunc(routes.RUN, controller.UserRun)
 
-	sseHandler := sse.GetSSEHandler(*logger, redisClient)
+	sseHandler := sse.GetSSEHandler(*logger)
 	mux.HandleFunc(routes.LOGS, sseHandler)
 	logger.Info(fmt.Sprintf("SSE endpoint registered at %s using Redis Pub/Sub", routes.LOGS))
 
@@ -110,12 +110,7 @@ func main() {
 	}
 
 	// Close Redis Client.
-	logger.Info("Shutting down Redis client...")
-	if redisErr := redisClient.Close(); redisErr != nil {
-		logger.Error(fmt.Sprintf("Redis client shutdown error: %v", redisErr))
-	} else {
-		logger.Info("Redis client shutdown complete.")
-	}
+	util.ShutDownRedisClient(*logger)
 
 	logger.Info("Server exiting.")
 }

modules/sse/server.go

@@ -7,7 +7,6 @@ import (
 	"evolve/util"
 	"fmt"
 	"net/http"
-	"os"
 	"strings"
 	"time"
 
@@ -33,41 +32,17 @@ type redisLogPayload struct {
 	RunID  string `json:"runId"`  // From EOF message.
 }
 
-// GetRedisClient initializes a Redis client.
-func GetRedisClient(logger util.Logger) (*redis.Client, error) {
-	redisURL := os.Getenv("REDIS_URL")
-	if redisURL == "" {
-		redisURL = "redis://localhost:6379/0"
-		logger.Warn(fmt.Sprintf("REDIS_URL not set, using default: %s", redisURL))
-	}
-	opts, err := redis.ParseURL(redisURL)
-	if err != nil {
-		logger.Error(fmt.Sprintf("Failed to parse REDIS_URL '%s': %v", redisURL, err))
-		return nil, err
-	}
-	rdb := redis.NewClient(opts)
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-	_, err = rdb.Ping(ctx).Result()
-	if err != nil {
-		logger.Error(fmt.Sprintf("Failed to connect to Redis at %s: %v", opts.Addr, err))
-		return nil, err
-	}
-	logger.Info(fmt.Sprintf("Successfully connected to Redis at %s", opts.Addr))
-	return rdb, nil
-}
-
 // GetSSEHandler returns an HTTP handler
 // for Server-Sent Events (SSE) using Redis Streams.
-func GetSSEHandler(logger util.Logger, redisClient *redis.Client) http.HandlerFunc {
-	if redisClient == nil {
+func GetSSEHandler(logger util.Logger) http.HandlerFunc {
+	if util.RedisClient == nil {
 		logger.Error("GetSSEHandler requires a non-nil Redis client")
 		return func(w http.ResponseWriter, r *http.Request) {
 			http.Error(w, "Internal Server Error: Redis client not configured", http.StatusInternalServerError)
 		}
 	}
 	return func(w http.ResponseWriter, r *http.Request) {
-		serveSSEWithStream(logger, redisClient, w, r) // Call the new function
+		serveSSEWithStream(logger, w, r) // Call the new function
 	}
 }
 
@@ -93,7 +68,7 @@ func sendSSEData(w http.ResponseWriter, rc *http.ResponseController, payload str
 }
 
 // serveSSEWithStream handles the SSE stream for a given run ID.
-func serveSSEWithStream(logger util.Logger, redisClient *redis.Client, w http.ResponseWriter, r *http.Request) {
+func serveSSEWithStream(logger util.Logger, w http.ResponseWriter, r *http.Request) {
 	ctx := r.Context()
 	logger.Info("[SSE Stream Handler] Entered serveSSEWithStream")
 
@@ -174,7 +149,7 @@ func serveSSEWithStream(logger util.Logger, redisClient *redis.Client, w http.Re
 	for {
 		// Use XRead to get batches of historical data
 		// We don't block here, just read what's available.
-		cmd := redisClient.XRead(ctx, &redis.XReadArgs{
+		cmd := util.RedisClient.XRead(ctx, &redis.XReadArgs{
 			Streams: []string{redisStreamName, lastProcessedID},
 			Count:   streamReadCount,
 		})
@@ -251,7 +226,7 @@ func serveSSEWithStream(logger util.Logger, redisClient *redis.Client, w http.Re
 		}
 
 		// logger.Info(fmt.Sprintf("[SSE Stream Handler] Blocking read on stream '%s' from ID: %s", redisStreamName, lastProcessedID))
-		cmd := redisClient.XRead(ctx, &redis.XReadArgs{
+		cmd := util.RedisClient.XRead(ctx, &redis.XReadArgs{
 			Streams: []string{redisStreamName, lastProcessedID},
 			Count:   streamReadCount,
 			Block:   blockTimeout,

util/enqueue.go

@@ -6,8 +6,6 @@ import (
 	"fmt"
 	"os"
 	"time"
-
-	amqp "github.com/rabbitmq/amqp091-go"
 )
 
 func EnqueueRunRequest(ctx context.Context, runID string, fileName string, extension string) error {
@@ -21,39 +19,11 @@ func EnqueueRunRequest(ctx context.Context, runID string, fileName string, exten
 		Timestamp time.Time `json:"timestamp"`
 	}
 
-	// Get RabbitMQ connection string from environment variable or use default
-	rabbitMQURL := os.Getenv("RABBITMQ_URL")
-	if rabbitMQURL == "" {
-		rabbitMQURL = "amqp://guest:guest@localhost:5672/"
-	}
-
-	// Connect to RabbitMQ server
-	conn, err := amqp.Dial(rabbitMQURL)
-	if err != nil {
-		logger.Error(fmt.Sprintf("Failed to connect to RabbitMQ: %v", err))
-	}
-	defer conn.Close()
-
-	// Create a channel
-	ch, err := conn.Channel()
-	if err != nil {
-		logger.Error(fmt.Sprintf("Failed to open a channel: %v", err))
-	}
-	defer ch.Close()
-
 	// Declare a queue
-	queueName := "task_queue"
-	q, err := ch.QueueDeclare(
-		queueName, // name
-		true,      // durable
-		false,     // delete when unused
-		false,     // exclusive
-		false,     // no-wait
-		nil,       // arguments
-	)
-
-	if err != nil {
-		logger.Error(fmt.Sprintf("Failed to declare a queue: %v", err))
+	queueName := os.Getenv("REDIS_QUEUE_NAME")
+	if queueName == "" {
+		queueName = "task_queue"
+		logger.Warn(fmt.Sprintf("REDIS_QUEUE_NAME not set, using default: %s", queueName))
 	}
 
 	// Create a new message
@@ -68,20 +38,11 @@ func EnqueueRunRequest(ctx context.Context, runID string, fileName string, exten
 	body, err := json.Marshal(msg)
 	if err != nil {
 		logger.Error(fmt.Sprintf("Error marshaling message: %v", err))
+		return err
 	}
 
-	// Publish message
-	err = ch.PublishWithContext(
-		ctx,
-		"",     // exchange
-		q.Name, // routing key
-		false,  // mandatory
-		false,  // immediate
-		amqp.Publishing{
-			DeliveryMode: amqp.Persistent,
-			Body:         body,
-		},
-	)
+	// Push message to Redis List (LPUSH = enqueue at head)
+	err = RedisClient.LPush(ctx, queueName, string(body)).Err();
 
 	if err != nil {
 		logger.Error(fmt.Sprintf("Failed to publish message: %v", err))

util/redis.go

@@ -0,0 +1,47 @@
+package util
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+  
+var RedisClient *redis.Client
+
+// InitRedisClient initializes a Redis client.
+func InitRedisClient(logger Logger) (error) {
+	redisURL := os.Getenv("REDIS_URL")
+	if redisURL == "" {
+		redisURL = "redis://localhost:6379/0"
+		logger.Warn(fmt.Sprintf("REDIS_URL not set, using default: %s", redisURL))
+	}
+	opts, err := redis.ParseURL(redisURL)
+	if err != nil {
+		logger.Error(fmt.Sprintf("Failed to parse REDIS_URL '%s': %v", redisURL, err))
+		return err
+	}
+	rdb := redis.NewClient(opts)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	_, err = rdb.Ping(ctx).Result()
+	if err != nil {
+		logger.Error(fmt.Sprintf("Failed to connect to Redis at %s: %v", opts.Addr, err))
+		return err
+	}
+	logger.Info(fmt.Sprintf("Successfully connected to Redis at %s", opts.Addr))
+	RedisClient = rdb
+	return nil
+}
+
+// ShutDownRedisClient shuts down the Redis client
+func ShutDownRedisClient(logger Logger) {
+	logger.Info("Shutting down Redis client...")
+	if redisErr := RedisClient.Close(); redisErr != nil {
+		logger.Error(fmt.Sprintf("Redis client shutdown error: %v", redisErr))
+	} else {
+		logger.Info("Redis client shutdown complete.")
+	}
+}