orchard

  1package web
  2
  3import (
  4	"bytes"
  5	"context"
  6	"log/slog"
  7	"sync"
  8	"testing"
  9	"time"
 10)
 11
 12// collector is a Sink that remembers what it was handed.
 13type collector struct {
 14	mu      sync.Mutex
 15	records []Record
 16	flushes int
 17}
 18
 19func (c *collector) sink(source string, records []Record) {
 20	c.mu.Lock()
 21	defer c.mu.Unlock()
 22	c.records = append(c.records, records...)
 23	c.flushes++
 24}
 25
 26func (c *collector) all() []Record {
 27	c.mu.Lock()
 28	defer c.mu.Unlock()
 29	return append([]Record{}, c.records...)
 30}
 31
 32// withShipper installs a shipper over a buffer-backed JSON handler and restores
 33// the process logger afterwards. ShipLogs replaces slog's default, which is
 34// global state, so a test that did not restore it would silently change every
 35// test that ran after it.
 36func withShipper(t *testing.T, source string, sink Sink) (*bytes.Buffer, *Shipper) {
 37	t.Helper()
 38
 39	var buf bytes.Buffer
 40	previous := slog.Default()
 41	slog.SetDefault(slog.New(slog.NewJSONHandler(&buf, nil)))
 42
 43	s := ShipLogs(source, sink)
 44	t.Cleanup(func() {
 45		s.Close()
 46		slog.SetDefault(previous)
 47	})
 48	return &buf, s
 49}
 50
 51// The whole design is a tee, not a replacement. stdout stays the source of
 52// truth, so a record has to appear in both places.
 53func TestShipLogsTeesRatherThanReplaces(t *testing.T) {
 54	c := &collector{}
 55	buf, shipper := withShipper(t, "blog", c.sink)
 56
 57	slog.Info("request",
 58		slog.Int("status", 200),
 59		slog.String("path", "/feed.atom"),
 60		slog.Float64("ms", 1.25))
 61
 62	// Close is the deterministic flush. The ticker would get there in five
 63	// seconds, which is not a thing to put in a test.
 64	shipper.Close()
 65
 66	if !bytes.Contains(buf.Bytes(), []byte(`"path":"/feed.atom"`)) {
 67		t.Errorf("the record did not reach the original handler:\n%s", buf.String())
 68	}
 69
 70	got := c.all()
 71	if len(got) != 1 {
 72		t.Fatalf("shipped %d records, want 1", len(got))
 73	}
 74	r := got[0]
 75	if r.Msg != "request" || r.Level != "INFO" {
 76		t.Errorf("msg/level = %q/%q", r.Msg, r.Level)
 77	}
 78	if r.Attrs["status"] != int64(200) {
 79		t.Errorf("status = %#v, want int64(200)", r.Attrs["status"])
 80	}
 81	if r.Attrs["ms"] != 1.25 {
 82		t.Errorf("ms = %#v, want 1.25", r.Attrs["ms"])
 83	}
 84	// Milliseconds, not nanoseconds and not seconds. The ingest side stores
 85	// this straight into a column and a wrong unit here would put every
 86	// record either in 1970 or fifty thousand years out.
 87	if delta := time.Now().UTC().UnixMilli() - r.Time; delta < 0 || delta > 60_000 {
 88		t.Errorf("Time = %d, which is not unix milliseconds near now", r.Time)
 89	}
 90}
 91
 92// Attribute kinds have to survive the crossing intact: a duration that became
 93// the string "1.042ms" is exactly what the hardening pass moved away from.
 94func TestShipLogsFlattensAttributeKinds(t *testing.T) {
 95	c := &collector{}
 96	_, shipper := withShipper(t, "blog", c.sink)
 97
 98	slog.Info("mixed",
 99		slog.Bool("ok", true),
100		slog.Int64("n", 42),
101		slog.Float64("f", 0.5),
102		slog.String("s", "text"),
103		slog.Duration("d", 1500*time.Millisecond),
104		slog.Any("err", context.Canceled),
105	)
106	shipper.Close()
107
108	got := c.all()
109	if len(got) != 1 {
110		t.Fatalf("shipped %d records, want 1", len(got))
111	}
112	a := got[0].Attrs
113	if a["ok"] != true || a["n"] != int64(42) || a["f"] != 0.5 || a["s"] != "text" {
114		t.Errorf("scalar attributes did not survive: %#v", a)
115	}
116	if a["d"] != "1.5s" {
117		t.Errorf("duration = %#v, want \"1.5s\"", a["d"])
118	}
119	// An arbitrary value is formatted rather than marshalled, because one
120	// value that cannot marshal must not cost the whole batch.
121	if a["err"] != "context canceled" {
122		t.Errorf("error = %#v, want its message", a["err"])
123	}
124}
125
126// A group is flattened to dotted keys. JSON has no notion of a slog group, and
127// a nested object would mean the ingest side could not pull "status" out with
128// one lookup.
129func TestShipLogsFlattensGroups(t *testing.T) {
130	c := &collector{}
131	_, shipper := withShipper(t, "blog", c.sink)
132
133	slog.Info("grouped", slog.Group("http", slog.Int("status", 503)))
134	shipper.Close()
135
136	got := c.all()
137	if len(got) != 1 {
138		t.Fatalf("shipped %d records, want 1", len(got))
139	}
140	if got[0].Attrs["http.status"] != int64(503) {
141		t.Errorf("attrs = %#v, want http.status", got[0].Attrs)
142	}
143}
144
145// WithAttrs and WithGroup have to return something that still tees. An embedded
146// handler's versions return the inner handler, and the failure mode is logs
147// that quietly stop being shipped from whichever subsystem used a child logger.
148func TestShipLogsSurvivesWithAttrsAndWithGroup(t *testing.T) {
149	c := &collector{}
150	_, shipper := withShipper(t, "status", c.sink)
151
152	child := slog.Default().With(slog.String("component", "crawler"))
153	child.Warn("crawl slow")
154
155	nested := slog.Default().WithGroup("job").With(slog.Int("attempt", 2))
156	nested.Info("retrying")
157
158	shipper.Close()
159
160	got := c.all()
161	if len(got) != 2 {
162		t.Fatalf("shipped %d records, want 2: a child logger stopped teeing", len(got))
163	}
164	if got[0].Attrs["component"] != "crawler" {
165		t.Errorf("With attrs lost: %#v", got[0].Attrs)
166	}
167	if got[1].Attrs["job.attempt"] != int64(2) {
168		t.Errorf("WithGroup attrs lost: %#v", got[1].Attrs)
169	}
170}
171
172// The failure this pins has no symptom until somebody uses a child logger, and
173// then it is silent: an attribute added before a WithGroup must NOT gain that
174// group's prefix. If it does, the ingest side stops recognising "component" and
175// "status" as hot columns, the record stores an empty component and a zero
176// status, and its hourly rollup is keyed wrong, while stdout shows the same
177// line correctly.
178func TestWithGroupDoesNotRePrefixEarlierAttrs(t *testing.T) {
179	c := &collector{}
180	_, shipper := withShipper(t, "status", c.sink)
181
182	logger := slog.Default().
183		With(slog.String("component", "crawler")).
184		WithGroup("http")
185	logger.Info("request", slog.Int("status", 503))
186
187	shipper.Close()
188
189	got := c.all()
190	if len(got) != 1 {
191		t.Fatalf("shipped %d records, want 1", len(got))
192	}
193	a := got[0].Attrs
194	if a["component"] != "crawler" {
195		t.Errorf("component = %#v, want \"crawler\" unprefixed; got attrs %#v", a["component"], a)
196	}
197	// The record's own attribute is inside the group, so it
198	// correctly gains the prefix. Only the earlier one must not.
199	if a["http.status"] != int64(503) {
200		t.Errorf("http.status = %#v, want int64(503); got attrs %#v", a["http.status"], a)
201	}
202	if _, bad := a["http.component"]; bad {
203		t.Errorf("component was retroactively moved into the group: %#v", a)
204	}
205}
206
207// An empty group name is a no-op in slog. Without the short-circuit the prefix
208// becomes "g." and every key under it gains a double dot.
209func TestWithGroupEmptyNameIsNoOp(t *testing.T) {
210	c := &collector{}
211	_, shipper := withShipper(t, "blog", c.sink)
212
213	h := slog.Default().Handler().WithGroup("").WithAttrs([]slog.Attr{slog.Int("n", 1)})
214	slog.New(h).Info("x")
215	shipper.Close()
216
217	got := c.all()
218	if len(got) != 1 {
219		t.Fatalf("shipped %d records, want 1", len(got))
220	}
221	if got[0].Attrs["n"] != int64(1) {
222		t.Errorf("attrs = %#v, want a bare \"n\"", got[0].Attrs)
223	}
224}
225
226// Nothing here may ever block the caller. A full queue drops, because the
227// record is already safely on stdout and a stalled site is a worse outcome than
228// a gap in a dashboard.
229func TestEnqueueDropsRatherThanBlocking(t *testing.T) {
230	s := &Shipper{ch: make(chan Record, 2)}
231
232	done := make(chan struct{})
233	go func() {
234		for i := 0; i < 100; i++ {
235			s.enqueue(Record{Msg: "x"})
236		}
237		close(done)
238	}()
239
240	select {
241	case <-done:
242	case <-time.After(2 * time.Second):
243		t.Fatal("enqueue blocked on a full queue")
244	}
245
246	if len(s.ch) != 2 {
247		t.Errorf("queue holds %d, want 2", len(s.ch))
248	}
249}
250
251// Handle can run after Close, on a shutdown path where something logs from a
252// later defer. A send on a closed channel panics, which would be the one way a
253// log shipper could take a site down with it, so the channel is never closed.
254func TestLoggingAfterCloseDoesNotPanic(t *testing.T) {
255	c := &collector{}
256	_, shipper := withShipper(t, "blog", c.sink)
257
258	shipper.Close()
259	shipper.Close() // idempotent, because a defer plus an explicit call happens
260
261	slog.Info("logged during shutdown")
262	// And a queue's worth more, to reach the point a closed channel would
263	// have been written to rather than merely selected on.
264	for i := 0; i < shipQueue+10; i++ {
265		slog.Info("more")
266	}
267}
268
269// A wedged sink must not hold shutdown open. Before this bound, a full queue
270// against a sink that accepts and never answers cost nine flushes at ten
271// seconds each: 100 seconds, against a Docker stop grace of ten, so every site
272// shipping to a hung logging container took a SIGKILL.
273func TestCloseIsBoundedWhenTheSinkHangs(t *testing.T) {
274	wedged := func(source string, records []Record) {
275		// Longer than any plausible Close budget, and longer than the whole
276		// test would tolerate if the bound were missing.
277		time.Sleep(30 * time.Second)
278	}
279	_, shipper := withShipper(t, "blog", wedged)
280
281	for i := 0; i < shipBatch*3; i++ {
282		slog.Info("burst")
283	}
284
285	start := time.Now()
286	shipper.Close()
287	elapsed := time.Since(start)
288
289	if elapsed > closeTimeout*3 {
290		t.Errorf("Close took %v against a wedged sink, want bounded near %v", elapsed, closeTimeout)
291	}
292}
293
294// Close flushes what is queued. A deploy kills these processes constantly and
295// the last few seconds of records are the ones that say why.
296func TestCloseFlushesPendingRecords(t *testing.T) {
297	c := &collector{}
298	_, shipper := withShipper(t, "blog", c.sink)
299
300	for i := 0; i < 10; i++ {
301		slog.Info("pending")
302	}
303	shipper.Close()
304
305	if got := len(c.all()); got != 10 {
306		t.Errorf("flushed %d records on close, want 10", got)
307	}
308}
309
310// The batch cap has to actually cap. Without it a burst would be handed to the
311// sink as one enormous POST that the ingest side would refuse with 413.
312func TestFlushesAtBatchSize(t *testing.T) {
313	c := &collector{}
314	_, shipper := withShipper(t, "blog", c.sink)
315
316	for i := 0; i < shipBatch+5; i++ {
317		slog.Info("burst")
318	}
319	shipper.Close()
320
321	c.mu.Lock()
322	flushes, total := c.flushes, len(c.records)
323	c.mu.Unlock()
324
325	if flushes < 2 {
326		t.Errorf("flushes = %d, want at least 2 past a batch of %d", flushes, shipBatch)
327	}
328	if total != shipBatch+5 {
329		t.Errorf("shipped %d records, want %d", total, shipBatch+5)
330	}
331}