From 8193d37eaf2c099de199fc6194739b9f22952731 Mon Sep 17 00:00:00 2001 From: arsam Date: Fri, 18 Sep 2026 23:19:54 +0330 Subject: [PATCH] Import source: internal/engine/engine Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality. --- internal/engine/engine.go | 290 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 290 insertions(+) create mode 100644 internal/engine/engine.go diff --git a/internal/engine/engine.go b/internal/engine/engine.go new file mode 100644 index 0000000..faeecdd --- /dev/null +++ b/internal/engine/engine.go @@ -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 +}