orchard

  1package main
  2
  3import (
  4	"context"
  5	"crypto/tls"
  6	"database/sql"
  7	"encoding/json"
  8	"errors"
  9	"fmt"
 10	"log/slog"
 11	"net"
 12	"net/http"
 13	"net/url"
 14	"strconv"
 15	"strings"
 16	"time"
 17
 18	"golang.org/x/net/http2"
 19)
 20
 21// This probe measures what a first-time visitor pays: DNS, TCP, the TLS
 22// handshake and the wait for the first byte, each timed by hand. Do not swap it
 23// for a pooled http.Client; the phase chart exists because nothing is reused.
 24
 25const (
 26	// A real Chrome UA: sites behave differently for something announcing
 27	// itself as a monitor.
 28	probeUserAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " +
 29		"(KHTML, like Gecko) Chrome/102.0.5005.115 Safari/537.36 Status/2.0.0"
 30	probeTimeout = 10 * time.Second
 31	maxRedirects = 5
 32)
 33
 34// PhaseTimings is one probe's breakdown. A nil phase never ran. TotalMS is wall
 35// clock for the first hop and is what goes into checks.response_ms.
 36type PhaseTimings struct {
 37	DNSMS   *int64
 38	TCPMS   *int64
 39	TLSMS   *int64
 40	TTFBMS  *int64
 41	TotalMS int64
 42}
 43
 44type probeOutcome struct {
 45	statusCode  int64
 46	headersJSON string
 47	timings     PhaseTimings
 48
 49	// A 200 from an edge cache says nothing about the origin.
 50	cacheStatus string
 51	age         *int64
 52
 53	// originUnreachable is set when a cache answered and a direct probe of the
 54	// live origin could have left it; see classifyCache.
 55	originUnreachable bool
 56}
 57
 58// hopResult is one request and response, with enough detail to follow a redirect.
 59type hopResult struct {
 60	statusCode  int64
 61	headers     map[string]string
 62	headersJSON string
 63	timings     PhaseTimings
 64}
 65
 66// errPlainHTTP is returned for an http:// URL: the probe is HTTP/2 over TLS
 67// only. Property creation rejects http:// too, so this is the second fence.
 68var errPlainHTTP = errors.New("plain HTTP not supported; use https:// (HTTP/2 only)")
 69
 70// atLeast1ms reports a sub-millisecond phase as 1 rather than 0. Probing a site
 71// that shares this machine goes over lo and truncates to zero, which reads on
 72// the chart as a phase that never happened.
 73func atLeast1ms(d time.Duration) int64 {
 74	ms := d.Milliseconds()
 75	if ms == 0 && d > 0 {
 76		return 1
 77	}
 78	return ms
 79}
 80
 81func ptr(v int64) *int64 { return &v }
 82
 83// parseHTTPURL parses a URL and insists it is absolute and http(s). url.Parse
 84// reads "example.com" as a relative path with no host and returns no error.
 85func parseHTTPURL(raw string) (*url.URL, error) {
 86	u, err := url.Parse(strings.TrimSpace(raw))
 87	if err != nil {
 88		return nil, err
 89	}
 90	if u.Scheme != "http" && u.Scheme != "https" {
 91		return nil, fmt.Errorf("URL scheme is %q, want http or https", u.Scheme)
 92	}
 93	if u.Host == "" {
 94		return nil, errors.New("URL has no host")
 95	}
 96	return u, nil
 97}
 98
 99// looksLikeTLSError picks between 526 for a certificate problem and 408 for a
100// generic timeout. Matching on error text, to avoid unwrapping driver types.
101func looksLikeTLSError(err error) bool {
102	var certErr *tls.CertificateVerificationError
103	if errors.As(err, &certErr) {
104		return true
105	}
106	var recordErr tls.RecordHeaderError
107	if errors.As(err, &recordErr) {
108		return true
109	}
110	s := strings.ToLower(err.Error())
111	for _, needle := range []string{"certificate", "tls", "handshake", "x509"} {
112		if strings.Contains(s, needle) {
113			return true
114		}
115	}
116	return false
117}
118
119// runCheck performs one probe and writes the row. It returns the status code
120// the alert state machine should act on.
121func runCheck(ctx context.Context, db *sql.DB, p *Property) (int64, error) {
122	started := time.Now()
123
124	outcome, err := probeWithRedirects(ctx, p.URL)
125	if err != nil {
126		code := int64(408)
127		if looksLikeTLSError(err) {
128			code = 526
129		}
130		// Record how long the failure took. An NXDOMAIN fails in milliseconds
131		// and charting it as the full timeout would poison the average.
132		outcome = &probeOutcome{
133			statusCode:  code,
134			headersJSON: "{}",
135			timings:     PhaseTimings{TotalMS: time.Since(started).Milliseconds()},
136		}
137	}
138
139	// The stored code is what the monitor concluded, not what the edge sent.
140	// advanceAlertState looks for a second failure by re-reading status_code out
141	// of checks, so a code that is only returned can never trigger a down.
142	effective := outcome.statusCode
143	if outcome.originUnreachable {
144		effective = statusOriginStale
145		slog.Info("origin unreachable behind cache",
146			slog.String("component", "checker"),
147			slog.String("url", p.URL),
148			slog.String("cf_cache_status", outcome.cacheStatus),
149			slog.Int64("age", derefAge(outcome.age)),
150			slog.Int64("edge_status", outcome.statusCode),
151		)
152	}
153
154	_, err = db.ExecContext(ctx,
155		`INSERT INTO checks (property_id, status_code, response_ms, headers,
156		                     dns_ms, tcp_ms, tls_ms, ttfb_ms,
157		                     cf_cache_status, age, created_at)
158		 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
159		p.ID[:], effective, outcome.timings.TotalMS, outcome.headersJSON,
160		outcome.timings.DNSMS, outcome.timings.TCPMS, outcome.timings.TLSMS,
161		outcome.timings.TTFBMS, nullableString(outcome.cacheStatus), outcome.age,
162		nowMS())
163	if err != nil {
164		return 0, fmt.Errorf("insert check: %w", err)
165	}
166	return effective, nil
167}
168
169// statusOriginStale is reported when a cache answered for an origin that has
170// stopped. It is not a real HTTP code; nothing sent one.
171const statusOriginStale = 523
172
173func nullableString(s string) any {
174	if s == "" {
175		return nil
176	}
177	return s
178}
179
180func derefAge(age *int64) int64 {
181	if age == nil {
182		return -1
183	}
184	return *age
185}
186
187// probeWithRedirects times the first hop in full, then follows up to
188// maxRedirects 3xx hops for the final status code, since the alert machine keys
189// on 200 and an apex-to-www property would otherwise sit at 301 forever.
190func probeWithRedirects(ctx context.Context, rawURL string) (*probeOutcome, error) {
191	current, err := parseHTTPURL(rawURL)
192	if err != nil {
193		return nil, err
194	}
195
196	ctx, cancel := context.WithTimeout(ctx, probeTimeout)
197	defer cancel()
198
199	first, err := phasedHop(ctx, current)
200	if err != nil {
201		return nil, err
202	}
203
204	status := first.statusCode
205	headersJSON := first.headersJSON
206	headers := first.headers
207
208	for hops := 0; isRedirect(status) && hops < maxRedirects; hops++ {
209		loc := headers["location"]
210		if loc == "" {
211			break
212		}
213		next, err := current.Parse(loc)
214		if err != nil {
215			break
216		}
217		current = next
218		hop, err := phasedHop(ctx, current)
219		if err != nil {
220			// Keep the 3xx actually observed: the site answered and pointed
221			// somewhere broken.
222			break
223		}
224		status = hop.statusCode
225		headersJSON = hop.headersJSON
226		headers = hop.headers
227	}
228
229	// Only pay for the extra request when the edge answered by itself, which is
230	// the one case where this response says nothing about the origin.
231	cacheStatus, age, cached := classifyCache(headers)
232	unreachable := false
233	if cached {
234		alive, known := originAnswered(ctx, current)
235		unreachable = known && !alive
236	}
237	return &probeOutcome{
238		statusCode:        status,
239		headersJSON:       headersJSON,
240		timings:           first.timings,
241		cacheStatus:       cacheStatus,
242		age:               age,
243		originUnreachable: unreachable,
244	}, nil
245}
246
247// classifyCache reports whether the edge answered out of its own copy. A hit is
248// not evidence either way about the origin, and Age cannot stand in for one: the
249// Edge TTL comes from a Cache Rule the origin never sees, so Age climbs past the
250// origin's own max-age on a site that is perfectly healthy.
251func classifyCache(headers map[string]string) (status string, age *int64, cached bool) {
252	status = headers["cf-cache-status"]
253
254	if raw, ok := headers["age"]; ok {
255		if n, err := strconv.ParseInt(raw, 10, 64); err == nil {
256			age = &n
257		}
258	}
259
260	switch strings.ToUpper(status) {
261	case "HIT", "UPDATING", "STALE":
262		return status, age, true
263	}
264	return status, age, false
265}
266
267// originAnswered asks the origin directly, with a query string nothing has
268// cached and against the one path every site here serves no-store. It reports
269// whether the origin answered and whether the question could be asked at all,
270// so a property with no health endpoint gets no opinion rather than a permanent
271// alarm.
272func originAnswered(ctx context.Context, u *url.URL) (alive, known bool) {
273	probe := *u
274	probe.Path = "/healthz"
275	probe.RawQuery = "cb=" + strconv.FormatInt(time.Now().UnixNano(), 36)
276
277	hop, err := phasedHop(ctx, &probe)
278	if err != nil {
279		return false, true
280	}
281	switch hop.statusCode {
282	case http.StatusOK:
283		return true, true
284	case http.StatusNotFound:
285		return false, false
286	}
287	return false, true
288}
289
290func isRedirect(code int64) bool {
291	switch code {
292	case 301, 302, 303, 307, 308:
293		return true
294	}
295	return false
296}
297
298// phasedHop performs one request, timing each phase separately.
299func phasedHop(ctx context.Context, u *url.URL) (*hopResult, error) {
300	host := u.Hostname()
301	port := u.Port()
302	if port == "" {
303		if u.Scheme == "https" {
304			port = "443"
305		} else {
306			port = "80"
307		}
308	}
309
310	totalStart := time.Now()
311
312	dnsStart := time.Now()
313	addrs, err := net.DefaultResolver.LookupIPAddr(ctx, host)
314	if err != nil {
315		return nil, fmt.Errorf("dns lookup: %w", err)
316	}
317	if len(addrs) == 0 {
318		return nil, fmt.Errorf("no addresses for %s", host)
319	}
320	dnsMS := atLeast1ms(time.Since(dnsStart))
321
322	tcpStart := time.Now()
323	var dialer net.Dialer
324	conn, err := dialer.DialContext(ctx, "tcp", net.JoinHostPort(addrs[0].IP.String(), port))
325	if err != nil {
326		return nil, fmt.Errorf("tcp connect: %w", err)
327	}
328	defer conn.Close()
329	if tcpConn, ok := conn.(*net.TCPConn); ok {
330		// Nagle would fold the request write into the wait being measured.
331		_ = tcpConn.SetNoDelay(true)
332	}
333	tcpMS := atLeast1ms(time.Since(tcpStart))
334
335	if u.Scheme != "https" {
336		return nil, errPlainHTTP
337	}
338
339	// ALPN is pinned to h2 alone, so a server that does not speak HTTP/2 fails
340	// the handshake rather than being measured over HTTP/1.1.
341	tlsStart := time.Now()
342	tlsConn := tls.Client(conn, &tls.Config{
343		ServerName: host,
344		NextProtos: []string{"h2"},
345		MinVersion: tls.VersionTLS12,
346	})
347	if err := tlsConn.HandshakeContext(ctx); err != nil {
348		return nil, fmt.Errorf("tls handshake: %w", err)
349	}
350	if proto := tlsConn.ConnectionState().NegotiatedProtocol; proto != "h2" {
351		return nil, fmt.Errorf("tls handshake: server did not negotiate h2 (got %q)", proto)
352	}
353	tlsMS := atLeast1ms(time.Since(tlsStart))
354
355	status, headers, headersJSON, ttfbMS, err := h2Request(ctx, tlsConn, u)
356	if err != nil {
357		return nil, err
358	}
359
360	return &hopResult{
361		statusCode:  status,
362		headers:     headers,
363		headersJSON: headersJSON,
364		timings: PhaseTimings{
365			DNSMS:   ptr(dnsMS),
366			TCPMS:   ptr(tcpMS),
367			TLSMS:   ptr(tlsMS),
368			TTFBMS:  ptr(ttfbMS),
369			TotalMS: time.Since(totalStart).Milliseconds(),
370		},
371	}, nil
372}
373
374// h2Request runs an HTTP/2 GET over an already-handshaked TLS connection, which
375// http.Client cannot do without dialing itself. TTFB runs from the h2 SETTINGS
376// exchange to the response HEADERS frame, so it includes protocol setup.
377func h2Request(ctx context.Context, conn *tls.Conn, u *url.URL) (
378	status int64, headers map[string]string, headersJSON string, ttfbMS int64, err error,
379) {
380	ttfbStart := time.Now()
381
382	tr := &http2.Transport{}
383	cc, err := tr.NewClientConn(conn)
384	if err != nil {
385		return 0, nil, "", 0, fmt.Errorf("h2 handshake: %w", err)
386	}
387	defer cc.Close()
388
389	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
390	if err != nil {
391		return 0, nil, "", 0, fmt.Errorf("h2 request build: %w", err)
392	}
393	req.Header.Set("user-agent", probeUserAgent)
394	req.Header.Set("accept", "*/*")
395
396	resp, err := cc.RoundTrip(req)
397	if err != nil {
398		return 0, nil, "", 0, fmt.Errorf("h2 response: %w", err)
399	}
400	ttfbMS = atLeast1ms(time.Since(ttfbStart))
401
402	// The body is never read: this measures time to first byte. Closing without
403	// reading sends RST_STREAM, which is correct and cheap.
404	defer resp.Body.Close()
405
406	headers = make(map[string]string, len(resp.Header))
407	for k, v := range resp.Header {
408		if len(v) > 0 {
409			headers[strings.ToLower(k)] = v[0]
410		}
411	}
412
413	// encoding/json sorts map keys, so the stored blob is byte-stable across
414	// probes rather than reshuffled by map iteration order.
415	encoded, err := json.Marshal(headers)
416	if err != nil {
417		encoded = []byte("{}")
418	}
419
420	return int64(resp.StatusCode), headers, string(encoded), ttfbMS, nil
421}
422
423// processCheck runs a probe, stores it, then advances the alert state machine.
424func processCheck(ctx context.Context, db *sql.DB, notifier *Notifier, p *Property) error {
425	status, err := runCheck(ctx, db, p)
426	if err != nil {
427		return err
428	}
429	return advanceAlertState(ctx, db, notifier, p, status)
430}
431
432// advanceAlertState is the debounce that decides when to wake somebody up.
433//
434//	up   -> down: two consecutive non-200 checks
435//	down -> up:   immediately, on any 200
436//
437// The state commits before the notification is sent, so a crash between the two
438// loses an alert rather than repeating one forever.
439func advanceAlertState(ctx context.Context, db *sql.DB, notifier *Notifier, p *Property, statusCode int64) error {
440	isUp := statusCode == 200
441
442	tx, err := db.BeginTx(ctx, nil)
443	if err != nil {
444		return err
445	}
446	defer func() { _ = tx.Rollback() }()
447
448	var currentState string
449	err = tx.QueryRowContext(ctx,
450		"SELECT alert_state FROM properties WHERE id = ?", p.ID[:]).Scan(&currentState)
451	if err == sql.ErrNoRows {
452		// Deleted between the probe being scheduled and finishing.
453		return nil
454	}
455	if err != nil {
456		return err
457	}
458
459	transition := ""
460	switch {
461	case isUp && currentState == "down":
462		transition = "recovery"
463	case !isUp && currentState == "up":
464		// The check just inserted is one of the two; look for a second.
465		rows, err := tx.QueryContext(ctx,
466			"SELECT status_code FROM checks WHERE property_id = ? ORDER BY created_at DESC LIMIT 2",
467			p.ID[:])
468		if err != nil {
469			return err
470		}
471		var recent []int64
472		for rows.Next() {
473			var code int64
474			if err := rows.Scan(&code); err != nil {
475				rows.Close()
476				return err
477			}
478			recent = append(recent, code)
479		}
480		rows.Close()
481		if err := rows.Err(); err != nil {
482			return err
483		}
484		if len(recent) >= 2 && recent[0] != 200 && recent[1] != 200 {
485			transition = "down"
486		}
487	}
488
489	if transition == "" {
490		return tx.Commit()
491	}
492
493	newState := "down"
494	if transition == "recovery" {
495		newState = "up"
496	}
497	now := nowMS()
498	if _, err := tx.ExecContext(ctx,
499		"UPDATE properties SET alert_state = ?, last_alert_sent = ?, updated_at = ? WHERE id = ?",
500		newState, now, now, p.ID[:]); err != nil {
501		return err
502	}
503	if err := tx.Commit(); err != nil {
504		return err
505	}
506
507	avg, err := recentAvgResponseMS(ctx, db, p.ID)
508	if err != nil {
509		slog.Info(fmt.Sprintf("alert: average response time for %s: %v", p.URL, err))
510	}
511
512	// Fire and forget: a slow notifier must not hold up the scheduler tick.
513	go notifier.Fire(transition, AlertContext{
514		ID:            p.ID.String(),
515		Name:          p.Name(),
516		URL:           p.URL,
517		CurrentStatus: statusCode,
518		AvgResponseMS: avg,
519	})
520	return nil
521}
522
523// recentAvgResponseMS averages the most recent 31 checks, mirroring the
524// dashboard tile so the alert quotes the number the operator will see.
525func recentAvgResponseMS(ctx context.Context, db *sql.DB, id [16]byte) (int64, error) {
526	var avg sql.NullFloat64
527	err := db.QueryRowContext(ctx,
528		`SELECT AVG(response_ms) FROM (
529		   SELECT response_ms FROM checks WHERE property_id = ?
530		   ORDER BY created_at DESC LIMIT 31
531		 )`, id[:]).Scan(&avg)
532	if err != nil || !avg.Valid {
533		return 0, err
534	}
535	return int64(avg.Float64), nil
536}
537
538// next3MinBoundary returns the next wall-clock time divisible by three minutes,
539// so every property shares one cadence and two charts line up on one x-axis.
540func next3MinBoundary() int64 {
541	now := time.Now().UTC()
542	aligned := now.Truncate(checkInterval)
543	return aligned.Add(checkInterval).UnixMilli()
544}