Import source: internal/store/store
Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality.
This commit is contained in:
arsam
@@ -0,0 +1,320 @@
|
|||||||
|
// Package store persists check results and incidents in SQLite.
|
||||||
|
package store
|
||||||
|
|
||||||
|
import (
|
||||||
|
"database/sql"
|
||||||
|
"fmt"
|
||||||
|
"sort"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
_ "modernc.org/sqlite"
|
||||||
|
)
|
||||||
|
|
||||||
|
type Store struct {
|
||||||
|
db *sql.DB
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check is one persisted probe result.
|
||||||
|
type Check struct {
|
||||||
|
TS int64 `json:"ts"` // unix ms
|
||||||
|
OK bool `json:"ok"`
|
||||||
|
LatencyMS float64 `json:"latency_ms"`
|
||||||
|
Detail string `json:"detail,omitempty"`
|
||||||
|
Err string `json:"error,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// Incident is one contiguous down period for a target.
|
||||||
|
type Incident struct {
|
||||||
|
ID int64 `json:"id"`
|
||||||
|
TargetID string `json:"target_id"`
|
||||||
|
StartedAt int64 `json:"started_at"` // unix ms
|
||||||
|
EndedAt *int64 `json:"ended_at,omitempty"` // nil = ongoing
|
||||||
|
Err string `json:"error,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// Bucket aggregates checks over a fixed window (e.g. one hour).
|
||||||
|
type Bucket struct {
|
||||||
|
Start int64 `json:"start"` // unix ms
|
||||||
|
Total int `json:"total"`
|
||||||
|
Failed int `json:"failed"`
|
||||||
|
AvgLatencyMS float64 `json:"avg_latency_ms"`
|
||||||
|
MaxLatencyMS float64 `json:"max_latency_ms"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// LatencyStats summarises successful-check latency over a window.
|
||||||
|
type LatencyStats struct {
|
||||||
|
AvgMS float64 `json:"avg_ms"`
|
||||||
|
P50MS float64 `json:"p50_ms"`
|
||||||
|
P95MS float64 `json:"p95_ms"`
|
||||||
|
MinMS float64 `json:"min_ms"`
|
||||||
|
MaxMS float64 `json:"max_ms"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func Open(path string) (*Store, error) {
|
||||||
|
db, err := sql.Open("sqlite", "file:"+path+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=synchronous(NORMAL)")
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
// modernc.org/sqlite is happiest with a single writer connection.
|
||||||
|
db.SetMaxOpenConns(1)
|
||||||
|
s := &Store{db: db}
|
||||||
|
if err := s.migrate(); err != nil {
|
||||||
|
db.Close()
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return s, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Store) migrate() error {
|
||||||
|
_, err := s.db.Exec(`
|
||||||
|
CREATE TABLE IF NOT EXISTS checks (
|
||||||
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||||
|
target_id TEXT NOT NULL,
|
||||||
|
ts INTEGER NOT NULL,
|
||||||
|
ok INTEGER NOT NULL,
|
||||||
|
latency_ms REAL NOT NULL,
|
||||||
|
detail TEXT NOT NULL DEFAULT '',
|
||||||
|
error TEXT NOT NULL DEFAULT ''
|
||||||
|
);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_checks_target_ts ON checks(target_id, ts);
|
||||||
|
|
||||||
|
CREATE TABLE IF NOT EXISTS incidents (
|
||||||
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||||
|
target_id TEXT NOT NULL,
|
||||||
|
started_at INTEGER NOT NULL,
|
||||||
|
ended_at INTEGER,
|
||||||
|
error TEXT NOT NULL DEFAULT ''
|
||||||
|
);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_incidents_target ON incidents(target_id, started_at);
|
||||||
|
`)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Store) Close() error { return s.db.Close() }
|
||||||
|
|
||||||
|
func (s *Store) InsertCheck(targetID string, c Check) error {
|
||||||
|
_, err := s.db.Exec(
|
||||||
|
`INSERT INTO checks (target_id, ts, ok, latency_ms, detail, error) VALUES (?, ?, ?, ?, ?, ?)`,
|
||||||
|
targetID, c.TS, c.OK, c.LatencyMS, c.Detail, c.Err)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// History returns checks in [from, to], downsampled to at most maxPoints by
|
||||||
|
// striding (keeps every Nth row, newest data intact).
|
||||||
|
func (s *Store) History(targetID string, from, to int64, maxPoints int) ([]Check, error) {
|
||||||
|
rows, err := s.db.Query(
|
||||||
|
`SELECT ts, ok, latency_ms, detail, error FROM checks
|
||||||
|
WHERE target_id = ? AND ts >= ? AND ts <= ? ORDER BY ts ASC`,
|
||||||
|
targetID, from, to)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
|
||||||
|
var all []Check
|
||||||
|
for rows.Next() {
|
||||||
|
var c Check
|
||||||
|
if err := rows.Scan(&c.TS, &c.OK, &c.LatencyMS, &c.Detail, &c.Err); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
all = append(all, c)
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if maxPoints <= 0 || len(all) <= maxPoints {
|
||||||
|
return all, nil
|
||||||
|
}
|
||||||
|
stride := (len(all) + maxPoints - 1) / maxPoints
|
||||||
|
out := make([]Check, 0, maxPoints)
|
||||||
|
for i := 0; i < len(all); i += stride {
|
||||||
|
// Prefer surfacing a failure within the stride so incidents
|
||||||
|
// survive downsampling.
|
||||||
|
pick := all[i]
|
||||||
|
for j := i; j < i+stride && j < len(all); j++ {
|
||||||
|
if !all[j].OK {
|
||||||
|
pick = all[j]
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
out = append(out, pick)
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Buckets aggregates checks in [from, to] into windows of bucketMS,
|
||||||
|
// aligned to bucket boundaries.
|
||||||
|
func (s *Store) Buckets(targetID string, from, to, bucketMS int64) ([]Bucket, error) {
|
||||||
|
rows, err := s.db.Query(
|
||||||
|
`SELECT (ts / ?) * ? AS bucket,
|
||||||
|
COUNT(*),
|
||||||
|
SUM(CASE WHEN ok = 0 THEN 1 ELSE 0 END),
|
||||||
|
AVG(CASE WHEN ok = 1 THEN latency_ms END),
|
||||||
|
MAX(CASE WHEN ok = 1 THEN latency_ms END)
|
||||||
|
FROM checks
|
||||||
|
WHERE target_id = ? AND ts >= ? AND ts <= ?
|
||||||
|
GROUP BY bucket ORDER BY bucket ASC`,
|
||||||
|
bucketMS, bucketMS, targetID, from, to)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
|
||||||
|
var out []Bucket
|
||||||
|
for rows.Next() {
|
||||||
|
var b Bucket
|
||||||
|
var avg, max sql.NullFloat64
|
||||||
|
if err := rows.Scan(&b.Start, &b.Total, &b.Failed, &avg, &max); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
b.AvgLatencyMS = avg.Float64
|
||||||
|
b.MaxLatencyMS = max.Float64
|
||||||
|
out = append(out, b)
|
||||||
|
}
|
||||||
|
return out, rows.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Uptime returns (ok, total) check counts in [from, now].
|
||||||
|
func (s *Store) Uptime(targetID string, from int64) (okCount, total int, err error) {
|
||||||
|
err = s.db.QueryRow(
|
||||||
|
`SELECT COALESCE(SUM(ok), 0), COUNT(*) FROM checks WHERE target_id = ? AND ts >= ?`,
|
||||||
|
targetID, from).Scan(&okCount, &total)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Latency computes latency stats over successful checks in [from, now].
|
||||||
|
func (s *Store) Latency(targetID string, from int64) (LatencyStats, error) {
|
||||||
|
rows, err := s.db.Query(
|
||||||
|
`SELECT latency_ms FROM checks WHERE target_id = ? AND ts >= ? AND ok = 1 ORDER BY latency_ms ASC`,
|
||||||
|
targetID, from)
|
||||||
|
if err != nil {
|
||||||
|
return LatencyStats{}, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
|
||||||
|
var vals []float64
|
||||||
|
var sum float64
|
||||||
|
for rows.Next() {
|
||||||
|
var v float64
|
||||||
|
if err := rows.Scan(&v); err != nil {
|
||||||
|
return LatencyStats{}, err
|
||||||
|
}
|
||||||
|
vals = append(vals, v)
|
||||||
|
sum += v
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
return LatencyStats{}, err
|
||||||
|
}
|
||||||
|
if len(vals) == 0 {
|
||||||
|
return LatencyStats{}, nil
|
||||||
|
}
|
||||||
|
pct := func(p float64) float64 {
|
||||||
|
idx := int(p * float64(len(vals)-1))
|
||||||
|
return vals[idx]
|
||||||
|
}
|
||||||
|
return LatencyStats{
|
||||||
|
AvgMS: sum / float64(len(vals)),
|
||||||
|
P50MS: pct(0.50),
|
||||||
|
P95MS: pct(0.95),
|
||||||
|
MinMS: vals[0],
|
||||||
|
MaxMS: vals[len(vals)-1],
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Store) OpenIncident(targetID string, startedAt int64, errStr string) (int64, error) {
|
||||||
|
res, err := s.db.Exec(
|
||||||
|
`INSERT INTO incidents (target_id, started_at, error) VALUES (?, ?, ?)`,
|
||||||
|
targetID, startedAt, errStr)
|
||||||
|
if err != nil {
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
return res.LastInsertId()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Store) CloseIncident(id, endedAt int64) error {
|
||||||
|
_, err := s.db.Exec(`UPDATE incidents SET ended_at = ? WHERE id = ?`, endedAt, id)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// OngoingIncident returns the open incident for a target, or nil.
|
||||||
|
func (s *Store) OngoingIncident(targetID string) (*Incident, error) {
|
||||||
|
var inc Incident
|
||||||
|
err := s.db.QueryRow(
|
||||||
|
`SELECT id, target_id, started_at, error FROM incidents
|
||||||
|
WHERE target_id = ? AND ended_at IS NULL ORDER BY started_at DESC LIMIT 1`,
|
||||||
|
targetID).Scan(&inc.ID, &inc.TargetID, &inc.StartedAt, &inc.Err)
|
||||||
|
if err == sql.ErrNoRows {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return &inc, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Incidents lists incidents overlapping [from, now], newest first,
|
||||||
|
// ongoing ones always included and first.
|
||||||
|
func (s *Store) Incidents(from int64) ([]Incident, error) {
|
||||||
|
rows, err := s.db.Query(
|
||||||
|
`SELECT id, target_id, started_at, ended_at, error FROM incidents
|
||||||
|
WHERE ended_at IS NULL OR ended_at >= ?
|
||||||
|
ORDER BY (ended_at IS NULL) DESC, started_at DESC LIMIT 200`,
|
||||||
|
from)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
|
||||||
|
out := []Incident{}
|
||||||
|
for rows.Next() {
|
||||||
|
var inc Incident
|
||||||
|
var ended sql.NullInt64
|
||||||
|
if err := rows.Scan(&inc.ID, &inc.TargetID, &inc.StartedAt, &ended, &inc.Err); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if ended.Valid {
|
||||||
|
inc.EndedAt = &ended.Int64
|
||||||
|
}
|
||||||
|
out = append(out, inc)
|
||||||
|
}
|
||||||
|
return out, rows.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
// IncidentsFor lists recent incidents for one target, newest first.
|
||||||
|
func (s *Store) IncidentsFor(targetID string, limit int) ([]Incident, error) {
|
||||||
|
rows, err := s.db.Query(
|
||||||
|
`SELECT id, target_id, started_at, ended_at, error FROM incidents
|
||||||
|
WHERE target_id = ? ORDER BY started_at DESC LIMIT ?`,
|
||||||
|
targetID, limit)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
|
||||||
|
out := []Incident{}
|
||||||
|
for rows.Next() {
|
||||||
|
var inc Incident
|
||||||
|
var ended sql.NullInt64
|
||||||
|
if err := rows.Scan(&inc.ID, &inc.TargetID, &inc.StartedAt, &ended, &inc.Err); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if ended.Valid {
|
||||||
|
inc.EndedAt = &ended.Int64
|
||||||
|
}
|
||||||
|
out = append(out, inc)
|
||||||
|
}
|
||||||
|
// Ongoing first, then newest.
|
||||||
|
sort.SliceStable(out, func(i, j int) bool {
|
||||||
|
return out[i].EndedAt == nil && out[j].EndedAt != nil
|
||||||
|
})
|
||||||
|
return out, rows.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Prune deletes checks older than the retention window.
|
||||||
|
func (s *Store) Prune(olderThan time.Time) error {
|
||||||
|
if _, err := s.db.Exec(`DELETE FROM checks WHERE ts < ?`, olderThan.UnixMilli()); err != nil {
|
||||||
|
return fmt.Errorf("prune checks: %w", err)
|
||||||
|
}
|
||||||
|
_, err := s.db.Exec(`DELETE FROM incidents WHERE ended_at IS NOT NULL AND ended_at < ?`, olderThan.UnixMilli())
|
||||||
|
return err
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user