diff --git a/flush_error_test.go b/flush_error_test.go new file mode 100644 index 0000000..1024db3 --- /dev/null +++ b/flush_error_test.go @@ -0,0 +1,91 @@ +package quickwit_test + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/moonrhythm/quickwit" +) + +type capturingHandler struct { + mu *sync.Mutex + recs *[]slog.Record +} + +func (h capturingHandler) Enabled(context.Context, slog.Level) bool { return true } +func (h capturingHandler) Handle(_ context.Context, r slog.Record) error { + h.mu.Lock() + defer h.mu.Unlock() + *h.recs = append(*h.recs, r.Clone()) + return nil +} +func (h capturingHandler) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h capturingHandler) WithGroup(string) slog.Handler { return h } + +// The "flush failed, retrying indefinitely" line must carry the failure cause so +// an operator can tell a 5xx/backpressure apart from a transport error. +func TestFlushFailure_RetryLogIncludesCause(t *testing.T) { + var mu sync.Mutex + var recs []slog.Record + prev := slog.Default() + slog.SetDefault(slog.New(capturingHandler{mu: &mu, recs: &recs})) + defer slog.SetDefault(prev) + + var attempts atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + io.Copy(io.Discard, r.Body) + if attempts.Add(1) < 3 { + w.WriteHeader(http.StatusInternalServerError) // transient: two 500s + return + } + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + c := quickwit.NewClient(server.URL + "/api/v1/test") + c.SetConcurrent(1) + c.SetBatchSize(1) + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := c.IngestSync(ctx, map[string]any{"index": 0}); err != nil { + t.Fatalf("IngestSync returned %v, want nil after retries", err) + } + + mu.Lock() + defer mu.Unlock() + var found bool + for _, r := range recs { + if r.Message != "quickwit: flush failed, retrying indefinitely" { + continue + } + found = true + var cause string + r.Attrs(func(a slog.Attr) bool { + if a.Key == "error" { + cause = a.Value.String() + return false + } + return true + }) + if cause == "" || cause == "" { + t.Errorf("retry log has no error cause, want the 5xx status; record=%+v", r) + continue + } + if !strings.Contains(cause, "500") { + t.Errorf("retry log error = %q, want it to name the 500 status", cause) + } + } + if !found { + t.Fatal("did not capture a 'flush failed, retrying indefinitely' log line") + } +} diff --git a/quickwit.go b/quickwit.go index a4419d1..dab2b84 100644 --- a/quickwit.go +++ b/quickwit.go @@ -799,6 +799,12 @@ func (c *Client) loop() { // reset at the top of every flush so it only reflects the most recent attempt. var retryAfter time.Duration + // lastErr carries the reason the most recent flush failed out of flush() and + // into retryFlush()'s log line, so "flush failed, retrying indefinitely" says + // why (transport error, timeout, non-2xx status, …). Reset at the top of every + // flush like retryAfter, so it only reflects the most recent attempt. + var lastErr error + base := strings.TrimSuffix(c.endpoint, "/") + "/ingest" // Two precomputed URLs: the default (commit=auto, used for fire-and-forget) // and the tracked variant carrying the configured commit mode. They are equal @@ -813,6 +819,7 @@ func (c *Client) loop() { } retryAfter = 0 + lastErr = nil buf.Reset() // encodeFailures can only ever hold fire-and-forget items (ack == nil): @@ -861,10 +868,12 @@ func (c *Client) loop() { gzw.Reset(&gzBuf) if _, err := gzw.Write(buf.Bytes()); err != nil { slog.Error("quickwit: failed to gzip ingest body", "error", err) + lastErr = err return false } if err := gzw.Close(); err != nil { slog.Error("quickwit: failed to finalize gzip ingest body", "error", err) + lastErr = err return false } body = gzBuf.Bytes() @@ -878,6 +887,7 @@ func (c *Client) loop() { defer cancel() req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL, bytes.NewReader(body)) if err != nil { + lastErr = err return false } if useGzip { @@ -887,6 +897,7 @@ func (c *Client) loop() { resp, err := c.httpClient().Do(req) if err != nil { + lastErr = fmt.Errorf("quickwit: ingest request failed: %w", err) return false } @@ -895,6 +906,7 @@ func (c *Client) loop() { resp.Body.Close() slog.Error("quickwit: ingest status not ok", "status", resp.Status) + lastErr = fmt.Errorf("quickwit: ingest status %s", resp.Status) if resp.StatusCode == http.StatusRequestEntityTooLarge { if c.autoReduceBatchSize { @@ -952,6 +964,7 @@ func (c *Client) loop() { if readErr != nil { // A truncated/incomplete 200 is ambiguous — retry rather than settle. slog.Error("quickwit: failed to read ingest response", "error", readErr) + lastErr = fmt.Errorf("quickwit: failed to read ingest response: %w", readErr) return false } @@ -1039,7 +1052,7 @@ func (c *Client) loop() { } attempt++ - slog.Info("quickwit: flush failed, retrying indefinitely", "attempt", attempt) + slog.Info("quickwit: flush failed, retrying indefinitely", "attempt", attempt, "error", lastErr) // Equal-jittered backoff, raised to a server-requested Retry-After // when present. The sleep is interruptible so Close (or a long @@ -1085,7 +1098,7 @@ func (c *Client) loop() { } attempt++ - slog.Info("quickwit: flush failed while closing, retrying", "attempt", attempt) + slog.Info("quickwit: flush failed while closing, retrying", "attempt", attempt, "error", lastErr) sleep := jitterBackoff(backoff) if sleep > remaining { sleep = remaining // never sleep past the deadline