diff --git a/README.md b/README.md index 4fd4e55..c26e654 100644 --- a/README.md +++ b/README.md @@ -114,6 +114,10 @@ type Option struct { // optional: records buffered while a batch is in flight (default: 1000). // When the queue is full, records are dropped and reported through OnError. QueueSize int + // optional: retries after a network error, 429 or 5xx (default: 3, -1 disables) + MaxRetries int + // optional: delay before the first retry, doubled on each following one, with random jitter (default: 500ms) + RetryBackoff time.Duration // optional: called when a record cannot be delivered // (default: one line on stderr) OnError func(err error) diff --git a/dispatcher.go b/dispatcher.go index d16bb4f..f243752 100644 --- a/dispatcher.go +++ b/dispatcher.go @@ -2,6 +2,7 @@ package slogbetterstack import ( "errors" + "math/rand" "sync" "time" ) @@ -10,6 +11,11 @@ const ( defaultBatchSize = 100 defaultFlushInterval = time.Second defaultQueueSize = 1000 + defaultMaxRetries = 3 + defaultRetryBackoff = 500 * time.Millisecond + + // maxRetryBackoff bounds the doubling delay (before jitter, which adds up to 50%), so a high MaxRetries does not stall Close for minutes. + maxRetryBackoff = 30 * time.Second ) // ErrQueueFull is reported through OnError when a record is dropped because @@ -87,7 +93,7 @@ func (d *dispatcher) run() { return } - err := send(d.option.Endpoint, d.option.Token, d.option.Timeout, d.option.Marshaler, batch) + err := d.sendWithRetry(batch) if err != nil { d.option.OnError(err) } @@ -113,3 +119,36 @@ func (d *dispatcher) run() { } } } + +// sendWithRetry retries retryable failures with jittered exponential backoff and +// returns the last error. A negative MaxRetries disables retries. +func (d *dispatcher) sendWithRetry(batch []map[string]any) error { + backoff := d.option.RetryBackoff + + var err error + for attempt := 0; ; attempt++ { + err = send(d.option.Endpoint, d.option.Token, d.option.Timeout, d.option.Marshaler, batch) + if err == nil || !retryable(err) || attempt >= d.option.MaxRetries { + return err + } + + time.Sleep(jitter(backoff)) + + backoff *= 2 + if backoff > maxRetryBackoff { + backoff = maxRetryBackoff + } + } +} + +// jitter returns a random duration in [d, d+d/2]. Without it, handlers that failed +// together (a server outage hitting every replica) would retry in lockstep. +// It only ever lengthens the wait, so the backoff stays a lower bound. +func jitter(d time.Duration) time.Duration { + spread := d / 2 + if spread <= 0 { + return d + } + + return d + time.Duration(rand.Int63n(int64(spread)+1)) +} diff --git a/handler.go b/handler.go index 3d9c3cf..6223d7c 100644 --- a/handler.go +++ b/handler.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -43,6 +44,14 @@ type Option struct { // default: 1000 QueueSize int + // optional: retries after a failed attempt (network error, 429 or 5xx) + // default: 3 + MaxRetries int + // optional: delay before the first retry, doubled on each following one, + // randomized between 100% and 150% to avoid synchronized retries + // default: 500ms + RetryBackoff time.Duration + // optional: called when a record cannot be delivered (marshaling failure, // network error, non-2xx response). // default: print one line on stderr @@ -95,6 +104,14 @@ func (o Option) NewBetterstackHandler() *BetterstackHandler { o.QueueSize = defaultQueueSize } + if o.MaxRetries == 0 { + o.MaxRetries = defaultMaxRetries + } + + if o.RetryBackoff <= 0 { + o.RetryBackoff = defaultRetryBackoff + } + if o.Converter == nil { o.Converter = DefaultConverter } @@ -175,7 +192,7 @@ func send(endpoint string, token string, timeout time.Duration, marshaler func(v json, err := marshaler(payload) if err != nil { - return err + return &permanentError{err} } body := bytes.NewBuffer(json) @@ -186,7 +203,7 @@ func send(endpoint string, token string, timeout time.Duration, marshaler func(v // @TODO: maintain a pool of tcp connections req, err := http.NewRequestWithContext(ctx, "POST", endpoint, body) if err != nil { - return err + return &permanentError{err} } req.Header.Add("authorization", `Bearer `+token) @@ -203,12 +220,44 @@ func send(endpoint string, token string, timeout time.Duration, marshaler func(v if resp.StatusCode < 200 || resp.StatusCode > 299 { // best effort: the status code alone is already actionable snippet, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBodyBytes)) - return fmt.Errorf("betterstack rejected logs: status %d: %s", resp.StatusCode, bytes.TrimSpace(snippet)) + return &statusError{code: resp.StatusCode, body: string(bytes.TrimSpace(snippet))} } return nil } +// permanentError marks a failure that a retry cannot fix. +type permanentError struct{ err error } + +func (e *permanentError) Error() string { return e.err.Error() } +func (e *permanentError) Unwrap() error { return e.err } + +// statusError is a non-2xx response. +type statusError struct { + code int + body string +} + +func (e *statusError) Error() string { + return fmt.Sprintf("betterstack rejected logs: status %d: %s", e.code, e.body) +} + +// retryable reports whether sending the same batch again can succeed: network +// errors, throttling and server errors. Other 4xx (bad token, bad payload) fail the same way every time. +func retryable(err error) bool { + var permanent *permanentError + if errors.As(err, &permanent) { + return false + } + + var status *statusError + if errors.As(err, &status) { + return status.code == http.StatusTooManyRequests || status.code >= 500 + } + + return true +} + func defaultOnError(err error) { fmt.Fprintf(os.Stderr, "slog-betterstack: %v\n", err) } diff --git a/handler_test.go b/handler_test.go index 6f0e854..2780425 100644 --- a/handler_test.go +++ b/handler_test.go @@ -21,7 +21,7 @@ func newTestHandler(t *testing.T, status int, body string, onError func(error)) })) t.Cleanup(srv.Close) - h := Option{Token: "t", Endpoint: srv.URL, OnError: onError, FlushInterval: 10 * time.Millisecond}.NewBetterstackHandler() + h := Option{Token: "t", Endpoint: srv.URL, OnError: onError, FlushInterval: 10 * time.Millisecond, MaxRetries: -1}.NewBetterstackHandler() t.Cleanup(func() { _ = h.Close() }) return h @@ -70,7 +70,7 @@ func TestOnError_NotCalledOnSuccess(t *testing.T) { close(done) })) - h := Option{Token: "t", Endpoint: srv.URL, OnError: func(error) { called <- struct{}{} }, FlushInterval: 10 * time.Millisecond}.NewBetterstackHandler() + h := Option{Token: "t", Endpoint: srv.URL, OnError: func(error) { called <- struct{}{} }, FlushInterval: 10 * time.Millisecond, MaxRetries: -1}.NewBetterstackHandler() slog.New(h).Info("hello") select { @@ -96,7 +96,7 @@ func TestOnError_NetworkAndMarshalFailure(t *testing.T) { endpoint := srv.URL srv.Close() // connection refused from here on - refused := Option{Token: "t", Endpoint: endpoint, OnError: func(err error) { errs <- err }, FlushInterval: 10 * time.Millisecond}.NewBetterstackHandler() + refused := Option{Token: "t", Endpoint: endpoint, OnError: func(err error) { errs <- err }, FlushInterval: 10 * time.Millisecond, MaxRetries: -1}.NewBetterstackHandler() defer refused.Close() //nolint:errcheck slog.New(refused).Info("hello") @@ -106,7 +106,7 @@ func TestOnError_NetworkAndMarshalFailure(t *testing.T) { Endpoint: endpoint, Marshaler: func(any) ([]byte, error) { return nil, marshalErr }, OnError: func(err error) { errs <- err }, - FlushInterval: 10 * time.Millisecond, + FlushInterval: 10 * time.Millisecond, MaxRetries: -1, }.NewBetterstackHandler() defer failing.Close() //nolint:errcheck slog.New(failing).Info("hello") diff --git a/retry_test.go b/retry_test.go new file mode 100644 index 0000000..8fb0998 --- /dev/null +++ b/retry_test.go @@ -0,0 +1,190 @@ +package slogbetterstack + +import ( + "errors" + "log/slog" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +const testBackoff = time.Millisecond + +// flakyServer answers with the given statuses in order, then 202. +func flakyServer(statuses ...int) (*httptest.Server, *atomic.Int32) { + var calls atomic.Int32 + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + n := int(calls.Add(1)) - 1 + if n < len(statuses) { + w.WriteHeader(statuses[n]) + return + } + w.WriteHeader(http.StatusAccepted) + })) + + return srv, &calls +} + +func logOnceAndClose(t *testing.T, o Option) []error { + t.Helper() + + var errs []error + o.Token = "t" + o.RetryBackoff = testBackoff + o.OnError = func(err error) { errs = append(errs, err) } + + h := o.NewBetterstackHandler() + slog.New(h).Info("hello") + _ = h.Close() // Close waits for the worker, so errs is safe to read afterwards + + return errs +} + +func TestRetry_RecoversFromRetryableStatuses(t *testing.T) { + for _, status := range []int{http.StatusTooManyRequests, http.StatusInternalServerError, http.StatusBadGateway, http.StatusServiceUnavailable} { + srv, calls := flakyServer(status, status) + + errs := logOnceAndClose(t, Option{Endpoint: srv.URL}) + srv.Close() + + if len(errs) != 0 || calls.Load() != 3 { + t.Fatalf("status %d: expected 3 calls and no error, got %d calls, errors %v", status, calls.Load(), errs) + } + } +} + +func TestRetry_DoesNotRetryClientErrors(t *testing.T) { + for _, status := range []int{http.StatusBadRequest, http.StatusUnauthorized, http.StatusForbidden, http.StatusRequestEntityTooLarge} { + srv, calls := flakyServer(status, status, status) + + errs := logOnceAndClose(t, Option{Endpoint: srv.URL}) + srv.Close() + + if len(errs) != 1 || calls.Load() != 1 { + t.Fatalf("status %d: expected 1 call and 1 error, got %d calls, errors %v", status, calls.Load(), errs) + } + } +} + +func TestRetry_GivesUpAfterMaxRetries(t *testing.T) { + srv, calls := flakyServer(500, 500, 500, 500, 500, 500) + defer srv.Close() + + errs := logOnceAndClose(t, Option{Endpoint: srv.URL, MaxRetries: 2}) + + if calls.Load() != 3 { + t.Fatalf("expected 1 attempt + 2 retries, got %d calls", calls.Load()) + } + if len(errs) != 1 { + t.Fatalf("expected the last error to be reported once, got %v", errs) + } +} + +func TestRetry_NegativeMaxRetriesDisablesRetries(t *testing.T) { + srv, calls := flakyServer(500, 500) + defer srv.Close() + + errs := logOnceAndClose(t, Option{Endpoint: srv.URL, MaxRetries: -1}) + + if calls.Load() != 1 || len(errs) != 1 { + t.Fatalf("expected 1 call and 1 error, got %d calls, errors %v", calls.Load(), errs) + } +} + +func TestRetry_NetworkErrorIsRetried(t *testing.T) { + srv := httptest.NewServer(http.NotFoundHandler()) + endpoint := srv.URL + srv.Close() + + var attempts atomic.Int32 + errs := logOnceAndClose(t, Option{ + Endpoint: endpoint, + MaxRetries: 2, + Marshaler: func(v any) ([]byte, error) { + attempts.Add(1) // marshaling runs once per attempt + return []byte("[]"), nil + }, + }) + + if attempts.Load() != 3 || len(errs) != 1 { + t.Fatalf("expected 3 attempts and 1 error, got %d attempts, errors %v", attempts.Load(), errs) + } +} + +func TestRetry_MarshalErrorIsNotRetried(t *testing.T) { + var attempts atomic.Int32 + boom := errors.New("boom") + + errs := logOnceAndClose(t, Option{ + Endpoint: "http://127.0.0.1:1", + Marshaler: func(any) ([]byte, error) { + attempts.Add(1) + return nil, boom + }, + }) + + if attempts.Load() != 1 || len(errs) != 1 || !errors.Is(errs[0], boom) { + t.Fatalf("expected 1 attempt and the marshal error, got %d attempts, errors %v", attempts.Load(), errs) + } +} + +func TestRetryable(t *testing.T) { + cases := []struct { + err error + want bool + }{ + {&statusError{code: 199}, false}, + {&statusError{code: 400}, false}, + {&statusError{code: 428}, false}, + {&statusError{code: 429}, true}, + {&statusError{code: 499}, false}, + {&statusError{code: 500}, true}, + {&statusError{code: 599}, true}, + {&permanentError{errors.New("x")}, false}, + {errors.New("connection refused"), true}, + } + + for _, c := range cases { + if got := retryable(c.err); got != c.want { + t.Errorf("retryable(%v) = %v, want %v", c.err, got, c.want) + } + } +} + +func TestJitter_StaysWithinBackoffAndOneAndAHalf(t *testing.T) { + const backoff = 100 * time.Millisecond + + min, max := 2*backoff, time.Duration(0) + for i := 0; i < 10000; i++ { + got := jitter(backoff) + if got < backoff || got > backoff+backoff/2 { + t.Fatalf("jitter(%v) = %v, outside [%v, %v]", backoff, got, backoff, backoff+backoff/2) + } + if got < min { + min = got + } + if got > max { + max = got + } + } + + // a constant result would mean jitter is not applied + if max-min < backoff/4 { + t.Fatalf("jitter barely varies: min %v, max %v", min, max) + } +} + +func TestJitter_TinyDurationsDoNotPanic(t *testing.T) { + for _, d := range []time.Duration{-time.Second, 0, 1, 2, 3} { + got := jitter(d) + if d > 0 && (got < d || got > d+d/2) { + t.Fatalf("jitter(%v) = %v, outside [%v, %v]", d, got, d, d+d/2) + } + if d <= 0 && got != d { + t.Fatalf("jitter(%v) = %v, want unchanged", d, got) + } + } +}