nmon/internal/poller/devloop.go

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