orchard

  1package web
  2
  3import (
  4	"bytes"
  5	"context"
  6	"encoding/json"
  7	"fmt"
  8	"io"
  9	"log/slog"
 10	"net/http"
 11	"os"
 12	"sync"
 13	"time"
 14)
 15
 16// Log shipping: a slog.Handler that tees every record to logging.bythewood.me
 17// on top of the existing stdout handler. stdout stays the source of truth, so
 18// nothing here ever blocks the caller. A full queue drops and a failed POST
 19// drops.
 20
 21// ShipEndpoint is a container name on the orchard-edge bridge, never the public
 22// hostname. Anything that can reach it is already inside the network, so there
 23// is no token to configure.
 24const ShipEndpoint = "http://orchard-logging:8000/ingest"
 25
 26const (
 27	// Bounded, so the failure mode is dropping rather than blocking.
 28	shipQueue = 4096
 29	// Whichever comes first. Each flush is one POST that the logging site
 30	// itself logs, so a faster cadence would make ingest the loudest thing
 31	// in the database.
 32	shipBatch   = 500
 33	shipEvery   = 5 * time.Second
 34	shipTimeout = 10 * time.Second
 35
 36	// The hard ceiling on Close, and what keeps a wedged logging site from
 37	// reaching the sites it watches. A sink that accepts the connection and
 38	// never answers costs shipTimeout per flush, which unbounded runs past
 39	// Docker's stop grace and turns one hung container into a SIGKILL for
 40	// every other site, skipping their db.Close(). A healthy sink drains in
 41	// milliseconds and Close returns immediately either way.
 42	closeTimeout = 2 * time.Second
 43)
 44
 45// Record is one slog record on the wire, and the shape the ingest endpoint
 46// parses. Short keys because this is machine to machine at volume.
 47type Record struct {
 48	// Unix milliseconds, UTC.
 49	Time  int64          `json:"t"`
 50	Level string         `json:"l"`
 51	Msg   string         `json:"m"`
 52	Attrs map[string]any `json:"a,omitempty"`
 53}
 54
 55// Batch is one POST body.
 56type Batch struct {
 57	Source  string   `json:"source"`
 58	Records []Record `json:"records"`
 59}
 60
 61// Sink consumes a flush. HTTPSink posts to the logging site, which passes its
 62// own database writer instead so it never posts to itself.
 63type Sink func(source string, records []Record)
 64
 65// Shipper owns the queue and the goroutine that drains it.
 66type Shipper struct {
 67	source string
 68	sink   Sink
 69	ch     chan Record
 70
 71	quit chan struct{}
 72	done chan struct{}
 73	stop sync.Once
 74}
 75
 76// ShipLogs installs the tee on top of whatever slog.Default() already is, so
 77// SetupLogging must have run first. The returned Shipper flushes what it holds
 78// on Close.
 79func ShipLogs(source string, sink Sink) *Shipper {
 80	s := &Shipper{
 81		source: source,
 82		sink:   sink,
 83		ch:     make(chan Record, shipQueue),
 84		quit:   make(chan struct{}),
 85		done:   make(chan struct{}),
 86	}
 87	slog.SetDefault(slog.New(&teeHandler{next: slog.Default().Handler(), ship: s}))
 88	go s.run()
 89	return s
 90}
 91
 92// enqueue never blocks. A full queue means the record is already safely on
 93// stdout either way.
 94//
 95// The channel is never closed. Handle can still run after Close, from a later
 96// defer, and a send on a closed channel panics, which is the one way a log
 97// shipper could take a site down with it.
 98func (s *Shipper) enqueue(r Record) {
 99	select {
100	case s.ch <- r:
101	default:
102	}
103}
104
105func (s *Shipper) run() {
106	defer close(s.done)
107
108	tick := time.NewTicker(shipEvery)
109	defer tick.Stop()
110
111	batch := make([]Record, 0, shipBatch)
112	flush := func() {
113		if len(batch) == 0 {
114			return
115		}
116		s.sink(s.source, batch)
117		batch = batch[:0]
118	}
119
120	for {
121		select {
122		case r := <-s.ch:
123			batch = append(batch, r)
124			if len(batch) >= shipBatch {
125				flush()
126			}
127		case <-tick.C:
128			flush()
129		case <-s.quit:
130			// Drain what is queued, then flush once. Anything enqueued
131			// after this drops.
132			for {
133				select {
134				case r := <-s.ch:
135					batch = append(batch, r)
136					if len(batch) >= shipBatch {
137						flush()
138					}
139					continue
140				default:
141				}
142				break
143			}
144			flush()
145			return
146		}
147	}
148}
149
150// Close drains what is queued and waits for the last flush, but never longer
151// than closeTimeout.
152func (s *Shipper) Close() {
153	s.stop.Do(func() { close(s.quit) })
154
155	t := time.NewTimer(closeTimeout)
156	defer t.Stop()
157	select {
158	case <-s.done:
159	case <-t.C:
160		// Sink is not answering. Those records are on stdout anyway.
161		fmt.Fprintf(os.Stderr, "log shipping: gave up draining after %s\n", closeTimeout)
162	}
163}
164
165// teeHandler writes through to the real handler and copies to the queue. It
166// cannot embed the next handler, because an embedded WithAttrs or WithGroup
167// returns the inner handler and silently stops teeing.
168type teeHandler struct {
169	next slog.Handler
170	ship *Shipper
171	// Each attribute keeps the group prefix in force when it was added, not
172	// the current one. A flat []slog.Attr would retroactively re-prefix
173	// anything attached before a later WithGroup, so .With(component=crawler)
174	// then .WithGroup("http") would ship "http.component" where slog says it
175	// must stay "component", and the ingest side matches keys by exact name.
176	attrs []groupedAttr
177	group string
178}
179
180// groupedAttr is one attribute frozen with the prefix it was added under.
181type groupedAttr struct {
182	prefix string
183	attr   slog.Attr
184}
185
186func (h *teeHandler) Enabled(ctx context.Context, l slog.Level) bool {
187	return h.next.Enabled(ctx, l)
188}
189
190func (h *teeHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
191	out := &teeHandler{next: h.next.WithAttrs(attrs), ship: h.ship, group: h.group}
192	out.attrs = append([]groupedAttr{}, h.attrs...)
193	for _, a := range attrs {
194		out.attrs = append(out.attrs, groupedAttr{prefix: h.group, attr: a})
195	}
196	return out
197}
198
199func (h *teeHandler) WithGroup(name string) slog.Handler {
200	// An empty name is a no-op, matching slog.Logger.WithGroup. Without it
201	// every key under the group gains a double dot.
202	if name == "" {
203		return h
204	}
205	prefix := name
206	if h.group != "" {
207		prefix = h.group + "." + name
208	}
209	out := &teeHandler{next: h.next.WithGroup(name), ship: h.ship, group: prefix}
210	out.attrs = append([]groupedAttr{}, h.attrs...)
211	return out
212}
213
214func (h *teeHandler) Handle(ctx context.Context, r slog.Record) error {
215	// stdout first and unconditionally.
216	err := h.next.Handle(ctx, r)
217
218	out := Record{
219		Time:  r.Time.UTC().UnixMilli(),
220		Level: r.Level.String(),
221		Msg:   r.Message,
222	}
223	attrs := make(map[string]any, r.NumAttrs()+len(h.attrs))
224	for _, a := range h.attrs {
225		flatten(attrs, a.prefix, a.attr)
226	}
227	r.Attrs(func(a slog.Attr) bool {
228		flatten(attrs, h.group, a)
229		return true
230	})
231	if len(attrs) > 0 {
232		out.Attrs = attrs
233	}
234	h.ship.enqueue(out)
235
236	return err
237}
238
239// flatten writes one attribute into the map, resolving LogValuer and turning a
240// group into dotted keys, so the ingest side can pull "status" out with one
241// lookup.
242func flatten(dst map[string]any, prefix string, a slog.Attr) {
243	a.Value = a.Value.Resolve()
244	key := a.Key
245	if prefix != "" {
246		key = prefix + "." + key
247	}
248
249	switch a.Value.Kind() {
250	case slog.KindGroup:
251		for _, sub := range a.Value.Group() {
252			flatten(dst, key, sub)
253		}
254	case slog.KindTime:
255		dst[key] = a.Value.Time().UTC().Format(time.RFC3339Nano)
256	case slog.KindDuration:
257		dst[key] = a.Value.Duration().String()
258	case slog.KindBool:
259		dst[key] = a.Value.Bool()
260	case slog.KindInt64:
261		dst[key] = a.Value.Int64()
262	case slog.KindUint64:
263		dst[key] = a.Value.Uint64()
264	case slog.KindFloat64:
265		dst[key] = a.Value.Float64()
266	case slog.KindString:
267		dst[key] = a.Value.String()
268	default:
269		// Formatted rather than marshalled, because an arbitrary value can
270		// fail to marshal and one bad attribute must not cost the batch.
271		dst[key] = fmt.Sprint(a.Value.Any())
272	}
273}
274
275// HTTPSink posts batches to the logging site. It never calls slog, because a
276// shipper that logged its own failures would enqueue a record about failing to
277// ship. State changes go straight to stderr instead.
278func HTTPSink() Sink {
279	client := &http.Client{Timeout: shipTimeout}
280	var (
281		mu      sync.Mutex
282		healthy = true
283		dropped int
284	)
285
286	note := func(ok bool, detail string, n int) {
287		mu.Lock()
288		defer mu.Unlock()
289		if ok == healthy {
290			if !ok {
291				dropped += n
292			}
293			return
294		}
295		healthy = ok
296		if ok {
297			fmt.Fprintf(os.Stderr, "log shipping recovered after dropping %d records\n", dropped+n)
298			dropped = 0
299			return
300		}
301		dropped = n
302		fmt.Fprintf(os.Stderr, "log shipping unavailable, dropping records: %s\n", detail)
303	}
304
305	return func(source string, records []Record) {
306		body, err := json.Marshal(Batch{Source: source, Records: records})
307		if err != nil {
308			note(false, err.Error(), len(records))
309			return
310		}
311
312		ctx, cancel := context.WithTimeout(context.Background(), shipTimeout)
313		defer cancel()
314
315		req, err := http.NewRequestWithContext(ctx, http.MethodPost, ShipEndpoint, bytes.NewReader(body))
316		if err != nil {
317			note(false, err.Error(), len(records))
318			return
319		}
320		req.Header.Set("Content-Type", "application/json")
321
322		resp, err := client.Do(req)
323		if err != nil {
324			note(false, err.Error(), len(records))
325			return
326		}
327		defer resp.Body.Close()
328		// Drained as well as closed, or the connection leaks out of the
329		// pool, and this one is reused every five seconds forever.
330		_, _ = io.Copy(io.Discard, resp.Body)
331
332		// 429 is the logging site shedding load, a healthy answer, so
333		// drop the batch without marking the sink down.
334		if resp.StatusCode == http.StatusTooManyRequests {
335			return
336		}
337		if resp.StatusCode >= 300 {
338			note(false, resp.Status, len(records))
339			return
340		}
341		note(true, "", 0)
342	}
343}