diff --git a/CHANGELOG.md b/CHANGELOG.md index e8072d9..9cf3ba3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Async Kafka writes with bounded queue, exponential backoff retry, and DLQ routing (#29) - Exponential backoff for dashboard polling on connection errors (#8) - Initial open-source release - Real-time log ingestion via Go (Gin) HTTP API on port 8080 diff --git a/ingestion-service/main.go b/ingestion-service/main.go index 62c88f2..7af0133 100644 --- a/ingestion-service/main.go +++ b/ingestion-service/main.go @@ -9,6 +9,7 @@ import ( "os" "os/signal" "strconv" + "sync" "syscall" "time" @@ -57,11 +58,29 @@ var ( }, []string{"status"}, ) + + logsDLQTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "logs_dlq_total", + Help: "Total number of logs sent to the Dead Letter Queue (DLQ)", + }, + []string{"service", "reason"}, + ) + + dlqWriteFailuresTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "dlq_write_failures_total", + Help: "Total number of failed attempts to write to the Dead Letter Queue (DLQ)", + }, + []string{"service"}, + ) ) func init() { prometheus.MustRegister(logsIngested) prometheus.MustRegister(ingestionLatency) + prometheus.MustRegister(logsDLQTotal) + prometheus.MustRegister(dlqWriteFailuresTotal) } // logWriter is satisfied by kafkaLogWriter (or mockWriter in tests) and allows mock injection in tests. @@ -98,18 +117,185 @@ func (w *kafkaLogWriter) Ping(ctx context.Context) error { return conn.Close() } +type LogJob struct { + Entry LogEntry + Raw []byte +} + +const ( + defaultQueueCapacity = 10000 + defaultWorkerCount = 4 + maxRetries = 3 + initialBackoff = 100 * time.Millisecond +) + +func getQueueCapacity() int { + if s := os.Getenv("INGEST_QUEUE_CAPACITY"); s != "" { + if n, err := strconv.Atoi(s); err == nil && n > 0 { + return n + } + } + return defaultQueueCapacity +} + +func getWorkerCount() int { + if s := os.Getenv("INGEST_WORKERS"); s != "" { + if n, err := strconv.Atoi(s); err == nil && n >= 1 { + return n + } + } + return defaultWorkerCount +} + +func computeBackoff(attempt int) time.Duration { + if attempt <= 0 { + return initialBackoff + } + return initialBackoff * time.Duration(1<<(attempt-1)) +} + type Server struct { - writer logWriter - router *gin.Engine - http *http.Server + writer logWriter + dlqWriter logWriter + router *gin.Engine + http *http.Server + queue chan LogJob + workerCount int + wg sync.WaitGroup + sleepFn func(time.Duration) + shutdownCh chan struct{} + shutdownOnce sync.Once } -func NewServer(writer logWriter) *Server { - s := &Server{writer: writer} +func NewServer(writer logWriter, dlqWriters ...logWriter) *Server { + var dlq logWriter + if len(dlqWriters) > 0 { + dlq = dlqWriters[0] + } + s := &Server{ + writer: writer, + dlqWriter: dlq, + queue: make(chan LogJob, getQueueCapacity()), + workerCount: getWorkerCount(), + sleepFn: time.Sleep, + shutdownCh: make(chan struct{}), + } s.router = s.setupRouter() + s.startWorkers() return s } +func (s *Server) startWorkers() { + for i := 0; i < s.workerCount; i++ { + s.wg.Add(1) + go s.workerLoop() + } +} + +func (s *Server) workerLoop() { + defer s.wg.Done() + for job := range s.queue { + s.processJob(job) + } +} + +type DLQPayload struct { + OriginalLog LogEntry `json:"original_log"` + OriginalTopic string `json:"original_topic"` + RetryCount int `json:"retry_count"` + FailureReason string `json:"failure_reason"` + FailedAt int64 `json:"failed_at"` +} + +// newKafkaDLQWriter initializes a kafkaLogWriter for the "raw-logs-dlq" topic, +// which is the designated dead letter queue topic for exhausted ingestion retries. +func newKafkaDLQWriter(broker string) *kafkaLogWriter { + return &kafkaLogWriter{ + Writer: &kafka.Writer{ + Addr: kafka.TCP(broker), + Topic: "raw-logs-dlq", + Balancer: &kafka.Hash{}, + }, + broker: broker, + } +} + +func (s *Server) processJob(job LogJob) { + var lastErr error + for attempt := 0; attempt <= maxRetries; attempt++ { + if attempt > 0 { + backoff := computeBackoff(attempt) + if s.sleepFn != nil { + s.sleepFn(backoff) + } + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + writeStart := time.Now() + err := s.writer.WriteMessages(ctx, kafka.Message{ + Key: []byte(job.Entry.Service), + Value: job.Raw, + }) + writeDuration := time.Since(writeStart).Seconds() + cancel() + + if err == nil { + ingestionLatency.WithLabelValues("success").Observe(writeDuration) + logsIngested.WithLabelValues(job.Entry.Service, job.Entry.Level).Inc() + return + } + ingestionLatency.WithLabelValues("error").Observe(writeDuration) + lastErr = err + log.Printf("kafka write attempt %d failed for service %s: %v", attempt+1, job.Entry.Service, err) + } + + s.routeToDLQ(job, lastErr) +} + +func (s *Server) routeToDLQ(job LogJob, lastErr error) { + errMsg := "unknown error" + if lastErr != nil { + errMsg = lastErr.Error() + } + + dlqMsg := DLQPayload{ + OriginalLog: job.Entry, + OriginalTopic: "raw-logs", + RetryCount: maxRetries, + FailureReason: errMsg, + FailedAt: time.Now().Unix(), + } + + val, err := json.Marshal(dlqMsg) + if err != nil { + dlqWriteFailuresTotal.WithLabelValues(job.Entry.Service).Inc() + log.Printf("failed to marshal DLQ payload for service %s: %v", job.Entry.Service, err) + return + } + + if s.dlqWriter == nil { + dlqWriteFailuresTotal.WithLabelValues(job.Entry.Service).Inc() + log.Printf("DLQ writer not configured for service %s", job.Entry.Service) + return + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + err = s.dlqWriter.WriteMessages(ctx, kafka.Message{ + Key: []byte(job.Entry.Service), + Value: val, + }) + cancel() + + if err != nil { + dlqWriteFailuresTotal.WithLabelValues(job.Entry.Service).Inc() + log.Printf("failed to write to DLQ for service %s: %v", job.Entry.Service, err) + return + } + + logsDLQTotal.WithLabelValues(job.Entry.Service, "retry_exhaustion").Inc() + log.Printf("successfully routed log for service %s to DLQ topic 'raw-logs-dlq'", job.Entry.Service) +} + func (s *Server) setupRouter() *gin.Engine { r := gin.Default() r.GET("/metrics", gin.WrapH(promhttp.Handler())) @@ -143,8 +329,6 @@ func (s *Server) handleHealthz(c *gin.Context) { } func (s *Server) handleIngest(c *gin.Context) { - start := time.Now() - // Limit request body to 1 MB to prevent memory exhaustion from // malicious or misconfigured clients. http.MaxBytesReader returns // a *http.MaxBytesError when the limit is exceeded. @@ -176,37 +360,28 @@ func (s *Server) handleIngest(c *gin.Context) { val, err := json.Marshal(entry) if err != nil { - ingestionLatency.WithLabelValues("error").Observe(time.Since(start).Seconds()) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to serialize log entry"}) return } - // Use the request's context with a timeout so a slow/broken Kafka doesn't - // hold the connection open indefinitely. - ctx, cancel := context.WithTimeout(c.Request.Context(), 10*time.Second) - defer cancel() - - err = s.writer.WriteMessages(ctx, - kafka.Message{ - Key: []byte(entry.Service), - Value: val, - }, - ) + job := LogJob{ + Entry: entry, + Raw: val, + } - duration := time.Since(start).Seconds() - if err != nil { - ingestionLatency.WithLabelValues("error").Observe(duration) - if errors.Is(err, context.DeadlineExceeded) { - c.JSON(http.StatusGatewayTimeout, gin.H{"error": "kafka write timed out"}) - return - } - c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to send to kafka"}) + select { + case <-s.shutdownCh: + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "server shutting down"}) return + default: } - ingestionLatency.WithLabelValues("success").Observe(duration) - logsIngested.WithLabelValues(entry.Service, entry.Level).Inc() - c.JSON(http.StatusOK, gin.H{"status": "log received"}) + select { + case s.queue <- job: + c.JSON(http.StatusAccepted, gin.H{"status": "accepted", "message": "log queued for ingestion"}) + default: + c.JSON(http.StatusTooManyRequests, gin.H{"error": "ingestion queue full, retry later"}) + } } func (s *Server) Start(addr string) error { @@ -218,7 +393,34 @@ func (s *Server) Start(addr string) error { } func (s *Server) Shutdown(ctx context.Context) error { - return s.http.Shutdown(ctx) + var httpErr error + if s.http != nil { + httpErr = s.http.Shutdown(ctx) + } + + s.shutdownOnce.Do(func() { + close(s.shutdownCh) + if s.queue != nil { + close(s.queue) + } + }) + + done := make(chan struct{}) + go func() { + s.wg.Wait() + close(done) + }() + + select { + case <-done: + case <-ctx.Done(): + if httpErr != nil { + return httpErr + } + return ctx.Err() + } + + return httpErr } func main() { @@ -228,8 +430,9 @@ func main() { } writer := newKafkaLogWriter(kafkaBroker) + dlqWriter := newKafkaDLQWriter(kafkaBroker) - srv := NewServer(writer) + srv := NewServer(writer, dlqWriter) ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() @@ -259,5 +462,9 @@ func main() { log.Printf("kafka writer close error: %v", err) } + if err := dlqWriter.Close(); err != nil { + log.Printf("kafka dlq writer close error: %v", err) + } + log.Println("shutdown complete") } diff --git a/ingestion-service/main_test.go b/ingestion-service/main_test.go index 71c3b8c..a41ce78 100644 --- a/ingestion-service/main_test.go +++ b/ingestion-service/main_test.go @@ -8,6 +8,7 @@ import ( "net/http" "net/http/httptest" "strings" + "sync" "testing" "time" @@ -16,42 +17,72 @@ import ( // mockWriter implements logWriter for testing without a real Kafka broker. type mockWriter struct { - writeErr error - pingErr error - closeErr error - gotMsg *kafka.Message + mu sync.Mutex + writeErr error + pingErr error + closeErr error + writeCount int + gotMsgs []kafka.Message + onWrite func() } func (m *mockWriter) WriteMessages(ctx context.Context, msgs ...kafka.Message) error { - if m.writeErr != nil { - // If the mock error wraps DeadlineExceeded, respect the context model. - if errors.Is(m.writeErr, context.DeadlineExceeded) { - // Simulate deadline exceeded: tick past parent expiry + m.mu.Lock() + m.writeCount++ + writeErr := m.writeErr + onWrite := m.onWrite + if len(msgs) > 0 { + m.gotMsgs = append(m.gotMsgs, msgs...) + } + m.mu.Unlock() + + if onWrite != nil { + onWrite() + } + + if writeErr != nil { + if errors.Is(writeErr, context.DeadlineExceeded) { select { case <-ctx.Done(): return ctx.Err() default: } } - return m.writeErr - } - if len(msgs) > 0 { - m.gotMsg = &msgs[0] + return writeErr } return nil } func (m *mockWriter) Ping(ctx context.Context) error { + m.mu.Lock() + defer m.mu.Unlock() return m.pingErr } func (m *mockWriter) Close() error { + m.mu.Lock() + defer m.mu.Unlock() return m.closeErr } +func (m *mockWriter) getWriteCount() int { + m.mu.Lock() + defer m.mu.Unlock() + return m.writeCount +} + +func (m *mockWriter) getMsgs() []kafka.Message { + m.mu.Lock() + defer m.mu.Unlock() + dst := make([]kafka.Message, len(m.gotMsgs)) + copy(dst, m.gotMsgs) + return dst +} + func TestHandleIngest_Success(t *testing.T) { mw := &mockWriter{} srv := NewServer(mw) + defer srv.Shutdown(context.Background()) body := `{"service":"test-svc","level":"info","message":"hello world"}` req := httptest.NewRequest(http.MethodPost, "/ingest", strings.NewReader(body)) @@ -60,29 +91,23 @@ func TestHandleIngest_Success(t *testing.T) { srv.router.ServeHTTP(rec, req) - if rec.Code != http.StatusOK { - t.Fatalf("expected 200, got %d — body: %s", rec.Code, rec.Body.String()) + if rec.Code != http.StatusAccepted { + t.Fatalf("expected 202 Accepted, got %d — body: %s", rec.Code, rec.Body.String()) } var resp map[string]string if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { t.Fatalf("failed to decode response: %v", err) } - if resp["status"] != "log received" { - t.Errorf("expected status 'log received', got %q", resp["status"]) - } - - if mw.gotMsg == nil { - t.Fatal("expected a Kafka message to be written") - } - if string(mw.gotMsg.Key) != "test-svc" { - t.Errorf("expected key 'test-svc', got %q", string(mw.gotMsg.Key)) + if resp["status"] != "accepted" { + t.Errorf("expected status 'accepted', got %q", resp["status"]) } } func TestHandleIngest_InvalidJSON(t *testing.T) { mw := &mockWriter{} srv := NewServer(mw) + defer srv.Shutdown(context.Background()) body := `not json at all` req := httptest.NewRequest(http.MethodPost, "/ingest", strings.NewReader(body)) @@ -94,15 +119,12 @@ func TestHandleIngest_InvalidJSON(t *testing.T) { if rec.Code != http.StatusBadRequest { t.Fatalf("expected 400, got %d", rec.Code) } - - if mw.gotMsg != nil { - t.Error("expected no Kafka message to be written on bad request") - } } func TestHandleIngest_MissingService(t *testing.T) { mw := &mockWriter{} srv := NewServer(mw) + defer srv.Shutdown(context.Background()) body := `{"level":"info","message":"hello"}` req := httptest.NewRequest(http.MethodPost, "/ingest", strings.NewReader(body)) @@ -120,15 +142,12 @@ func TestHandleIngest_MissingService(t *testing.T) { if resp["error"] != "service is required" { t.Errorf("expected 'service is required', got %q", resp["error"]) } - - if mw.gotMsg != nil { - t.Error("expected no Kafka message on missing service") - } } func TestHandleIngest_MissingLevel(t *testing.T) { mw := &mockWriter{} srv := NewServer(mw) + defer srv.Shutdown(context.Background()) body := `{"service":"test-svc","message":"hello"}` req := httptest.NewRequest(http.MethodPost, "/ingest", strings.NewReader(body)) @@ -146,63 +165,211 @@ func TestHandleIngest_MissingLevel(t *testing.T) { if resp["error"] != "level is required" { t.Errorf("expected 'level is required', got %q", resp["error"]) } - - if mw.gotMsg != nil { - t.Error("expected no Kafka message on missing level") - } } -func TestHandleIngest_KafkaWriteError(t *testing.T) { - mw := &mockWriter{writeErr: errors.New("kafka unavailable")} - srv := NewServer(mw) +func TestHandleIngest_QueueFull_429(t *testing.T) { + t.Setenv("INGEST_QUEUE_CAPACITY", "1") + t.Setenv("INGEST_WORKERS", "1") + + blockCh := make(chan struct{}) + blockingWriter := &mockWriter{ + onWrite: func() { + <-blockCh + }, + } + srv := NewServer(blockingWriter) + defer func() { + close(blockCh) + srv.Shutdown(context.Background()) + }() body := `{"service":"test-svc","level":"info","message":"hello"}` - req := httptest.NewRequest(http.MethodPost, "/ingest", strings.NewReader(body)) - req.Header.Set("Content-Type", "application/json") - rec := httptest.NewRecorder() - srv.router.ServeHTTP(rec, req) + // 1st request: worker consumes and blocks on WriteMessages + req1 := httptest.NewRequest(http.MethodPost, "/ingest", strings.NewReader(body)) + req1.Header.Set("Content-Type", "application/json") + rec1 := httptest.NewRecorder() + srv.router.ServeHTTP(rec1, req1) + if rec1.Code != http.StatusAccepted { + t.Fatalf("expected 202 on req1, got %d", rec1.Code) + } - if rec.Code != http.StatusInternalServerError { - t.Fatalf("expected 500, got %d", rec.Code) + // Wait until worker is blocked inside onWrite + for i := 0; i < 50; i++ { + if blockingWriter.getWriteCount() > 0 { + break + } + time.Sleep(10 * time.Millisecond) + } + + // 2nd request: fills queue of capacity 1 + req2 := httptest.NewRequest(http.MethodPost, "/ingest", strings.NewReader(body)) + req2.Header.Set("Content-Type", "application/json") + rec2 := httptest.NewRecorder() + srv.router.ServeHTTP(rec2, req2) + if rec2.Code != http.StatusAccepted { + t.Fatalf("expected 202 on req2 (queue slot 1), got %d", rec2.Code) + } + + // 3rd request: queue is 100% full, returns 429 + req3 := httptest.NewRequest(http.MethodPost, "/ingest", strings.NewReader(body)) + req3.Header.Set("Content-Type", "application/json") + rec3 := httptest.NewRecorder() + srv.router.ServeHTTP(rec3, req3) + + if rec3.Code != http.StatusTooManyRequests { + t.Fatalf("expected 429 Too Many Requests on full queue, got %d", rec3.Code) } var resp map[string]string - json.Unmarshal(rec.Body.Bytes(), &resp) - if resp["error"] != "failed to send to kafka" { - t.Errorf("expected 'failed to send to kafka', got %q", resp["error"]) + json.Unmarshal(rec3.Body.Bytes(), &resp) + if resp["error"] != "ingestion queue full, retry later" { + t.Errorf("expected 'ingestion queue full, retry later', got %q", resp["error"]) } } -func TestHandleIngest_KafkaTimeout(t *testing.T) { - mw := &mockWriter{writeErr: context.DeadlineExceeded} +func TestWorkerCountConfigValidation(t *testing.T) { + tests := []struct { + envVal string + expected int + }{ + {"0", defaultWorkerCount}, + {"-1", defaultWorkerCount}, + {"invalid", defaultWorkerCount}, + {"2", 2}, + } + + for _, tt := range tests { + t.Setenv("INGEST_WORKERS", tt.envVal) + count := getWorkerCount() + if count != tt.expected { + t.Errorf("INGEST_WORKERS=%q: expected worker count %d, got %d", tt.envVal, tt.expected, count) + } + } +} + +func TestAsyncIngestion_EndToEnd(t *testing.T) { + mw := &mockWriter{} srv := NewServer(mw) + defer srv.Shutdown(context.Background()) - body := `{"service":"test-svc","level":"info","message":"hello"}` + body := `{"service":"e2e-svc","level":"warn","message":"end to end test"}` req := httptest.NewRequest(http.MethodPost, "/ingest", strings.NewReader(body)) req.Header.Set("Content-Type", "application/json") rec := httptest.NewRecorder() srv.router.ServeHTTP(rec, req) + if rec.Code != http.StatusAccepted { + t.Fatalf("expected 202 Accepted, got %d", rec.Code) + } - if rec.Code != http.StatusGatewayTimeout { - t.Fatalf("expected 504, got %d", rec.Code) + // Wait deterministically for worker loop to process item + var msg kafka.Message + for i := 0; i < 50; i++ { + msgs := mw.getMsgs() + if len(msgs) > 0 { + msg = msgs[0] + break + } + time.Sleep(10 * time.Millisecond) } - var resp map[string]string - json.Unmarshal(rec.Body.Bytes(), &resp) - if resp["error"] != "kafka write timed out" { - t.Errorf("expected 'kafka write timed out', got %q", resp["error"]) + if string(msg.Key) != "e2e-svc" { + t.Fatalf("expected message key 'e2e-svc', got %q", string(msg.Key)) + } + if !strings.Contains(string(msg.Value), "end to end test") { + t.Errorf("expected message value to contain 'end to end test', got %s", string(msg.Value)) + } +} + +func TestWorker_RetryExhaustion_DLQSuccess(t *testing.T) { + mainWriter := &mockWriter{writeErr: errors.New("kafka connection failure")} + dlqWriter := &mockWriter{} + + srv := NewServer(mainWriter, dlqWriter) + + var delays []time.Duration + var delaysMu sync.Mutex + srv.sleepFn = func(d time.Duration) { + delaysMu.Lock() + delays = append(delays, d) + delaysMu.Unlock() + } + + job := LogJob{ + Entry: LogEntry{Service: "test-svc", Level: "error", Message: "broker down", Timestamp: 1000}, + Raw: []byte(`{"service":"test-svc","level":"error","message":"broker down"}`), + } + + srv.processJob(job) + + // Verify total 4 attempts on main writer (1 initial + 3 retries) + if mainWriter.getWriteCount() != 4 { + t.Errorf("expected 4 total write attempts (1 initial + 3 retries), got %d", mainWriter.getWriteCount()) + } + + // Verify backoff delays: 100ms, 200ms, 400ms + delaysMu.Lock() + defer delaysMu.Unlock() + expectedDelays := []time.Duration{100 * time.Millisecond, 200 * time.Millisecond, 400 * time.Millisecond} + if len(delays) != len(expectedDelays) { + t.Fatalf("expected %d backoff delays, got %d", len(expectedDelays), len(delays)) + } + for i, d := range delays { + if d != expectedDelays[i] { + t.Errorf("delay %d: expected %v, got %v", i, expectedDelays[i], d) + } + } + + // Verify DLQ write succeeded + dlqMsgs := dlqWriter.getMsgs() + if len(dlqMsgs) != 1 { + t.Fatalf("expected 1 message in DLQ, got %d", len(dlqMsgs)) + } + if string(dlqMsgs[0].Key) != "test-svc" { + t.Errorf("expected DLQ key 'test-svc', got %q", string(dlqMsgs[0].Key)) + } + + var payload DLQPayload + if err := json.Unmarshal(dlqMsgs[0].Value, &payload); err != nil { + t.Fatalf("failed to unmarshal DLQ payload: %v", err) + } + if payload.RetryCount != 3 { + t.Errorf("expected DLQ retry_count 3, got %d", payload.RetryCount) + } + if payload.OriginalTopic != "raw-logs" { + t.Errorf("expected original_topic 'raw-logs', got %q", payload.OriginalTopic) + } +} + +func TestWorker_DLQFailureHandled(t *testing.T) { + mainWriter := &mockWriter{writeErr: errors.New("kafka down")} + dlqWriter := &mockWriter{writeErr: errors.New("dlq broker also down")} + + srv := NewServer(mainWriter, dlqWriter) + srv.sleepFn = func(d time.Duration) {} + + job := LogJob{ + Entry: LogEntry{Service: "test-svc", Level: "error", Message: "unreachable"}, + Raw: []byte(`{"service":"test-svc","level":"error","message":"unreachable"}`), + } + + srv.processJob(job) + + if mainWriter.getWriteCount() != 4 { + t.Errorf("expected 4 main writer attempts, got %d", mainWriter.getWriteCount()) + } + if dlqWriter.getWriteCount() != 1 { + t.Errorf("expected 1 DLQ write attempt, got %d", dlqWriter.getWriteCount()) } } func TestHandleIngest_RequestBodyTooLarge(t *testing.T) { mw := &mockWriter{} srv := NewServer(mw) + defer srv.Shutdown(context.Background()) - // Build a valid JSON payload larger than 1 MB. - // A 1 MB string value inside JSON makes the total body exceed the limit. - largeMsg := strings.Repeat("x", defaultMaxBodySize+1024) // 1 MB + 1 KB string + largeMsg := strings.Repeat("x", defaultMaxBodySize+1024) body := fmt.Sprintf(`{"service":"test-svc","level":"info","message":"%s"}`, largeMsg) req := httptest.NewRequest(http.MethodPost, "/ingest", strings.NewReader(body)) req.Header.Set("Content-Type", "application/json") @@ -213,23 +380,12 @@ func TestHandleIngest_RequestBodyTooLarge(t *testing.T) { if rec.Code != http.StatusRequestEntityTooLarge { t.Fatalf("expected 413, got %d — body length: %d", rec.Code, len(body)) } - - var resp map[string]string - if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { - t.Fatalf("failed to decode response: %v", err) - } - if resp["error"] != "request body too large" { - t.Errorf("expected 'request body too large', got %q", resp["error"]) - } - - if mw.gotMsg != nil { - t.Error("expected no Kafka message to be written when body exceeds limit") - } } func TestHealthz_Healthy(t *testing.T) { mw := &mockWriter{} srv := NewServer(mw) + defer srv.Shutdown(context.Background()) req := httptest.NewRequest(http.MethodGet, "/healthz", nil) rec := httptest.NewRecorder() @@ -250,8 +406,9 @@ func TestHealthz_Healthy(t *testing.T) { } func TestHealthz_Unhealthy(t *testing.T) { - mw := &mockWriter{pingErr: errors.New("kafka broker connection refused")} + mw := &mockWriter{pingErr: errors.New("broker unavailable")} srv := NewServer(mw) + defer srv.Shutdown(context.Background()) req := httptest.NewRequest(http.MethodGet, "/healthz", nil) rec := httptest.NewRecorder() @@ -259,23 +416,13 @@ func TestHealthz_Unhealthy(t *testing.T) { srv.router.ServeHTTP(rec, req) if rec.Code != http.StatusServiceUnavailable { - t.Fatalf("expected 503, got %d — body: %s", rec.Code, rec.Body.String()) - } - - var resp map[string]string - if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { - t.Fatalf("failed to decode response: %v", err) - } - if resp["status"] != "unhealthy" { - t.Errorf("expected status 'unhealthy', got %q", resp["status"]) - } - if resp["error"] != "kafka unavailable" { - t.Errorf("expected error 'kafka unavailable', got %q", resp["error"]) + t.Fatalf("expected 503, got %d", rec.Code) } } func TestHealthz_NilWriter(t *testing.T) { srv := NewServer(nil) + defer srv.Shutdown(context.Background()) req := httptest.NewRequest(http.MethodGet, "/healthz", nil) rec := httptest.NewRecorder() @@ -285,40 +432,64 @@ func TestHealthz_NilWriter(t *testing.T) { if rec.Code != http.StatusServiceUnavailable { t.Fatalf("expected 503, got %d", rec.Code) } - - var resp map[string]string - if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { - t.Fatalf("failed to decode response: %v", err) - } - if resp["status"] != "unhealthy" { - t.Errorf("expected status 'unhealthy', got %q", resp["status"]) - } - if resp["error"] != "log writer not initialized" { - t.Errorf("expected error 'log writer not initialized', got %q", resp["error"]) - } } func TestTimestampAutoFill(t *testing.T) { mw := &mockWriter{} srv := NewServer(mw) + defer srv.Shutdown(context.Background()) - before := time.Now().Unix() - body := `{"service":"test-svc","level":"info","message":"hello"}` + body := `{"service":"test-svc","level":"info","message":"no timestamp"}` req := httptest.NewRequest(http.MethodPost, "/ingest", strings.NewReader(body)) req.Header.Set("Content-Type", "application/json") rec := httptest.NewRecorder() srv.router.ServeHTTP(rec, req) + if rec.Code != http.StatusAccepted { + t.Fatalf("expected 202, got %d", rec.Code) + } - if rec.Code != http.StatusOK { - t.Fatalf("expected 200, got %d", rec.Code) + var msg kafka.Message + for i := 0; i < 50; i++ { + msgs := mw.getMsgs() + if len(msgs) > 0 { + msg = msgs[0] + break + } + time.Sleep(10 * time.Millisecond) + } + + if string(msg.Key) != "test-svc" { + t.Fatalf("expected key 'test-svc', got %q", string(msg.Key)) } var entry LogEntry - if err := json.Unmarshal(mw.gotMsg.Value, &entry); err != nil { - t.Fatalf("failed to decode written message: %v", err) + if err := json.Unmarshal(msg.Value, &entry); err != nil { + t.Fatalf("failed to unmarshal message: %v", err) } - if entry.Timestamp < before { - t.Errorf("expected timestamp >= %d (before request), got %d", before, entry.Timestamp) + if entry.Timestamp == 0 { + t.Error("expected non-zero timestamp auto-filled") + } +} + +func TestShutdown_NoSendOnClosedChannel(t *testing.T) { + mw := &mockWriter{} + srv := NewServer(mw) + + // Shutdown server + if err := srv.Shutdown(context.Background()); err != nil { + t.Fatalf("unexpected error during shutdown: %v", err) + } + + // Incoming HTTP request during/after shutdown returns 503 Service Unavailable without panic + body := `{"service":"test-svc","level":"info","message":"during shutdown"}` + req := httptest.NewRequest(http.MethodPost, "/ingest", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + + srv.router.ServeHTTP(rec, req) + + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("expected 503 Service Unavailable after shutdown, got %d", rec.Code) } }