// Package sched runs the collection loops and holds the latest snapshot. package sched import ( "context" "log" "os" "sort" "sync" "time" "nasmon/internal/alert" "nasmon/internal/collect" "nasmon/internal/config" "nasmon/internal/history" "nasmon/internal/store" ) const ( metricsEvery = 10 * time.Second storageEvery = 60 * time.Second smartRetain = 365 * 24 * time.Hour alertRetain = 90 * 24 * time.Hour selfTestSlack = 6 * time.Hour selfTestWin = 3 // hours after SelfTests.Hour during which tests may start ) type DiskView struct { collect.Disk Smart *collect.Smart `json:"smart,omitempty"` // latest snapshot with data Standby bool `json:"standby"` Error string `json:"error,omitempty"` } type Status struct { Host string `json:"host"` Version string `json:"version"` Uptime float64 `json:"uptime_s"` UpdatedAt time.Time `json:"updated_at"` CPU collect.CPU `json:"cpu"` Mem collect.Memory `json:"mem"` Temps []collect.Temp `json:"temps"` CPUTemp float64 `json:"cpu_temp"` HasCPUTemp bool `json:"has_cpu_temp"` Disks []DiskView `json:"disks"` DisksErr string `json:"disks_err,omitempty"` Pools []collect.Pool `json:"pools"` PoolsErr string `json:"pools_err,omitempty"` Datasets []collect.Dataset `json:"datasets"` RootFS collect.FSUsage `json:"rootfs"` Alerts []store.Alert `json:"alerts"` Services collect.ShareServices `json:"services"` LastSmartCheck time.Time `json:"last_smart_check"` NextSmartCheck time.Time `json:"next_smart_check"` SmartRunning bool `json:"smart_running"` LastMailErr string `json:"last_mail_err,omitempty"` } type Scheduler struct { cfg *config.Store st *store.Store ring *history.Ring alerts *alert.Manager version string started time.Time cpu collect.CPUSampler wake chan struct{} runMu sync.Mutex // serialises SMART runs evalMu sync.Mutex // serialises rule evaluation (storage loop vs SMART pass) mu sync.RWMutex status Status // latest SMART data per serial, kept across standby reads smart map[string]*collect.Smart // serials whose latest read was standby (temperature is stale) asleep map[string]bool } func New(cfg *config.Store, st *store.Store, ring *history.Ring, am *alert.Manager, version string) *Scheduler { host, _ := os.Hostname() return &Scheduler{ cfg: cfg, st: st, ring: ring, alerts: am, version: version, started: time.Now(), wake: make(chan struct{}, 1), smart: map[string]*collect.Smart{}, asleep: map[string]bool{}, status: Status{Host: host, Version: version}, } } func (s *Scheduler) Status() Status { s.mu.RLock() defer s.mu.RUnlock() st := s.status st.Uptime = time.Since(s.started).Seconds() return st } // SetStatus replaces the snapshot; intended for tests and tooling. func (s *Scheduler) SetStatus(st Status) { s.mu.Lock() defer s.mu.Unlock() st.Host, st.Version = s.status.Host, s.status.Version s.status = st } // Reload re-plans the next SMART pass after the interval was changed. func (s *Scheduler) Reload() { cfg := s.cfg.Get() s.mu.Lock() if !s.status.LastSmartCheck.IsZero() { s.status.NextSmartCheck = s.status.LastSmartCheck.Add(time.Duration(cfg.SmartIntervalMin) * time.Minute) } s.mu.Unlock() s.poke() } // RunSmartNow triggers an immediate SMART pass (non-blocking). func (s *Scheduler) RunSmartNow() { s.mu.Lock() s.status.NextSmartCheck = time.Now() s.mu.Unlock() s.poke() } func (s *Scheduler) poke() { select { case s.wake <- struct{}{}: default: } } func (s *Scheduler) Run(ctx context.Context) { s.collectMetrics() s.collectStorage() go s.metricsLoop(ctx) go s.storageLoop(ctx) s.smartLoop(ctx) } func (s *Scheduler) metricsLoop(ctx context.Context) { t := time.NewTicker(metricsEvery) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: s.collectMetrics() } } } func (s *Scheduler) storageLoop(ctx context.Context) { t := time.NewTicker(storageEvery) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: s.collectStorage() s.evaluate() } } } func (s *Scheduler) smartLoop(ctx context.Context) { // first pass shortly after start so the UI has data s.mu.Lock() s.status.NextSmartCheck = time.Now().Add(5 * time.Second) s.mu.Unlock() lastPrune := time.Time{} for { s.mu.RLock() next := s.status.NextSmartCheck s.mu.RUnlock() wait := time.Until(next) if wait < 0 { wait = 0 } select { case <-ctx.Done(): return case <-s.wake: continue case <-time.After(wait): } s.runSmart() if time.Since(lastPrune) > 24*time.Hour { if err := s.st.PruneSmart(time.Now().Add(-smartRetain)); err != nil { log.Printf("prune smart history: %v", err) } if err := s.st.PruneAlerts(time.Now().Add(-alertRetain)); err != nil { log.Printf("prune alert log: %v", err) } lastPrune = time.Now() } } } func (s *Scheduler) collectMetrics() { cpu, err := s.cpu.Sample() if err != nil { log.Printf("cpu: %v", err) } mem, err := collect.ReadMemory() if err != nil { log.Printf("mem: %v", err) } temps := collect.ReadTemps() cpuTemp, hasCPU := collect.CPUTemp(temps) s.mu.Lock() s.status.CPU, s.status.Mem, s.status.Temps = cpu, mem, temps s.status.CPUTemp, s.status.HasCPUTemp = cpuTemp, hasCPU s.status.UpdatedAt = time.Now() s.mu.Unlock() s.ring.Add(history.Sample{T: time.Now().Unix(), CPU: cpu.UsagePct, MemUsed: mem.Used, ARC: mem.ARCSize, CPUTemp: cpuTemp}) } // RefreshAlerts reloads the active alert list after the user acknowledged one. func (s *Scheduler) RefreshAlerts() { alerts, err := s.st.ActiveAlerts() if err != nil { return } s.mu.Lock() s.status.Alerts = alerts s.mu.Unlock() } // RefreshStorage re-reads disks, pools and datasets right away (after a // dataset was created or destroyed from the UI). func (s *Scheduler) RefreshStorage() { s.collectStorage() } func (s *Scheduler) collectStorage() { disks, derr := collect.ListDisks() pools, perr := collect.ReadPools() datasets, dserr := collect.ReadDatasets() root, _ := collect.ReadFS("/") alerts, _ := s.st.ActiveAlerts() services := collect.ReadShareServices() s.mu.Lock() defer s.mu.Unlock() if derr != nil { s.status.DisksErr = derr.Error() } else { s.status.DisksErr = "" s.status.Disks = s.buildViews(disks) } if perr != nil { s.status.PoolsErr = perr.Error() } else { s.status.PoolsErr = "" s.status.Pools = pools } if dserr == nil { // keep the last good list if zfs list failed transiently s.status.Datasets = datasets } s.status.RootFS = root s.status.Alerts = alerts s.status.Services = services } // buildViews merges lsblk disks with cached SMART data; caller holds s.mu. func (s *Scheduler) buildViews(disks []collect.Disk) []DiskView { views := make([]DiskView, 0, len(disks)) for _, d := range disks { v := DiskView{Disk: d} if sm := s.smart[d.Serial]; sm != nil { v.Smart = sm v.Standby = s.asleep[d.Serial] if sm.Error != "" && !sm.Supported { v.Error = sm.Error } } for _, old := range s.status.Disks { if old.Serial == d.Serial && v.Smart == nil { v.Standby, v.Error = old.Standby, old.Error } } views = append(views, v) } sort.Slice(views, func(i, j int) bool { return views[i].Name < views[j].Name }) return views } func (s *Scheduler) runSmart() { s.runMu.Lock() defer s.runMu.Unlock() cfg := s.cfg.Get() now := time.Now() s.mu.Lock() s.status.SmartRunning = true s.status.NextSmartCheck = now.Add(time.Duration(cfg.SmartIntervalMin) * time.Minute) s.mu.Unlock() defer func() { s.mu.Lock() s.status.SmartRunning = false s.status.LastSmartCheck = time.Now() s.mu.Unlock() }() disks, err := collect.ListDisks() if err != nil { log.Printf("lsblk: %v", err) return } for _, d := range disks { sm := collect.ReadSmart(d.Path, cfg.SmartSkipStandby) serial := d.Serial if sm.Serial == "" { sm.Serial = serial } if sm.Standby { s.mu.Lock() s.asleep[serial] = true if s.smart[serial] == nil { if prev, _ := s.st.LastSmart(serial); prev != nil { s.smart[serial] = prev } } s.mu.Unlock() continue } if sm.Error != "" && !sm.Supported { log.Printf("smartctl %s: %s", d.Path, sm.Error) s.mu.Lock() s.asleep[serial] = false if s.smart[serial] == nil { s.smart[serial] = &sm } s.mu.Unlock() continue } if err := s.st.AddSmart(sm); err != nil { log.Printf("store smart: %v", err) } _ = s.st.TouchDisk(serial, d.Model, d.Path, now) s.mu.Lock() s.asleep[serial] = false s.smart[serial] = &sm s.mu.Unlock() s.maybeSelfTest(cfg.SelfTests, d, &sm, now) } s.mu.Lock() s.status.Disks = s.buildViews(disks) s.mu.Unlock() s.evaluate() } func (s *Scheduler) maybeSelfTest(c config.SelfTests, d collect.Disk, sm *collect.Smart, now time.Time) { if !c.Enabled || !sm.Supported || d.Serial == "" { return } if sm.LastTest != nil && sm.LastTest.InProgress { return } h := now.Hour() if h < c.Hour || h >= c.Hour+selfTestWin { return } for _, k := range []struct { kind string days int }{{"long", c.LongDays}, {"short", c.ShortDays}} { if k.days <= 0 { continue } last, err := s.st.LastSelfTest(d.Serial, k.kind) if err != nil { continue } if now.Sub(last) < time.Duration(k.days)*24*time.Hour-selfTestSlack { continue } if err := collect.StartSelfTest(d.Path, k.kind); err != nil { log.Printf("self-test: %v", err) return } log.Printf("started %s self-test on %s (%s)", k.kind, d.Path, d.Serial) _ = s.st.SetSelfTest(d.Serial, k.kind, now) return } } // LoadCachedSmart primes the in-memory SMART cache from the database at start. func (s *Scheduler) LoadCachedSmart() { seen, err := s.st.SeenDisks() if err != nil { return } s.mu.Lock() defer s.mu.Unlock() for _, d := range seen { if sm, _ := s.st.LastSmart(d.Serial); sm != nil { s.smart[d.Serial] = sm } } } // evaluate runs alert rules against the current snapshot. func (s *Scheduler) evaluate() { s.evalMu.Lock() defer s.evalMu.Unlock() cfg := s.cfg.Get() now := time.Now() s.mu.RLock() st := s.status in := alert.Input{ Now: now, Th: cfg.Thresholds, Disks: nil, Pools: st.Pools, PoolsOK: st.PoolsErr == "", CPUTemp: st.CPUTemp, HasCPU: st.HasCPUTemp, RootFS: st.RootFS, RootOK: st.RootFS.Total > 0, Prev: map[string]*collect.Smart{}, } for _, v := range st.Disks { in.Disks = append(in.Disks, v.Disk) if v.Smart == nil { in.Frozen = append(in.Frozen, "smart:"+v.Serial+":") continue } sm := *v.Smart if v.Standby { sm.HasTemp = false // stale temperature while asleep } in.Smarts = append(in.Smarts, sm) } s.mu.RUnlock() // previous snapshots for growth rules for _, sm := range in.Smarts { if prev := s.prevSnapshot(sm); prev != nil { in.Prev[sm.Serial] = prev } } if st.DisksErr != "" { in.Frozen = append(in.Frozen, "disk:", "smart:") } else { in.Seen, _ = s.st.SeenDisks() } if !in.PoolsOK { in.Frozen = append(in.Frozen, "zfs:") } events := alert.Evaluate(in) out, err := s.alerts.Reconcile(in, events) if err != nil { log.Printf("alerts: %v", err) return } alerts, _ := s.st.ActiveAlerts() s.mu.Lock() s.status.Alerts = alerts if out.MailErr != nil { s.status.LastMailErr = out.MailErr.Error() } else if out.Mailed { s.status.LastMailErr = "" } s.mu.Unlock() } // prevSnapshot returns the oldest snapshot from the last 24 hours, so a // "value increased" alert stays open for a day after the growth was seen. func (s *Scheduler) prevSnapshot(cur collect.Smart) *collect.Smart { hist, err := s.st.SmartHistory(cur.Serial, cur.CheckedAt.Add(-24*time.Hour)) if err != nil || len(hist) < 2 { return nil } p := hist[0] return &collect.Smart{Serial: cur.Serial, Attrs: []collect.SmartAttr{{ID: 199, Raw: p.CRC}}} }