Triggers 500 error, increments error counter
@@ -549,9 +652,746 @@ func (r *statusRecorder) WriteHeader(code int) {
r.ResponseWriter.WriteHeader(code)
}
+// initRabbitMQ starts background RabbitMQ connection manager with retry logic
+func initRabbitMQ(ctx context.Context) error {
+ host := os.Getenv("RABBITMQ_HOST")
+
+ // If no host configured, RabbitMQ is optional
+ if host == "" {
+ logger.Info("RabbitMQ not configured, skipping initialization")
+ rabbitConnectionStatus.Set(0)
+ 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"
+ }
+ if vhost == "" {
+ vhost = "/"
+ }
+
+ url := fmt.Sprintf("amqp://%s:%s@%s:%s/%s", user, pass, host, port, vhost)
+
+ 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("vhost", vhost),
+ )
+
+ 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)
+
+ select {
+ case <-ctx.Done():
+ return
+ case <-time.After(backoff):
+ }
+
+ // Exponential backoff
+ backoff = backoff * 2
+ if backoff > maxBackoff {
+ backoff = maxBackoff
+ }
+ continue
+ }
+
+ 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),
+ slog.Bool("consumer_mode", consumerMode),
+ )
+
+ // 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)
+ 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
+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")
+}
+
+// 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,
+ })
+
+ // Retry connection with exponential backoff
+ // Valkey may not be ready immediately during pod startup
+ maxRetries := 5
+ backoff := time.Second
+
+ 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()),
+ )
+
+ if attempt < maxRetries {
+ time.Sleep(backoff)
+ backoff *= 2 // Exponential backoff
+ }
+ }
+
+ logger.Error("Valkey connection failed after all retries",
+ slog.String("addr", addr),
+ slog.Int("db", db),
+ )
+ valkeyConnectionStatus.Set(0)
+}
+
+// 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()
+ // 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"
@@ -572,15 +1412,35 @@ func main() {
}()
}
+ // Start RabbitMQ connection manager (non-blocking, handles retries internally)
+ initRabbitMQ(ctx)
+ defer closeRabbitMQ()
+
+ // Initialize Valkey/Redis connection
+ initValkey()
+ defer closeValkey()
+
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))
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))
+ // 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{
Addr: ":" + port,
Handler: mux,
@@ -593,6 +1453,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 +1465,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