clawbot
77a708d819
check / check (push) Waiting to run
The prefix handler kept one update per prefix in each batch across all peers, chosen by a timestamp cut to whole seconds, so most peers' announcements and withdrawals were lost and the first update in a second won. A batch now keeps the last update per prefix and peer in feed order. A message's withdrawals are queued before its announcements, so a prefix both withdrawn and announced in one message keeps its route (RFC 4271 section 4.3). Tests run each case through the handler into a real database and check the in-memory counts. Model: opus-5-5
481 lines
13 KiB
Go
481 lines
13 KiB
Go
package routewatch
|
|
|
|
import (
|
|
"net"
|
|
"net/netip"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.eeqj.de/sneak/routewatch/internal/database"
|
|
"git.eeqj.de/sneak/routewatch/internal/logger"
|
|
"git.eeqj.de/sneak/routewatch/internal/metrics"
|
|
"git.eeqj.de/sneak/routewatch/internal/ristypes"
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
const (
|
|
// prefixHandlerQueueSize is the queue capacity for prefix tracking
|
|
// operations, about 4 seconds of feed at peak. The streamer drops rather
|
|
// than blocks when a queue is full, so this bounds memory. Batches still
|
|
// flush on a timer (prefixBatchTimeout).
|
|
prefixHandlerQueueSize = 20000
|
|
|
|
// prefixBatchSize is the number of prefix updates to batch together
|
|
prefixBatchSize = 25000
|
|
|
|
// prefixBatchTimeout is the maximum time to wait before flushing a batch
|
|
// DO NOT reduce this timeout - larger batches are more efficient
|
|
prefixBatchTimeout = 1 * time.Second
|
|
|
|
// IP version constants
|
|
ipv4Version = 4
|
|
ipv6Version = 6
|
|
)
|
|
|
|
// PrefixHandler tracks BGP prefixes and maintains a live routing table in the database.
|
|
// Routes are added on announcement and deleted on withdrawal.
|
|
type PrefixHandler struct {
|
|
db database.Store
|
|
logger *logger.Logger
|
|
metrics *metrics.Tracker
|
|
|
|
// Batching
|
|
mu sync.Mutex
|
|
batch []prefixUpdate
|
|
lastFlush time.Time
|
|
stopCh chan struct{}
|
|
wg sync.WaitGroup
|
|
}
|
|
|
|
type prefixUpdate struct {
|
|
prefix string
|
|
originASN int
|
|
peer string
|
|
messageType string
|
|
timestamp time.Time
|
|
path []int
|
|
}
|
|
|
|
// NewPrefixHandler creates a new batched prefix tracking handler
|
|
func NewPrefixHandler(db database.Store, logger *logger.Logger) *PrefixHandler {
|
|
h := &PrefixHandler{
|
|
db: db,
|
|
logger: logger,
|
|
batch: make([]prefixUpdate, 0, prefixBatchSize),
|
|
lastFlush: time.Now(),
|
|
stopCh: make(chan struct{}),
|
|
}
|
|
|
|
// Start the flush timer goroutine
|
|
h.wg.Add(1)
|
|
go h.flushLoop()
|
|
|
|
return h
|
|
}
|
|
|
|
// SetMetricsTracker sets the metrics tracker for recording route updates
|
|
func (h *PrefixHandler) SetMetricsTracker(metrics *metrics.Tracker) {
|
|
h.metrics = metrics
|
|
}
|
|
|
|
// WantsMessage returns true if this handler wants to process messages of the given type
|
|
func (h *PrefixHandler) WantsMessage(messageType string) bool {
|
|
// We only care about UPDATE messages for the routing table
|
|
return messageType == "UPDATE"
|
|
}
|
|
|
|
// QueueCapacity returns the desired queue capacity for this handler
|
|
func (h *PrefixHandler) QueueCapacity() int {
|
|
// Batching allows us to use a larger queue
|
|
return prefixHandlerQueueSize
|
|
}
|
|
|
|
// HandleMessage processes a message to track prefix information
|
|
func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
|
|
// Use the pre-parsed timestamp
|
|
timestamp := msg.ParsedTimestamp
|
|
|
|
// Get origin ASN from path (last element)
|
|
var originASN int
|
|
if len(msg.Path) > 0 {
|
|
originASN = msg.Path[len(msg.Path)-1]
|
|
}
|
|
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
|
|
// Withdrawals are queued before announcements, because the batch keeps
|
|
// each peer's last update for a prefix: a message that lists a prefix as
|
|
// both withdrawn and announced keeps the route (RFC 4271 section 4.3).
|
|
for _, prefix := range msg.Withdrawals {
|
|
h.batch = append(h.batch, prefixUpdate{
|
|
prefix: canonicalPrefix(prefix),
|
|
originASN: originASN, // Use the originASN from path if available
|
|
peer: msg.Peer,
|
|
messageType: "withdrawal",
|
|
timestamp: timestamp,
|
|
path: msg.Path,
|
|
})
|
|
// Record withdrawal in metrics
|
|
if h.metrics != nil {
|
|
h.metrics.RecordWithdrawal()
|
|
}
|
|
}
|
|
|
|
for _, announcement := range msg.Announcements {
|
|
for _, prefix := range announcement.Prefixes {
|
|
h.batch = append(h.batch, prefixUpdate{
|
|
prefix: canonicalPrefix(prefix),
|
|
originASN: originASN,
|
|
peer: msg.Peer,
|
|
messageType: "announcement",
|
|
timestamp: timestamp,
|
|
path: msg.Path,
|
|
})
|
|
// Record announcement in metrics
|
|
if h.metrics != nil {
|
|
h.metrics.RecordAnnouncement()
|
|
}
|
|
}
|
|
}
|
|
|
|
// Check if we need to flush
|
|
if len(h.batch) >= prefixBatchSize {
|
|
h.flushBatchLocked()
|
|
}
|
|
}
|
|
|
|
// flushLoop runs in a goroutine and periodically flushes batches
|
|
func (h *PrefixHandler) flushLoop() {
|
|
defer h.wg.Done()
|
|
ticker := time.NewTicker(prefixBatchTimeout)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
h.mu.Lock()
|
|
if time.Since(h.lastFlush) >= prefixBatchTimeout {
|
|
h.flushBatchLocked()
|
|
}
|
|
h.mu.Unlock()
|
|
case <-h.stopCh:
|
|
// Final flush
|
|
h.mu.Lock()
|
|
h.flushBatchLocked()
|
|
h.mu.Unlock()
|
|
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// flushBatchLocked flushes the prefix batch to the database (must be called with mutex held)
|
|
func (h *PrefixHandler) flushBatchLocked() {
|
|
if len(h.batch) == 0 {
|
|
return
|
|
}
|
|
|
|
startTime := time.Now()
|
|
batchSize := len(h.batch)
|
|
|
|
// For each prefix and peer, keep only the last update the feed delivered.
|
|
// Updates from different peers for the same prefix are all kept. Arrival
|
|
// order decides, not the timestamp: timestamps are whole seconds, so
|
|
// updates within one second carry the same one.
|
|
type prefixAndPeer struct{ prefix, peer string }
|
|
lastUpdates := make(map[prefixAndPeer]prefixUpdate)
|
|
for _, update := range h.batch {
|
|
lastUpdates[prefixAndPeer{update.prefix, update.peer}] = update
|
|
}
|
|
|
|
// Collect routes to upsert and delete
|
|
var routesToUpsert []*database.LiveRoute
|
|
var routesToDelete []database.LiveRouteDeletion
|
|
|
|
// Collect unique prefixes to update
|
|
prefixesToUpdate := make(map[string]time.Time)
|
|
|
|
for _, update := range lastUpdates {
|
|
// Track prefix for both announcements and withdrawals
|
|
if _, exists := prefixesToUpdate[update.prefix]; !exists || update.timestamp.After(prefixesToUpdate[update.prefix]) {
|
|
prefixesToUpdate[update.prefix] = update.timestamp
|
|
}
|
|
|
|
if update.messageType == "announcement" && update.originASN > 0 {
|
|
// Create live route for batch upsert
|
|
route := h.createLiveRoute(update)
|
|
if route != nil {
|
|
routesToUpsert = append(routesToUpsert, route)
|
|
}
|
|
} else if update.messageType == "withdrawal" {
|
|
// Parse CIDR to get IP version
|
|
_, ipVersion, err := parseCIDR(update.prefix)
|
|
if err != nil {
|
|
h.logger.Error("Failed to parse CIDR for withdrawal", "prefix", update.prefix, "error", err)
|
|
|
|
continue
|
|
}
|
|
|
|
// Create deletion record for batch delete
|
|
routesToDelete = append(routesToDelete, database.LiveRouteDeletion{
|
|
Prefix: update.prefix,
|
|
OriginASN: update.originASN,
|
|
PeerIP: update.peer,
|
|
IPVersion: ipVersion,
|
|
})
|
|
}
|
|
}
|
|
|
|
// Process batch operations
|
|
successCount := 0
|
|
if len(routesToUpsert) > 0 {
|
|
if err := h.db.UpsertLiveRouteBatch(routesToUpsert); err != nil {
|
|
h.logger.Error("Failed to upsert route batch", "error", err, "count", len(routesToUpsert))
|
|
} else {
|
|
successCount += len(routesToUpsert)
|
|
}
|
|
}
|
|
|
|
if len(routesToDelete) > 0 {
|
|
if err := h.db.DeleteLiveRouteBatch(routesToDelete); err != nil {
|
|
h.logger.Error("Failed to delete route batch", "error", err, "count", len(routesToDelete))
|
|
} else {
|
|
successCount += len(routesToDelete)
|
|
}
|
|
}
|
|
|
|
// Update prefix tables
|
|
if len(prefixesToUpdate) > 0 {
|
|
if err := h.db.UpdatePrefixesBatch(prefixesToUpdate); err != nil {
|
|
h.logger.Error("Failed to update prefix batch", "error", err, "count", len(prefixesToUpdate))
|
|
}
|
|
}
|
|
|
|
elapsed := time.Since(startTime)
|
|
h.logger.Debug("Flushed prefix batch",
|
|
"batch_size", batchSize,
|
|
"unique_prefixes", len(prefixesToUpdate),
|
|
"success", successCount,
|
|
"duration_ms", elapsed.Milliseconds(),
|
|
)
|
|
|
|
// Clear batch
|
|
h.batch = h.batch[:0]
|
|
h.lastFlush = time.Now()
|
|
}
|
|
|
|
// canonicalPrefix returns prefix in the text form net/netip prints, the form
|
|
// the IP lookup builds when it looks a prefix up. The feed sends IPv6
|
|
// announcements compressed ("2001:db8::/32") but IPv6 withdrawals uncompressed
|
|
// ("2001:db8:0:0:0:0:0:0/32"); stored as received, a withdrawal would not match
|
|
// the route its announcement stored. A prefix that does not parse is returned
|
|
// unchanged, and the batch flush reports it.
|
|
func canonicalPrefix(prefix string) string {
|
|
p, err := netip.ParsePrefix(prefix)
|
|
if err != nil {
|
|
return prefix
|
|
}
|
|
|
|
return p.Masked().String()
|
|
}
|
|
|
|
// parseCIDR extracts the mask length and IP version from a prefix string
|
|
func parseCIDR(prefix string) (maskLength int, ipVersion int, err error) {
|
|
_, ipNet, err := net.ParseCIDR(prefix)
|
|
if err != nil {
|
|
return 0, 0, err
|
|
}
|
|
|
|
ones, _ := ipNet.Mask.Size()
|
|
if strings.Contains(prefix, ":") {
|
|
return ones, ipv6Version, nil
|
|
}
|
|
|
|
return ones, ipv4Version, nil
|
|
}
|
|
|
|
// processAnnouncement handles storing an announcement in the database
|
|
// nolint:unused // kept for potential future use
|
|
func (h *PrefixHandler) processAnnouncement(_ *database.Prefix, update prefixUpdate) {
|
|
// Parse CIDR to get mask length
|
|
maskLength, ipVersion, err := parseCIDR(update.prefix)
|
|
if err != nil {
|
|
h.logger.Error("Failed to parse CIDR",
|
|
"prefix", update.prefix,
|
|
"error", err,
|
|
)
|
|
|
|
return
|
|
}
|
|
|
|
// Track route update metrics
|
|
if h.metrics != nil {
|
|
if ipVersion == ipv4Version {
|
|
h.metrics.RecordIPv4Update()
|
|
} else {
|
|
h.metrics.RecordIPv6Update()
|
|
}
|
|
}
|
|
|
|
// Create live route record
|
|
liveRoute := &database.LiveRoute{
|
|
ID: uuid.New(),
|
|
Prefix: update.prefix,
|
|
MaskLength: maskLength,
|
|
IPVersion: ipVersion,
|
|
OriginASN: update.originASN,
|
|
PeerIP: update.peer,
|
|
ASPath: update.path,
|
|
NextHop: update.peer, // Using peer as next hop
|
|
LastUpdated: update.timestamp,
|
|
}
|
|
|
|
if err := h.db.UpsertLiveRoute(liveRoute); err != nil {
|
|
h.logger.Error("Failed to upsert live route",
|
|
"prefix", update.prefix,
|
|
"error", err,
|
|
)
|
|
}
|
|
}
|
|
|
|
// createLiveRoute creates a LiveRoute from a prefix update
|
|
func (h *PrefixHandler) createLiveRoute(update prefixUpdate) *database.LiveRoute {
|
|
// Parse CIDR to get mask length
|
|
maskLength, ipVersion, err := parseCIDR(update.prefix)
|
|
if err != nil {
|
|
h.logger.Error("Failed to parse CIDR",
|
|
"prefix", update.prefix,
|
|
"error", err,
|
|
)
|
|
|
|
return nil
|
|
}
|
|
|
|
// Track route update metrics
|
|
if h.metrics != nil {
|
|
if ipVersion == ipv4Version {
|
|
h.metrics.RecordIPv4Update()
|
|
} else {
|
|
h.metrics.RecordIPv6Update()
|
|
}
|
|
}
|
|
|
|
// Create live route record
|
|
liveRoute := &database.LiveRoute{
|
|
ID: uuid.New(),
|
|
Prefix: update.prefix,
|
|
MaskLength: maskLength,
|
|
IPVersion: ipVersion,
|
|
OriginASN: update.originASN,
|
|
PeerIP: update.peer,
|
|
ASPath: update.path,
|
|
NextHop: update.peer, // Using peer as next hop
|
|
LastUpdated: update.timestamp,
|
|
}
|
|
|
|
return liveRoute
|
|
}
|
|
|
|
// processAnnouncementDirect handles storing an announcement directly without prefix table
|
|
// nolint:unused // kept for potential future use
|
|
func (h *PrefixHandler) processAnnouncementDirect(update prefixUpdate) {
|
|
// Parse CIDR to get mask length
|
|
maskLength, ipVersion, err := parseCIDR(update.prefix)
|
|
if err != nil {
|
|
h.logger.Error("Failed to parse CIDR",
|
|
"prefix", update.prefix,
|
|
"error", err,
|
|
)
|
|
|
|
return
|
|
}
|
|
|
|
// Track route update metrics
|
|
if h.metrics != nil {
|
|
if ipVersion == ipv4Version {
|
|
h.metrics.RecordIPv4Update()
|
|
} else {
|
|
h.metrics.RecordIPv6Update()
|
|
}
|
|
}
|
|
|
|
// Create live route record
|
|
liveRoute := &database.LiveRoute{
|
|
ID: uuid.New(),
|
|
Prefix: update.prefix,
|
|
MaskLength: maskLength,
|
|
IPVersion: ipVersion,
|
|
OriginASN: update.originASN,
|
|
PeerIP: update.peer,
|
|
ASPath: update.path,
|
|
NextHop: update.peer, // Using peer as next hop
|
|
LastUpdated: update.timestamp,
|
|
}
|
|
|
|
if err := h.db.UpsertLiveRoute(liveRoute); err != nil {
|
|
h.logger.Error("Failed to upsert live route",
|
|
"prefix", update.prefix,
|
|
"error", err,
|
|
)
|
|
}
|
|
}
|
|
|
|
// processWithdrawalDirect handles removing a route directly without prefix table
|
|
// nolint:unused // kept for potential future use
|
|
func (h *PrefixHandler) processWithdrawalDirect(update prefixUpdate) {
|
|
// For withdrawals, we need to delete the route from live_routes
|
|
if update.originASN > 0 {
|
|
if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil {
|
|
h.logger.Error("Failed to delete live route",
|
|
"prefix", update.prefix,
|
|
"origin_asn", update.originASN,
|
|
"peer", update.peer,
|
|
"error", err,
|
|
)
|
|
}
|
|
} else {
|
|
// If no origin ASN, just delete all routes for this prefix from this peer
|
|
if err := h.db.DeleteLiveRoute(update.prefix, 0, update.peer); err != nil {
|
|
h.logger.Error("Failed to delete live route (no origin ASN)",
|
|
"prefix", update.prefix,
|
|
"peer", update.peer,
|
|
"error", err,
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
// processWithdrawal handles removing a route from the live routing table
|
|
// nolint:unused // kept for potential future use
|
|
func (h *PrefixHandler) processWithdrawal(_ *database.Prefix, update prefixUpdate) {
|
|
// For withdrawals, we need to delete the route from live_routes
|
|
// Since we have the origin ASN from the update, we can delete the specific route
|
|
if update.originASN > 0 {
|
|
if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil {
|
|
h.logger.Error("Failed to delete live route",
|
|
"prefix", update.prefix,
|
|
"origin_asn", update.originASN,
|
|
"peer", update.peer,
|
|
"error", err,
|
|
)
|
|
}
|
|
} else {
|
|
// If no origin ASN, just delete all routes for this prefix from this peer
|
|
if err := h.db.DeleteLiveRoute(update.prefix, 0, update.peer); err != nil {
|
|
h.logger.Error("Failed to delete live route (no origin ASN)",
|
|
"prefix", update.prefix,
|
|
"peer", update.peer,
|
|
"error", err,
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Stop gracefully stops the handler and flushes remaining batches
|
|
func (h *PrefixHandler) Stop() {
|
|
close(h.stopCh)
|
|
h.wg.Wait()
|
|
}
|