660 lines
21 KiB
Go
660 lines
21 KiB
Go
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<?`, time.Now().Unix())
|
|
if s.retentionDays <= 0 {
|
|
return
|
|
}
|
|
cut := time.Now().Add(-time.Duration(s.retentionDays) * 24 * time.Hour).Unix()
|
|
_, _ = s.db.ExecContext(ctx, `DELETE FROM monitor_checks WHERE checked_at<?`, cut)
|
|
}
|
|
func ParseID(v string) (int64, error) {
|
|
id, e := strconv.ParseInt(v, 10, 64)
|
|
if e != nil || id < 1 {
|
|
return 0, fmt.Errorf("invalid id")
|
|
}
|
|
return id, nil
|
|
}
|
|
|
|
func bytesTrimSpace(b []byte) []byte { return []byte(strings.TrimSpace(string(b))) }
|