Sandbar/internal/sandbar/monitor.go

package sandbar

import (
	"context"
	"crypto/tls"
	"database/sql"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"

	"git.veillax.com/Veillax/reef/netguard"
)

// Monitors check that things are up. A check runs every interval; the monitor
// goes down after Retries failures in a row (so a blip is ignored) and raises an
// alert, and recovers on the first success. Heartbeat monitors work the other
// way round: your cron job or service calls /hb/<token>, and the monitor goes
// down when the calls stop.

type Monitor struct {
	ID           string `json:"id"`
	Name         string `json:"name"`
	Kind         string `json:"kind"`   // http, tcp, dns, tls, heartbeat
	Target       string `json:"target"` // URL, host:port, hostname, or host[:port] for tls
	IntervalS    int    `json:"intervalS"`
	TimeoutS     int    `json:"timeoutS"`
	Retries      int    `json:"retries"`  // failures in a row before it is down (default 2)
	Expect       string `json:"expect"`   // http: accepted status, e.g. "200-299" (default)
	Keyword      string `json:"keyword"`  // http: text the body must contain
	WarnDays     int    `json:"warnDays"` // tls: fail when the certificate expires within this many days (default 14)
	GraceS       int    `json:"graceS"`   // heartbeat: extra time allowed past the interval
	Token        string `json:"token"`    // heartbeat: the secret in the ping URL
	PolicyID     string `json:"policyId"` // escalation policy for the alert it raises
	Severity     string `json:"severity"` // alert severity (default critical)
	Public       bool   `json:"public"`   // shown on the status page
	AllowPrivate bool   `json:"allowPrivate"`
	Enabled      bool   `json:"enabled"`
}

// MonitorState is what we know about a monitor right now.
type MonitorState struct {
	MonitorID string  `json:"monitorId"`
	Status    string  `json:"status"` // pending, up, down
	Since     int64   `json:"since"`
	LastCheck int64   `json:"lastCheck"`
	LastMS    int     `json:"lastMs"`
	Fails     int     `json:"fails"`
	LastPing  int64   `json:"lastPing"`
	Detail    string  `json:"detail"`
	Uptime30  float64 `json:"uptime30"` // percent of checks that passed in the retention window; -1 if none
}

// interval is how often the monitor is checked. A heartbeat is evaluated often so
// a missed one is noticed soon after its deadline, not up to a whole interval later.
func (m *Monitor) interval() time.Duration {
	if m.Kind == "heartbeat" {
		return 15 * time.Second
	}
	if m.IntervalS < 10 {
		return time.Minute
	}
	return time.Duration(m.IntervalS) * time.Second
}

func (m *Monitor) timeout() time.Duration {
	if m.TimeoutS <= 0 || m.TimeoutS > 120 {
		return 10 * time.Second
	}
	return time.Duration(m.TimeoutS) * time.Second
}

func (m *Monitor) retries() int {
	if m.Retries < 1 {
		return 2
	}
	return m.Retries
}

func validateMonitor(m *Monitor) error {
	m.Name, m.Target = strings.TrimSpace(m.Name), strings.TrimSpace(m.Target)
	if m.Name == "" {
		return errors.New("give the monitor a name")
	}
	switch m.Kind {
	case "http":
		u, err := url.Parse(m.Target)
		if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
			return errors.New("an http monitor needs an http:// or https:// URL")
		}
	case "tcp":
		if _, _, err := net.SplitHostPort(m.Target); err != nil {
			return errors.New("a tcp monitor needs host:port")
		}
	case "dns", "tls":
		if m.Target == "" || strings.ContainsAny(m.Target, " /\\") {
			return errors.New("enter a hostname")
		}
	case "heartbeat":
		if m.IntervalS < 10 {
			return errors.New("a heartbeat interval is at least 10 seconds")
		}
		if m.Token == "" {
			m.Token = newToken()
		}
	default:
		return errors.New("kind must be http, tcp, dns, tls or heartbeat")
	}
	if m.Severity == "" {
		m.Severity = "critical"
	}
	if _, _, err := parseRange(m.Expect); err != nil {
		return err
	}
	return nil
}

// parseRange reads "200-299", "204" or "" (= 200-299).
func parseRange(s string) (lo, hi int, err error) {
	s = strings.TrimSpace(s)
	if s == "" {
		return 200, 299, nil
	}
	a, b, found := strings.Cut(s, "-")
	if lo, err = strconv.Atoi(strings.TrimSpace(a)); err != nil {
		return 0, 0, errors.New(`expected status must look like "200-299" or "204"`)
	}
	hi = lo
	if found {
		if hi, err = strconv.Atoi(strings.TrimSpace(b)); err != nil || hi < lo {
			return 0, 0, errors.New(`expected status must look like "200-299" or "204"`)
		}
	}
	return lo, hi, nil
}

// ---------- the checks ----------

// runCheck performs one probe and returns how long it took and an error if it failed.
func (s *Server) runCheck(ctx context.Context, m *Monitor) (time.Duration, error) {
	ctx, cancel := context.WithTimeout(ctx, m.timeout())
	defer cancel()
	start := time.Now()
	var err error
	switch m.Kind {
	case "http":
		err = s.checkHTTP(ctx, m)
	case "tcp":
		var c net.Conn
		if c, err = netguard.DialContext(m.AllowPrivate)(ctx, "tcp", m.Target); err == nil {
			c.Close()
		}
	case "dns":
		var addrs []string
		if addrs, err = net.DefaultResolver.LookupHost(ctx, m.Target); err == nil && len(addrs) == 0 {
			err = errors.New("no addresses")
		}
	case "tls":
		err = checkTLS(ctx, m, s.now())
	}
	return time.Since(start), err
}

func (s *Server) checkHTTP(ctx context.Context, m *Monitor) error {
	req, err := http.NewRequestWithContext(ctx, "GET", m.Target, nil)
	if err != nil {
		return err
	}
	req.Header.Set("User-Agent", "Sandbar-Monitor/1")
	c := *s.httpClient(m.AllowPrivate)
	c.Timeout = 0 // the context carries the deadline
	c.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }
	resp, err := c.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	lo, hi, _ := parseRange(m.Expect)
	if resp.StatusCode < lo || resp.StatusCode > hi {
		return fmt.Errorf("status %d, expected %d-%d", resp.StatusCode, lo, hi)
	}
	if m.Keyword != "" {
		body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
		if !strings.Contains(string(body), m.Keyword) {
			return fmt.Errorf("the page doesn't contain %q", m.Keyword)
		}
	}
	return nil
}

func checkTLS(ctx context.Context, m *Monitor, now time.Time) error {
	host, port, err := net.SplitHostPort(m.Target)
	if err != nil {
		host, port = m.Target, "443"
	}
	d := tls.Dialer{Config: &tls.Config{ServerName: host}}
	conn, err := d.DialContext(ctx, "tcp", net.JoinHostPort(host, port))
	if err != nil {
		return err
	}
	defer conn.Close()
	certs := conn.(*tls.Conn).ConnectionState().PeerCertificates
	if len(certs) == 0 {
		return errors.New("no certificate")
	}
	warn := m.WarnDays
	if warn <= 0 {
		warn = 14
	}
	left := certs[0].NotAfter.Sub(now)
	if left < time.Duration(warn)*24*time.Hour {
		return fmt.Errorf("the certificate expires in %d days (%s)", int(left.Hours()/24), certs[0].NotAfter.Format("2006-01-02"))
	}
	return nil
}

// ---------- scheduling and state ----------

func (s *Server) monitorState(id string) MonitorState {
	st := MonitorState{MonitorID: id, Status: "pending", Uptime30: -1}
	s.st.db.QueryRow(`SELECT status,since,last_check,last_ms,fails,last_ping,detail FROM mstate WHERE monitor_id=?`, id).
		Scan(&st.Status, &st.Since, &st.LastCheck, &st.LastMS, &st.Fails, &st.LastPing, &st.Detail)
	var total, ok sql.NullInt64
	s.st.db.QueryRow(`SELECT COUNT(*), SUM(ok) FROM checks WHERE monitor_id=?`, id).Scan(&total, &ok)
	if total.Int64 > 0 {
		st.Uptime30 = float64(ok.Int64) * 100 / float64(total.Int64)
	}
	return st
}

func (s *Server) saveState(st MonitorState) {
	s.st.db.Exec(`INSERT INTO mstate(monitor_id,status,since,last_check,last_ms,fails,last_ping,detail) VALUES(?,?,?,?,?,?,?,?)
		ON CONFLICT(monitor_id) DO UPDATE SET status=excluded.status, since=excluded.since, last_check=excluded.last_check,
		last_ms=excluded.last_ms, fails=excluded.fails, last_ping=excluded.last_ping, detail=excluded.detail`,
		st.MonitorID, st.Status, st.Since, st.LastCheck, st.LastMS, st.Fails, st.LastPing, st.Detail)
}

// runDueMonitors starts a check for every enabled monitor whose interval has elapsed.
func (s *Server) runDueMonitors(now time.Time) {
	mons, err := listOf[Monitor](s.st, "monitor")
	if err != nil {
		return
	}
	for i := range mons {
		m := mons[i]
		if !m.Enabled {
			continue
		}
		st := s.monitorState(m.ID)
		if now.Sub(time.Unix(st.LastCheck, 0)) < m.interval() {
			continue
		}
		s.mu.Lock()
		busy := s.running[m.ID]
		s.running[m.ID] = true
		s.mu.Unlock()
		if busy {
			continue
		}
		go func() {
			defer func() {
				s.mu.Lock()
				delete(s.running, m.ID)
				s.mu.Unlock()
			}()
			s.checkMonitor(&m, now)
		}()
	}
}

// checkMonitor runs one probe (or evaluates a heartbeat) and moves state forward.
func (s *Server) checkMonitor(m *Monitor, now time.Time) {
	st := s.monitorState(m.ID)
	var dur time.Duration
	var err error
	if m.Kind == "heartbeat" {
		if st.LastPing == 0 {
			// Never pinged: give it a full interval from when we first looked.
			if st.LastCheck == 0 {
				st.LastCheck, st.Status = now.Unix(), "pending"
				st.Since = now.Unix()
				s.saveState(st)
				return
			}
			if now.Unix()-st.Since < int64(m.IntervalS+m.GraceS) {
				st.LastCheck = now.Unix() // still within the first interval: nothing to judge yet
				s.saveState(st)
				return
			}
			err = errors.New("no heartbeat received yet")
		} else if now.Unix()-st.LastPing > int64(m.IntervalS+m.GraceS) {
			err = fmt.Errorf("no heartbeat for %s", now.Sub(time.Unix(st.LastPing, 0)).Round(time.Second))
		}
	} else {
		dur, err = s.runCheck(context.Background(), m)
	}
	s.recordResult(m, st, now, dur, err)
}

func (s *Server) recordResult(m *Monitor, st MonitorState, now time.Time, dur time.Duration, err error) {
	ok := err == nil
	okInt := 0
	if ok {
		okInt = 1
	}
	s.st.db.Exec(`INSERT INTO checks(monitor_id,time,ok,ms,detail) VALUES(?,?,?,?,?)`, m.ID, now.Unix(), okInt, dur.Milliseconds(), errText(err))
	st.LastCheck, st.LastMS, st.Detail = now.Unix(), int(dur.Milliseconds()), errText(err)
	key := "monitor:" + m.ID
	if ok {
		wasDown := st.Status == "down"
		if st.Status != "up" {
			st.Status, st.Since = "up", now.Unix()
		}
		st.Fails = 0
		s.saveState(st)
		if wasDown {
			s.ResolveAlertKey(key, "monitor")
		}
		return
	}
	st.Fails++
	if st.Fails >= m.retries() && st.Status != "down" {
		st.Status, st.Since = "down", now.Unix()
	}
	s.saveState(st)
	if st.Status == "down" {
		s.RaiseAlert(key, "monitor", m.Name+" is down", m.Name+" ("+m.Kind+" "+m.Target+"): "+st.Detail, m.Severity, m.PolicyID)
	}
}

func errText(err error) string {
	if err == nil {
		return ""
	}
	return err.Error()
}

// Heartbeat records a ping for the monitor with this token. It returns false if there is none.
func (s *Server) Heartbeat(token string) bool {
	if token == "" {
		return false
	}
	mons, _ := listOf[Monitor](s.st, "monitor")
	for i := range mons {
		m := mons[i]
		if m.Kind == "heartbeat" && m.Token == token && m.Enabled {
			st := s.monitorState(m.ID)
			st.LastPing = s.now().Unix()
			if st.Since == 0 {
				st.Since = st.LastPing
			}
			s.saveState(st)
			// A ping on a down monitor recovers it at once rather than at the next check.
			if st.Status == "down" {
				s.recordResult(&m, st, s.now(), 0, nil)
			}
			return true
		}
	}
	return false
}

// deleteMonitorData drops what we keep about a monitor and closes its alert.
func (s *Server) deleteMonitorData(id string) {
	s.st.db.Exec(`DELETE FROM mstate WHERE monitor_id=?`, id)
	s.st.db.Exec(`DELETE FROM checks WHERE monitor_id=?`, id)
	s.ResolveAlertKey("monitor:"+id, "monitor deleted")
}

func (s *Server) recentChecks(id string, limit int) []map[string]any {
	out := []map[string]any{}
	rows, err := s.st.db.Query(`SELECT time,ok,ms,detail FROM checks WHERE monitor_id=? ORDER BY time DESC LIMIT ?`, id, limit)
	if err != nil {
		return out
	}
	defer rows.Close()
	for rows.Next() {
		var t, ok, ms int64
		var d string
		if rows.Scan(&t, &ok, &ms, &d) == nil {
			out = append(out, map[string]any{"time": t, "ok": ok == 1, "ms": ms, "detail": d})
		}
	}
	return out
}