feat(torch): add queue to process the nodes when torch detects a even…

…t from k8s

Signed-off-by: Jose Ramon Mañes <jose@celestia.org>

pkg/db/redis/config.go

@@ -0,0 +1,59 @@
+package redis
+
+import (
+	"os"
+
+	log "github.com/sirupsen/logrus"
+)
+
+var (
+	redisHost    = ""
+	redisPort    = ""
+	redisPass    = ""
+	redisFullUrl = ""
+)
+
+// InitRedisConfig checks env vars and add default values in case we need
+func InitRedisConfig() *RedisClient {
+	// redis config
+	redisHost = GetRedisHost()
+	redisPort = GetRedisPort()
+	redisPass = GetRedisPass()
+	redisFullUrl = GetRedisFullURL()
+
+	log.Info("Redis host to connect: ", redisFullUrl)
+
+	return NewRedisClient(redisFullUrl, redisPass, 0)
+}
+
+// GetRedisHost returns the redis host to connect
+func GetRedisHost() string {
+	redisHost = os.Getenv("REDIS_HOST")
+	if redisHost == "" {
+		redisHost = "localhost"
+	}
+	return redisHost
+}
+
+// GetRedisPort returns the redis port
+func GetRedisPort() string {
+	redisPort = os.Getenv("REDIS_PORT")
+	if redisPort == "" {
+		redisPort = "6379"
+	}
+	return redisPort
+}
+
+// GetRedisFullURL returns the full url
+func GetRedisFullURL() string {
+	return redisHost + ":" + redisPort
+}
+
+// GetRedisPass returns the redis pass
+func GetRedisPass() string {
+	redisPass = os.Getenv("REDIS_PASS")
+	if redisPass == "" {
+		redisPass = ""
+	}
+	return redisPass
+}

pkg/db/redis/manager.go

@@ -7,8 +7,8 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-// SaveNodeId stores the values in redis.
-func SaveNodeId(
+// SetNodeId stores the values in redis.
+func SetNodeId(
 	podName string,
 	r *RedisClient,
 	ctx context.Context,
@@ -29,6 +29,8 @@ func SaveNodeId(
 			log.Error("Error adding the node to redis: ", err)
 			return err
 		}
+	} else {
+		log.Info("Node ", "["+podName+"]"+" found in Redis")
 	}
 
 	return nil

pkg/db/redis/producer.go

@@ -0,0 +1,38 @@
+package redis
+
+import (
+	"github.com/adjust/rmq/v5"
+	log "github.com/sirupsen/logrus"
+)
+
+// Producer add data into the queue.
+func Producer(data, queueName string) error {
+	log.Info("Adding STS [", data, "] node to the queue: [", queueName, "]")
+	data += "-0"
+	log.Info("Getting the pod from the STS [", data, "]")
+
+	connection, err := rmq.OpenConnection(
+		"producer",
+		"tcp",
+		GetRedisFullURL(),
+		2,
+		nil,
+	)
+	if err != nil {
+		log.Error("Error: ", err)
+		return err
+	}
+
+	queue, err := connection.OpenQueue(queueName)
+	if err != nil {
+		log.Error("Error: ", err)
+		return err
+	}
+
+	if err := queue.Publish(data); err != nil {
+		log.Error("Error, failed to publish: ", err)
+		return err
+	}
+
+	return nil
+}

pkg/db/redis/redis.go

@@ -2,7 +2,6 @@ package redis
 
 import (
 	"context"
-	"os"
 	"time"
 
 	"github.com/redis/go-redis/v9"
@@ -13,28 +12,6 @@ type RedisClient struct {
 	client *redis.Client
 }
 
-// InitRedisConfig checks env vars and add default values in case we need
-func InitRedisConfig() *RedisClient {
-	// redis config
-	redisHost := os.Getenv("REDIS_HOST")
-	if redisHost == "" {
-		redisHost = "localhost"
-	}
-	redisPort := os.Getenv("REDIS_PORT")
-	if redisPort == "" {
-		redisPort = "6379"
-	}
-	redisHost = redisHost + ":" + redisPort
-	log.Info("Redis host to connect: ", redisHost)
-
-	redisPass := os.Getenv("REDIS_PASS")
-	if redisHost == "" {
-		redisPass = ""
-	}
-
-	return NewRedisClient(redisHost, redisPass, 0)
-}
-
 // NewRedisClient returns a Redis client connection
 func NewRedisClient(addr string, password string, db int) *RedisClient {
 	client := redis.NewClient(&redis.Options{