Import source: internal/engine/engine
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,290 @@
|
||||
// Package engine schedules checks, runs flap detection and groups incidents.
|
||||
package engine
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"math/rand"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"dash/internal/checker"
|
||||
"dash/internal/config"
|
||||
"dash/internal/store"
|
||||
)
|
||||
|
||||
// State is the flap-filtered status of a target.
|
||||
type State string
|
||||
|
||||
const (
|
||||
StateUnknown State = "unknown"
|
||||
StateUp State = "up"
|
||||
StateDegraded State = "degraded"
|
||||
StateDown State = "down"
|
||||
)
|
||||
|
||||
// CheckEvent is broadcast on every completed probe.
|
||||
type CheckEvent struct {
|
||||
TargetID string `json:"target_id"`
|
||||
TS int64 `json:"ts"`
|
||||
OK bool `json:"ok"`
|
||||
LatencyMS float64 `json:"latency_ms"`
|
||||
Detail string `json:"detail,omitempty"`
|
||||
Err string `json:"error,omitempty"`
|
||||
State State `json:"state"`
|
||||
}
|
||||
|
||||
// StateEvent is broadcast when flap detection confirms a transition.
|
||||
type StateEvent struct {
|
||||
TargetID string `json:"target_id"`
|
||||
TS int64 `json:"ts"`
|
||||
State State `json:"state"`
|
||||
PrevState State `json:"prev_state"`
|
||||
Incident *store.Incident `json:"incident,omitempty"`
|
||||
}
|
||||
|
||||
// Event is a named SSE payload.
|
||||
type Event struct {
|
||||
Name string // "check" | "state"
|
||||
Data any
|
||||
}
|
||||
|
||||
// Hub fans events out to SSE subscribers.
|
||||
type Hub struct {
|
||||
mu sync.Mutex
|
||||
subs map[chan Event]struct{}
|
||||
}
|
||||
|
||||
func NewHub() *Hub {
|
||||
return &Hub{subs: map[chan Event]struct{}{}}
|
||||
}
|
||||
|
||||
func (h *Hub) Subscribe() chan Event {
|
||||
ch := make(chan Event, 64)
|
||||
h.mu.Lock()
|
||||
h.subs[ch] = struct{}{}
|
||||
h.mu.Unlock()
|
||||
return ch
|
||||
}
|
||||
|
||||
func (h *Hub) Unsubscribe(ch chan Event) {
|
||||
h.mu.Lock()
|
||||
delete(h.subs, ch)
|
||||
h.mu.Unlock()
|
||||
}
|
||||
|
||||
func (h *Hub) Broadcast(e Event) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
for ch := range h.subs {
|
||||
select {
|
||||
case ch <- e:
|
||||
default: // slow consumer: drop rather than block the engine
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TargetStatus is the live in-memory view of one target.
|
||||
type TargetStatus struct {
|
||||
Target config.Target
|
||||
State State
|
||||
Last *checker.Result
|
||||
Fails int // consecutive failures
|
||||
Incident *store.Incident
|
||||
}
|
||||
|
||||
// Engine owns one scheduler goroutine per target.
|
||||
type Engine struct {
|
||||
cfg *config.Config
|
||||
store *store.Store
|
||||
hub *Hub
|
||||
|
||||
mu sync.RWMutex
|
||||
runtime map[string]*targetRuntime
|
||||
}
|
||||
|
||||
type targetRuntime struct {
|
||||
cfg config.Target
|
||||
|
||||
state State
|
||||
last *checker.Result
|
||||
fails int
|
||||
oks int
|
||||
firstFailTS int64
|
||||
incident *store.Incident
|
||||
}
|
||||
|
||||
func New(cfg *config.Config, st *store.Store, hub *Hub) *Engine {
|
||||
e := &Engine{cfg: cfg, store: st, hub: hub, runtime: map[string]*targetRuntime{}}
|
||||
for _, t := range cfg.Targets {
|
||||
rt := &targetRuntime{cfg: t, state: StateUnknown}
|
||||
// Resume an incident that was open when the process last stopped.
|
||||
if inc, err := st.OngoingIncident(t.ID); err == nil && inc != nil {
|
||||
rt.incident = inc
|
||||
}
|
||||
e.runtime[t.ID] = rt
|
||||
}
|
||||
return e
|
||||
}
|
||||
|
||||
// Run blocks until ctx is cancelled.
|
||||
func (e *Engine) Run(ctx context.Context) {
|
||||
var wg sync.WaitGroup
|
||||
for _, t := range e.cfg.Targets {
|
||||
wg.Add(1)
|
||||
go func(t config.Target) {
|
||||
defer wg.Done()
|
||||
e.loop(ctx, t)
|
||||
}(t)
|
||||
}
|
||||
|
||||
// Daily retention pruning.
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
tick := time.NewTicker(24 * time.Hour)
|
||||
defer tick.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-tick.C:
|
||||
cutoff := time.Now().AddDate(0, 0, -e.cfg.Settings.HistoryDays)
|
||||
if err := e.store.Prune(cutoff); err != nil {
|
||||
log.Printf("prune: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
func (e *Engine) loop(ctx context.Context, t config.Target) {
|
||||
// Jitter the first run so all targets don't fire at once.
|
||||
jitter := time.Duration(rand.Int63n(int64(2 * time.Second)))
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(jitter):
|
||||
}
|
||||
|
||||
tick := time.NewTicker(t.Interval.Duration)
|
||||
defer tick.Stop()
|
||||
for {
|
||||
e.runOnce(ctx, t)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-tick.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (e *Engine) runOnce(ctx context.Context, t config.Target) {
|
||||
res := checker.Run(ctx, t)
|
||||
ts := res.TS.UnixMilli()
|
||||
|
||||
if err := e.store.InsertCheck(t.ID, store.Check{
|
||||
TS: ts, OK: res.OK, LatencyMS: res.LatencyMS, Detail: res.Detail, Err: res.Err,
|
||||
}); err != nil {
|
||||
log.Printf("store check %s: %v", t.ID, err)
|
||||
}
|
||||
|
||||
e.mu.Lock()
|
||||
rt := e.runtime[t.ID]
|
||||
prev := rt.state
|
||||
rt.last = &res
|
||||
|
||||
if res.OK {
|
||||
rt.oks++
|
||||
rt.fails = 0
|
||||
} else {
|
||||
if rt.fails == 0 {
|
||||
rt.firstFailTS = ts
|
||||
}
|
||||
rt.fails++
|
||||
rt.oks = 0
|
||||
}
|
||||
|
||||
var incidentEvent *store.Incident
|
||||
switch {
|
||||
case rt.state == StateDown:
|
||||
// Recovery needs RecoverThreshold consecutive successes.
|
||||
if res.OK && rt.oks >= t.RecoverThreshold {
|
||||
rt.state = stateForOK(res)
|
||||
if rt.incident != nil {
|
||||
if err := e.store.CloseIncident(rt.incident.ID, ts); err != nil {
|
||||
log.Printf("close incident %s: %v", t.ID, err)
|
||||
}
|
||||
closed := *rt.incident
|
||||
closed.EndedAt = &ts
|
||||
incidentEvent = &closed
|
||||
rt.incident = nil
|
||||
}
|
||||
}
|
||||
case !res.OK:
|
||||
if rt.fails >= t.FailThreshold || (prev == StateUnknown && rt.fails >= t.FailThreshold) {
|
||||
rt.state = StateDown
|
||||
if rt.incident == nil {
|
||||
// The incident spans the whole losing streak, starting
|
||||
// at the first failed check, not the confirming one.
|
||||
id, err := e.store.OpenIncident(t.ID, rt.firstFailTS, res.Err)
|
||||
if err != nil {
|
||||
log.Printf("open incident %s: %v", t.ID, err)
|
||||
} else {
|
||||
rt.incident = &store.Incident{
|
||||
ID: id, TargetID: t.ID, StartedAt: rt.firstFailTS, Err: res.Err,
|
||||
}
|
||||
}
|
||||
}
|
||||
incidentEvent = rt.incident
|
||||
} else {
|
||||
// Failing but not yet confirmed down: flapping.
|
||||
rt.state = StateDegraded
|
||||
}
|
||||
default: // res.OK while up/degraded/unknown
|
||||
rt.state = stateForOK(res)
|
||||
// A stale incident from a previous process run: close it.
|
||||
if rt.incident != nil && prev == StateUnknown {
|
||||
if err := e.store.CloseIncident(rt.incident.ID, ts); err != nil {
|
||||
log.Printf("close stale incident %s: %v", t.ID, err)
|
||||
}
|
||||
rt.incident = nil
|
||||
}
|
||||
}
|
||||
state := rt.state
|
||||
e.mu.Unlock()
|
||||
|
||||
e.hub.Broadcast(Event{Name: "check", Data: CheckEvent{
|
||||
TargetID: t.ID, TS: ts, OK: res.OK, LatencyMS: res.LatencyMS,
|
||||
Detail: res.Detail, Err: res.Err, State: state,
|
||||
}})
|
||||
if state != prev {
|
||||
e.hub.Broadcast(Event{Name: "state", Data: StateEvent{
|
||||
TargetID: t.ID, TS: ts, State: state, PrevState: prev, Incident: incidentEvent,
|
||||
}})
|
||||
}
|
||||
}
|
||||
|
||||
func stateForOK(res checker.Result) State {
|
||||
if res.Degraded {
|
||||
return StateDegraded
|
||||
}
|
||||
return StateUp
|
||||
}
|
||||
|
||||
// Snapshot returns the live status of every target, in config order.
|
||||
func (e *Engine) Snapshot() []TargetStatus {
|
||||
e.mu.RLock()
|
||||
defer e.mu.RUnlock()
|
||||
out := make([]TargetStatus, 0, len(e.cfg.Targets))
|
||||
for _, t := range e.cfg.Targets {
|
||||
rt := e.runtime[t.ID]
|
||||
out = append(out, TargetStatus{
|
||||
Target: rt.cfg, State: rt.state, Last: rt.last,
|
||||
Fails: rt.fails, Incident: rt.incident,
|
||||
})
|
||||
}
|
||||
return out
|
||||
}
|
||||
Reference in New Issue
Block a user