nmon/internal/api/bgp.go

package api

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/netip"
	"slices"
	"strconv"
	"strings"

	"github.com/danielgtaylor/huma/v2"

	"github.com/v10networks/nmon/internal/profile"
	"github.com/v10networks/nmon/internal/store"
)

// The profile metrics BGP peers are built from. Every BGP profile uses these names whatever MIB feeds them, so
// the rule, the API and the UI never care which vendor a peer comes from.
const (
	bgpState    = "bgp_peer_state"
	bgpUptime   = "bgp_peer_uptime"
	bgpPrefixes = "bgp_prefixes_accepted"
	bgpInUpd    = "bgp_peer_in_updates"
	bgpOutUpd   = "bgp_peer_out_updates"

	stateEstablished = "established"
	stateUnknown     = "unknown"
	// rateWindow is long enough to hold several 5-minute samples of an update counter.
	rateWindow = "15m"
)

func (s *server) registerBGP(api huma.API) {
	huma.Register(api, huma.Operation{
		OperationID: "list-bgp-peers", Method: http.MethodGet, Path: "/api/v1/bgp/peers", Tags: []string{"bgp"},
		Summary: "BGP sessions of every device",
		Description: "Every BGP peer the profiles collect, with the latest state, uptime, accepted prefixes per address family and " +
			"update rates from the hot store, and the peer AS, local AS, router ID and description from discovery. Sessions that " +
			"are down come first. state is established, down (anything else) or an FSM state name.",
		Errors: []int{http.StatusTooManyRequests, http.StatusServiceUnavailable},
	}, s.listBGPPeers)
}

// bgpDevice is a device whose profile collects BGP, with its peer rows.
type bgpDevice struct {
	dev   store.Device
	state profile.Metric
	comps []store.Component
}

func (s *server) listBGPPeers(ctx context.Context, in *struct {
	Device string `query:"device" doc:"Only this device (slug)."`
	State  string `query:"state" doc:"established, down, or an FSM state: idle, connect, active, opensent, openconfirm."`
	AS     string `query:"as" pattern:"^[0-9]{1,10}$" doc:"Only sessions with this remote AS."`
	Family string `query:"family" enum:"ipv4,ipv6" doc:"Only IPv4 or only IPv6 peers."`
	Q      string `query:"q" doc:"Substring of the peer address, description or device name."`
	PageParams
}) (*struct{ Body BGPPeerList }, error) {
	devs, err := s.bgpDevices(ctx, in.Device)
	if err != nil {
		return nil, err
	}
	peers, err := s.bgpPeers(ctx, devs)
	if err != nil {
		return nil, err
	}
	kept := peers[:0]
	var counts BGPCounts
	for _, p := range peers {
		if !bgpMatches(p, optInt(in.AS), in.Family, in.Q) {
			continue
		}
		counts.Total++
		if p.Established {
			counts.Established++
		} else {
			counts.Down++
		}
		if bgpStateMatches(p, in.State) {
			kept = append(kept, p)
		}
	}
	sortPeers(kept)
	total := int64(len(kept))
	lo := min(in.offset(), total)
	page := kept[lo:min(lo+in.limit(), total)]
	out := BGPPeerList{Items: slices.Clone(page), Page: in.pageNo(), PerPage: in.limit(), Total: total, Counts: counts}
	return &struct{ Body BGPPeerList }{out}, nil
}

// bgpDevices lists the devices whose profile collects BGP and which have peer rows.
func (s *server) bgpDevices(ctx context.Context, only string) ([]bgpDevice, error) {
	if s.Profiles == nil {
		return nil, nil
	}
	var out []bgpDevice
	for off := int64(0); ; off += store.MaxLimit {
		rows, n, err := s.Store.ListDevices(ctx, store.DeviceFilter{Kind: store.KindNetwork}, store.Page{Offset: off, Limit: store.MaxLimit})
		if err != nil {
			return nil, s.storeErr(err)
		}
		for _, d := range rows {
			if only != "" && d.Slug != only {
				continue
			}
			if bd, ok, err := s.bgpDeviceOf(ctx, d); err != nil {
				return nil, s.storeErr(err)
			} else if ok {
				out = append(out, bd)
			}
		}
		if off+store.MaxLimit >= n {
			return out, nil
		}
	}
}

func (s *server) bgpDeviceOf(ctx context.Context, d store.Device) (bgpDevice, bool, error) {
	res := s.Profiles.Get(d.Profile)
	if res == nil {
		return bgpDevice{}, false, nil
	}
	m, ok := res.Metric(bgpState)
	if !ok || !m.Pollable() {
		return bgpDevice{}, false, nil
	}
	all, err := s.Store.ListComponents(ctx, d.ID)
	if err != nil {
		return bgpDevice{}, false, err
	}
	bd := bgpDevice{dev: d, state: m}
	for _, c := range all {
		if c.TableName == m.TableKey() {
			bd.comps = append(bd.comps, c)
		}
	}
	return bd, len(bd.comps) > 0, nil
}

// bgpPeers builds the peers of the devices from the inventory and the hot store.
func (s *server) bgpPeers(ctx context.Context, devs []bgpDevice) ([]BGPPeer, error) {
	if len(devs) == 0 {
		return []BGPPeer{}, nil
	}
	slugs := make([]string, len(devs))
	for i, d := range devs {
		slugs[i] = d.dev.Slug
	}
	match := `device=~` + quote(regexAlt(slugs))
	latest, err := s.queryLatest(ctx, match, []string{bgpState, bgpUptime, bgpPrefixes})
	if err != nil {
		return nil, err
	}
	rates, err := s.queryRates(ctx, match, []string{bgpInUpd, bgpOutUpd})
	if err != nil {
		return nil, err
	}
	out := []BGPPeer{}
	for _, d := range devs {
		out = append(out, peersOf(d, latest[d.dev.Slug], rates[d.dev.Slug])...)
	}
	return out, nil
}

// rateReadings is metric -> component -> per-second rate.
type rateReadings map[string]map[string]float64

// queryRates reads the per-second rate of counters over rateWindow for every device in match.
func (s *server) queryRates(ctx context.Context, match string, names []string) (map[string]rateReadings, error) {
	parts := make([]string, len(names))
	for i, n := range names {
		parts[i] = `label_set(rate(` + n + `{` + match + `}[` + rateWindow + `]), "k", ` + quote(n) + `)`
	}
	release, err := s.admit(ctx, false)
	if err != nil {
		return nil, err
	}
	defer release()
	qctx, cancel := context.WithTimeout(ctx, QueryTimeout)
	defer cancel()
	body, err := s.Querier.Query(qctx, "hot", strings.Join(parts, " or "), s.Clock.Now())
	if err != nil {
		return nil, s.vmErr(qctx, err)
	}
	rows, err := parseVM(body)
	if err != nil {
		return nil, fmt.Errorf("bgp rates: %w", err)
	}
	out := map[string]rateReadings{}
	for _, r := range rows {
		if r.Value == nil || !r.Value.OK {
			continue
		}
		dev, k := r.Metric["device"], r.Metric["k"]
		if out[dev] == nil {
			out[dev] = rateReadings{}
		}
		if out[dev][k] == nil {
			out[dev][k] = map[string]float64{}
		}
		out[dev][k][r.Metric["component"]] = r.Value.V
	}
	return out, nil
}

// peersOf builds the peers of one device.
func peersOf(d bgpDevice, rd readings, rates rateReadings) []BGPPeer {
	seen := map[string]int{}
	for _, c := range d.comps {
		seen[componentLabel(c)]++
	}
	out := make([]BGPPeer, 0, len(d.comps))
	for _, c := range d.comps {
		peer := componentLabel(c)
		series := peer // the poller keeps rows with equal labels apart with the index
		if seen[peer] > 1 {
			series += "#" + c.Idx
		}
		out = append(out, buildPeer(d, c, peer, series, rd, rates))
	}
	return out
}

func componentLabel(c store.Component) string {
	if c.Label != "" {
		return c.Label
	}
	return c.Idx
}

func buildPeer(d bgpDevice, c store.Component, peer, series string, rd readings, rates rateReadings) BGPPeer {
	var attrs map[string]string
	_ = json.Unmarshal([]byte(c.Attrs), &attrs) // attrs are written by discovery; unreadable ones are simply absent
	p := BGPPeer{
		Device: DeviceRef{Slug: d.dev.Slug, Name: d.dev.Name}, Peer: peer, Component: series, Family: peerFamily(peer),
		RemoteAS: parseAS(attrs["remote_as"]), LocalAS: parseAS(attrs["local_as"]), RouterID: attrs["router_id"],
		Description: attrs["description"], State: stateUnknown, StateClass: stateUnknown, Prefixes: []BGPFamily{},
	}
	if r := rd.at(bgpState, series); r != nil {
		name := d.state.Enum[int(r.v)]
		if name == "" {
			name = stateUnknown
		}
		p.State, p.StateClass, p.Established = name, stateOf(d.state, name), name == stateEstablished
		t := r.ts
		p.Sampled = &t
	}
	if r := rd.at(bgpUptime, series); r != nil {
		v := r.v
		p.UptimeS = &v
	}
	p.InUpdatesPerS, p.OutUpdatesPerS = rateAt(rates, bgpInUpd, series), rateAt(rates, bgpOutUpd, series)
	p.Prefixes = prefixFamilies(rd[bgpPrefixes], series)
	return p
}

func rateAt(rates rateReadings, metric, comp string) *float64 {
	if v, ok := rates[metric][comp]; ok {
		return &v
	}
	return nil
}

// prefixFamilies reads the "<peer>|<family>" components of the prefix metric that belong to the peer.
func prefixFamilies(by map[string]*reading, series string) []BGPFamily {
	out := []BGPFamily{}
	for comp, r := range by {
		if fam, ok := strings.CutPrefix(comp, series+"|"); ok {
			v := r.v
			out = append(out, BGPFamily{Family: fam, Accepted: &v})
		}
	}
	slices.SortFunc(out, func(a, b BGPFamily) int { return strings.Compare(a.Family, b.Family) })
	return out
}

func peerFamily(peer string) string {
	if a, err := netip.ParseAddr(peer); err == nil && a.Is6() {
		return "ipv6"
	}
	return "ipv4"
}

func parseAS(s string) *int64 {
	if v, err := strconv.ParseInt(s, 10, 64); err == nil && v >= 0 {
		return &v
	}
	return nil
}

func bgpMatches(p BGPPeer, as *int64, family, q string) bool {
	if as != nil && (p.RemoteAS == nil || *p.RemoteAS != *as) {
		return false
	}
	if family != "" && p.Family != family {
		return false
	}
	if q == "" {
		return true
	}
	q = strings.ToLower(q)
	for _, f := range []string{p.Peer, p.Description, p.Device.Name, p.Device.Slug, p.RouterID} {
		if strings.Contains(strings.ToLower(f), q) {
			return true
		}
	}
	return false
}

func bgpStateMatches(p BGPPeer, state string) bool {
	switch state {
	case "":
		return true
	case "down":
		return !p.Established
	}
	return p.State == state
}

// sortPeers puts the sessions that are down first, then orders by device and peer.
func sortPeers(ps []BGPPeer) {
	slices.SortStableFunc(ps, func(a, b BGPPeer) int {
		if a.Established != b.Established {
			if a.Established {
				return 1
			}
			return -1
		}
		if c := strings.Compare(a.Device.Slug, b.Device.Slug); c != 0 {
			return c
		}
		return strings.Compare(a.Peer, b.Peer)
	})
}