nasmon/internal/sched/sched.go

// 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}}}
}