From 67b8d6f1ba3f3ec75bde49df87136f0c4c5ba159 Mon Sep 17 00:00:00 2001 From: Hardik Kaurani Date: Thu, 20 Aug 2026 11:09:53 +0530 Subject: [PATCH 1/9] feat(ingestion): add dlq and worker metrics --- ingestion-service/main.go | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/ingestion-service/main.go b/ingestion-service/main.go index 62c88f2..cc29bed 100644 --- a/ingestion-service/main.go +++ b/ingestion-service/main.go @@ -57,11 +57,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. From 25b828b2fb4a7d066bc4f923185f42860a3381db Mon Sep 17 00:00:00 2001 From: Hardik Kaurani Date: Thu, 20 Aug 2026 11:10:07 +0530 Subject: [PATCH 2/9] feat(ingestion): implement bounded log queue and async handler --- ingestion-service/main.go | 57 ++++++++++++++++++++++----------------- 1 file changed, 32 insertions(+), 25 deletions(-) diff --git a/ingestion-service/main.go b/ingestion-service/main.go index cc29bed..393423b 100644 --- a/ingestion-service/main.go +++ b/ingestion-service/main.go @@ -116,14 +116,34 @@ func (w *kafkaLogWriter) Ping(ctx context.Context) error { return conn.Close() } +type LogJob struct { + Entry LogEntry + Raw []byte +} + +const defaultQueueCapacity = 10000 + +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 +} + type Server struct { writer logWriter router *gin.Engine http *http.Server + queue chan LogJob } func NewServer(writer logWriter) *Server { - s := &Server{writer: writer} + s := &Server{ + writer: writer, + queue: make(chan LogJob, getQueueCapacity()), + } s.router = s.setupRouter() return s } @@ -199,32 +219,19 @@ func (s *Server) handleIngest(c *gin.Context) { 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, - }, - ) - - 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"}) - return + job := LogJob{ + Entry: entry, + Raw: val, } - 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: + ingestionLatency.WithLabelValues("success").Observe(time.Since(start).Seconds()) + c.JSON(http.StatusAccepted, gin.H{"status": "accepted", "message": "log queued for ingestion"}) + default: + ingestionLatency.WithLabelValues("rate_limited").Observe(time.Since(start).Seconds()) + c.JSON(http.StatusTooManyRequests, gin.H{"error": "ingestion queue full, retry later"}) + } } func (s *Server) Start(addr string) error { From db84e046c1c4fa2bf90a1d2e87916d4a97653108 Mon Sep 17 00:00:00 2001 From: Hardik Kaurani Date: Thu, 20 Aug 2026 11:10:19 +0530 Subject: [PATCH 3/9] feat(ingestion): add worker pool with exponential backoff retry --- ingestion-service/main.go | 98 +++++++++++++++++++++++++++++++++++---- 1 file changed, 90 insertions(+), 8 deletions(-) diff --git a/ingestion-service/main.go b/ingestion-service/main.go index 393423b..bff4e57 100644 --- a/ingestion-service/main.go +++ b/ingestion-service/main.go @@ -121,7 +121,13 @@ type LogJob struct { Raw []byte } -const defaultQueueCapacity = 10000 +const ( + defaultQueueCapacity = 10000 + defaultWorkerCount = 4 + maxRetries = 3 + initialBackoff = 100 * time.Millisecond + maxBackoff = 2000 * time.Millisecond +) func getQueueCapacity() int { if s := os.Getenv("INGEST_QUEUE_CAPACITY"); s != "" { @@ -132,22 +138,98 @@ func getQueueCapacity() int { return defaultQueueCapacity } +func getWorkerCount() int { + if s := os.Getenv("INGEST_WORKERS"); s != "" { + if n, err := strconv.Atoi(s); err == nil && n > 0 { + return n + } + } + return defaultWorkerCount +} + +func computeBackoff(attempt int) time.Duration { + if attempt <= 0 { + return initialBackoff + } + delay := initialBackoff * time.Duration(1<<(attempt-1)) + if delay > maxBackoff { + return maxBackoff + } + return delay +} + type Server struct { - writer logWriter - router *gin.Engine - http *http.Server - queue chan LogJob + writer logWriter + dlqWriter logWriter + router *gin.Engine + http *http.Server + queue chan LogJob + workerCount int + wg sync.WaitGroup + sleepFn func(time.Duration) } -func NewServer(writer logWriter) *Server { +func NewServer(writer logWriter, dlqWriters ...logWriter) *Server { + var dlq logWriter + if len(dlqWriters) > 0 { + dlq = dlqWriters[0] + } s := &Server{ - writer: writer, - queue: make(chan LogJob, getQueueCapacity()), + writer: writer, + dlqWriter: dlq, + queue: make(chan LogJob, getQueueCapacity()), + workerCount: getWorkerCount(), + sleepFn: time.Sleep, } 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) + } +} + +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) + err := s.writer.WriteMessages(ctx, kafka.Message{ + Key: []byte(job.Entry.Service), + Value: job.Raw, + }) + cancel() + + if err == nil { + logsIngested.WithLabelValues(job.Entry.Service, job.Entry.Level).Inc() + return + } + lastErr = err + log.Printf("kafka write attempt %d failed for service %s: %v", attempt+1, job.Entry.Service, err) + } + + if lastErr != nil { + log.Printf("retries exhausted for service %s: %v", job.Entry.Service, lastErr) + } +} + func (s *Server) setupRouter() *gin.Engine { r := gin.Default() r.GET("/metrics", gin.WrapH(promhttp.Handler())) From f2d7e4bd75a47fbd28d88193ab3bbee874ab7351 Mon Sep 17 00:00:00 2001 From: Hardik Kaurani Date: Thu, 20 Aug 2026 11:10:33 +0530 Subject: [PATCH 4/9] feat(ingestion): implement dlq fallback for retry exhaustion --- ingestion-service/main.go | 63 ++++++++++++++++++++++++++++++++++++++- 1 file changed, 62 insertions(+), 1 deletion(-) diff --git a/ingestion-service/main.go b/ingestion-service/main.go index bff4e57..8930a39 100644 --- a/ingestion-service/main.go +++ b/ingestion-service/main.go @@ -200,6 +200,25 @@ func (s *Server) workerLoop() { } } +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"` +} + +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++ { @@ -225,9 +244,51 @@ func (s *Server) processJob(job LogJob) { 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 { - log.Printf("retries exhausted for service %s: %v", job.Entry.Service, lastErr) + 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 { From 199c8f5be83d135aa2781cc0331ea4a8ed54bff4 Mon Sep 17 00:00:00 2001 From: Hardik Kaurani Date: Thu, 20 Aug 2026 11:10:49 +0530 Subject: [PATCH 5/9] feat(ingestion): add graceful worker shutdown and queue drain --- ingestion-service/main.go | 33 +++++++++++++++++++++++++++++++-- 1 file changed, 31 insertions(+), 2 deletions(-) diff --git a/ingestion-service/main.go b/ingestion-service/main.go index 8930a39..97992a6 100644 --- a/ingestion-service/main.go +++ b/ingestion-service/main.go @@ -386,7 +386,31 @@ 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) + } + + 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() { @@ -396,8 +420,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() @@ -427,5 +452,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") } From 894b3a81af7548d6e368140d51a7a962044b3251 Mon Sep 17 00:00:00 2001 From: Hardik Kaurani Date: Thu, 20 Aug 2026 11:12:40 +0530 Subject: [PATCH 6/9] test(ingestion): add unit test suite for async queue, retry, dlq, and shutdown --- ingestion-service/main.go | 3 +- ingestion-service/main_test.go | 272 ++++++++++++++++----------------- 2 files changed, 134 insertions(+), 141 deletions(-) diff --git a/ingestion-service/main.go b/ingestion-service/main.go index 97992a6..f434a04 100644 --- a/ingestion-service/main.go +++ b/ingestion-service/main.go @@ -9,6 +9,7 @@ import ( "os" "os/signal" "strconv" + "sync" "syscall" "time" @@ -140,7 +141,7 @@ func getQueueCapacity() int { func getWorkerCount() int { if s := os.Getenv("INGEST_WORKERS"); s != "" { - if n, err := strconv.Atoi(s); err == nil && n > 0 { + if n, err := strconv.Atoi(s); err == nil && n >= 0 { return n } } diff --git a/ingestion-service/main_test.go b/ingestion-service/main_test.go index 71c3b8c..e7970c8 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,17 +17,20 @@ 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 } func (m *mockWriter) WriteMessages(ctx context.Context, msgs ...kafka.Message) error { + m.mu.Lock() + defer m.mu.Unlock() + m.writeCount++ 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 select { case <-ctx.Done(): return ctx.Err() @@ -36,22 +40,41 @@ func (m *mockWriter) WriteMessages(ctx context.Context, msgs ...kafka.Message) e return m.writeErr } if len(msgs) > 0 { - m.gotMsg = &msgs[0] + m.gotMsgs = append(m.gotMsgs, msgs...) } 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 +83,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 +111,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 +134,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,166 +157,147 @@ 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")} +func TestHandleIngest_QueueFull_429(t *testing.T) { + t.Setenv("INGEST_QUEUE_CAPACITY", "1") + t.Setenv("INGEST_WORKERS", "0") + mw := &mockWriter{} srv := NewServer(mw) + defer 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) + // Saturate queue of capacity 1 + req1 := httptest.NewRequest(http.MethodPost, "/ingest", strings.NewReader(body)) + req1.Header.Set("Content-Type", "application/json") + rec1 := httptest.NewRecorder() + srv.router.ServeHTTP(rec1, req1) + + // Second request when queue is full returns 429 + 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 rec.Code != http.StatusInternalServerError { - t.Fatalf("expected 500, got %d", rec.Code) + if rec2.Code != http.StatusTooManyRequests { + t.Fatalf("expected 429 Too Many Requests on full queue, got %d", rec2.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(rec2.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} - srv := NewServer(mw) +func TestWorker_RetryExhaustion_DLQSuccess(t *testing.T) { + mainWriter := &mockWriter{writeErr: errors.New("kafka connection failure")} + dlqWriter := &mockWriter{} - 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 := NewServer(mainWriter, dlqWriter) - srv.router.ServeHTTP(rec, req) - - if rec.Code != http.StatusGatewayTimeout { - t.Fatalf("expected 504, got %d", rec.Code) + var delays []time.Duration + var delaysMu sync.Mutex + srv.sleepFn = func(d time.Duration) { + delaysMu.Lock() + delays = append(delays, d) + delaysMu.Unlock() } - 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"]) + job := LogJob{ + Entry: LogEntry{Service: "test-svc", Level: "error", Message: "broker down", Timestamp: 1000}, + Raw: []byte(`{"service":"test-svc","level":"error","message":"broker down"}`), } -} -func TestHandleIngest_RequestBodyTooLarge(t *testing.T) { - mw := &mockWriter{} - srv := NewServer(mw) - - // 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 - 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") - rec := httptest.NewRecorder() + srv.processJob(job) - srv.router.ServeHTTP(rec, req) - - if rec.Code != http.StatusRequestEntityTooLarge { - t.Fatalf("expected 413, got %d — body length: %d", rec.Code, len(body)) + // 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()) } - var resp map[string]string - if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { - t.Fatalf("failed to decode response: %v", err) + // 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)) } - if resp["error"] != "request body too large" { - t.Errorf("expected 'request body too large', got %q", resp["error"]) + for i, d := range delays { + if d != expectedDelays[i] { + t.Errorf("delay %d: expected %v, got %v", i, expectedDelays[i], d) + } } - if mw.gotMsg != nil { - t.Error("expected no Kafka message to be written when body exceeds limit") + // Verify DLQ write succeeded + dlqMsgs := dlqWriter.getMsgs() + if len(dlqMsgs) != 1 { + t.Fatalf("expected 1 message in DLQ, got %d", len(dlqMsgs)) } -} - -func TestHealthz_Healthy(t *testing.T) { - mw := &mockWriter{} - srv := NewServer(mw) - - req := httptest.NewRequest(http.MethodGet, "/healthz", nil) - rec := httptest.NewRecorder() - - srv.router.ServeHTTP(rec, req) - - if rec.Code != http.StatusOK { - t.Fatalf("expected 200, got %d", rec.Code) + if string(dlqMsgs[0].Key) != "test-svc" { + t.Errorf("expected DLQ key 'test-svc', got %q", string(dlqMsgs[0].Key)) } - var resp map[string]string - if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { - t.Fatalf("failed to decode response: %v", err) + var payload DLQPayload + if err := json.Unmarshal(dlqMsgs[0].Value, &payload); err != nil { + t.Fatalf("failed to unmarshal DLQ payload: %v", err) } - if resp["status"] != "ok" { - t.Errorf("expected status 'ok', got %q", resp["status"]) + 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 TestHealthz_Unhealthy(t *testing.T) { - mw := &mockWriter{pingErr: errors.New("kafka broker connection refused")} - srv := NewServer(mw) +func TestWorker_DLQFailureHandled(t *testing.T) { + mainWriter := &mockWriter{writeErr: errors.New("kafka down")} + dlqWriter := &mockWriter{writeErr: errors.New("dlq broker also down")} - req := httptest.NewRequest(http.MethodGet, "/healthz", nil) - rec := httptest.NewRecorder() - - srv.router.ServeHTTP(rec, req) + srv := NewServer(mainWriter, dlqWriter) + srv.sleepFn = func(d time.Duration) {} - if rec.Code != http.StatusServiceUnavailable { - t.Fatalf("expected 503, got %d — body: %s", rec.Code, rec.Body.String()) + job := LogJob{ + Entry: LogEntry{Service: "test-svc", Level: "error", Message: "unreachable"}, + Raw: []byte(`{"service":"test-svc","level":"error","message":"unreachable"}`), } - 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"]) + // Should process retries and attempt DLQ write without crashing or deadlocking + srv.processJob(job) + + if mainWriter.getWriteCount() != 4 { + t.Errorf("expected 4 main writer attempts, got %d", mainWriter.getWriteCount()) } - if resp["error"] != "kafka unavailable" { - t.Errorf("expected error 'kafka unavailable', got %q", resp["error"]) + if dlqWriter.getWriteCount() != 1 { + t.Errorf("expected 1 DLQ write attempt, got %d", dlqWriter.getWriteCount()) } } -func TestHealthz_NilWriter(t *testing.T) { - srv := NewServer(nil) +func TestHandleIngest_RequestBodyTooLarge(t *testing.T) { + mw := &mockWriter{} + srv := NewServer(mw) + defer srv.Shutdown(context.Background()) - req := httptest.NewRequest(http.MethodGet, "/healthz", nil) + 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") rec := httptest.NewRecorder() srv.router.ServeHTTP(rec, req) - 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"]) + if rec.Code != http.StatusRequestEntityTooLarge { + t.Fatalf("expected 413, got %d — body length: %d", rec.Code, len(body)) } } -func TestTimestampAutoFill(t *testing.T) { +func TestHealthz_Healthy(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"}` - req := httptest.NewRequest(http.MethodPost, "/ingest", strings.NewReader(body)) - req.Header.Set("Content-Type", "application/json") + req := httptest.NewRequest(http.MethodGet, "/healthz", nil) rec := httptest.NewRecorder() srv.router.ServeHTTP(rec, req) @@ -314,11 +306,11 @@ func TestTimestampAutoFill(t *testing.T) { t.Fatalf("expected 200, got %d", rec.Code) } - var entry LogEntry - if err := json.Unmarshal(mw.gotMsg.Value, &entry); err != nil { - t.Fatalf("failed to decode written message: %v", err) + var resp map[string]string + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatalf("failed to decode response: %v", err) } - if entry.Timestamp < before { - t.Errorf("expected timestamp >= %d (before request), got %d", before, entry.Timestamp) + if resp["status"] != "ok" { + t.Errorf("expected status 'ok', got %q", resp["status"]) } } From bc4e34f8b97077f88b3cef14a4035b0e500ffd15 Mon Sep 17 00:00:00 2001 From: Hardik Kaurani Date: Thu, 20 Aug 2026 11:12:52 +0530 Subject: [PATCH 7/9] docs(changelog): add entry for issue #29 async ingestion --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) 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 From 7af8023cb6a689c431038df698e603f2ae05a98f Mon Sep 17 00:00:00 2001 From: Hardik Kaurani Date: Fri, 21 Aug 2026 10:13:12 +0530 Subject: [PATCH 8/9] fix(ingestion): address async pipeline review feedback --- ingestion-service/main.go | 50 ++++---- ingestion-service/main_test.go | 213 ++++++++++++++++++++++++++++++--- 2 files changed, 223 insertions(+), 40 deletions(-) diff --git a/ingestion-service/main.go b/ingestion-service/main.go index f434a04..54d80c8 100644 --- a/ingestion-service/main.go +++ b/ingestion-service/main.go @@ -127,7 +127,6 @@ const ( defaultWorkerCount = 4 maxRetries = 3 initialBackoff = 100 * time.Millisecond - maxBackoff = 2000 * time.Millisecond ) func getQueueCapacity() int { @@ -141,7 +140,7 @@ func getQueueCapacity() int { func getWorkerCount() int { if s := os.Getenv("INGEST_WORKERS"); s != "" { - if n, err := strconv.Atoi(s); err == nil && n >= 0 { + if n, err := strconv.Atoi(s); err == nil && n >= 1 { return n } } @@ -152,22 +151,20 @@ func computeBackoff(attempt int) time.Duration { if attempt <= 0 { return initialBackoff } - delay := initialBackoff * time.Duration(1<<(attempt-1)) - if delay > maxBackoff { - return maxBackoff - } - return delay + return initialBackoff * time.Duration(1<<(attempt-1)) } type Server struct { - writer logWriter - dlqWriter logWriter - router *gin.Engine - http *http.Server - queue chan LogJob - workerCount int - wg sync.WaitGroup - sleepFn func(time.Duration) + 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, dlqWriters ...logWriter) *Server { @@ -181,6 +178,7 @@ func NewServer(writer logWriter, dlqWriters ...logWriter) *Server { queue: make(chan LogJob, getQueueCapacity()), workerCount: getWorkerCount(), sleepFn: time.Sleep, + shutdownCh: make(chan struct{}), } s.router = s.setupRouter() s.startWorkers() @@ -209,6 +207,8 @@ type DLQPayload struct { 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{ @@ -231,16 +231,20 @@ func (s *Server) processJob(job LogJob) { } 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) } @@ -325,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. @@ -358,7 +360,6 @@ 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 } @@ -370,10 +371,10 @@ func (s *Server) handleIngest(c *gin.Context) { select { case s.queue <- job: - ingestionLatency.WithLabelValues("success").Observe(time.Since(start).Seconds()) c.JSON(http.StatusAccepted, gin.H{"status": "accepted", "message": "log queued for ingestion"}) + case <-s.shutdownCh: + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "server shutting down"}) default: - ingestionLatency.WithLabelValues("rate_limited").Observe(time.Since(start).Seconds()) c.JSON(http.StatusTooManyRequests, gin.H{"error": "ingestion queue full, retry later"}) } } @@ -392,9 +393,12 @@ func (s *Server) Shutdown(ctx context.Context) error { httpErr = s.http.Shutdown(ctx) } - if s.queue != nil { - close(s.queue) - } + s.shutdownOnce.Do(func() { + close(s.shutdownCh) + if s.queue != nil { + close(s.queue) + } + }) done := make(chan struct{}) go func() { diff --git a/ingestion-service/main_test.go b/ingestion-service/main_test.go index e7970c8..a41ce78 100644 --- a/ingestion-service/main_test.go +++ b/ingestion-service/main_test.go @@ -23,24 +23,32 @@ type mockWriter struct { closeErr error writeCount int gotMsgs []kafka.Message + onWrite func() } func (m *mockWriter) WriteMessages(ctx context.Context, msgs ...kafka.Message) error { m.mu.Lock() - defer m.mu.Unlock() m.writeCount++ - if m.writeErr != nil { - if errors.Is(m.writeErr, context.DeadlineExceeded) { + 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.gotMsgs = append(m.gotMsgs, msgs...) + return writeErr } return nil } @@ -161,36 +169,119 @@ func TestHandleIngest_MissingLevel(t *testing.T) { func TestHandleIngest_QueueFull_429(t *testing.T) { t.Setenv("INGEST_QUEUE_CAPACITY", "1") - t.Setenv("INGEST_WORKERS", "0") - mw := &mockWriter{} - srv := NewServer(mw) - defer srv.Shutdown(context.Background()) + 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"}` - // Saturate queue of capacity 1 + // 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) + } - // Second request when queue is full returns 429 + // 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) + } - if rec2.Code != http.StatusTooManyRequests { - t.Fatalf("expected 429 Too Many Requests on full queue, 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(rec2.Body.Bytes(), &resp) + 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 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":"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) + } + + // 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) + } + + 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{} @@ -263,7 +354,6 @@ func TestWorker_DLQFailureHandled(t *testing.T) { Raw: []byte(`{"service":"test-svc","level":"error","message":"unreachable"}`), } - // Should process retries and attempt DLQ write without crashing or deadlocking srv.processJob(job) if mainWriter.getWriteCount() != 4 { @@ -314,3 +404,92 @@ func TestHealthz_Healthy(t *testing.T) { t.Errorf("expected status 'ok', got %q", resp["status"]) } } + +func TestHealthz_Unhealthy(t *testing.T) { + 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() + + srv.router.ServeHTTP(rec, req) + + if rec.Code != http.StatusServiceUnavailable { + 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() + + srv.router.ServeHTTP(rec, req) + + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("expected 503, got %d", rec.Code) + } +} + +func TestTimestampAutoFill(t *testing.T) { + mw := &mockWriter{} + srv := NewServer(mw) + defer srv.Shutdown(context.Background()) + + 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) + } + + 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(msg.Value, &entry); err != nil { + t.Fatalf("failed to unmarshal message: %v", err) + } + 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) + } +} From 117f5112924fcb96448c22ff79e6f4fb25bd509b Mon Sep 17 00:00:00 2001 From: Hardik Kaurani Date: Fri, 21 Aug 2026 10:24:27 +0530 Subject: [PATCH 9/9] fix(ingestion): check shutdownCh before queue send in handleIngest --- ingestion-service/main.go | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/ingestion-service/main.go b/ingestion-service/main.go index 54d80c8..7af0133 100644 --- a/ingestion-service/main.go +++ b/ingestion-service/main.go @@ -370,10 +370,15 @@ func (s *Server) handleIngest(c *gin.Context) { } select { - case s.queue <- job: - c.JSON(http.StatusAccepted, gin.H{"status": "accepted", "message": "log queued for ingestion"}) case <-s.shutdownCh: c.JSON(http.StatusServiceUnavailable, gin.H{"error": "server shutting down"}) + return + default: + } + + 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"}) }