package poller import ( "context" "errors" "sync" "sync/atomic" "time" "github.com/v10networks/nmon/internal/reach" "github.com/v10networks/nmon/internal/selfmon" "github.com/v10networks/nmon/internal/store" ) var errDeadline = errors.New("poller: cycle deadline") // devLoop is the pair of goroutines serving one device. type devLoop struct { p *Poller id int64 slug string cancel context.CancelFunc tgt atomic.Pointer[target] wake chan struct{} // the target changed kick chan struct{} // poll the primary group now duration selfmon.Gauge missed, skipped, mismatch, reboots, discontins selfmon.Counter mu sync.Mutex // guards st st devState // Touched only by the SNMP goroutine. snmpUp *bool // last state logged persisted string up uptimeTracker disc map[string]int64 // last ifCounterDiscontinuityTime (ticks) by interface ID supply map[string]float64 // last printer supply percent by component } func (l *devLoop) setTarget(t *target) { l.tgt.Store(t) select { case l.wake <- struct{}{}: default: } } func (l *devLoop) status() DeviceStatus { t := l.tgt.Load() l.mu.Lock() defer l.mu.Unlock() s := DeviceStatus{ Device: l.slug, DeviceID: l.id, State: l.st.state, Varbinds: l.st.varbinds, ConsecutiveFailures: l.st.consecFail, MissedDeadlines: l.st.consecMissed, LastCycleDuration: l.st.lastDur, LastCycle: l.st.lastCycle, NextCycle: l.st.next, LastError: l.st.lastErr, } if t != nil { s.Profile = t.profile s.Interfaces = t.ifaces if len(t.groups) > 0 { s.BaseInterval = t.groups[0].base s.Interval = l.st.interval(s.BaseInterval) } } return s } func (l *devLoop) varbinds() int { l.mu.Lock() defer l.mu.Unlock() return l.st.varbinds } // halve shrinks the device's varbind count after a request it could not carry, and returns the new count. func (l *devLoop) halve() int { l.mu.Lock() defer l.mu.Unlock() return l.st.halve() } func (l *devLoop) effInterval(g *group) time.Duration { l.mu.Lock() defer l.mu.Unlock() return l.st.interval(g.base) } func (l *devLoop) failing() bool { l.mu.Lock() defer l.mu.Unlock() return l.st.consecFail > 0 } // persistState stores a change of poll state, keeping the device's warnings and // leaving a device in maintenance alone. func (l *devLoop) persistState(ctx context.Context, want string) { if want == l.persisted { return } d, err := l.p.o.Store.GetDeviceByID(ctx, l.id) if err != nil { l.p.log.Warn("read poll state", "device", l.slug, "err", err) return } seen := want if d.PollState == store.PollMaintenance { seen = reach.PollMaintenance } l.tell(func(o reach.Observer, at time.Time) { o.ObservePoll(l.slug, seen, at) }) if d.PollState != store.PollMaintenance && d.PollState != want { if err := l.p.o.Store.SetPollState(ctx, l.id, want, []string(d.Warnings)); err != nil { l.p.log.Warn("persist poll state", "device", l.slug, "state", want, "err", err) return } l.p.log.Info("poll state", "device", l.slug, "state", want) } l.persisted = want } func (l *devLoop) rebooted(ctx context.Context, ev rebootEvent) { l.reboots.Inc() l.p.log.Warn("device rebooted", "device", l.slug, "uptime_before_s", ev.before, "uptime_after_s", ev.after) err := l.p.o.Store.AppendAudit(ctx, "poller", "device_rebooted", l.slug, map[string]any{"uptime_seconds": ev.before}, map[string]any{"uptime_seconds": ev.after}) if err != nil { l.p.log.Warn("audit reboot", "device", l.slug, "err", err) } } func (l *devLoop) discontinuity(ctx context.Context, t *target, ev discEvent) { if !ev.first { l.discontins.Inc() l.p.log.Info("counter discontinuity", "device", l.slug, "iface", ev.ifID, "ticks", ev.ticks) } if cs, ok := l.p.o.Store.(store.CounterDiscontinuities); ok { if err := cs.SetCounterDiscontinuity(ctx, ev.ifID, ev.ticks); err != nil { l.p.log.Warn("record counter discontinuity", "device", l.slug, "iface", ev.ifID, "err", err) } } }