nmon/internal/poller/fetch.go

package poller

import (
	"context"
	"errors"
	"fmt"
	"math"
	"sync"
	"time"

	"github.com/v10networks/nmon/internal/metrics"
	"github.com/v10networks/nmon/internal/profile"
	"github.com/v10networks/nmon/internal/snmp"
)

// poll fetches the group's OIDs and converts the answers.
func (l *devLoop) poll(ctx context.Context, t *target, g *group) pollOut {
	p := l.p
	select {
	case p.sem <- struct{}{}:
		defer func() { <-p.sem }()
	case <-ctx.Done():
		return pollOut{err: context.Cause(ctx)}
	}
	rec := l.debugRecorder(t)
	vbs, got, shrunk, err := (&fetcher{p: p, slug: l.slug, sz: l, rec: rec}).fetch(ctx, t, g)
	out := pollOut{shrunk: shrunk, err: err}
	l.recordDebug(rec, t, g, vbs, got, err)
	ts := p.clk.Now()
	out.samples = make([]metrics.Sample, 0, len(vbs))
	for i, it := range g.items {
		if !got[i] {
			continue
		}
		l.convert(t, it, vbs[i], ts, &out)
	}
	return out
}

// convert turns one answer into a sample, or counts it as skipped, and feeds
// uptime and discontinuity readings to their trackers.
func (l *devLoop) convert(t *target, it item, vb snmp.Varbind, ts time.Time, out *pollOut) {
	if vb.OID != it.oid {
		out.skipped++
		return
	}
	v, ok, reason := profile.Convert(it.m, vb)
	if !ok {
		out.skipped++
		if reason == profile.ReasonTypeMismatch {
			out.mismatch++
			l.p.log.Debug("type mismatch", "device", l.slug, "metric", it.label, "oid", it.oid, "snmp_type", vb.Type)
		}
		return
	}
	if it.lk == "" {
		out.samples = append(out.samples, metrics.New(it.label, v, ts, metrics.LabelDevice, l.slug))
	} else {
		out.samples = append(out.samples, metrics.New(it.label, v, ts, metrics.LabelDevice, l.slug, it.lk, it.lv))
	}
	l.observe(t, it, v, out)
}

// observe feeds a converted reading to the tracker its role names.
func (l *devLoop) observe(t *target, it item, v float64, out *pollOut) {
	switch it.role {
	case roleSysUp:
		if re, before := l.up.observeSys(v, t.hasHr); re {
			out.reboots = append(out.reboots, rebootEvent{before, v})
		}
	case roleHrUp:
		if re, before := l.up.observeHr(v); re {
			out.reboots = append(out.reboots, rebootEvent{before, v})
		}
	case roleDisc:
		if ev, ok := l.observeDisc(t, it.ifID, v); ok {
			out.discs = append(out.discs, ev)
		}
	}
}

// observeDisc compares an ifCounterDiscontinuityTime reading with the last one.
// The first reading is a baseline unless the store already holds a different value.
func (l *devLoop) observeDisc(t *target, ifID string, seconds float64) (discEvent, bool) {
	ticks := int64(math.Round(seconds * 100))
	prev, seen := l.disc[ifID]
	l.disc[ifID] = ticks
	if !seen {
		prev = t.discBase[ifID]
		if prev == 0 {
			return discEvent{ifID, ticks, true}, ticks != 0
		}
	}
	if ticks == prev {
		return discEvent{}, false
	}
	return discEvent{ifID: ifID, ticks: ticks}, true
}

// snmpTarget is how to reach t: its address, credential and the poller's timeout and retries.
func (p *Poller) snmpTarget(t *target) snmp.Target {
	st := snmp.Target{
		Addr: t.addr, Timeout: p.o.Timeout, Retries: p.o.Retries, MaxVarbinds: HardMaxVarbinds, Clock: p.clk,
	}
	c := t.cred
	st.Cred = snmp.Credential{
		Community: c.Community, V3User: c.V3User, V3AuthProto: c.V3AuthProto, V3AuthPass: c.V3AuthPass,
		V3PrivProto: c.V3PrivProto, V3PrivPass: c.V3PrivPass, V3Context: c.V3Context,
	}
	st.Cred.Version = trimV(c.Version)
	return st
}

func trimV(s string) string {
	if len(s) > 0 && s[0] == 'v' {
		return s[1:]
	}
	return s
}

// rangeFetch is one fetch: the shared cursor over the group's OIDs and the
// first error any worker hit.
type rangeFetch struct {
	f      *fetcher
	g      *group
	vbs    []snmp.Varbind
	got    []bool
	shrunk syncFlag

	mu    sync.Mutex
	pos   int
	first error
}

// take hands out the next range of at most the device's current varbind count.
func (f *rangeFetch) take() (lo, hi int, ok bool) {
	f.mu.Lock()
	defer f.mu.Unlock()
	n := len(f.g.oids)
	if f.pos >= n || f.first != nil {
		return 0, 0, false
	}
	lo = f.pos
	f.pos = min(f.pos+f.f.sz.varbinds(), n)
	return lo, f.pos, true
}

// fail records the first error.
func (f *rangeFetch) fail(e error) {
	f.mu.Lock()
	if f.first == nil {
		f.first = e
	}
	f.mu.Unlock()
}

// worker dials its own session and fetches ranges until none are left or
// something fails, in which case it cancels the other workers.
func (f *rangeFetch) worker(ctx context.Context, cancel context.CancelFunc, tg snmp.Target, worker int) {
	sess, err := f.f.dial(ctx, tg, worker)
	if err != nil {
		f.fail(err)
		cancel()
		return
	}
	defer func() { _ = sess.Close() }()
	for {
		lo, hi, ok := f.take()
		if !ok {
			return
		}
		if err := f.f.getRange(ctx, sess, f, lo, hi); err != nil {
			f.fail(err)
			cancel()
			return
		}
	}
}

// sizer is the varbind count a device's requests carry. A polling device keeps it between cycles; a test poll
// starts from the default and forgets it.
type sizer interface {
	varbinds() int
	halve() int
}

// fetcher is the one fetch path: a polled cycle, a debug-recorded cycle and a test poll all go through it,
// so a test reflects what polling does. rec, when set, records every request.
type fetcher struct {
	p    *Poller
	slug string
	sz   sizer
	rec  *traceRec
}

// dial opens a poll session, wrapped to record its requests when tracing.
func (f *fetcher) dial(ctx context.Context, tg snmp.Target, worker int) (Session, error) {
	s, err := f.p.o.Dial(ctx, tg)
	if err != nil {
		f.rec.dialFailed(worker, err)
		return nil, err
	}
	if f.rec == nil {
		return s, nil
	}
	return &tracedSession{Session: s, rec: f.rec, worker: worker}, nil
}

// fetch GETs every OID of the group in requests of at most the device's
// current varbind count, with up to t.concurrency requests in flight. On
// tooBig or a lost response the count halves and that request is retried once
// in pieces of the new size. It returns whatever arrived, even on error.
func (f *fetcher) fetch(ctx context.Context, t *target, g *group) (vbs []snmp.Varbind, got []bool, shrunk bool, err error) {
	n := len(g.oids)
	rf := &rangeFetch{f: f, g: g, vbs: make([]snmp.Varbind, n), got: make([]bool, n)}
	wctx, wcancel := context.WithCancel(ctx)
	defer wcancel()
	workers := min(t.concurrency, (n+f.sz.varbinds()-1)/f.sz.varbinds())
	tg := f.p.snmpTarget(t)
	var wg sync.WaitGroup
	for w := range max(workers, 1) {
		wg.Add(1)
		go func() {
			defer wg.Done()
			rf.worker(wctx, wcancel, tg, w)
		}()
	}
	wg.Wait()
	first := rf.first
	// The cycle context says why a context error happened (deadline vs shutdown).
	if cause := context.Cause(ctx); first != nil && cause != nil && errors.Is(first, context.Canceled) {
		first = cause
	}
	return rf.vbs, rf.got, rf.shrunk.get(), first
}

type syncFlag struct {
	mu sync.Mutex
	v  bool
}

func (f *syncFlag) set() { f.mu.Lock(); f.v = true; f.mu.Unlock() }
func (f *syncFlag) get() bool {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.v
}

func retriable(err error) bool {
	return errors.Is(err, snmp.ErrTooBig) || errors.Is(err, snmp.ErrTimeout)
}

// getRange GETs oids[lo:hi] into the fetch. When the request is too big or
// lost it halves the device's varbind count and retries once in smaller pieces.
func (f *fetcher) getRange(ctx context.Context, s Session, rf *rangeFetch, lo, hi int) error {
	oids := rf.g.oids
	res, err := s.Get(ctx, oids[lo:hi])
	if err == nil {
		return store1(rf.vbs, rf.got, lo, res, hi-lo)
	}
	if !retriable(err) || hi-lo <= 1 {
		return err
	}
	size := f.sz.halve()
	rf.shrunk.set()
	f.p.log.Info("varbinds halved", "device", f.slug, "varbinds", size, "err", err)
	if m, ok := s.(retryMarker); ok {
		m.markRetry(size)
	}
	for at := lo; at < hi; at += size {
		end := min(at+size, hi)
		res, err = s.Get(ctx, oids[at:end])
		if err != nil {
			return err
		}
		if err := store1(rf.vbs, rf.got, at, res, end-at); err != nil {
			return err
		}
	}
	return nil
}

func store1(vbs []snmp.Varbind, got []bool, at int, res []snmp.Varbind, want int) error {
	if len(res) != want {
		return fmt.Errorf("asked for %d varbinds, got %d", want, len(res))
	}
	copy(vbs[at:], res)
	for i := range res {
		got[at+i] = true
	}
	return nil
}