Initial import: nasmon health monitor, split out of home_nas
Go service (nasmon/) plus nfpm packaging (packaging/), carried over from the home_nas monorepo. Adds rpm packaging alongside the existing deb build (Fedora/RHEL/Rocky dependency names via nfpm overrides, per-format systemd unit path, dpkg+rpm-compatible maintainer scripts) and a Gitea Actions workflow that builds both formats on every push. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 5
@@ -0,0 +1,479 @@
|
||||
// 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}}}
|
||||
}
|
||||
Reference in New Issue
Block a user