From c8ed236773a21f7b2c1666a2e0698183455f59be Mon Sep 17 00:00:00 2001 From: Petr Heinz Date: Tue, 6 Oct 2026 15:31:06 +0200 Subject: [PATCH 01/11] T-20356 Add failing tests: deliver every record, report every failure The handler sends each record in its own goroutine with no way to wait for it, ignores the response status and the error, retries nothing and has no limit on the goroutines it starts. Records logged right before the process exits are lost, a wrong token is never noticed, and an outage either drops batches silently or stacks up goroutines. These tests pin the behaviour the next commits implement, on top of the shape and option tests that stay as they are: - Close delivers what is still queued and Flush waits for delivery. - Records are batched by size and interval and gzip-compressed, with the JSON array body and the Marshaler contract unchanged. - Derived handlers share one queue. - 408, 429, 5xx and network errors are retried with backoff, honouring Retry-After; other 4xx are terminal. A 413 batch is split. - Delivery failures and drops are reported through OnError; the default reporter writes to stderr. Stats count every record. - A full queue drops instead of blocking the application; a shutdown timeout bounds Close; Timeout bounds each attempt. - Concurrent Handle calls on a derived handler do not race. The tests do not compile against the current API (no Close, Flush, Stats or the new Option fields), so this commit is expected to fail CI. Co-Authored-By: Claude Fable 5.1 --- handler_test.go | 258 +++++++++++++++++-------- transport_test.go | 480 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 660 insertions(+), 78 deletions(-) create mode 100644 transport_test.go diff --git a/handler_test.go b/handler_test.go index aeba805..d45dcb5 100644 --- a/handler_test.go +++ b/handler_test.go @@ -1,6 +1,7 @@ package slogbetterstack import ( + "compress/gzip" "context" "encoding/json" "errors" @@ -11,43 +12,64 @@ import ( "reflect" "sort" "strings" + "sync" "testing" "time" ) // These tests pin the behaviour applications rely on: the shape of the records that reach -// Better Stack, the request headers, and what every option does. The transport may change -// underneath them, so they only ever look at what arrives at the server, never at how. +// Better Stack, the request headers, and what every option does. They only ever look at what +// arrives at the server, never at how it got there. type request struct { header http.Header records []map[string]any } -// newServer stands in for Better Stack. It reports every request it receives with the records -// decoded from its JSON array body. -func newServer(t *testing.T) (*httptest.Server, <-chan request) { +// responder writes the status Better Stack would answer with. The records are the decoded body. +type responder func(w http.ResponseWriter, r *http.Request, records []map[string]any) + +func accepted(w http.ResponseWriter, _ *http.Request, _ []map[string]any) { + w.WriteHeader(http.StatusAccepted) +} + +func status(code int) responder { + return func(w http.ResponseWriter, _ *http.Request, _ []map[string]any) { w.WriteHeader(code) } +} + +// newServer stands in for Better Stack: it decodes every request's JSON array body, gzip-compressed +// or not, lets respond answer it and reports the request once it is answered. +func newServer(t *testing.T, respond responder) (*httptest.Server, <-chan request) { t.Helper() - requests := make(chan request, 16) + requests := make(chan request, 1024) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - body, err := io.ReadAll(r.Body) + var body io.Reader = r.Body + if r.Header.Get("Content-Encoding") == "gzip" { + unzipped, err := gzip.NewReader(r.Body) + if err != nil { + t.Errorf("Content-Encoding is gzip but the body is not: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + defer unzipped.Close() + body = unzipped + } + raw, err := io.ReadAll(body) if err != nil { t.Error(err) + w.WriteHeader(http.StatusBadRequest) return } var records []map[string]any - if err := json.Unmarshal(body, &records); err != nil { - t.Errorf("body is not a JSON array of records: %v\n%s", err, body) + if err := json.Unmarshal(raw, &records); err != nil { + t.Errorf("body is not a JSON array of records: %v\n%s", err, raw) + w.WriteHeader(http.StatusBadRequest) return } + respond(w, r, records) requests <- request{header: r.Header.Clone(), records: records} })) - t.Cleanup(func() { - server.Close() - // The handler sends through http.DefaultTransport; drop its idle connections so their - // goroutines are gone before goleak looks. - http.DefaultTransport.(*http.Transport).CloseIdleConnections() - }) + t.Cleanup(server.Close) return server, requests } @@ -62,6 +84,25 @@ func receive(t *testing.T, requests <-chan request) request { } } +// receiveN waits for n requests. +func receiveN(t *testing.T, requests <-chan request, n int) []request { + t.Helper() + var got []request + for len(got) < n { + got = append(got, receive(t, requests)) + } + return got +} + +func nothingWithin(t *testing.T, requests <-chan request, d time.Duration) { + t.Helper() + select { + case r := <-requests: + t.Fatalf("a request with %d records reached the server, want none", len(r.records)) + case <-time.After(d): + } +} + func oneRecord(t *testing.T, requests <-chan request) map[string]any { t.Helper() r := receive(t, requests) @@ -71,6 +112,14 @@ func oneRecord(t *testing.T, requests <-chan request) map[string]any { return r.records[0] } +func messages(records []map[string]any) []string { + var out []string + for _, r := range records { + out = append(out, r["message"].(string)) + } + return out +} + func extraOf(t *testing.T, record map[string]any) map[string]any { t.Helper() extra, ok := record["extra"].(map[string]any) @@ -80,15 +129,57 @@ func extraOf(t *testing.T, record map[string]any) map[string]any { return extra } -func newLogger(server *httptest.Server, option Option) *slog.Logger { +// errorList collects what the handler reports through OnError. +type errorList struct { + mu sync.Mutex + errs []error +} + +func (l *errorList) add(err error) { + l.mu.Lock() + defer l.mu.Unlock() + l.errs = append(l.errs, err) +} + +func (l *errorList) all() []error { + l.mu.Lock() + defer l.mu.Unlock() + return append([]error(nil), l.errs...) +} + +func (l *errorList) joined() string { + var parts []string + for _, err := range l.all() { + parts = append(parts, err.Error()) + } + return strings.Join(parts, "\n") +} + +// newHandler builds a handler that sends to server and is closed when the test ends. Batches wait +// 10ms instead of a second and retries back off for a millisecond, unless the test says otherwise. +func newHandler(t *testing.T, server *httptest.Server, option Option) *BetterstackHandler { + t.Helper() option.Token = "test-token" option.Endpoint = server.URL - return slog.New(option.NewBetterstackHandler()) + if option.BatchInterval == 0 { + option.BatchInterval = 10 * time.Millisecond + } + if option.RetryBackoff == 0 { + option.RetryBackoff = time.Millisecond + } + handler := option.NewBetterstackHandler() + t.Cleanup(func() { _ = handler.Close() }) + return handler +} + +func newLogger(t *testing.T, server *httptest.Server, option Option) *slog.Logger { + t.Helper() + return slog.New(newHandler(t, server, option)) } func TestRecordShape(t *testing.T) { - server, requests := newServer(t) - logger := newLogger(server, Option{}).With("release", "v1.0.0") + server, requests := newServer(t, accepted) + logger := newLogger(t, server, Option{}).With("release", "v1.0.0") before := time.Now() logger. @@ -157,16 +248,17 @@ func TestRecordShape(t *testing.T) { } func TestRequestHeaders(t *testing.T) { - server, requests := newServer(t) - logger := newLogger(server, Option{}) + server, requests := newServer(t, accepted) + logger := newLogger(t, server, Option{}) logger.Info("hello") got := receive(t, requests) for header, want := range map[string]string{ - "Authorization": "Bearer test-token", - "Content-Type": "application/json", - "User-Agent": "BetterStackHQ/slog-betterstack", + "Authorization": "Bearer test-token", + "Content-Type": "application/json", + "Content-Encoding": "gzip", + "User-Agent": "BetterStackHQ/slog-betterstack/" + version, } { if value := got.header.Get(header); value != want { t.Errorf("%s = %q, want %q", header, value, want) @@ -180,19 +272,54 @@ func TestRequestHeaders(t *testing.T) { } } -func TestDefaults(t *testing.T) { - handler, ok := Option{Token: "x"}.NewBetterstackHandler().(*BetterstackHandler) - if !ok { - t.Fatal("NewBetterstackHandler does not return a *BetterstackHandler") +func TestDisableCompression(t *testing.T) { + server, requests := newServer(t, accepted) + logger := newLogger(t, server, Option{DisableCompression: true}) + + logger.Info("plain") + + got := receive(t, requests) + if encoding := got.header.Get("Content-Encoding"); encoding != "" { + t.Errorf("Content-Encoding = %q, want none with DisableCompression", encoding) } - if handler.option.Endpoint != "https://in.logs.betterstack.com/" { - t.Errorf("Endpoint = %q, want the Better Stack ingesting endpoint", handler.option.Endpoint) + if want := []string{"plain"}; !reflect.DeepEqual(messages(got.records), want) { + t.Errorf("messages = %v, want %v", messages(got.records), want) } - if handler.option.Timeout != 10*time.Second { - t.Errorf("Timeout = %s, want 10s", handler.option.Timeout) +} + +func TestDefaults(t *testing.T) { + handler := Option{Token: "x"}.NewBetterstackHandler() + option := handler.option + for name, got := range map[string]any{ + "Endpoint": option.Endpoint, + "Timeout": option.Timeout, + "Level": option.Level.Level(), + "BatchSize": option.BatchSize, + "BatchInterval": option.BatchInterval, + "MaxQueueSize": option.MaxQueueSize, + "MaxRetries": option.MaxRetries, + "RetryBackoff": option.RetryBackoff, + "MaxInFlight": option.MaxInFlight, + "ShutdownTimeout": option.ShutdownTimeout, + } { + want := map[string]any{ + "Endpoint": "https://in.logs.betterstack.com/", + "Timeout": 10 * time.Second, + "Level": slog.LevelDebug, + "BatchSize": 1000, + "BatchInterval": time.Second, + "MaxQueueSize": 100_000, + "MaxRetries": 5, + "RetryBackoff": 300 * time.Millisecond, + "MaxInFlight": 5, + "ShutdownTimeout": 15 * time.Second, + }[name] + if !reflect.DeepEqual(got, want) { + t.Errorf("%s = %v, want %v", name, got, want) + } } - if handler.option.Level.Level() != slog.LevelDebug { - t.Errorf("Level = %s, want DEBUG", handler.option.Level.Level()) + if option.OnError == nil { + t.Error("OnError = nil, want the default reporter") } } @@ -222,8 +349,8 @@ func TestLevel(t *testing.T) { } func TestWithGroupNestsAttributes(t *testing.T) { - server, requests := newServer(t) - logger := newLogger(server, Option{}).WithGroup("request") + server, requests := newServer(t, accepted) + logger := newLogger(t, server, Option{}).WithGroup("request") logger.Info("handled", "id", "r-1", slog.Group("response", "status", 200)) @@ -238,9 +365,16 @@ func TestWithGroupNestsAttributes(t *testing.T) { } } +func TestWithGroupEmptyNameIsANoOp(t *testing.T) { + handler := Option{Token: "x"}.NewBetterstackHandler() + if handler.WithGroup("") != handler { + t.Error(`WithGroup("") returned a new handler, want the same one`) + } +} + func TestAddSource(t *testing.T) { - server, requests := newServer(t) - logger := newLogger(server, Option{AddSource: true}) + server, requests := newServer(t, accepted) + logger := newLogger(t, server, Option{AddSource: true}) logger.Info("where am I") @@ -260,8 +394,8 @@ func TestAddSource(t *testing.T) { } func TestReplaceAttr(t *testing.T) { - server, requests := newServer(t) - logger := newLogger(server, Option{ + server, requests := newServer(t, accepted) + logger := newLogger(t, server, Option{ ReplaceAttr: func(groups []string, a slog.Attr) slog.Attr { if a.Key == "password" { return slog.String("password", "[redacted]") @@ -280,8 +414,8 @@ func TestReplaceAttr(t *testing.T) { func TestAttrFromContext(t *testing.T) { type key struct{} - server, requests := newServer(t) - logger := newLogger(server, Option{ + server, requests := newServer(t, accepted) + logger := newLogger(t, server, Option{ AttrFromContext: []func(ctx context.Context) []slog.Attr{ func(ctx context.Context) []slog.Attr { return []slog.Attr{slog.String("request_id", ctx.Value(key{}).(string))} @@ -302,8 +436,8 @@ func TestContextKey(t *testing.T) { ContextKey = "context" t.Cleanup(func() { ContextKey = previous }) - server, requests := newServer(t) - logger := newLogger(server, Option{}) + server, requests := newServer(t, accepted) + logger := newLogger(t, server, Option{}) logger.Info("hello", "a", 1) @@ -318,9 +452,9 @@ func TestContextKey(t *testing.T) { } func TestConverterAndMarshaler(t *testing.T) { - server, requests := newServer(t) + server, requests := newServer(t, accepted) marshaled := make(chan any, 1) - logger := newLogger(server, Option{ + logger := newLogger(t, server, Option{ Converter: func(addSource bool, replaceAttr func(groups []string, a slog.Attr) slog.Attr, loggerAttr []slog.Attr, groups []string, record *slog.Record) map[string]any { return map[string]any{"custom": record.Message} }, @@ -345,35 +479,3 @@ func TestConverterAndMarshaler(t *testing.T) { t.Error("Marshaler was not called") } } - -func TestWithGroupEmptyNameIsANoOp(t *testing.T) { - handler := Option{Token: "x"}.NewBetterstackHandler() - if handler.WithGroup("") != handler { - t.Error(`WithGroup("") returned a new handler, want the same one`) - } -} - -func TestSendErrors(t *testing.T) { - payload := []map[string]any{{"message": "m"}} - - t.Run("marshaling", func(t *testing.T) { - failing := func(any) ([]byte, error) { return nil, errors.New("cannot marshal") } - if err := send("http://127.0.0.1:0/", "x", time.Second, failing, payload); err == nil || err.Error() != "cannot marshal" { - t.Errorf("err = %v, want the Marshaler's error", err) - } - }) - - t.Run("invalid endpoint", func(t *testing.T) { - if err := send("://not-a-url", "x", time.Second, json.Marshal, payload); err == nil { - t.Error("err = nil, want a request error for an invalid endpoint") - } - }) - - t.Run("unreachable endpoint", func(t *testing.T) { - server := httptest.NewServer(http.NotFoundHandler()) - server.Close() // nothing listens on this URL any more - if err := send(server.URL, "x", time.Second, json.Marshal, payload); err == nil { - t.Error("err = nil, want a connection error for a closed endpoint") - } - }) -} diff --git a/transport_test.go b/transport_test.go new file mode 100644 index 0000000..6696142 --- /dev/null +++ b/transport_test.go @@ -0,0 +1,480 @@ +package slogbetterstack + +import ( + "context" + "errors" + "fmt" + "log/slog" + "net/http" + "reflect" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// These tests pin how records travel: batching, flushing on Close, retries, what gets reported +// and what gets dropped. Every record handed to the handler ends up either acknowledged by the +// server or counted as dropped for a reason, and nothing here blocks the application. + +func TestCloseDeliversPendingRecords(t *testing.T) { + server, requests := newServer(t, accepted) + handler := newHandler(t, server, Option{BatchInterval: time.Hour}) + logger := slog.New(handler) + + for i := 0; i < 3; i++ { + logger.Info("pending", "i", i) + } + nothingWithin(t, requests, 50*time.Millisecond) // the batch waits for its interval or for Close + + if err := handler.Close(); err != nil { + t.Fatalf("Close = %v, want nil", err) + } + + got := receive(t, requests) + if want := []string{"pending", "pending", "pending"}; !reflect.DeepEqual(messages(got.records), want) { + t.Errorf("messages = %v, want %v in one request", messages(got.records), want) + } + if stats := handler.Stats(); stats.Enqueued != 3 || stats.Sent != 3 { + t.Errorf("stats = %+v, want Enqueued 3 and Sent 3", stats) + } +} + +func TestBatchSizeTriggersASend(t *testing.T) { + server, requests := newServer(t, accepted) + logger := newLogger(t, server, Option{BatchSize: 2, BatchInterval: time.Hour}) + + logger.Info("one") + logger.Info("two") + logger.Info("three") + + got := receive(t, requests) + if want := []string{"one", "two"}; !reflect.DeepEqual(messages(got.records), want) { + t.Errorf("messages = %v, want %v", messages(got.records), want) + } + nothingWithin(t, requests, 50*time.Millisecond) // "three" waits for the next full batch, the interval or Close +} + +func TestBatchIntervalTriggersASend(t *testing.T) { + server, requests := newServer(t, accepted) + logger := newLogger(t, server, Option{BatchInterval: 20 * time.Millisecond}) + + logger.Info("alone") + + got := receive(t, requests) // without Flush or Close + if want := []string{"alone"}; !reflect.DeepEqual(messages(got.records), want) { + t.Errorf("messages = %v, want %v", messages(got.records), want) + } +} + +func TestFlushWaitsForDelivery(t *testing.T) { + server, requests := newServer(t, accepted) + handler := newHandler(t, server, Option{BatchInterval: time.Hour}) + logger := slog.New(handler) + + logger.Info("a") + logger.Info("b") + if err := handler.Flush(context.Background()); err != nil { + t.Fatalf("Flush = %v, want nil", err) + } + + select { + case got := <-requests: + if want := []string{"a", "b"}; !reflect.DeepEqual(messages(got.records), want) { + t.Errorf("messages = %v, want %v", messages(got.records), want) + } + default: + t.Fatal("Flush returned before the records reached the server") + } +} + +func TestDerivedHandlersShareOneQueue(t *testing.T) { + server, requests := newServer(t, accepted) + handler := newHandler(t, server, Option{BatchInterval: time.Hour}) + base := slog.New(handler) + + base.Info("base") + base.With("x", 1).Info("with") + base.WithGroup("g").Info("group", "k", "v") + if err := handler.Close(); err != nil { + t.Fatal(err) + } + + got := receive(t, requests) + if want := []string{"base", "with", "group"}; !reflect.DeepEqual(messages(got.records), want) { + t.Errorf("messages = %v, want %v in one request", messages(got.records), want) + } +} + +func TestRejectedBatchIsReportedAndNotRetried(t *testing.T) { + errs := &errorList{} + server, requests := newServer(t, status(http.StatusUnauthorized)) + handler := newHandler(t, server, Option{OnError: errs.add}) + + slog.New(handler).Info("secret") + if err := handler.Close(); err != nil { + t.Fatal(err) + } + + receive(t, requests) + nothingWithin(t, requests, 50*time.Millisecond) // a bad token is not retried + if got := errs.joined(); !strings.Contains(got, "401") { + t.Errorf("OnError got %q, want the 401 status", got) + } + if stats := handler.Stats(); stats.DroppedRejected != 1 || stats.Sent != 0 { + t.Errorf("stats = %+v, want DroppedRejected 1 and Sent 0", stats) + } +} + +func TestTransientFailuresAreRetried(t *testing.T) { + var calls atomic.Int32 + server, requests := newServer(t, func(w http.ResponseWriter, _ *http.Request, _ []map[string]any) { + if calls.Add(1) <= 2 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusAccepted) + }) + errs := &errorList{} + handler := newHandler(t, server, Option{OnError: errs.add}) + + slog.New(handler).Info("eventually") + if err := handler.Close(); err != nil { + t.Fatal(err) + } + + got := receiveN(t, requests, 3) + if want := []string{"eventually"}; !reflect.DeepEqual(messages(got[2].records), want) { + t.Errorf("third request carried %v, want %v", messages(got[2].records), want) + } + if stats := handler.Stats(); stats.Sent != 1 || stats.Retries != 2 { + t.Errorf("stats = %+v, want Sent 1 and Retries 2", stats) + } + if got := errs.all(); len(got) != 0 { + t.Errorf("OnError got %v, want nothing for retries that succeed", got) + } +} + +func TestRetryAfterIsHonoured(t *testing.T) { + var mu sync.Mutex + var times []time.Time + server, _ := newServer(t, func(w http.ResponseWriter, _ *http.Request, _ []map[string]any) { + mu.Lock() + times = append(times, time.Now()) + first := len(times) == 1 + mu.Unlock() + if first { + w.Header().Set("Retry-After", "1") + w.WriteHeader(http.StatusTooManyRequests) + return + } + w.WriteHeader(http.StatusAccepted) + }) + handler := newHandler(t, server, Option{RetryBackoff: time.Millisecond}) + + slog.New(handler).Info("throttled") + if err := handler.Close(); err != nil { + t.Fatal(err) + } + + mu.Lock() + defer mu.Unlock() + if len(times) != 2 { + t.Fatalf("got %d requests, want 2", len(times)) + } + if gap := times[1].Sub(times[0]); gap < 900*time.Millisecond { + t.Errorf("retried after %s, want at least the Retry-After second", gap) + } +} + +func TestRetriesRunOut(t *testing.T) { + errs := &errorList{} + server, requests := newServer(t, status(http.StatusServiceUnavailable)) + handler := newHandler(t, server, Option{MaxRetries: 2, OnError: errs.add}) + + slog.New(handler).Info("doomed") + if err := handler.Close(); err != nil { + t.Fatal(err) + } + + receiveN(t, requests, 3) // the first attempt and two retries + nothingWithin(t, requests, 50*time.Millisecond) + if got := errs.joined(); !strings.Contains(got, "3 attempts") || !strings.Contains(got, "503") { + t.Errorf("OnError got %q, want the attempt count and the last status", got) + } + if stats := handler.Stats(); stats.DroppedRejected != 1 || stats.Retries != 2 { + t.Errorf("stats = %+v, want DroppedRejected 1 and Retries 2", stats) + } +} + +func TestOversizedBatchIsSplit(t *testing.T) { + server, requests := newServer(t, func(w http.ResponseWriter, _ *http.Request, records []map[string]any) { + if len(records) > 1 { + w.WriteHeader(http.StatusRequestEntityTooLarge) + return + } + w.WriteHeader(http.StatusAccepted) + }) + errs := &errorList{} + handler := newHandler(t, server, Option{BatchInterval: time.Hour, OnError: errs.add}) + logger := slog.New(handler) + + for i := 0; i < 4; i++ { + logger.Info(fmt.Sprintf("record %d", i)) + } + if err := handler.Close(); err != nil { + t.Fatal(err) + } + + var delivered []string + for _, r := range receiveN(t, requests, 7) { // 4, then 2+2, then 1+1+1+1 + if len(r.records) == 1 { + delivered = append(delivered, messages(r.records)...) + } + } + if want := []string{"record 0", "record 1", "record 2", "record 3"}; !reflect.DeepEqual(delivered, want) { + t.Errorf("delivered = %v, want %v", delivered, want) + } + if stats := handler.Stats(); stats.Sent != 4 || stats.DroppedOversize != 0 { + t.Errorf("stats = %+v, want Sent 4 and nothing dropped", stats) + } + if got := errs.all(); len(got) != 0 { + t.Errorf("OnError got %v, want nothing for a batch that was split and delivered", got) + } +} + +func TestSingleOversizedRecordIsDropped(t *testing.T) { + errs := &errorList{} + server, requests := newServer(t, status(http.StatusRequestEntityTooLarge)) + handler := newHandler(t, server, Option{OnError: errs.add}) + + slog.New(handler).Info("huge") + if err := handler.Close(); err != nil { + t.Fatal(err) + } + + receive(t, requests) + nothingWithin(t, requests, 50*time.Millisecond) // nothing to split, nothing to retry + if got := errs.joined(); !strings.Contains(got, "larger") { + t.Errorf("OnError got %q, want a message about the record being too large", got) + } + if stats := handler.Stats(); stats.DroppedOversize != 1 { + t.Errorf("stats = %+v, want DroppedOversize 1", stats) + } +} + +func TestFullQueueDropsInsteadOfBlocking(t *testing.T) { + gate := make(chan struct{}) + server, _ := newServer(t, func(w http.ResponseWriter, _ *http.Request, _ []map[string]any) { + <-gate + w.WriteHeader(http.StatusAccepted) + }) + errs := &errorList{} + handler := newHandler(t, server, Option{ + BatchSize: 1, + BatchInterval: time.Hour, + MaxQueueSize: 2, + MaxInFlight: 1, + OnError: errs.add, + }) + logger := slog.New(handler) + + start := time.Now() + for i := 0; i < 10; i++ { + logger.Info("burst", "i", i) + } + if elapsed := time.Since(start); elapsed > 500*time.Millisecond { + t.Fatalf("10 log calls took %s while delivery was stalled, want them not to block", elapsed) + } + + close(gate) + if err := handler.Close(); err != nil { + t.Fatal(err) + } + + stats := handler.Stats() + if stats.DroppedQueueFull == 0 { + t.Error("nothing was dropped although the queue holds 2 records and delivery was stalled") + } + if stats.Sent+stats.DroppedQueueFull != 10 { + t.Errorf("stats = %+v, want Sent + DroppedQueueFull = 10", stats) + } + if got := errs.joined(); !strings.Contains(got, "queue is full") { + t.Errorf("OnError got %q, want a summary of the queue-full drops", got) + } +} + +func TestConcurrentHandlesDoNotRace(t *testing.T) { + type key struct{} + server, _ := newServer(t, accepted) + handler := newHandler(t, server, Option{ + BatchInterval: time.Hour, + AttrFromContext: []func(ctx context.Context) []slog.Attr{ + func(ctx context.Context) []slog.Attr { + return []slog.Attr{slog.String("worker", ctx.Value(key{}).(string))} + }, + }, + }) + // Repeated keys leave spare capacity in the attribute slice the derived handler shares with + // every goroutine that logs through it. + logger := slog.New(handler).With("a", 1).With("b", 2).With("a", 3) + + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + ctx := context.WithValue(context.Background(), key{}, fmt.Sprint(i)) + for j := 0; j < 200; j++ { + logger.InfoContext(ctx, "concurrent") + } + }(i) + } + wg.Wait() + if err := handler.Close(); err != nil { + t.Fatal(err) + } + if stats := handler.Stats(); stats.Sent != 1600 { + t.Errorf("stats = %+v, want Sent 1600", stats) + } +} + +func TestCloseIsFinal(t *testing.T) { + server, requests := newServer(t, accepted) + handler := newHandler(t, server, Option{BatchInterval: time.Hour}) + ctx := context.Background() + + slog.New(handler).Info("before") + if err := handler.Close(); err != nil { + t.Fatalf("Close = %v, want nil", err) + } + if err := handler.Close(); err != nil { + t.Errorf("second Close = %v, want nil", err) + } + + if err := handler.Handle(ctx, slog.NewRecord(time.Now(), slog.LevelInfo, "after", 0)); !errors.Is(err, ErrClosed) { + t.Errorf("Handle after Close = %v, want ErrClosed", err) + } + if err := handler.Flush(ctx); !errors.Is(err, ErrClosed) { + t.Errorf("Flush after Close = %v, want ErrClosed", err) + } + + if want := []string{"before"}; !reflect.DeepEqual(messages(receive(t, requests).records), want) { + t.Errorf("delivered %v, want %v", messages(receive(t, requests).records), want) + } + nothingWithin(t, requests, 50*time.Millisecond) + if stats := handler.Stats(); stats.Sent != 1 || stats.DroppedClosed != 1 { + t.Errorf("stats = %+v, want Sent 1 and DroppedClosed 1", stats) + } +} + +func TestShutdownTimeoutBoundsClose(t *testing.T) { + gate := make(chan struct{}) + server, _ := newServer(t, func(w http.ResponseWriter, r *http.Request, _ []map[string]any) { + select { + case <-gate: + case <-r.Context().Done(): + } + w.WriteHeader(http.StatusAccepted) + }) + t.Cleanup(func() { close(gate) }) + errs := &errorList{} + handler := newHandler(t, server, Option{ShutdownTimeout: 100 * time.Millisecond, OnError: errs.add}) + + slog.New(handler).Info("stuck") + start := time.Now() + err := handler.Close() + if err == nil || !strings.Contains(err.Error(), "shutdown timeout") { + t.Errorf("Close = %v, want an error naming the shutdown timeout", err) + } + if elapsed := time.Since(start); elapsed > 2*time.Second { + t.Errorf("Close took %s, want about the 100ms shutdown timeout", elapsed) + } + if stats := handler.Stats(); stats.DroppedClosed != 1 || stats.Sent != 0 { + t.Errorf("stats = %+v, want DroppedClosed 1 and Sent 0", stats) + } +} + +func TestTimeoutIsHonoured(t *testing.T) { + server, _ := newServer(t, func(w http.ResponseWriter, r *http.Request, _ []map[string]any) { + select { + case <-time.After(2 * time.Second): + case <-r.Context().Done(): + } + w.WriteHeader(http.StatusAccepted) + }) + errs := &errorList{} + handler := newHandler(t, server, Option{Timeout: 50 * time.Millisecond, MaxRetries: -1, OnError: errs.add}) + + slog.New(handler).Info("slow") + start := time.Now() + if err := handler.Close(); err != nil { + t.Fatal(err) + } + if elapsed := time.Since(start); elapsed > time.Second { + t.Errorf("Close took %s, want the 50ms Timeout to cut the request short", elapsed) + } + if got := errs.joined(); !strings.Contains(strings.ToLower(got), "timeout") && !strings.Contains(got, "deadline") { + t.Errorf("OnError got %q, want a timeout error", got) + } + if stats := handler.Stats(); stats.DroppedRejected != 1 || stats.Retries != 0 { + t.Errorf("stats = %+v, want DroppedRejected 1 and no retries with MaxRetries -1", stats) + } +} + +func TestUnreachableEndpointIsReported(t *testing.T) { + server, _ := newServer(t, accepted) + server.Close() // nothing listens on this URL any more + errs := &errorList{} + handler := newHandler(t, server, Option{MaxRetries: 1, OnError: errs.add}) + + slog.New(handler).Info("nowhere") + if err := handler.Close(); err != nil { + t.Fatal(err) + } + + if got := errs.joined(); !strings.Contains(got, "2 attempts") { + t.Errorf("OnError got %q, want the connection error after 2 attempts", got) + } + if stats := handler.Stats(); stats.DroppedRejected != 1 || stats.Retries != 1 { + t.Errorf("stats = %+v, want DroppedRejected 1 and Retries 1", stats) + } +} + +func TestEncodingFailureIsReported(t *testing.T) { + server, requests := newServer(t, accepted) + errs := &errorList{} + handler := newHandler(t, server, Option{ + Marshaler: func(any) ([]byte, error) { return nil, errors.New("cannot marshal") }, + OnError: errs.add, + }) + + slog.New(handler).Info("unencodable") + if err := handler.Close(); err != nil { + t.Fatal(err) + } + + nothingWithin(t, requests, 50*time.Millisecond) + if got := errs.joined(); !strings.Contains(got, "cannot marshal") { + t.Errorf("OnError got %q, want the Marshaler's error", got) + } + if stats := handler.Stats(); stats.DroppedRejected != 1 { + t.Errorf("stats = %+v, want DroppedRejected 1", stats) + } +} + +func TestOnErrorPanicsAreContained(t *testing.T) { + server, requests := newServer(t, status(http.StatusUnauthorized)) + handler := newHandler(t, server, Option{OnError: func(error) { panic("reporter bug") }}) + + slog.New(handler).Info("secret") + if err := handler.Close(); err != nil { + t.Fatal(err) + } + + receive(t, requests) + if stats := handler.Stats(); stats.DroppedRejected != 1 { + t.Errorf("stats = %+v, want DroppedRejected 1 with the reporter's panic contained", stats) + } +} From 06053ac7bb74989b1a11ebbf3069d649630414c4 Mon Sep 17 00:00:00 2001 From: Petr Heinz Date: Tue, 6 Oct 2026 15:38:42 +0200 Subject: [PATCH 02/11] T-20356 Batch, flush on Close, retry and report delivery failures Records now go through a bounded queue to one sender goroutine that uploads them in gzip-compressed batches of 1000 or after a second, with up to five uploads in flight. 408, 429, 5xx and network errors are retried with exponential backoff and jitter, honouring Retry-After; other 4xx are terminal; a 413 batch is split. Every failure and every drop is reported through OnError, one line on stderr by default, and counted in Stats. NewBetterstackHandler returns the concrete *BetterstackHandler so that Close and Flush can be called; it still satisfies slog.Handler, so slog.New callers do not change. The API, the record shape (extra, logger.name, logger.version) and the Debug default stay as they were, and a custom Marshaler still receives the []map[string]any of a batch. Also: Timeout bounds each attempt instead of being capped by a hard-coded client timeout, Handle no longer appends into the attribute slice that derived handlers share (a data race under concurrent logging), the User-Agent carries the version, and the example closes the handler instead of sleeping. Co-Authored-By: Claude Fable 5.1 --- doc.go | 19 ++ example-project/main.go | 13 +- handler.go | 162 +++++++----- transport.go | 538 ++++++++++++++++++++++++++++++++++++++++ version.go | 4 + 5 files changed, 676 insertions(+), 60 deletions(-) create mode 100644 doc.go create mode 100644 transport.go diff --git a/doc.go b/doc.go new file mode 100644 index 0000000..bd5fb8a --- /dev/null +++ b/doc.go @@ -0,0 +1,19 @@ +// Package slogbetterstack sends the logs of a Go application to Better Stack through a +// [log/slog] handler. +// +// handler := slogbetterstack.Option{ +// Token: "$SOURCE_TOKEN", +// Endpoint: "https://$INGESTING_HOST/", +// }.NewBetterstackHandler() +// defer handler.Close() +// +// logger := slog.New(handler) +// logger.Info("Hello from Better Stack!", "service", "UserService") +// +// Records are queued and uploaded in batches by a background goroutine, so logging never +// waits for the network. Close delivers what is still queued and must run before the program +// exits; os.Exit and log.Fatal skip deferred calls. Delivery failures are reported through +// [Option.OnError], on stderr by default, and counted in [Stats]. +// +// See https://betterstack.com/docs/logs/go/ for the full documentation. +package slogbetterstack diff --git a/example-project/main.go b/example-project/main.go index ee75632..24875c7 100644 --- a/example-project/main.go +++ b/example-project/main.go @@ -27,7 +27,8 @@ func main() { option.Endpoint = "https://" + host + "/" } - logger := slog.New(option.NewBetterstackHandler()) + handler := option.NewBetterstackHandler() + logger := slog.New(handler) logger = logger.With("release", "v1.0.0") logger.Debug("Debugging user service.", "service", "UserService") @@ -44,8 +45,12 @@ func main() { With("error", fmt.Errorf("an error")). Error("a message", slog.Int("count", 1)) - // Logs are sent asynchronously: give the handler a moment before the process exits. - time.Sleep(5 * time.Second) + // Records are sent in batches: Close delivers what is still queued before the program exits. + if err := handler.Close(); err != nil { + fmt.Fprintln(os.Stderr, "close:", err) + os.Exit(1) + } - fmt.Println("Sent 3 log records. Open Better Stack → Live tail to see them.") + stats := handler.Stats() + fmt.Printf("Sent %d log records. Open Better Stack → Live tail to see them.\n", stats.Sent) } diff --git a/handler.go b/handler.go index d82ed82..7ca73de 100644 --- a/handler.go +++ b/handler.go @@ -1,10 +1,10 @@ package slogbetterstack import ( - "bytes" "context" "encoding/json" "net/http" + "slices" "time" "log/slog" @@ -22,12 +22,12 @@ type Option struct { Token string // optional: endpoint Endpoint string - // default: 10s + // optional: how long one upload attempt may take (default: 10s) Timeout time.Duration // optional: customize record builder Converter Converter - // optional: custom marshaler + // optional: custom marshaler, called with the []map[string]any of a batch's records Marshaler func(v any) ([]byte, error) // optional: fetch attributes from context AttrFromContext []func(ctx context.Context) []slog.Attr @@ -35,9 +35,40 @@ type Option struct { // optional: see slog.HandlerOptions AddSource bool ReplaceAttr func(groups []string, a slog.Attr) slog.Attr + + // Delivery. Records are queued and uploaded in batches by a background goroutine, so + // logging never waits for the network. Call Close before the program exits to deliver + // what is still queued. + + // optional: records per upload (default: 1000) + BatchSize int + // optional: how long a partial batch waits before it is uploaded (default: 1s) + BatchInterval time.Duration + // optional: records the queue holds while uploads are behind; further records are + // dropped and counted rather than blocking the application (default: 100000) + MaxQueueSize int + // optional: concurrent uploads (default: 5) + MaxInFlight int + // optional: retries after a failed attempt, for 408, 429, 5xx and network errors; + // negative disables retries (default: 5) + MaxRetries int + // optional: base delay before a retry, doubled on every attempt with jitter; a + // Retry-After header is honoured instead (default: 300ms) + RetryBackoff time.Duration + // optional: how long Close waits for queued and in-flight records (default: 15s) + ShutdownTimeout time.Duration + // optional: send the JSON uncompressed instead of gzip-compressed + DisableCompression bool + // optional: receives every delivery failure and drop summary, from a background + // goroutine; it must not log through this handler (default: one line on stderr) + OnError func(err error) + // optional: the HTTP client to upload with; Timeout still applies to every request + HTTPClient *http.Client } -func (o Option) NewBetterstackHandler() slog.Handler { +// NewBetterstackHandler returns a handler that sends records to Better Stack. The handler is +// also a [slog.Handler]; keep the returned value to call Close before the program exits. +func (o Option) NewBetterstackHandler() *BetterstackHandler { if o.Level == nil { o.Level = slog.LevelDebug } @@ -51,7 +82,7 @@ func (o Option) NewBetterstackHandler() slog.Handler { } if o.Timeout == 0 { - o.Timeout = 10 * time.Second + o.Timeout = defaultTimeout } if o.Converter == nil { @@ -66,43 +97,76 @@ func (o Option) NewBetterstackHandler() slog.Handler { o.AttrFromContext = []func(ctx context.Context) []slog.Attr{} } + if o.BatchSize <= 0 { + o.BatchSize = defaultBatchSize + } + if o.BatchInterval <= 0 { + o.BatchInterval = defaultBatchInterval + } + if o.MaxQueueSize <= 0 { + o.MaxQueueSize = defaultMaxQueueSize + } + if o.MaxInFlight <= 0 { + o.MaxInFlight = defaultMaxInFlight + } + switch { + case o.MaxRetries == 0: + o.MaxRetries = defaultMaxRetries + case o.MaxRetries < 0: + o.MaxRetries = 0 + } + if o.RetryBackoff <= 0 { + o.RetryBackoff = defaultRetryBackoff + } + if o.ShutdownTimeout <= 0 { + o.ShutdownTimeout = defaultShutdownTimeout + } + if o.OnError == nil { + o.OnError = defaultOnError + } + return &BetterstackHandler{ - option: o, - attrs: []slog.Attr{}, - groups: []string{}, + option: o, + attrs: []slog.Attr{}, + groups: []string{}, + transport: newTransport(o), } } var _ slog.Handler = (*BetterstackHandler)(nil) +// BetterstackHandler is a [slog.Handler] that sends records to Better Stack. Handlers derived +// with WithAttrs and WithGroup share the queue and the uploads of the handler they came from, +// and Close on any of them closes all of them. type BetterstackHandler struct { - option Option - attrs []slog.Attr - groups []string + option Option + attrs []slog.Attr + groups []string + transport *transport } func (h *BetterstackHandler) Enabled(_ context.Context, level slog.Level) bool { return level >= h.option.Level.Level() } +// Handle converts the record and queues it for upload. It never waits for the network: when +// the queue is full the record is dropped and counted. After Close it returns ErrClosed. func (h *BetterstackHandler) Handle(ctx context.Context, record slog.Record) error { fromContext := slogcommon.ContextExtractor(ctx, h.option.AttrFromContext) - payload := h.option.Converter(h.option.AddSource, h.option.ReplaceAttr, append(h.attrs, fromContext...), h.groups, &record) - - // non-blocking - go func() { - // @TODO: batching ? - _ = send(h.option.Endpoint, h.option.Token, h.option.Timeout, h.option.Marshaler, []map[string]any{payload}) - }() + // Every goroutine logging through this handler shares h.attrs, so appending must never + // write into its spare capacity. + attrs := append(slices.Clip(h.attrs), fromContext...) + payload := h.option.Converter(h.option.AddSource, h.option.ReplaceAttr, attrs, h.groups, &record) - return nil + return h.transport.enqueue(payload) } func (h *BetterstackHandler) WithAttrs(attrs []slog.Attr) slog.Handler { return &BetterstackHandler{ - option: h.option, - attrs: slogcommon.AppendAttrsToGroup(h.groups, h.attrs, attrs...), - groups: h.groups, + option: h.option, + attrs: slogcommon.AppendAttrsToGroup(h.groups, h.attrs, attrs...), + groups: h.groups, + transport: h.transport, } } @@ -113,43 +177,29 @@ func (h *BetterstackHandler) WithGroup(name string) slog.Handler { } return &BetterstackHandler{ - option: h.option, - attrs: h.attrs, - groups: append(h.groups, name), + option: h.option, + attrs: h.attrs, + groups: append(h.groups, name), + transport: h.transport, } } -func send(endpoint string, token string, timeout time.Duration, marshaler func(v any) ([]byte, error), payload []map[string]any) error { - client := http.Client{ - Timeout: time.Duration(10) * time.Second, - } - - json, err := marshaler(payload) - if err != nil { - return err - } - - body := bytes.NewBuffer(json) - - ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() - - // @TODO: maintain a pool of tcp connections - req, err := http.NewRequestWithContext(ctx, "POST", endpoint, body) - if err != nil { - return err - } - - req.Header.Add("authorization", `Bearer `+token) - req.Header.Add("content-type", `application/json`) - req.Header.Add("user-agent", name) - - resp, err := client.Do(req) - if err != nil { - return err - } +// Flush uploads every record queued so far and returns once Better Stack has acknowledged +// them, a delivery failed for good, or ctx is done. Failures are reported through OnError. +func (h *BetterstackHandler) Flush(ctx context.Context) error { + return h.transport.flush(ctx) +} - defer resp.Body.Close() //nolint:errcheck +// Close delivers what is still queued, waits for the uploads in flight up to ShutdownTimeout +// and stops the background goroutine. It must run before the program exits: records are +// batched, so without it the last ones are lost. Note that os.Exit and log.Fatal skip deferred +// calls. Close is safe to call more than once; later calls return the first result. +func (h *BetterstackHandler) Close() error { + return h.transport.close() +} - return nil +// Stats reports what happened to the records handed to this handler and the ones derived +// from it. +func (h *BetterstackHandler) Stats() Stats { + return h.transport.stats.snapshot() } diff --git a/transport.go b/transport.go new file mode 100644 index 0000000..a11dbf8 --- /dev/null +++ b/transport.go @@ -0,0 +1,538 @@ +package slogbetterstack + +import ( + "bytes" + "compress/gzip" + "context" + "errors" + "fmt" + "io" + "math/rand" + "net/http" + "os" + "strconv" + "sync" + "sync/atomic" + "time" +) + +// ErrClosed is returned by Handle and Flush once the handler has been closed. +var ErrClosed = errors.New("slog-betterstack: the handler is closed") + +const ( + defaultBatchSize = 1000 + defaultBatchInterval = time.Second + defaultMaxQueueSize = 100_000 + defaultMaxInFlight = 5 + defaultMaxRetries = 5 + defaultRetryBackoff = 300 * time.Millisecond + defaultTimeout = 10 * time.Second + defaultShutdownTimeout = 15 * time.Second + + maxBackoff = 30 * time.Second + maxRetryAfter = time.Minute + dropReportInterval = 5 * time.Second +) + +// Stats counts what happened to the records handed to a handler. Once Close has returned, +// Enqueued equals Sent plus all the Dropped counters. +type Stats struct { + Enqueued uint64 // records handed to Handle + Sent uint64 // records acknowledged by Better Stack + Retries uint64 // upload attempts after the first + + DroppedQueueFull uint64 // the application logged faster than records could be delivered + DroppedRejected uint64 // Better Stack refused them, the retries ran out, or they could not be encoded + DroppedOversize uint64 // a single record larger than Better Stack accepts + DroppedClosed uint64 // logged after Close, or not delivered before the shutdown timeout +} + +type counters struct { + enqueued, sent, retries atomic.Uint64 + droppedQueueFull, droppedRejected, droppedOversize, droppedClosed atomic.Uint64 +} + +func (c *counters) snapshot() Stats { + return Stats{ + Enqueued: c.enqueued.Load(), + Sent: c.sent.Load(), + Retries: c.retries.Load(), + DroppedQueueFull: c.droppedQueueFull.Load(), + DroppedRejected: c.droppedRejected.Load(), + DroppedOversize: c.droppedOversize.Load(), + DroppedClosed: c.droppedClosed.Load(), + } +} + +func defaultOnError(err error) { + fmt.Fprintln(os.Stderr, err) +} + +// transport moves records from Handle to Better Stack: a bounded queue, one sender goroutine +// that assembles batches, and up to MaxInFlight upload goroutines that retry. Handlers derived +// from one another share a transport. +type transport struct { + option Option + client *http.Client + owned *http.Transport // nil when the caller supplied the HTTP client + + queue chan map[string]any + flushes chan flushRequest // unbuffered: a request is only ever taken by a running sender + closing chan struct{} // closed by close: the sender drains, uploads and exits + exited chan struct{} // closed by the sender on its way out + slots chan struct{} // one token per upload in flight + + ctx context.Context // cancelled when the shutdown timeout runs out + cancel context.CancelFunc + + mu sync.Mutex // guards started, closed and the queue send in enqueue + started bool + closed bool + + closeOnce sync.Once + closeErr error + inFlight sync.WaitGroup + + stats counters + reportedQueueFull uint64 // queue-full drops already summarised through OnError +} + +type flushRequest struct { + done chan struct{} // closed once every upload dispatched before the request has finished +} + +func newTransport(option Option) *transport { + ctx, cancel := context.WithCancel(context.Background()) + t := &transport{ + option: option, + queue: make(chan map[string]any, option.MaxQueueSize), + flushes: make(chan flushRequest), + closing: make(chan struct{}), + exited: make(chan struct{}), + slots: make(chan struct{}, option.MaxInFlight), + ctx: ctx, + cancel: cancel, + } + if option.HTTPClient != nil { + t.client = option.HTTPClient + } else { + t.owned = http.DefaultTransport.(*http.Transport).Clone() + t.client = &http.Client{Transport: t.owned} + } + return t +} + +// enqueue hands a record to the sender without ever blocking: when the queue is full the +// record is dropped and counted. The sender starts with the first record. +func (t *transport) enqueue(record map[string]any) error { + t.stats.enqueued.Add(1) + + t.mu.Lock() + defer t.mu.Unlock() + if t.closed { + t.stats.droppedClosed.Add(1) + return ErrClosed + } + if !t.started { + t.started = true + go t.run() + } + select { + case t.queue <- record: + default: + t.stats.droppedQueueFull.Add(1) + } + return nil +} + +// run is the sender: it takes records off the queue, uploads a batch when it is full or has +// waited BatchInterval, and serves Flush and Close. +func (t *transport) run() { + defer close(t.exited) + + var ( + batch []map[string]any + pending []chan struct{} // uploads dispatched since the last Flush + timer = time.NewTimer(time.Hour) + timerC <-chan time.Time // nil while no partial batch is waiting + ) + timer.Stop() + reports := time.NewTicker(dropReportInterval) + defer reports.Stop() + + dispatch := func() { + if timerC != nil { + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + timerC = nil + } + if len(batch) == 0 { + return + } + records := batch + batch = nil + pending = append(settled(pending), t.dispatch(records)) + } + add := func(record map[string]any) { + batch = append(batch, record) + if len(batch) == 1 { + timer.Reset(t.option.BatchInterval) + timerC = timer.C + } + if len(batch) >= t.option.BatchSize { + dispatch() + } + } + drain := func() { + for { + select { + case record := <-t.queue: + add(record) + default: + return + } + } + } + + for { + select { + case record := <-t.queue: + add(record) + case <-timerC: + timerC = nil + dispatch() + case request := <-t.flushes: + drain() + dispatch() + uploads := pending + pending = nil + go func() { + for _, done := range uploads { + <-done + } + close(request.done) + }() + case <-reports.C: + t.reportQueueFullDrops() + case <-t.closing: + drain() + dispatch() + return + } + } +} + +// settled drops the uploads that have finished, so pending only ever holds what is in flight. +func settled(uploads []chan struct{}) []chan struct{} { + kept := uploads[:0] + for _, done := range uploads { + select { + case <-done: + default: + kept = append(kept, done) + } + } + return kept +} + +// dispatch starts the upload of a batch once an upload slot is free, which is the only point +// where the sender waits: while every slot is taken the queue fills, and then enqueue drops. +func (t *transport) dispatch(records []map[string]any) chan struct{} { + done := make(chan struct{}) + t.slots <- struct{}{} + t.inFlight.Add(1) + go func() { + defer func() { + <-t.slots + t.inFlight.Done() + close(done) + }() + t.upload(records) + }() + return done +} + +// upload sends a batch, retrying what is worth retrying, splitting what is too large and +// reporting what cannot be delivered. +func (t *transport) upload(records []map[string]any) { + body, err := t.option.Marshaler(records) + if err != nil { + t.drop(&t.stats.droppedRejected, len(records), fmt.Errorf("slog-betterstack: dropped %s that could not be encoded: %w", plural(len(records), "record"), err)) + return + } + if !t.option.DisableCompression { + if body, err = compress(body); err != nil { + t.drop(&t.stats.droppedRejected, len(records), fmt.Errorf("slog-betterstack: dropped %s that could not be compressed: %w", plural(len(records), "record"), err)) + return + } + } + + for attempt := 0; ; attempt++ { + if t.ctx.Err() != nil { + t.drop(&t.stats.droppedClosed, len(records), nil) + return + } + + code, status, header, err := t.post(body) + var retryAfter time.Duration + if err == nil { + switch { + case code >= 200 && code < 300: + t.stats.sent.Add(uint64(len(records))) + return + case code == http.StatusRequestEntityTooLarge: + if len(records) > 1 { + half := len(records) / 2 + t.upload(records[:half]) + t.upload(records[half:]) + return + } + t.drop(&t.stats.droppedOversize, 1, fmt.Errorf("slog-betterstack: dropped a record that is larger than Better Stack accepts (%s)", status)) + return + case !retryable(code): + t.drop(&t.stats.droppedRejected, len(records), fmt.Errorf("slog-betterstack: Better Stack rejected %s: %s", plural(len(records), "record"), status)) + return + } + err = fmt.Errorf("Better Stack answered %s", status) + retryAfter = parseRetryAfter(header.Get("Retry-After")) + } + + if t.ctx.Err() != nil { + t.drop(&t.stats.droppedClosed, len(records), nil) + return + } + if attempt >= t.option.MaxRetries { + t.drop(&t.stats.droppedRejected, len(records), fmt.Errorf("slog-betterstack: dropped %s after %s: %w", plural(len(records), "record"), plural(attempt+1, "attempt"), err)) + return + } + t.stats.retries.Add(1) + if !t.wait(backoff(t.option.RetryBackoff, attempt, retryAfter)) { + t.drop(&t.stats.droppedClosed, len(records), nil) + return + } + } +} + +// post makes one upload attempt. It returns the status code and line and the response headers, +// or the error of an attempt that got no response. +func (t *transport) post(body []byte) (int, string, http.Header, error) { + ctx, cancel := context.WithTimeout(t.ctx, t.option.Timeout) + defer cancel() + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, t.option.Endpoint, bytes.NewReader(body)) + if err != nil { + return 0, "", nil, err + } + req.Header.Set("Authorization", "Bearer "+t.option.Token) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("User-Agent", userAgent) + if !t.option.DisableCompression { + req.Header.Set("Content-Encoding", "gzip") + } + + resp, err := t.client.Do(req) + if err != nil { + return 0, "", nil, err + } + defer func() { _ = resp.Body.Close() }() + // Reading the body to its end lets the connection be reused for the next upload. + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 64<<10)) + + return resp.StatusCode, resp.Status, resp.Header, nil +} + +// retryable reports whether a status is worth another attempt: a timeout, a throttle or a +// server-side failure. Anything else 4xx is a verdict on the request itself. +func retryable(code int) bool { + return code == http.StatusRequestTimeout || code == http.StatusTooManyRequests || code >= 500 +} + +// parseRetryAfter reads a Retry-After header given as seconds or as an HTTP date, capped so +// that a throttle cannot park an upload for longer than a minute. +func parseRetryAfter(value string) time.Duration { + if value == "" { + return 0 + } + var wait time.Duration + if seconds, err := strconv.Atoi(value); err == nil { + wait = time.Duration(seconds) * time.Second + } else if at, err := http.ParseTime(value); err == nil { + wait = time.Until(at) + } + if wait < 0 { + return 0 + } + if wait > maxRetryAfter { + return maxRetryAfter + } + return wait +} + +// backoff is the delay before the retry after the given attempt: what the server asked for, +// or base doubled per attempt with jitter, capped at maxBackoff. +func backoff(base time.Duration, attempt int, retryAfter time.Duration) time.Duration { + if retryAfter > 0 { + return retryAfter + } + delay := base + for i := 0; i < attempt && delay < maxBackoff; i++ { + delay *= 2 + } + if delay > maxBackoff { + delay = maxBackoff + } + if delay <= 1 { + return delay + } + return delay/2 + time.Duration(rand.Int63n(int64(delay/2))) +} + +// wait sleeps for d unless the transport is shutting down first. +func (t *transport) wait(d time.Duration) bool { + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-timer.C: + return true + case <-t.ctx.Done(): + return false + } +} + +func (t *transport) drop(counter *atomic.Uint64, n int, err error) { + counter.Add(uint64(n)) + if err != nil { + t.report(err) + } +} + +// report hands an error to OnError. A panic in the callback is contained: the host application +// must not go down because its error reporter has a bug. +func (t *transport) report(err error) { + defer func() { + if r := recover(); r != nil { + fmt.Fprintf(os.Stderr, "slog-betterstack: OnError panicked: %v\n", r) + } + }() + t.option.OnError(err) +} + +// reportQueueFullDrops summarises the records dropped at the queue since the last summary. +// It runs on the sender goroutine and, after the sender has exited, from close. +func (t *transport) reportQueueFullDrops() { + total := t.stats.droppedQueueFull.Load() + if n := total - t.reportedQueueFull; n > 0 { + t.reportedQueueFull = total + t.report(fmt.Errorf("slog-betterstack: dropped %s because the queue is full: the application logs faster than records can be delivered", plural(n, "record"))) + } +} + +// flush asks the sender to upload everything queued so far and waits for those uploads. +func (t *transport) flush(ctx context.Context) error { + t.mu.Lock() + started, closed := t.started, t.closed + t.mu.Unlock() + if closed { + return ErrClosed + } + if !started { + return nil + } + + request := flushRequest{done: make(chan struct{})} + select { + case t.flushes <- request: + case <-t.exited: + return ErrClosed + case <-ctx.Done(): + return ctx.Err() + } + select { + case <-request.done: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +// close stops accepting records, delivers what is queued, waits for the uploads in flight up +// to ShutdownTimeout and reports what was left behind. +func (t *transport) close() error { + t.closeOnce.Do(func() { + t.mu.Lock() + t.closed = true + started := t.started + t.mu.Unlock() + + deadline := time.NewTimer(t.option.ShutdownTimeout) + defer deadline.Stop() + timedOut := false + + if started { + close(t.closing) + select { + case <-t.exited: + case <-deadline.C: + timedOut = true + t.cancel() + <-t.exited + } + + // The sender has exited, so nothing adds to inFlight any more. + finished := make(chan struct{}) + go func() { + t.inFlight.Wait() + close(finished) + }() + if timedOut { + <-finished + } else { + select { + case <-finished: + case <-deadline.C: + timedOut = true + t.cancel() + <-finished + } + } + } + t.cancel() + if t.owned != nil { + t.owned.CloseIdleConnections() + } + + t.reportQueueFullDrops() + if timedOut { + undelivered := t.stats.droppedClosed.Load() + t.closeErr = fmt.Errorf("slog-betterstack: the shutdown timeout of %s ran out with %s undelivered", t.option.ShutdownTimeout, plural(undelivered, "record")) + t.report(t.closeErr) + } + }) + return t.closeErr +} + +func compress(body []byte) ([]byte, error) { + var buf bytes.Buffer + writer, err := gzip.NewWriterLevel(&buf, gzip.BestSpeed) + if err != nil { + return nil, err + } + if _, err := writer.Write(body); err != nil { + return nil, err + } + if err := writer.Close(); err != nil { + return nil, err + } + return buf.Bytes(), nil +} + +func plural[N int | uint64](n N, noun string) string { + if n == 1 { + return "1 " + noun + } + return fmt.Sprintf("%d %ss", n, noun) +} diff --git a/version.go b/version.go index a875fa9..52039f5 100644 --- a/version.go +++ b/version.go @@ -7,3 +7,7 @@ const name = "BetterStackHQ/slog-betterstack" // version is the latest released version and is sent in the "logger.version" field of every // record. The Release workflow bumps it and tags the commit, so edit it only there. const version = "1.4.4" + +// userAgent identifies the library and its version in every request, like the other Better +// Stack clients do. +const userAgent = name + "/" + version From 65b84bf3de732379455015d30d31a5a33d8a4044 Mon Sep 17 00:00:00 2001 From: Petr Heinz Date: Tue, 6 Oct 2026 15:38:42 +0200 Subject: [PATCH 03/11] T-20356 Cover Retry-After parsing, backoff and Flush edge cases Co-Authored-By: Claude Fable 5.1 --- retry_test.go | 85 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 85 insertions(+) create mode 100644 retry_test.go diff --git a/retry_test.go b/retry_test.go new file mode 100644 index 0000000..78e0b1d --- /dev/null +++ b/retry_test.go @@ -0,0 +1,85 @@ +package slogbetterstack + +import ( + "context" + "errors" + "log/slog" + "net/http" + "testing" + "time" +) + +func TestParseRetryAfter(t *testing.T) { + for name, tc := range map[string]struct { + value string + min time.Duration + max time.Duration + }{ + "absent": {"", 0, 0}, + "seconds": {"7", 7 * time.Second, 7 * time.Second}, + "http date": {time.Now().Add(30 * time.Second).UTC().Format(http.TimeFormat), 25 * time.Second, 30 * time.Second}, + "past date": {time.Now().Add(-time.Minute).UTC().Format(http.TimeFormat), 0, 0}, + "garbage": {"soon", 0, 0}, + "capped": {"3600", maxRetryAfter, maxRetryAfter}, + } { + t.Run(name, func(t *testing.T) { + if got := parseRetryAfter(tc.value); got < tc.min || got > tc.max { + t.Errorf("parseRetryAfter(%q) = %s, want between %s and %s", tc.value, got, tc.min, tc.max) + } + }) + } +} + +func TestBackoff(t *testing.T) { + base := 100 * time.Millisecond + + for attempt, want := range []time.Duration{base, 2 * base, 4 * base, 8 * base} { + for i := 0; i < 20; i++ { + if got := backoff(base, attempt, 0); got < want/2 || got > want { + t.Fatalf("backoff(attempt %d) = %s, want jitter between %s and %s", attempt, got, want/2, want) + } + } + } + + if got := backoff(base, 60, 0); got > maxBackoff || got < maxBackoff/2 { + t.Errorf("backoff(attempt 60) = %s, want it capped around %s", got, maxBackoff) + } + if got := backoff(base, 3, 9*time.Second); got != 9*time.Second { + t.Errorf("backoff with Retry-After = %s, want the server's 9s", got) + } + if got := backoff(1, 0, 0); got != 1 { + t.Errorf("backoff(1ns) = %s, want 1ns without jitter", got) + } +} + +func TestFlushBeforeAnyRecordIsANoOp(t *testing.T) { + server, _ := newServer(t, accepted) + handler := newHandler(t, server, Option{}) + + if err := handler.Flush(context.Background()); err != nil { + t.Errorf("Flush on an idle handler = %v, want nil", err) + } + if err := handler.Close(); err != nil { + t.Errorf("Close on an idle handler = %v, want nil", err) + } +} + +func TestFlushGivesUpWhenItsContextEnds(t *testing.T) { + gate := make(chan struct{}) + server, _ := newServer(t, func(w http.ResponseWriter, r *http.Request, _ []map[string]any) { + select { + case <-gate: + case <-r.Context().Done(): + } + w.WriteHeader(http.StatusAccepted) + }) + t.Cleanup(func() { close(gate) }) + handler := newHandler(t, server, Option{BatchInterval: time.Hour, ShutdownTimeout: 100 * time.Millisecond}) + + slog.New(handler).Info("stuck") + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + if err := handler.Flush(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("Flush = %v, want context.DeadlineExceeded while the upload is stuck", err) + } +} From 7034e5dce8a912c7466d01a57758a057d266d150 Mon Sep 17 00:00:00 2001 From: Petr Heinz Date: Tue, 6 Oct 2026 15:41:53 +0200 Subject: [PATCH 04/11] T-20356 Satisfy golangci-lint v2: check Close in the test helper, lowercase an error Co-Authored-By: Claude Fable 5.1 --- handler_test.go | 2 +- transport.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/handler_test.go b/handler_test.go index d45dcb5..0631347 100644 --- a/handler_test.go +++ b/handler_test.go @@ -51,7 +51,7 @@ func newServer(t *testing.T, respond responder) (*httptest.Server, <-chan reques w.WriteHeader(http.StatusBadRequest) return } - defer unzipped.Close() + defer func() { _ = unzipped.Close() }() body = unzipped } raw, err := io.ReadAll(body) diff --git a/transport.go b/transport.go index a11dbf8..34ad50b 100644 --- a/transport.go +++ b/transport.go @@ -297,7 +297,7 @@ func (t *transport) upload(records []map[string]any) { t.drop(&t.stats.droppedRejected, len(records), fmt.Errorf("slog-betterstack: Better Stack rejected %s: %s", plural(len(records), "record"), status)) return } - err = fmt.Errorf("Better Stack answered %s", status) + err = fmt.Errorf("the server answered %s", status) retryAfter = parseRetryAfter(header.Get("Retry-After")) } From 76cda55a5f1d536c6ce3c4cdfc81155e59aaab79 Mon Sep 17 00:00:00 2001 From: Petr Heinz Date: Tue, 6 Oct 2026 15:49:25 +0200 Subject: [PATCH 05/11] T-20356 Report a missing token instead of panicking; show the Info level Without a source token the handler now reports the omission once through OnError, returns ErrMissingToken from Handle and drops every record, so an unset environment variable cannot take the application down. The constructor keeps its signature. The package quickstart and the example project set Level: slog.LevelInfo with a note that Debug is the default. Co-Authored-By: Claude Fable 5.1 --- doc.go | 5 +++-- example-project/main.go | 7 +++++-- handler.go | 7 ++----- handler_test.go | 30 +++++++++++++++++++++++------- transport.go | 11 +++++++++++ 5 files changed, 44 insertions(+), 16 deletions(-) diff --git a/doc.go b/doc.go index bd5fb8a..7b0ed20 100644 --- a/doc.go +++ b/doc.go @@ -4,6 +4,7 @@ // handler := slogbetterstack.Option{ // Token: "$SOURCE_TOKEN", // Endpoint: "https://$INGESTING_HOST/", +// Level: slog.LevelInfo, // Debug if omitted // }.NewBetterstackHandler() // defer handler.Close() // @@ -12,8 +13,8 @@ // // Records are queued and uploaded in batches by a background goroutine, so logging never // waits for the network. Close delivers what is still queued and must run before the program -// exits; os.Exit and log.Fatal skip deferred calls. Delivery failures are reported through -// [Option.OnError], on stderr by default, and counted in [Stats]. +// exits; os.Exit and log.Fatal skip deferred calls. Delivery failures, and a missing token, +// are reported through [Option.OnError], on stderr by default, and counted in [Stats]. // // See https://betterstack.com/docs/logs/go/ for the full documentation. package slogbetterstack diff --git a/example-project/main.go b/example-project/main.go index 24875c7..f5a315d 100644 --- a/example-project/main.go +++ b/example-project/main.go @@ -22,7 +22,10 @@ func main() { os.Exit(1) } - option := slogbetterstack.Option{Level: slog.LevelDebug, Token: token} + option := slogbetterstack.Option{ + Token: token, + Level: slog.LevelInfo, // Debug if omitted + } if host := os.Getenv("BETTERSTACK_INGESTING_HOST"); host != "" { option.Endpoint = "https://" + host + "/" } @@ -31,7 +34,7 @@ func main() { logger := slog.New(handler) logger = logger.With("release", "v1.0.0") - logger.Debug("Debugging user service.", "service", "UserService") + logger.Info("Starting user service.", "service", "UserService") logger.With("userID", 123).Error("Unable to fetch user data.") diff --git a/handler.go b/handler.go index 7ca73de..46e6b6c 100644 --- a/handler.go +++ b/handler.go @@ -18,7 +18,8 @@ type Option struct { // log level (default: debug) Level slog.Leveler - // token + // source token; without it the handler reports the omission through OnError once and + // drops every record instead of sending anything Token string // optional: endpoint Endpoint string @@ -73,10 +74,6 @@ func (o Option) NewBetterstackHandler() *BetterstackHandler { o.Level = slog.LevelDebug } - if o.Token == "" { - panic("missing Betterstack token") - } - if o.Endpoint == "" { o.Endpoint = BetterstackEndpoint } diff --git a/handler_test.go b/handler_test.go index 0631347..186574d 100644 --- a/handler_test.go +++ b/handler_test.go @@ -323,13 +323,29 @@ func TestDefaults(t *testing.T) { } } -func TestMissingTokenPanics(t *testing.T) { - defer func() { - if recover() == nil { - t.Error("NewBetterstackHandler without a token did not panic") - } - }() - Option{}.NewBetterstackHandler() +func TestMissingTokenIsReportedOnceAndDropsRecords(t *testing.T) { + server, requests := newServer(t, accepted) + errs := &errorList{} + handler := Option{Endpoint: server.URL, OnError: errs.add}.NewBetterstackHandler() + t.Cleanup(func() { _ = handler.Close() }) + logger := slog.New(handler) + + logger.Info("first") + logger.Info("second") + if err := handler.Handle(context.Background(), slog.NewRecord(time.Now(), slog.LevelInfo, "third", 0)); !errors.Is(err, ErrMissingToken) { + t.Errorf("Handle = %v, want ErrMissingToken", err) + } + if err := handler.Close(); err != nil { + t.Errorf("Close = %v, want nil", err) + } + + nothingWithin(t, requests, 50*time.Millisecond) + if got := errs.all(); len(got) != 1 || !errors.Is(got[0], ErrMissingToken) { + t.Errorf("OnError got %v, want ErrMissingToken exactly once", got) + } + if stats := handler.Stats(); stats.DroppedRejected != 3 || stats.Sent != 0 { + t.Errorf("stats = %+v, want DroppedRejected 3 and Sent 0", stats) + } } func TestLevel(t *testing.T) { diff --git a/transport.go b/transport.go index 34ad50b..e1b0e50 100644 --- a/transport.go +++ b/transport.go @@ -19,6 +19,10 @@ import ( // ErrClosed is returned by Handle and Flush once the handler has been closed. var ErrClosed = errors.New("slog-betterstack: the handler is closed") +// ErrMissingToken is returned by Handle, and reported through OnError once, when the handler +// was built without a source token. Nothing is sent and every record is counted as dropped. +var ErrMissingToken = errors.New("slog-betterstack: no source token configured, records are dropped") + const ( defaultBatchSize = 1000 defaultBatchInterval = time.Second @@ -92,6 +96,7 @@ type transport struct { closeOnce sync.Once closeErr error inFlight sync.WaitGroup + tokenOnce sync.Once // reports the missing token a single time stats counters reportedQueueFull uint64 // queue-full drops already summarised through OnError @@ -127,6 +132,12 @@ func newTransport(option Option) *transport { func (t *transport) enqueue(record map[string]any) error { t.stats.enqueued.Add(1) + if t.option.Token == "" { + t.stats.droppedRejected.Add(1) + t.tokenOnce.Do(func() { t.report(ErrMissingToken) }) + return ErrMissingToken + } + t.mu.Lock() defer t.mu.Unlock() if t.closed { From cfbdcc0765e370035294788976387a3a940b668b Mon Sep 17 00:00:00 2001 From: Petr Heinz Date: Tue, 6 Oct 2026 15:53:06 +0200 Subject: [PATCH 06/11] =?UTF-8?q?T-20356=20Credit=20Tom=C3=A1=C5=A1=20Proc?= =?UTF-8?q?h=C3=A1zka=20in=20the=20README?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5.1 --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index f7177b7..724281a 100644 --- a/README.md +++ b/README.md @@ -16,9 +16,9 @@ Experience SQL-compatible structured log management based on ClickHouse. [Learn Please let us know at [hello@betterstack.com](mailto:hello@betterstack.com). We're happy to help! ## Credits -`slog-betterstack` was created and maintained by [Samuel Berthe](https://github.com/samber) and released under the MIT license. +`slog-betterstack` was created and maintained by [Samuel Berthe](https://github.com/samber) and released under the MIT license. [Tomáš Procházka](https://github.com/prochac) reported that records logged before exit were lost, and his analysis of the official clients shaped the delivery defaults and the drop accounting. -Thank you, Samuel! ❤️ +Thank you, Samuel and Tomáš! ❤️ --- From f346ada161ed9a4af754c395f4feb59414d5ff10 Mon Sep 17 00:00:00 2001 From: Petr Heinz Date: Tue, 6 Oct 2026 16:00:01 +0200 Subject: [PATCH 07/11] T-20356 Add failing tests from the end-to-end red team An end-to-end run against an unreachable endpoint showed that a batch abandoned at the shutdown timeout reports only the timeout, never the connection error behind it. A run that kept logging after Close showed that those records are counted but never mentioned, because slog.Logger discards Handle's error. A negative Timeout was not treated as unset. Co-Authored-By: Claude Fable 5.1 --- shutdown_test.go | 83 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 83 insertions(+) create mode 100644 shutdown_test.go diff --git a/shutdown_test.go b/shutdown_test.go new file mode 100644 index 0000000..a385cee --- /dev/null +++ b/shutdown_test.go @@ -0,0 +1,83 @@ +package slogbetterstack + +import ( + "log/slog" + "net/http" + "strings" + "testing" + "time" +) + +// Found by an end-to-end run against an unreachable endpoint: the batch was abandoned at the +// shutdown timeout and the only report was the timeout itself, so the connection error that +// caused it never reached the operator. +func TestAbandonedUploadReportsItsLastError(t *testing.T) { + gate := make(chan struct{}) + server, _ := newServer(t, func(w http.ResponseWriter, r *http.Request, _ []map[string]any) { + select { + case <-gate: + case <-r.Context().Done(): + } + w.WriteHeader(http.StatusAccepted) + }) + t.Cleanup(func() { close(gate) }) + errs := &errorList{} + handler := newHandler(t, server, Option{ + Timeout: 50 * time.Millisecond, // the first attempt times out + RetryBackoff: 10 * time.Second, // the retry is still waiting when Close gives up + ShutdownTimeout: 100 * time.Millisecond, + OnError: errs.add, + }) + + slog.New(handler).Info("stuck") + if err := handler.Close(); err == nil { + t.Fatal("Close = nil, want the shutdown timeout error") + } + + got := errs.joined() + if !strings.Contains(got, "at shutdown after 1 attempt") || !strings.Contains(got, "deadline") { + t.Errorf("OnError got %q, want the abandoned batch's last error", got) + } + if stats := handler.Stats(); stats.DroppedClosed != 1 || stats.Retries != 1 { + t.Errorf("stats = %+v, want DroppedClosed 1 and Retries 1", stats) + } +} + +// slog.Logger discards Handle's error, so a program that keeps logging after Close would +// never learn that those records go nowhere. +func TestRecordsAfterCloseAreReportedOnce(t *testing.T) { + server, _ := newServer(t, accepted) + errs := &errorList{} + handler := newHandler(t, server, Option{OnError: errs.add}) + logger := slog.New(handler) + + logger.Info("before") + if err := handler.Close(); err != nil { + t.Fatal(err) + } + logger.Info("after") + logger.Info("after again") + + var reports int + for _, err := range errs.all() { + if strings.Contains(err.Error(), "after Close") { + reports++ + } + } + if reports != 1 { + t.Errorf("got %d reports about records logged after Close, want exactly 1: %v", reports, errs.all()) + } + if stats := handler.Stats(); stats.DroppedClosed != 2 { + t.Errorf("stats = %+v, want DroppedClosed 2", stats) + } +} + +func TestNegativeTimeoutMeansTheDefault(t *testing.T) { + handler := Option{Token: "x", Timeout: -1}.NewBetterstackHandler() + if handler.option.Timeout != 10*time.Second { + t.Errorf("Timeout = %s, want the 10s default for a negative value", handler.option.Timeout) + } + if err := handler.Close(); err != nil { + t.Error(err) + } +} From c980ad60f3f5beba11c683d07fb69e0ed9b5d7fa Mon Sep 17 00:00:00 2001 From: Petr Heinz Date: Tue, 6 Oct 2026 16:00:01 +0200 Subject: [PATCH 08/11] T-20356 Report abandoned uploads' last error and records logged after Close A batch dropped at shutdown after at least one failed attempt now reports that attempt's error, so an outage is visible even when Close gives up on it. The first record logged after Close is reported once. A Timeout of zero or less means the default. OnError is documented as concurrent and non-blocking, and the constructor as one per process. Co-Authored-By: Claude Fable 5.1 --- handler.go | 9 ++++++--- transport.go | 18 +++++++++++++++--- 2 files changed, 21 insertions(+), 6 deletions(-) diff --git a/handler.go b/handler.go index 46e6b6c..4f4fe7f 100644 --- a/handler.go +++ b/handler.go @@ -60,8 +60,9 @@ type Option struct { ShutdownTimeout time.Duration // optional: send the JSON uncompressed instead of gzip-compressed DisableCompression bool - // optional: receives every delivery failure and drop summary, from a background - // goroutine; it must not log through this handler (default: one line on stderr) + // optional: receives every delivery failure and drop summary. It is called from + // background goroutines, possibly several at once, must return promptly and must not + // log through this handler (default: one line on stderr) OnError func(err error) // optional: the HTTP client to upload with; Timeout still applies to every request HTTPClient *http.Client @@ -69,6 +70,8 @@ type Option struct { // NewBetterstackHandler returns a handler that sends records to Better Stack. The handler is // also a [slog.Handler]; keep the returned value to call Close before the program exits. +// Create one handler per process rather than one per request: each handler owns a goroutine +// and a connection pool from its first record until Close. func (o Option) NewBetterstackHandler() *BetterstackHandler { if o.Level == nil { o.Level = slog.LevelDebug @@ -78,7 +81,7 @@ func (o Option) NewBetterstackHandler() *BetterstackHandler { o.Endpoint = BetterstackEndpoint } - if o.Timeout == 0 { + if o.Timeout <= 0 { o.Timeout = defaultTimeout } diff --git a/transport.go b/transport.go index e1b0e50..8ff4ca4 100644 --- a/transport.go +++ b/transport.go @@ -97,6 +97,7 @@ type transport struct { closeErr error inFlight sync.WaitGroup tokenOnce sync.Once // reports the missing token a single time + lateOnce sync.Once // reports records logged after Close a single time stats counters reportedQueueFull uint64 // queue-full drops already summarised through OnError @@ -142,6 +143,7 @@ func (t *transport) enqueue(record map[string]any) error { defer t.mu.Unlock() if t.closed { t.stats.droppedClosed.Add(1) + t.lateOnce.Do(func() { t.report(fmt.Errorf("slog-betterstack: records logged after Close are dropped")) }) return ErrClosed } if !t.started { @@ -282,9 +284,19 @@ func (t *transport) upload(records []map[string]any) { } } + // abandon drops the batch because the handler is shutting down. After a failed attempt the + // failure is reported, so that an outage is not hidden behind the shutdown timeout. + abandon := func(attempts int, lastErr error) { + var err error + if lastErr != nil { + err = fmt.Errorf("slog-betterstack: dropped %s at shutdown after %s: %w", plural(len(records), "record"), plural(attempts, "attempt"), lastErr) + } + t.drop(&t.stats.droppedClosed, len(records), err) + } + for attempt := 0; ; attempt++ { if t.ctx.Err() != nil { - t.drop(&t.stats.droppedClosed, len(records), nil) + abandon(attempt, nil) return } @@ -313,7 +325,7 @@ func (t *transport) upload(records []map[string]any) { } if t.ctx.Err() != nil { - t.drop(&t.stats.droppedClosed, len(records), nil) + abandon(attempt+1, err) return } if attempt >= t.option.MaxRetries { @@ -322,7 +334,7 @@ func (t *transport) upload(records []map[string]any) { } t.stats.retries.Add(1) if !t.wait(backoff(t.option.RetryBackoff, attempt, retryAfter)) { - t.drop(&t.stats.droppedClosed, len(records), nil) + abandon(attempt+1, err) return } } From 3e2afab1c755e1b8351af85490d3a73d53b93080 Mon Sep 17 00:00:00 2001 From: Petr Heinz Date: Tue, 6 Oct 2026 16:01:32 +0200 Subject: [PATCH 09/11] T-20356 Add a failing test for a record over Better Stack's size limit Sending a 12 MiB record to the real endpoint got a 2xx, and the record never appeared: the per-record limit of 10 MiB uncompressed is enforced after the request is accepted, so the client counted it as sent. Co-Authored-By: Claude Fable 5.1 --- shutdown_test.go | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/shutdown_test.go b/shutdown_test.go index a385cee..c9770ff 100644 --- a/shutdown_test.go +++ b/shutdown_test.go @@ -72,6 +72,36 @@ func TestRecordsAfterCloseAreReportedOnce(t *testing.T) { } } +// Found by sending a 12 MiB record to the real endpoint: it answered 2xx and the record never +// appeared, because the per-record limit is enforced after the request is accepted. +func TestRecordOverTheSizeLimitIsDroppedBeforeSending(t *testing.T) { + server, requests := newServer(t, accepted) + errs := &errorList{} + handler := newHandler(t, server, Option{BatchInterval: time.Hour, OnError: errs.add}) + logger := slog.New(handler) + + logger.Info("small before") + logger.Info("huge", "blob", strings.Repeat("x", maxRecordBytes)) + logger.Info("small after") + if err := handler.Close(); err != nil { + t.Fatal(err) + } + + var delivered []string + for len(delivered) < 2 { + delivered = append(delivered, messages(receive(t, requests).records)...) + } + if want := []string{"small before", "small after"}; strings.Join(delivered, ",") != strings.Join(want, ",") { + t.Errorf("delivered %v, want %v", delivered, want) + } + if got := errs.joined(); !strings.Contains(got, "larger than") { + t.Errorf("OnError got %q, want a report about the oversized record", got) + } + if stats := handler.Stats(); stats.Sent != 2 || stats.DroppedOversize != 1 { + t.Errorf("stats = %+v, want Sent 2 and DroppedOversize 1", stats) + } +} + func TestNegativeTimeoutMeansTheDefault(t *testing.T) { handler := Option{Token: "x", Timeout: -1}.NewBetterstackHandler() if handler.option.Timeout != 10*time.Second { From b5c3e60fa9336694c01177c077bb6e29d96b7146 Mon Sep 17 00:00:00 2001 From: Petr Heinz Date: Tue, 6 Oct 2026 16:01:32 +0200 Subject: [PATCH 10/11] T-20356 Drop and report a record over the size limit before sending it A batch whose JSON is over 10 MiB is split until the records that are over the limit on their own stand alone, and only those are dropped and reported; the rest is sent. The server would otherwise accept the request and discard the record silently. Co-Authored-By: Claude Fable 5.1 --- transport.go | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/transport.go b/transport.go index 8ff4ca4..33f8051 100644 --- a/transport.go +++ b/transport.go @@ -36,6 +36,11 @@ const ( maxBackoff = 30 * time.Second maxRetryAfter = time.Minute dropReportInterval = 5 * time.Second + + // maxRecordBytes is Better Stack's limit on one record's uncompressed JSON. The endpoint + // accepts a request whose compressed body fits and discards such a record afterwards + // without a word, so the check has to happen here for the drop to be visible. + maxRecordBytes = 10 << 20 ) // Stats counts what happened to the records handed to a handler. Once Close has returned, @@ -277,6 +282,18 @@ func (t *transport) upload(records []map[string]any) { t.drop(&t.stats.droppedRejected, len(records), fmt.Errorf("slog-betterstack: dropped %s that could not be encoded: %w", plural(len(records), "record"), err)) return } + if len(body) > maxRecordBytes { + // Too big for a request body as well as for a single record. Split until the records + // that are over the limit on their own stand alone, and drop just those. + if len(records) > 1 { + half := len(records) / 2 + t.upload(records[:half]) + t.upload(records[half:]) + return + } + t.drop(&t.stats.droppedOversize, 1, fmt.Errorf("slog-betterstack: dropped a record of %d bytes, larger than the %d bytes Better Stack accepts", len(body), maxRecordBytes)) + return + } if !t.option.DisableCompression { if body, err = compress(body); err != nil { t.drop(&t.stats.droppedRejected, len(records), fmt.Errorf("slog-betterstack: dropped %s that could not be compressed: %w", plural(len(records), "record"), err)) From f05463034dd8be20be213d844ed315a381016f8d Mon Sep 17 00:00:00 2001 From: Petr Heinz Date: Tue, 6 Oct 2026 16:02:19 +0200 Subject: [PATCH 11/11] T-20356 Let the size-limit test stand on its own The test named the new constant, so against the code it was written to fail on it did not compile instead of failing. A literal makes it fail for the right reason: the record is sent and counted. Co-Authored-By: Claude Fable 5.1 --- shutdown_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/shutdown_test.go b/shutdown_test.go index c9770ff..89c88b6 100644 --- a/shutdown_test.go +++ b/shutdown_test.go @@ -81,7 +81,7 @@ func TestRecordOverTheSizeLimitIsDroppedBeforeSending(t *testing.T) { logger := slog.New(handler) logger.Info("small before") - logger.Info("huge", "blob", strings.Repeat("x", maxRecordBytes)) + logger.Info("huge", "blob", strings.Repeat("x", 10<<20)) // over 10 MiB once encoded logger.Info("small after") if err := handler.Close(); err != nil { t.Fatal(err)