package monitor import ( "context" "crypto/tls" "database/sql" "encoding/json" "errors" "fmt" "io" "net" "net/http" "net/url" "os/exec" "strconv" "strings" "sync" "time" "git.send.nrw/sendnrw/dockwatch/internal/buildinfo" "git.send.nrw/sendnrw/dockwatch/internal/nodes" ) type Monitor struct { ID int64 `json:"id"` Name string `json:"name"` Type string `json:"type"` Target string `json:"target"` NodeID *int64 `json:"node_id,omitempty"` ServiceID *int64 `json:"service_id,omitempty"` IntervalSeconds int `json:"interval_seconds"` TimeoutMS int `json:"timeout_ms"` ExpectedMin int `json:"expected_min"` ExpectedMax int `json:"expected_max"` Method string `json:"method"` HeadersJSON string `json:"headers_json"` Body string `json:"body"` Keyword string `json:"keyword"` InvertKeyword bool `json:"invert_keyword"` IgnoreTLS bool `json:"ignore_tls"` RequireHealthy bool `json:"require_healthy"` Enabled bool `json:"enabled"` Status string `json:"status"` MaintenanceUntil *int64 `json:"maintenance_until,omitempty"` MaintenanceNote string `json:"maintenance_note"` LastCheckedAt *int64 `json:"last_checked_at,omitempty"` CreatedAt int64 `json:"created_at"` UpdatedAt int64 `json:"updated_at"` Uptime24h float64 `json:"uptime_24h"` LastLatencyMS int64 `json:"last_latency_ms"` LastMessage string `json:"last_message"` LastStatusCode int `json:"last_status_code"` } type Input struct { Name string `json:"name"` Type string `json:"type"` Target string `json:"target"` NodeID *int64 `json:"node_id"` ServiceID *int64 `json:"service_id"` IntervalSeconds int `json:"interval_seconds"` TimeoutMS int `json:"timeout_ms"` ExpectedMin int `json:"expected_min"` ExpectedMax int `json:"expected_max"` Method string `json:"method"` HeadersJSON string `json:"headers_json"` Body string `json:"body"` Keyword string `json:"keyword"` InvertKeyword bool `json:"invert_keyword"` IgnoreTLS bool `json:"ignore_tls"` RequireHealthy bool `json:"require_healthy"` Enabled *bool `json:"enabled"` } type MaintenanceInput struct { Until *int64 `json:"until"` Note string `json:"note"` } type Check struct { ID int64 `json:"id,omitempty"` MonitorID int64 `json:"monitor_id,omitempty"` OK bool `json:"ok"` StatusCode int `json:"status_code"` LatencyMS int64 `json:"latency_ms"` Message string `json:"message"` CheckedAt int64 `json:"checked_at"` } type Event struct { MonitorID int64 `json:"monitor_id"` Name string `json:"name"` Target string `json:"target"` From string `json:"from"` To string `json:"to"` Check Check `json:"check"` } type Service struct { db *sql.DB nodes *nodes.Manager workers chan struct{} mu sync.Mutex running map[int64]bool retentionDays int eventSink func(context.Context, Event) } func New(db *sql.DB, nm *nodes.Manager, c, r int) *Service { return &Service{db: db, nodes: nm, workers: make(chan struct{}, c), running: map[int64]bool{}, retentionDays: r} } func (s *Service) SetEventSink(fn func(context.Context, Event)) { s.eventSink = fn } const selectMonitor = `WITH stats AS ( SELECT monitor_id,100.0*AVG(ok) AS uptime_24h FROM monitor_checks WHERE checked_at>=? GROUP BY monitor_id ), latest AS ( SELECT monitor_id,MAX(id) AS id FROM monitor_checks GROUP BY monitor_id ) SELECT m.id,m.name,m.type,m.target,m.node_id,m.service_id,m.interval_seconds,m.timeout_ms,m.expected_min,m.expected_max,m.method,m.headers_json,m.body,m.keyword,m.invert_keyword,m.ignore_tls,m.require_healthy,m.enabled,m.status,m.maintenance_until,m.maintenance_note,m.last_checked_at,m.created_at,m.updated_at, COALESCE(stats.uptime_24h,0),COALESCE(c.latency_ms,0),COALESCE(c.message,''),COALESCE(c.status_code,0) FROM monitors m LEFT JOIN stats ON stats.monitor_id=m.id LEFT JOIN latest ON latest.monitor_id=m.id LEFT JOIN monitor_checks c ON c.id=latest.id` func scanMonitor(sc interface{ Scan(...any) error }) (Monitor, error) { var m Monitor var node, serviceID, last, maint sql.NullInt64 err := sc.Scan(&m.ID, &m.Name, &m.Type, &m.Target, &node, &serviceID, &m.IntervalSeconds, &m.TimeoutMS, &m.ExpectedMin, &m.ExpectedMax, &m.Method, &m.HeadersJSON, &m.Body, &m.Keyword, &m.InvertKeyword, &m.IgnoreTLS, &m.RequireHealthy, &m.Enabled, &m.Status, &maint, &m.MaintenanceNote, &last, &m.CreatedAt, &m.UpdatedAt, &m.Uptime24h, &m.LastLatencyMS, &m.LastMessage, &m.LastStatusCode) if node.Valid { m.NodeID = &node.Int64 } if serviceID.Valid { m.ServiceID = &serviceID.Int64 } if last.Valid { m.LastCheckedAt = &last.Int64 } if maint.Valid { m.MaintenanceUntil = &maint.Int64 } return m, err } func (s *Service) List(ctx context.Context) ([]Monitor, error) { rows, e := s.db.QueryContext(ctx, selectMonitor+` ORDER BY m.name`, time.Now().Add(-24*time.Hour).Unix()) if e != nil { return nil, e } defer rows.Close() out := []Monitor{} for rows.Next() { m, e := scanMonitor(rows) if e != nil { return nil, e } out = append(out, m) } return out, rows.Err() } func (s *Service) Get(ctx context.Context, id int64) (Monitor, error) { return scanMonitor(s.db.QueryRowContext(ctx, selectMonitor+` WHERE m.id=?`, time.Now().Add(-24*time.Hour).Unix(), id)) } const selectMonitorSchedule = `SELECT m.id,m.name,m.type,m.target,m.node_id,m.service_id,m.interval_seconds,m.timeout_ms,m.expected_min,m.expected_max,m.method,m.headers_json,m.body,m.keyword,m.invert_keyword,m.ignore_tls,m.require_healthy,m.enabled,m.status,m.maintenance_until,m.maintenance_note,m.last_checked_at,m.created_at,m.updated_at,0.0,0,'',0 FROM monitors m ORDER BY m.id` func (s *Service) listForSchedule(ctx context.Context) ([]Monitor, error) { rows, err := s.db.QueryContext(ctx, selectMonitorSchedule) if err != nil { return nil, err } defer rows.Close() out := []Monitor{} for rows.Next() { m, err := scanMonitor(rows) if err != nil { return nil, err } out = append(out, m) } return out, rows.Err() } func normalize(in *Input, requireName bool) error { in.Name = strings.TrimSpace(in.Name) in.Type = strings.ToLower(strings.TrimSpace(in.Type)) in.Target = strings.TrimSpace(in.Target) if in.Target == "" { return errors.New("target required") } if requireName && in.Name == "" { return errors.New("name required") } if strings.ContainsAny(in.Name, "\r\n") { return errors.New("invalid monitor name") } if len(in.Name) > 200 || len(in.Target) > 4096 { return errors.New("name or target too long") } if in.Type != "http" && in.Type != "tcp" && in.Type != "dns" && in.Type != "docker" { return errors.New("type must be http, tcp, dns or docker") } if in.NodeID != nil && *in.NodeID < 1 { return errors.New("node_id must be positive") } if in.ServiceID != nil && *in.ServiceID < 1 { return errors.New("service_id must be positive") } if in.IntervalSeconds == 0 { in.IntervalSeconds = 60 } if in.IntervalSeconds < 10 || in.IntervalSeconds > 86400 { return errors.New("interval_seconds must be 10..86400") } if in.TimeoutMS == 0 { in.TimeoutMS = 5000 } if in.TimeoutMS < 100 || in.TimeoutMS > 60000 { return errors.New("timeout_ms must be 100..60000") } if len(in.HeadersJSON) > 64<<10 || len(in.Body) > 256<<10 || len(in.Keyword) > 4096 { return errors.New("monitor headers/body/keyword too large") } in.Method = strings.ToUpper(strings.TrimSpace(in.Method)) if in.Method == "" { in.Method = "GET" } if in.HeadersJSON == "" { in.HeadersJSON = "{}" } var h map[string]string if err := json.Unmarshal([]byte(in.HeadersJSON), &h); err != nil { return errors.New("headers_json must be a JSON object with string values") } for k, v := range h { if strings.TrimSpace(k) == "" || strings.ContainsAny(k, "\r\n") || strings.ContainsAny(v, "\r\n") { return errors.New("HTTP headers must not contain empty names or newlines") } } switch in.Type { case "http": u, err := url.Parse(in.Target) if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" || u.User != nil { return errors.New("HTTP target must be an absolute http(s) URL without embedded credentials") } allowed := map[string]bool{"GET": true, "HEAD": true, "POST": true, "PUT": true, "PATCH": true, "DELETE": true, "OPTIONS": true} if !allowed[in.Method] { return errors.New("unsupported HTTP method") } if in.ExpectedMin == 0 { in.ExpectedMin = 200 } if in.ExpectedMax == 0 { in.ExpectedMax = 399 } if in.ExpectedMin < 100 || in.ExpectedMax > 599 || in.ExpectedMin > in.ExpectedMax { return errors.New("expected HTTP status range must be within 100..599") } case "tcp": host, port, err := net.SplitHostPort(in.Target) if err != nil || strings.TrimSpace(host) == "" || strings.TrimSpace(port) == "" { return errors.New("TCP target must be host:port") } if p, err := strconv.Atoi(port); err != nil || p < 1 || p > 65535 { return errors.New("TCP target port must be 1..65535") } case "dns": if strings.ContainsAny(in.Target, " /\\") { return errors.New("DNS target must be a hostname or IP address") } case "docker": if len(in.Target) > 255 || strings.ContainsAny(in.Target, "\r\n") || strings.HasPrefix(in.Target, "-") { return errors.New("invalid Docker container target") } } return nil } func (s *Service) Create(ctx context.Context, in Input, userID int64) (Monitor, error) { if e := normalize(&in, true); e != nil { return Monitor{}, e } enabled := true if in.Enabled != nil { enabled = *in.Enabled } now := time.Now().Unix() res, e := s.db.ExecContext(ctx, `INSERT INTO monitors(name,type,target,node_id,service_id,interval_seconds,timeout_ms,expected_min,expected_max,method,headers_json,body,keyword,invert_keyword,ignore_tls,require_healthy,enabled,status,created_by,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,'pending',?,?,?)`, in.Name, in.Type, in.Target, in.NodeID, in.ServiceID, in.IntervalSeconds, in.TimeoutMS, in.ExpectedMin, in.ExpectedMax, in.Method, in.HeadersJSON, in.Body, in.Keyword, in.InvertKeyword, in.IgnoreTLS, in.RequireHealthy, enabled, userID, now, now) if e != nil { return Monitor{}, e } id, _ := res.LastInsertId() return s.Get(ctx, id) } func (s *Service) Update(ctx context.Context, id int64, in Input) (Monitor, error) { if e := normalize(&in, true); e != nil { return Monitor{}, e } enabled := true if in.Enabled != nil { enabled = *in.Enabled } now := time.Now().Unix() res, e := s.db.ExecContext(ctx, `UPDATE monitors SET name=?,type=?,target=?,node_id=?,service_id=?,interval_seconds=?,timeout_ms=?,expected_min=?,expected_max=?,method=?,headers_json=?,body=?,keyword=?,invert_keyword=?,ignore_tls=?,require_healthy=?,enabled=?,updated_at=? WHERE id=?`, in.Name, in.Type, in.Target, in.NodeID, in.ServiceID, in.IntervalSeconds, in.TimeoutMS, in.ExpectedMin, in.ExpectedMax, in.Method, in.HeadersJSON, in.Body, in.Keyword, in.InvertKeyword, in.IgnoreTLS, in.RequireHealthy, enabled, now, id) if e != nil { return Monitor{}, e } n, _ := res.RowsAffected() if n == 0 { return Monitor{}, sql.ErrNoRows } if !enabled { _, _ = s.db.ExecContext(ctx, `UPDATE monitors SET status='paused' WHERE id=?`, id) } else { _, _ = s.db.ExecContext(ctx, `UPDATE monitors SET status=CASE WHEN status='paused' THEN 'pending' ELSE status END WHERE id=?`, id) } return s.Get(ctx, id) } func (s *Service) SetPaused(ctx context.Context, id int64, paused bool) error { enabled := !paused status := "pending" if paused { status = "paused" } res, e := s.db.ExecContext(ctx, `UPDATE monitors SET enabled=?,status=?,updated_at=? WHERE id=?`, enabled, status, time.Now().Unix(), id) if e != nil { return e } n, _ := res.RowsAffected() if n == 0 { return sql.ErrNoRows } return nil } func (s *Service) SetMaintenance(ctx context.Context, id int64, in MaintenanceInput) error { now := time.Now().Unix() if in.Until != nil && *in.Until <= now { return errors.New("maintenance end must be in the future") } in.Note = strings.TrimSpace(in.Note) if len(in.Note) > 2000 { return errors.New("maintenance note too long") } res, e := s.db.ExecContext(ctx, `UPDATE monitors SET maintenance_until=?,maintenance_note=?,status='maintenance',updated_at=? WHERE id=?`, in.Until, in.Note, now, id) if e != nil { return e } n, _ := res.RowsAffected() if n == 0 { return sql.ErrNoRows } return nil } func (s *Service) ClearMaintenance(ctx context.Context, id int64) error { r, e := s.db.ExecContext(ctx, `UPDATE monitors SET maintenance_until=NULL,maintenance_note='',status=CASE WHEN enabled=1 THEN 'pending' ELSE 'paused' END,updated_at=? WHERE id=?`, time.Now().Unix(), id) if e != nil { return e } n, _ := r.RowsAffected() if n == 0 { return sql.ErrNoRows } return nil } func (s *Service) Delete(ctx context.Context, id int64) error { _, e := s.db.ExecContext(ctx, `DELETE FROM monitors WHERE id=?`, id) return e } func (s *Service) Checks(ctx context.Context, id int64, limit int) ([]Check, error) { if limit < 1 { limit = 100 } if limit > 1000 { limit = 1000 } rows, e := s.db.QueryContext(ctx, `SELECT id,monitor_id,ok,status_code,latency_ms,message,checked_at FROM monitor_checks WHERE monitor_id=? ORDER BY checked_at DESC LIMIT ?`, id, limit) if e != nil { return nil, e } defer rows.Close() out := []Check{} for rows.Next() { var c Check if e := rows.Scan(&c.ID, &c.MonitorID, &c.OK, &c.StatusCode, &c.LatencyMS, &c.Message, &c.CheckedAt); e != nil { return nil, e } out = append(out, c) } return out, rows.Err() } func (s *Service) Run(ctx context.Context) { tick := time.NewTicker(2 * time.Second) cleanup := time.NewTicker(6 * time.Hour) defer tick.Stop() defer cleanup.Stop() for { select { case <-ctx.Done(): return case <-tick.C: s.schedule(ctx) case <-cleanup.C: s.cleanup(ctx) } } } func (s *Service) schedule(ctx context.Context) { ms, e := s.listForSchedule(ctx) if e != nil { return } now := time.Now().Unix() for _, m := range ms { if !m.Enabled { continue } if m.Status == "maintenance" && m.MaintenanceUntil == nil { continue } if m.MaintenanceUntil != nil { if *m.MaintenanceUntil == 0 || *m.MaintenanceUntil > now { if m.Status != "maintenance" { _, _ = s.db.ExecContext(ctx, `UPDATE monitors SET status='maintenance' WHERE id=?`, m.ID) } continue } _ = s.ClearMaintenance(ctx, m.ID) } if m.LastCheckedAt != nil && now-*m.LastCheckedAt < int64(m.IntervalSeconds) { continue } s.mu.Lock() if s.running[m.ID] { s.mu.Unlock() continue } s.running[m.ID] = true s.mu.Unlock() go func(mon Monitor) { select { case s.workers <- struct{}{}: case <-ctx.Done(): s.mu.Lock() delete(s.running, mon.ID) s.mu.Unlock() return } defer func() { <-s.workers; s.mu.Lock(); delete(s.running, mon.ID); s.mu.Unlock() }() c := s.probe(ctx, mon) if ctx.Err() != nil { return } x, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() _ = s.recordCheck(x, mon, c) }(m) } } func (s *Service) CheckNow(ctx context.Context, id int64) (Check, error) { s.mu.Lock() if s.running[id] { s.mu.Unlock() return Check{}, errors.New("monitor check already running") } s.running[id] = true s.mu.Unlock() defer func() { s.mu.Lock() delete(s.running, id) s.mu.Unlock() }() m, err := s.Get(ctx, id) if err != nil { return Check{}, err } c := s.probe(ctx, m) if err := ctx.Err(); err != nil { return c, err } // A manual diagnostic check must not implicitly resume a paused monitor or // end maintenance. We still keep the check in history, but preserve the // lifecycle state until the user explicitly changes it. updateState := m.Enabled && m.Status != "maintenance" if err = s.recordCheckResult(ctx, m, c, updateState); err != nil { return c, err } return c, nil } func (s *Service) recordCheck(ctx context.Context, m Monitor, c Check) error { return s.recordCheckResult(ctx, m, c, true) } func (s *Service) recordCheckResult(ctx context.Context, m Monitor, c Check, updateState bool) error { tx, err := s.db.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() if _, err = tx.ExecContext(ctx, `INSERT INTO monitor_checks(monitor_id,ok,status_code,latency_ms,message,checked_at) VALUES(?,?,?,?,?,?)`, m.ID, c.OK, c.StatusCode, c.LatencyMS, c.Message, c.CheckedAt); err != nil { return err } status := "down" if c.OK { status = "up" } if updateState { if _, err = tx.ExecContext(ctx, `UPDATE monitors SET status=?,last_checked_at=?,updated_at=? WHERE id=?`, status, c.CheckedAt, c.CheckedAt, m.ID); err != nil { return err } } else { if _, err = tx.ExecContext(ctx, `UPDATE monitors SET last_checked_at=?,updated_at=? WHERE id=?`, c.CheckedAt, c.CheckedAt, m.ID); err != nil { return err } } if err = tx.Commit(); err != nil { return err } if updateState && s.eventSink != nil && (m.Status == "up" || m.Status == "down") && m.Status != status { s.eventSink(context.Background(), Event{MonitorID: m.ID, Name: m.Name, Target: m.Target, From: m.Status, To: status, Check: c}) } return nil } func (s *Service) probe(ctx context.Context, m Monitor) Check { in := Input{Name: m.Name, Type: m.Type, Target: m.Target, TimeoutMS: m.TimeoutMS, ExpectedMin: m.ExpectedMin, ExpectedMax: m.ExpectedMax, Method: m.Method, HeadersJSON: m.HeadersJSON, Body: m.Body, Keyword: m.Keyword, InvertKeyword: m.InvertKeyword, IgnoreTLS: m.IgnoreTLS, RequireHealthy: m.RequireHealthy} if m.NodeID != nil { b, _, e := s.nodes.Do(ctx, *m.NodeID, http.MethodPost, "/agent/v1/probe", in) if e != nil { return Check{Message: e.Error(), CheckedAt: time.Now().Unix()} } var c Check if json.Unmarshal(b, &c) != nil { return Check{Message: "invalid agent response", CheckedAt: time.Now().Unix()} } return c } return Probe(ctx, in) } func Probe(ctx context.Context, in Input) Check { start := time.Now() c := Check{CheckedAt: start.Unix()} if err := normalize(&in, false); err != nil { c.Message = err.Error() return c } pctx, cancel := context.WithTimeout(ctx, time.Duration(in.TimeoutMS)*time.Millisecond) defer cancel() switch in.Type { case "http": tr := &http.Transport{TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12, InsecureSkipVerify: in.IgnoreTLS}} defer tr.CloseIdleConnections() client := &http.Client{Transport: tr} req, e := http.NewRequestWithContext(pctx, in.Method, in.Target, strings.NewReader(in.Body)) if e != nil { c.Message = e.Error() return c } var headers map[string]string _ = json.Unmarshal([]byte(in.HeadersJSON), &headers) for k, v := range headers { req.Header.Set(k, v) } req.Header.Set("User-Agent", "Dockwatch/"+buildinfo.Current().Version) resp, e := client.Do(req) if e != nil { c.Message = e.Error() return c } defer resp.Body.Close() body, _ := io.ReadAll(io.LimitReader(resp.Body, 512<<10)) c.StatusCode = resp.StatusCode c.OK = resp.StatusCode >= in.ExpectedMin && resp.StatusCode <= in.ExpectedMax if c.OK && in.Keyword != "" { found := strings.Contains(string(body), in.Keyword) if in.InvertKeyword { found = !found } c.OK = found if !found { c.Message = "keyword assertion failed" } } if !c.OK && c.Message == "" { c.Message = "unexpected HTTP status" } case "tcp": conn, e := (&net.Dialer{}).DialContext(pctx, "tcp", in.Target) if e != nil { c.Message = e.Error() return c } _ = conn.Close() c.OK = true case "docker": cmd := exec.CommandContext(pctx, "docker", "inspect", "--format", "{{json .State}}", in.Target) b, e := cmd.Output() if e != nil { c.Message = "docker inspect: " + e.Error() return c } var st struct { Running bool `json:"Running"` Status string `json:"Status"` Health *struct { Status string `json:"Status"` } `json:"Health"` } if e = json.Unmarshal(bytesTrimSpace(b), &st); e != nil { c.Message = "invalid docker state: " + e.Error() return c } if !st.Running { c.Message = "container is not running (" + st.Status + ")" return c } if in.RequireHealthy { if st.Health == nil { c.Message = "container has no healthcheck" return c } if strings.ToLower(st.Health.Status) != "healthy" { c.Message = "container health is " + st.Health.Status return c } } c.OK = true c.Message = "container running" if in.RequireHealthy { c.Message = "container running and healthy" } case "dns": _, e := net.DefaultResolver.LookupHost(pctx, in.Target) if e != nil { c.Message = e.Error() return c } c.OK = true default: c.Message = "unsupported monitor type" } c.LatencyMS = time.Since(start).Milliseconds() return c } func (s *Service) cleanup(ctx context.Context) { // Expired authentication sessions are always disposable, even when heartbeat // retention is configured as unlimited (0). _, _ = s.db.ExecContext(ctx, `DELETE FROM sessions WHERE expires_at