From 9dd1c6563196938b594af3b6759c3c593e21f75a Mon Sep 17 00:00:00 2001 From: EpochBoy Date: Thu, 29 Jan 2026 14:32:54 +0100 Subject: [PATCH 1/6] feat(rabbitmq): add RabbitMQ pub/sub demo integration Added: - github.com/rabbitmq/amqp091-go client library - /rabbitmq/status endpoint - connection status + consumed messages - /rabbitmq/publish endpoint - publish messages (GET or POST) - Background consumer goroutine for demo queue - Prometheus metrics: epochcloud_rabbitmq_{messages_published,messages_consumed,publish_errors,connected}_total - HTML template: RabbitMQ Demo section Features: - Graceful handling when RabbitMQ not configured (optional) - Auto-ack consumer with last 10 messages stored - Trace ID correlation in messages - OpenTelemetry spans for all RabbitMQ operations Env vars (optional): - RABBITMQ_HOST, RABBITMQ_PORT, RABBITMQ_USERNAME - RABBITMQ_PASSWORD, RABBITMQ_VHOST, RABBITMQ_URI, RABBITMQ_QUEUE --- go.mod | 1 + go.sum | 2 + main.go | 393 +++++++++++++++++++++++++++++++++++++++++++++++++++++--- 3 files changed, 377 insertions(+), 19 deletions(-) diff --git a/go.mod b/go.mod index 3c1f442..d5f7d55 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.24.0 require ( github.com/prometheus/client_golang v1.22.0 + github.com/rabbitmq/amqp091-go v1.10.0 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 go.opentelemetry.io/otel v1.35.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.35.0 diff --git a/go.sum b/go.sum index a73f927..e120490 100644 --- a/go.sum +++ b/go.sum @@ -37,6 +37,8 @@ github.com/prometheus/common v0.62.0 h1:xasJaQlnWAeyHdUBeGjXmutelfJHWMRr+Fg4QszZ github.com/prometheus/common v0.62.0/go.mod h1:vyBcEuLSvWos9B1+CyL7JZ2up+uFzXhkqml0W5zIY1I= github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc= github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk= +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/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= diff --git a/main.go b/main.go index dfc9a7d..b421eb8 100644 --- a/main.go +++ b/main.go @@ -10,9 +10,12 @@ import ( "os" "os/signal" "strconv" + "sync" "syscall" "time" + amqp "github.com/rabbitmq/amqp091-go" + "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" "github.com/prometheus/client_golang/prometheus/promhttp" @@ -40,6 +43,25 @@ var logger *slog.Logger // Tracer - OpenTelemetry tracer for Tempo var tracer trace.Tracer +// RabbitMQ connection and channel +var ( + rabbitConn *amqp.Connection + rabbitChannel *amqp.Channel + rabbitMu sync.RWMutex + rabbitEnabled bool + // Stores last N consumed messages for demo display + consumedMessages []ConsumedMessage + consumedMessagesMu sync.RWMutex + maxConsumedMessages = 10 +) + +// ConsumedMessage holds a consumed message for display +type ConsumedMessage struct { + Body string `json:"body"` + Timestamp time.Time `json:"timestamp"` + TraceID string `json:"trace_id"` +} + // Prometheus metrics var ( httpRequestsTotal = promauto.NewCounterVec( @@ -83,6 +105,35 @@ var ( }, []string{"type"}, ) + + // RabbitMQ metrics + rabbitMessagesPublished = promauto.NewCounter( + prometheus.CounterOpts{ + Name: "epochcloud_rabbitmq_messages_published_total", + Help: "Total number of messages published to RabbitMQ", + }, + ) + + rabbitMessagesConsumed = promauto.NewCounter( + prometheus.CounterOpts{ + Name: "epochcloud_rabbitmq_messages_consumed_total", + Help: "Total number of messages consumed from RabbitMQ", + }, + ) + + rabbitPublishErrors = promauto.NewCounter( + prometheus.CounterOpts{ + Name: "epochcloud_rabbitmq_publish_errors_total", + Help: "Total number of publish errors", + }, + ) + + rabbitConnectionStatus = promauto.NewGauge( + prometheus.GaugeOpts{ + Name: "epochcloud_rabbitmq_connected", + Help: "RabbitMQ connection status (1=connected, 0=disconnected)", + }, + ) ) type HealthResponse struct { @@ -254,7 +305,17 @@ const pageTemplate = `
-

🔥 Chaos Testing

+

� RabbitMQ Demo

+

Message broker integration with pub/sub demo:

+
+

Connection status & consumed messages

+

Publish message (GET or POST)

+

Publish custom message

+
+

Messages auto-consumed by background worker. View metrics: epochcloud_rabbitmq_*

+
+
+

�🔥 Chaos Testing

Test the AlertManager → ntfy notification pipeline:

Triggers 500 error, increments error counter

@@ -549,6 +610,306 @@ func (r *statusRecorder) WriteHeader(code int) { r.ResponseWriter.WriteHeader(code) } +// initRabbitMQ initializes RabbitMQ connection if credentials are available +func initRabbitMQ(ctx context.Context) error { + host := os.Getenv("RABBITMQ_HOST") + port := os.Getenv("RABBITMQ_PORT") + user := os.Getenv("RABBITMQ_USERNAME") + pass := os.Getenv("RABBITMQ_PASSWORD") + vhost := os.Getenv("RABBITMQ_VHOST") + + // If no host configured, RabbitMQ is optional + if host == "" { + logger.Info("RabbitMQ not configured, skipping initialization") + rabbitConnectionStatus.Set(0) + return nil + } + + if port == "" { + port = "5672" + } + if vhost == "" { + vhost = "/" + } + + url := fmt.Sprintf("amqp://%s:%s@%s:%s/%s", user, pass, host, port, vhost) + + conn, err := amqp.Dial(url) + if err != nil { + logger.Error("failed to connect to RabbitMQ", + slog.String("host", host), + slog.String("error", err.Error()), + ) + rabbitConnectionStatus.Set(0) + return err + } + + ch, err := conn.Channel() + if err != nil { + conn.Close() + logger.Error("failed to open RabbitMQ channel", slog.String("error", err.Error())) + rabbitConnectionStatus.Set(0) + return err + } + + rabbitMu.Lock() + rabbitConn = conn + rabbitChannel = ch + rabbitEnabled = true + rabbitMu.Unlock() + + rabbitConnectionStatus.Set(1) + logger.Info("RabbitMQ connected", + slog.String("host", host), + slog.String("vhost", vhost), + ) + + // Start background consumer for demo queue + go startConsumer(ctx) + + return nil +} + +// startConsumer starts a background consumer for the demo queue +func startConsumer(ctx context.Context) { + queueName := os.Getenv("RABBITMQ_QUEUE") + if queueName == "" { + queueName = "epochcloud-demo" + } + + rabbitMu.RLock() + ch := rabbitChannel + enabled := rabbitEnabled + rabbitMu.RUnlock() + + if !enabled || ch == nil { + return + } + + // Declare the queue (idempotent) + _, err := ch.QueueDeclare( + queueName, + true, // durable + false, // auto-delete + false, // exclusive + false, // no-wait + nil, // arguments + ) + if err != nil { + logger.Error("failed to declare queue", slog.String("queue", queueName), slog.String("error", err.Error())) + return + } + + msgs, err := ch.Consume( + queueName, + "epochcloud-test-consumer", // consumer tag + true, // auto-ack + false, // exclusive + false, // no-local + false, // no-wait + nil, // args + ) + if err != nil { + logger.Error("failed to start consumer", slog.String("queue", queueName), slog.String("error", err.Error())) + return + } + + logger.Info("RabbitMQ consumer started", slog.String("queue", queueName)) + + for { + select { + case <-ctx.Done(): + logger.Info("RabbitMQ consumer shutting down") + return + case msg, ok := <-msgs: + if !ok { + logger.Warn("RabbitMQ consumer channel closed") + rabbitConnectionStatus.Set(0) + return + } + rabbitMessagesConsumed.Inc() + + consumed := ConsumedMessage{ + Body: string(msg.Body), + Timestamp: time.Now(), + TraceID: msg.MessageId, // Use MessageId as trace correlation + } + + consumedMessagesMu.Lock() + consumedMessages = append([]ConsumedMessage{consumed}, consumedMessages...) + if len(consumedMessages) > maxConsumedMessages { + consumedMessages = consumedMessages[:maxConsumedMessages] + } + consumedMessagesMu.Unlock() + + logger.Info("message consumed", + slog.String("queue", queueName), + slog.String("body", string(msg.Body)), + ) + } + } +} + +// RabbitMQPublishRequest is the request body for publishing a message +type RabbitMQPublishRequest struct { + Message string `json:"message"` +} + +// RabbitMQResponse is the response for RabbitMQ operations +type RabbitMQResponse struct { + Success bool `json:"success"` + Message string `json:"message"` + TraceID string `json:"trace_id,omitempty"` + Consumed []ConsumedMessage `json:"consumed,omitempty"` + Connected bool `json:"connected"` + Timestamp string `json:"timestamp"` +} + +// rabbitStatusHandler returns RabbitMQ connection status and consumed messages +func rabbitStatusHandler(w http.ResponseWriter, r *http.Request) { + ctx := r.Context() + _, span := tracer.Start(ctx, "rabbitStatus") + defer span.End() + + rabbitMu.RLock() + connected := rabbitEnabled && rabbitConn != nil && !rabbitConn.IsClosed() + rabbitMu.RUnlock() + + consumedMessagesMu.RLock() + messages := make([]ConsumedMessage, len(consumedMessages)) + copy(messages, consumedMessages) + consumedMessagesMu.RUnlock() + + resp := RabbitMQResponse{ + Success: true, + Message: "RabbitMQ status", + Connected: connected, + Consumed: messages, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + + logger.InfoContext(ctx, "rabbitmq status check", + slog.Bool("connected", connected), + slog.Int("consumed_messages", len(messages)), + slog.String("trace_id", span.SpanContext().TraceID().String()), + ) + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(resp) +} + +// rabbitPublishHandler publishes a message to RabbitMQ +func rabbitPublishHandler(w http.ResponseWriter, r *http.Request) { + ctx := r.Context() + _, span := tracer.Start(ctx, "rabbitPublish") + defer span.End() + + rabbitMu.RLock() + ch := rabbitChannel + enabled := rabbitEnabled + rabbitMu.RUnlock() + + if !enabled || ch == nil { + resp := RabbitMQResponse{ + Success: false, + Message: "RabbitMQ not connected", + Connected: false, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusServiceUnavailable) + json.NewEncoder(w).Encode(resp) + return + } + + // Get message from query param or body + message := r.URL.Query().Get("message") + if message == "" && r.Method == http.MethodPost { + var req RabbitMQPublishRequest + if err := json.NewDecoder(r.Body).Decode(&req); err == nil { + message = req.Message + } + } + if message == "" { + message = fmt.Sprintf("Hello from EpochCloud at %s", time.Now().Format(time.RFC3339)) + } + + queueName := os.Getenv("RABBITMQ_QUEUE") + if queueName == "" { + queueName = "epochcloud-demo" + } + + traceID := span.SpanContext().TraceID().String() + + err := ch.PublishWithContext(ctx, + "", // exchange (default) + queueName, // routing key (queue name) + false, // mandatory + false, // immediate + amqp.Publishing{ + ContentType: "text/plain", + Body: []byte(message), + MessageId: traceID, // Trace correlation + Timestamp: time.Now(), + }, + ) + + if err != nil { + rabbitPublishErrors.Inc() + logger.ErrorContext(ctx, "failed to publish message", + slog.String("queue", queueName), + slog.String("error", err.Error()), + slog.String("trace_id", traceID), + ) + resp := RabbitMQResponse{ + Success: false, + Message: "Failed to publish: " + err.Error(), + Connected: true, + TraceID: traceID, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusInternalServerError) + json.NewEncoder(w).Encode(resp) + return + } + + rabbitMessagesPublished.Inc() + logger.InfoContext(ctx, "message published", + slog.String("queue", queueName), + slog.String("message", message), + slog.String("trace_id", traceID), + ) + + resp := RabbitMQResponse{ + Success: true, + Message: fmt.Sprintf("Published to %s: %s", queueName, message), + Connected: true, + TraceID: traceID, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(resp) +} + +// closeRabbitMQ closes RabbitMQ connections +func closeRabbitMQ() { + rabbitMu.Lock() + defer rabbitMu.Unlock() + + if rabbitChannel != nil { + rabbitChannel.Close() + rabbitChannel = nil + } + if rabbitConn != nil { + rabbitConn.Close() + rabbitConn = nil + } + rabbitEnabled = false + rabbitConnectionStatus.Set(0) + logger.Info("RabbitMQ connections closed") +} + func main() { initLogger() @@ -572,6 +933,14 @@ func main() { }() } + // Initialize RabbitMQ (optional - won't fail if not configured) + if err := initRabbitMQ(ctx); err != nil { + logger.Warn("RabbitMQ initialization failed, continuing without messaging", + slog.String("error", err.Error()), + ) + } + defer closeRabbitMQ() + appInfo.WithLabelValues(Version, Commit, BuildTime, getEnvironment()).Set(1) mux := http.NewServeMux() @@ -579,6 +948,9 @@ func main() { mux.Handle("/health", metricsMiddleware("/health", healthHandler)) mux.Handle("/version", metricsMiddleware("/version", versionHandler)) mux.Handle("/chaos", metricsMiddleware("/chaos", chaosHandler)) + // RabbitMQ demo endpoints + mux.Handle("/rabbitmq/status", metricsMiddleware("/rabbitmq/status", rabbitStatusHandler)) + mux.Handle("/rabbitmq/publish", metricsMiddleware("/rabbitmq/publish", rabbitPublishHandler)) mux.Handle("/metrics", promhttp.Handler()) server := &http.Server{ @@ -593,6 +965,7 @@ func main() { signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) <-sigChan logger.Info("shutting down") + cancel() // Cancel context to stop RabbitMQ consumer shutdownCtx, c := context.WithTimeout(context.Background(), 30*time.Second) defer c() server.Shutdown(shutdownCtx) @@ -604,21 +977,3 @@ func main() { os.Exit(1) } } -// test 1767625973 -// test2 1767626122 -// test3 1767626184 -// test1 1767626512 -// test2 1767626520 -// test3 1767626529 -// test-terminate1 1767626716 -// test-terminate2 1767626726 -// final-test1 1767626992 -// final-test2 1767627002 -// cancel test 1767634496 -// Test cancel-in-progress status fix 1767637323 -// Test cancel-in-progress status fix 1767637327 -// Trigger cancel 1767637344 -// Test timestamp fix 1767637721 -// Trigger cancel test 1767637774 -// Final test 1767639503 -// Cancel test 1767639554 From 63badd92415ce919d08a09815ec5130eb6b55e11 Mon Sep 17 00:00:00 2001 From: EpochBoy Date: Fri, 30 Jan 2026 14:37:04 +0100 Subject: [PATCH 2/6] fix: add RabbitMQ connection retry with exponential backoff The app now runs a background connection manager that: - Retries connection with exponential backoff (1s to 30s) - Automatically reconnects when connection is lost - Uses NotifyClose to detect connection drops - Non-blocking startup (app starts immediately, connects when RabbitMQ is ready) This makes the app resilient to RabbitMQ being unavailable at startup. --- main.go | 171 +++++++++++++++++++++++++++++++++++++++++++------------- 1 file changed, 132 insertions(+), 39 deletions(-) diff --git a/main.go b/main.go index b421eb8..b8106c6 100644 --- a/main.go +++ b/main.go @@ -610,13 +610,9 @@ func (r *statusRecorder) WriteHeader(code int) { r.ResponseWriter.WriteHeader(code) } -// initRabbitMQ initializes RabbitMQ connection if credentials are available +// initRabbitMQ starts background RabbitMQ connection manager with retry logic func initRabbitMQ(ctx context.Context) error { host := os.Getenv("RABBITMQ_HOST") - port := os.Getenv("RABBITMQ_PORT") - user := os.Getenv("RABBITMQ_USERNAME") - pass := os.Getenv("RABBITMQ_PASSWORD") - vhost := os.Getenv("RABBITMQ_VHOST") // If no host configured, RabbitMQ is optional if host == "" { @@ -625,6 +621,20 @@ func initRabbitMQ(ctx context.Context) error { return nil } + // Start background connection manager (non-blocking) + go rabbitConnectionManager(ctx) + + return nil +} + +// rabbitConnectionManager maintains RabbitMQ connection with exponential backoff retry +func rabbitConnectionManager(ctx context.Context) { + host := os.Getenv("RABBITMQ_HOST") + port := os.Getenv("RABBITMQ_PORT") + user := os.Getenv("RABBITMQ_USERNAME") + pass := os.Getenv("RABBITMQ_PASSWORD") + vhost := os.Getenv("RABBITMQ_VHOST") + if port == "" { port = "5672" } @@ -634,40 +644,127 @@ func initRabbitMQ(ctx context.Context) error { url := fmt.Sprintf("amqp://%s:%s@%s:%s/%s", user, pass, host, port, vhost) - conn, err := amqp.Dial(url) - if err != nil { - logger.Error("failed to connect to RabbitMQ", + backoff := time.Second + maxBackoff := 30 * time.Second + + for { + select { + case <-ctx.Done(): + logger.Info("RabbitMQ connection manager shutting down") + return + default: + } + + // Check if already connected + rabbitMu.RLock() + connected := rabbitEnabled && rabbitConn != nil && !rabbitConn.IsClosed() + rabbitMu.RUnlock() + + if connected { + // Already connected, wait and check again + select { + case <-ctx.Done(): + return + case <-time.After(5 * time.Second): + continue + } + } + + // Attempt connection + logger.Info("attempting RabbitMQ connection", slog.String("host", host), - slog.String("error", err.Error()), + slog.String("vhost", vhost), ) - rabbitConnectionStatus.Set(0) - return err - } - ch, err := conn.Channel() - if err != nil { - conn.Close() - logger.Error("failed to open RabbitMQ channel", slog.String("error", err.Error())) - rabbitConnectionStatus.Set(0) - return err - } + conn, err := amqp.Dial(url) + if err != nil { + logger.Warn("RabbitMQ connection failed, will retry", + slog.String("host", host), + slog.String("error", err.Error()), + slog.Duration("retry_in", backoff), + ) + rabbitConnectionStatus.Set(0) - rabbitMu.Lock() - rabbitConn = conn - rabbitChannel = ch - rabbitEnabled = true - rabbitMu.Unlock() - - rabbitConnectionStatus.Set(1) - logger.Info("RabbitMQ connected", - slog.String("host", host), - slog.String("vhost", vhost), - ) + select { + case <-ctx.Done(): + return + case <-time.After(backoff): + } - // Start background consumer for demo queue - go startConsumer(ctx) + // Exponential backoff + backoff = backoff * 2 + if backoff > maxBackoff { + backoff = maxBackoff + } + continue + } - return nil + ch, err := conn.Channel() + if err != nil { + conn.Close() + logger.Warn("RabbitMQ channel open failed, will retry", + slog.String("error", err.Error()), + slog.Duration("retry_in", backoff), + ) + rabbitConnectionStatus.Set(0) + + select { + case <-ctx.Done(): + return + case <-time.After(backoff): + } + + backoff = backoff * 2 + if backoff > maxBackoff { + backoff = maxBackoff + } + continue + } + + // Success - update state + rabbitMu.Lock() + rabbitConn = conn + rabbitChannel = ch + rabbitEnabled = true + rabbitMu.Unlock() + + rabbitConnectionStatus.Set(1) + backoff = time.Second // Reset backoff on success + + logger.Info("RabbitMQ connected", + slog.String("host", host), + slog.String("vhost", vhost), + ) + + // Start consumer for this connection + consumerCtx, consumerCancel := context.WithCancel(ctx) + go startConsumer(consumerCtx) + + // Wait for connection close notification + closeChan := make(chan *amqp.Error, 1) + conn.NotifyClose(closeChan) + + select { + case <-ctx.Done(): + consumerCancel() + return + case amqpErr := <-closeChan: + consumerCancel() + if amqpErr != nil { + logger.Warn("RabbitMQ connection lost, will reconnect", + slog.String("error", amqpErr.Error()), + ) + } else { + logger.Info("RabbitMQ connection closed, will reconnect") + } + rabbitMu.Lock() + rabbitEnabled = false + rabbitConn = nil + rabbitChannel = nil + rabbitMu.Unlock() + rabbitConnectionStatus.Set(0) + } + } } // startConsumer starts a background consumer for the demo queue @@ -933,12 +1030,8 @@ func main() { }() } - // Initialize RabbitMQ (optional - won't fail if not configured) - if err := initRabbitMQ(ctx); err != nil { - logger.Warn("RabbitMQ initialization failed, continuing without messaging", - slog.String("error", err.Error()), - ) - } + // Start RabbitMQ connection manager (non-blocking, handles retries internally) + initRabbitMQ(ctx) defer closeRabbitMQ() appInfo.WithLabelValues(Version, Commit, BuildTime, getEnvironment()).Set(1) From 60ca3d40beac67ecdfe989d9abf9b3752da800b5 Mon Sep 17 00:00:00 2001 From: EpochBoy Date: Sat, 31 Jan 2026 13:05:51 +0100 Subject: [PATCH 3/6] feat: add CONSUMER_MODE for KEDA-scaled consumer deployment When CONSUMER_MODE=true: - Only consumes messages from RabbitMQ queue - HTTP server runs minimal endpoints (/health, /metrics only) - No /rabbitmq/publish or other producer endpoints This enables: - Separate consumer pods scaled by KEDA based on queue depth - Producer pods that only publish (don't consume own messages) - True microservices pub/sub pattern demonstration --- main.go | 40 +++++++++++++++++++++++++++++++--------- 1 file changed, 31 insertions(+), 9 deletions(-) diff --git a/main.go b/main.go index b8106c6..2a7fd8e 100644 --- a/main.go +++ b/main.go @@ -49,6 +49,9 @@ var ( rabbitChannel *amqp.Channel rabbitMu sync.RWMutex rabbitEnabled bool + // Consumer mode - when true, only consume messages (no HTTP server except /health and /metrics) + // Used by KEDA-scaled consumer pods + consumerMode bool // Stores last N consumed messages for demo display consumedMessages []ConsumedMessage consumedMessagesMu sync.RWMutex @@ -734,11 +737,19 @@ func rabbitConnectionManager(ctx context.Context) { logger.Info("RabbitMQ connected", slog.String("host", host), slog.String("vhost", vhost), + slog.Bool("consumer_mode", consumerMode), ) - // Start consumer for this connection - consumerCtx, consumerCancel := context.WithCancel(ctx) - go startConsumer(consumerCtx) + // Only start consumer if in consumer mode + // This allows producer and consumer to be separate deployments + var consumerCancel context.CancelFunc + if consumerMode { + var consumerCtx context.Context + consumerCtx, consumerCancel = context.WithCancel(ctx) + go startConsumer(consumerCtx) + } else { + consumerCancel = func() {} // no-op + } // Wait for connection close notification closeChan := make(chan *amqp.Error, 1) @@ -1010,6 +1021,12 @@ func closeRabbitMQ() { func main() { initLogger() + // Check if running in consumer-only mode (for KEDA scaling) + consumerMode = os.Getenv("CONSUMER_MODE") == "true" + if consumerMode { + logger.Info("starting in CONSUMER mode - HTTP server disabled, consuming only") + } + port := os.Getenv("PORT") if port == "" { port = "8080" @@ -1037,15 +1054,20 @@ func main() { appInfo.WithLabelValues(Version, Commit, BuildTime, getEnvironment()).Set(1) mux := http.NewServeMux() - mux.Handle("/", metricsMiddleware("/", rootHandler)) + // Always register health and metrics for probes and monitoring mux.Handle("/health", metricsMiddleware("/health", healthHandler)) - mux.Handle("/version", metricsMiddleware("/version", versionHandler)) - mux.Handle("/chaos", metricsMiddleware("/chaos", chaosHandler)) - // RabbitMQ demo endpoints - mux.Handle("/rabbitmq/status", metricsMiddleware("/rabbitmq/status", rabbitStatusHandler)) - mux.Handle("/rabbitmq/publish", metricsMiddleware("/rabbitmq/publish", rabbitPublishHandler)) mux.Handle("/metrics", promhttp.Handler()) + if !consumerMode { + // Full HTTP API only in producer mode + mux.Handle("/", metricsMiddleware("/", rootHandler)) + mux.Handle("/version", metricsMiddleware("/version", versionHandler)) + mux.Handle("/chaos", metricsMiddleware("/chaos", chaosHandler)) + // RabbitMQ demo endpoints + mux.Handle("/rabbitmq/status", metricsMiddleware("/rabbitmq/status", rabbitStatusHandler)) + mux.Handle("/rabbitmq/publish", metricsMiddleware("/rabbitmq/publish", rabbitPublishHandler)) + } + server := &http.Server{ Addr: ":" + port, Handler: mux, From 00de2007656dba986e46b82db4a4cfa1e0ae0440 Mon Sep 17 00:00:00 2001 From: EpochBoy Date: Sun, 1 Feb 2026 14:28:20 +0100 Subject: [PATCH 4/6] feat: add Valkey cache integration with /cache/* endpoints --- go.mod | 2 + go.sum | 4 + main.go | 353 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 359 insertions(+) diff --git a/go.mod b/go.mod index d5f7d55..2aa4110 100644 --- a/go.mod +++ b/go.mod @@ -16,6 +16,7 @@ require ( github.com/beorn7/perks v1.0.1 // indirect github.com/cenkalti/backoff/v4 v4.3.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/felixge/httpsnoop v1.0.4 // indirect github.com/go-logr/logr v1.4.2 // indirect github.com/go-logr/stdr v1.2.2 // indirect @@ -25,6 +26,7 @@ require ( github.com/prometheus/client_model v0.6.1 // indirect github.com/prometheus/common v0.62.0 // indirect github.com/prometheus/procfs v0.15.1 // indirect + github.com/redis/go-redis/v9 v9.17.3 // indirect go.opentelemetry.io/auto/sdk v1.1.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.35.0 // indirect go.opentelemetry.io/otel/metric v1.35.0 // indirect diff --git a/go.sum b/go.sum index e120490..7849204 100644 --- a/go.sum +++ b/go.sum @@ -6,6 +6,8 @@ github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UF github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= @@ -39,6 +41,8 @@ github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0leargg github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk= 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.17.3 h1:fN29NdNrE17KttK5Ndf20buqfDZwGNgoUr9qjl1DQx4= +github.com/redis/go-redis/v9 v9.17.3/go.mod h1:u410H11HMLoB+TP67dz8rL9s6QW2j76l0//kSOd3370= github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= diff --git a/main.go b/main.go index 2a7fd8e..0e13d59 100644 --- a/main.go +++ b/main.go @@ -14,6 +14,8 @@ import ( "syscall" "time" + "github.com/redis/go-redis/v9" + amqp "github.com/rabbitmq/amqp091-go" "github.com/prometheus/client_golang/prometheus" @@ -43,6 +45,13 @@ var logger *slog.Logger // Tracer - OpenTelemetry tracer for Tempo var tracer trace.Tracer +// Valkey/Redis client +var ( + valkeyClient *redis.Client + valkeyMu sync.RWMutex + valkeyEnabled bool +) + // RabbitMQ connection and channel var ( rabbitConn *amqp.Connection @@ -137,6 +146,36 @@ var ( Help: "RabbitMQ connection status (1=connected, 0=disconnected)", }, ) + + // Valkey/Redis metrics + valkeyCacheHits = promauto.NewCounter( + prometheus.CounterOpts{ + Name: "epochcloud_valkey_cache_hits_total", + Help: "Total number of Valkey cache hits", + }, + ) + + valkeyCacheMisses = promauto.NewCounter( + prometheus.CounterOpts{ + Name: "epochcloud_valkey_cache_misses_total", + Help: "Total number of Valkey cache misses", + }, + ) + + valkeyOperations = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "epochcloud_valkey_operations_total", + Help: "Total number of Valkey operations by type", + }, + []string{"operation"}, + ) + + valkeyConnectionStatus = promauto.NewGauge( + prometheus.GaugeOpts{ + Name: "epochcloud_valkey_connected", + Help: "Valkey connection status (1=connected, 0=disconnected)", + }, + ) ) type HealthResponse struct { @@ -1018,6 +1057,312 @@ func closeRabbitMQ() { logger.Info("RabbitMQ connections closed") } +// initValkey initializes the Valkey/Redis client +func initValkey() { + host := os.Getenv("VALKEY_HOST") + port := os.Getenv("VALKEY_PORT") + password := os.Getenv("VALKEY_PASSWORD") + dbStr := os.Getenv("VALKEY_DATABASE") + + if host == "" { + logger.Info("VALKEY_HOST not set, Valkey disabled") + return + } + + db := 0 + if dbStr != "" { + if parsed, err := strconv.Atoi(dbStr); err == nil { + db = parsed + } + } + + if port == "" { + port = "6379" + } + + addr := fmt.Sprintf("%s:%s", host, port) + + client := redis.NewClient(&redis.Options{ + Addr: addr, + Password: password, + DB: db, + }) + + // Test connection + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + if err := client.Ping(ctx).Err(); err != nil { + logger.Warn("failed to connect to Valkey", + slog.String("addr", addr), + slog.Int("db", db), + slog.String("error", err.Error()), + ) + valkeyConnectionStatus.Set(0) + return + } + + valkeyMu.Lock() + valkeyClient = client + valkeyEnabled = true + valkeyMu.Unlock() + + valkeyConnectionStatus.Set(1) + logger.Info("Valkey connected", + slog.String("addr", addr), + slog.Int("db", db), + ) +} + +// closeValkey closes the Valkey connection +func closeValkey() { + valkeyMu.Lock() + defer valkeyMu.Unlock() + + if valkeyClient != nil { + valkeyClient.Close() + valkeyClient = nil + } + valkeyEnabled = false + valkeyConnectionStatus.Set(0) + logger.Info("Valkey connection closed") +} + +// ValkeyResponse for cache API responses +type ValkeyResponse struct { + Success bool `json:"success"` + Operation string `json:"operation,omitempty"` + Key string `json:"key,omitempty"` + Value string `json:"value,omitempty"` + Hit bool `json:"hit,omitempty"` + Message string `json:"message,omitempty"` + Connected bool `json:"connected"` + TraceID string `json:"trace_id,omitempty"` + Timestamp string `json:"timestamp"` +} + +// valkeyStatusHandler returns Valkey connection status +func valkeyStatusHandler(w http.ResponseWriter, r *http.Request) { + ctx := r.Context() + _, span := tracer.Start(ctx, "valkeyStatus") + defer span.End() + + valkeyMu.RLock() + client := valkeyClient + enabled := valkeyEnabled + valkeyMu.RUnlock() + + connected := false + if enabled && client != nil { + pingCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + if err := client.Ping(pingCtx).Err(); err == nil { + connected = true + } + } + + if connected { + valkeyConnectionStatus.Set(1) + } else { + valkeyConnectionStatus.Set(0) + } + + resp := ValkeyResponse{ + Success: connected, + Message: fmt.Sprintf("Valkey enabled=%v, connected=%v", enabled, connected), + Connected: connected, + TraceID: span.SpanContext().TraceID().String(), + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(resp) +} + +// valkeySetHandler sets a key-value pair in Valkey +func valkeySetHandler(w http.ResponseWriter, r *http.Request) { + ctx := r.Context() + _, span := tracer.Start(ctx, "valkeySet") + defer span.End() + + valkeyMu.RLock() + client := valkeyClient + enabled := valkeyEnabled + valkeyMu.RUnlock() + + if !enabled || client == nil { + resp := ValkeyResponse{ + Success: false, + Message: "Valkey not connected", + Connected: false, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusServiceUnavailable) + json.NewEncoder(w).Encode(resp) + return + } + + key := r.URL.Query().Get("key") + value := r.URL.Query().Get("value") + ttlStr := r.URL.Query().Get("ttl") + + if key == "" { + key = "demo-key" + } + if value == "" { + value = fmt.Sprintf("demo-value-%d", time.Now().Unix()) + } + + ttl := 5 * time.Minute + if ttlStr != "" { + if seconds, err := strconv.Atoi(ttlStr); err == nil { + ttl = time.Duration(seconds) * time.Second + } + } + + traceID := span.SpanContext().TraceID().String() + + err := client.Set(ctx, key, value, ttl).Err() + if err != nil { + logger.ErrorContext(ctx, "failed to set key in Valkey", + slog.String("key", key), + slog.String("error", err.Error()), + slog.String("trace_id", traceID), + ) + resp := ValkeyResponse{ + Success: false, + Operation: "SET", + Key: key, + Message: "Failed to set: " + err.Error(), + Connected: true, + TraceID: traceID, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusInternalServerError) + json.NewEncoder(w).Encode(resp) + return + } + + valkeyOperations.WithLabelValues("set").Inc() + logger.InfoContext(ctx, "key set in Valkey", + slog.String("key", key), + slog.Duration("ttl", ttl), + slog.String("trace_id", traceID), + ) + + resp := ValkeyResponse{ + Success: true, + Operation: "SET", + Key: key, + Value: value, + Message: fmt.Sprintf("Set key '%s' with TTL %v", key, ttl), + Connected: true, + TraceID: traceID, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(resp) +} + +// valkeyGetHandler gets a value from Valkey +func valkeyGetHandler(w http.ResponseWriter, r *http.Request) { + ctx := r.Context() + _, span := tracer.Start(ctx, "valkeyGet") + defer span.End() + + valkeyMu.RLock() + client := valkeyClient + enabled := valkeyEnabled + valkeyMu.RUnlock() + + if !enabled || client == nil { + resp := ValkeyResponse{ + Success: false, + Message: "Valkey not connected", + Connected: false, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusServiceUnavailable) + json.NewEncoder(w).Encode(resp) + return + } + + key := r.URL.Query().Get("key") + if key == "" { + key = "demo-key" + } + + traceID := span.SpanContext().TraceID().String() + + value, err := client.Get(ctx, key).Result() + if err == redis.Nil { + // Cache miss + valkeyCacheMisses.Inc() + valkeyOperations.WithLabelValues("get").Inc() + logger.InfoContext(ctx, "cache miss", + slog.String("key", key), + slog.String("trace_id", traceID), + ) + resp := ValkeyResponse{ + Success: true, + Operation: "GET", + Key: key, + Hit: false, + Message: fmt.Sprintf("Key '%s' not found (cache miss)", key), + Connected: true, + TraceID: traceID, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(resp) + return + } else if err != nil { + logger.ErrorContext(ctx, "failed to get key from Valkey", + slog.String("key", key), + slog.String("error", err.Error()), + slog.String("trace_id", traceID), + ) + resp := ValkeyResponse{ + Success: false, + Operation: "GET", + Key: key, + Message: "Failed to get: " + err.Error(), + Connected: true, + TraceID: traceID, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusInternalServerError) + json.NewEncoder(w).Encode(resp) + return + } + + // Cache hit + valkeyCacheHits.Inc() + valkeyOperations.WithLabelValues("get").Inc() + logger.InfoContext(ctx, "cache hit", + slog.String("key", key), + slog.String("trace_id", traceID), + ) + + resp := ValkeyResponse{ + Success: true, + Operation: "GET", + Key: key, + Value: value, + Hit: true, + Message: fmt.Sprintf("Key '%s' found (cache hit)", key), + Connected: true, + TraceID: traceID, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(resp) +} + func main() { initLogger() @@ -1051,6 +1396,10 @@ func main() { initRabbitMQ(ctx) defer closeRabbitMQ() + // Initialize Valkey/Redis connection + initValkey() + defer closeValkey() + appInfo.WithLabelValues(Version, Commit, BuildTime, getEnvironment()).Set(1) mux := http.NewServeMux() @@ -1066,6 +1415,10 @@ func main() { // RabbitMQ demo endpoints mux.Handle("/rabbitmq/status", metricsMiddleware("/rabbitmq/status", rabbitStatusHandler)) mux.Handle("/rabbitmq/publish", metricsMiddleware("/rabbitmq/publish", rabbitPublishHandler)) + // Valkey/Redis cache demo endpoints + mux.Handle("/cache/status", metricsMiddleware("/cache/status", valkeyStatusHandler)) + mux.Handle("/cache/set", metricsMiddleware("/cache/set", valkeySetHandler)) + mux.Handle("/cache/get", metricsMiddleware("/cache/get", valkeyGetHandler)) } server := &http.Server{ From b5871eefe0d8735ef19217f6b44fd463adaf2158 Mon Sep 17 00:00:00 2001 From: EpochBoy Date: Sun, 1 Feb 2026 16:52:55 +0100 Subject: [PATCH 5/6] fix(valkey): add retry with exponential backoff for connection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Valkey may not be ready immediately during pod startup, especially with Linkerd mTLS. Added retry logic (5 attempts, exponential backoff 1s→16s) to handle transient connection failures. --- main.go | 48 ++++++++++++++++++++++++++++++++++-------------- 1 file changed, 34 insertions(+), 14 deletions(-) diff --git a/main.go b/main.go index 0e13d59..f3c6b68 100644 --- a/main.go +++ b/main.go @@ -1088,30 +1088,50 @@ func initValkey() { DB: db, }) - // Test connection - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() + // Retry connection with exponential backoff + // Valkey may not be ready immediately during pod startup + maxRetries := 5 + backoff := time.Second - if err := client.Ping(ctx).Err(); err != nil { - logger.Warn("failed to connect to Valkey", + for attempt := 1; attempt <= maxRetries; attempt++ { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + err := client.Ping(ctx).Err() + cancel() + + if err == nil { + valkeyMu.Lock() + valkeyClient = client + valkeyEnabled = true + valkeyMu.Unlock() + + valkeyConnectionStatus.Set(1) + logger.Info("Valkey connected", + slog.String("addr", addr), + slog.Int("db", db), + ) + return + } + + logger.Warn("failed to connect to Valkey, retrying", slog.String("addr", addr), slog.Int("db", db), + slog.Int("attempt", attempt), + slog.Int("max_retries", maxRetries), + slog.Duration("backoff", backoff), slog.String("error", err.Error()), ) - valkeyConnectionStatus.Set(0) - return - } - valkeyMu.Lock() - valkeyClient = client - valkeyEnabled = true - valkeyMu.Unlock() + if attempt < maxRetries { + time.Sleep(backoff) + backoff *= 2 // Exponential backoff + } + } - valkeyConnectionStatus.Set(1) - logger.Info("Valkey connected", + logger.Error("Valkey connection failed after all retries", slog.String("addr", addr), slog.Int("db", db), ) + valkeyConnectionStatus.Set(0) } // closeValkey closes the Valkey connection From 95e1133d7b69dffda1ba5632ed93cca427277932 Mon Sep 17 00:00:00 2001 From: Renovate Bot Date: Tue, 3 Feb 2026 15:36:56 +0000 Subject: [PATCH 6/6] chore(deps): update module go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc to v1.40.0 --- go.mod | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/go.mod b/go.mod index 2aa4110..598338a 100644 --- a/go.mod +++ b/go.mod @@ -7,7 +7,7 @@ require ( github.com/rabbitmq/amqp091-go v1.10.0 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 go.opentelemetry.io/otel v1.35.0 - go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.35.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.40.0 go.opentelemetry.io/otel/sdk v1.35.0 go.opentelemetry.io/otel/trace v1.35.0 )