325 lines
8.2 KiB
Go
325 lines
8.2 KiB
Go
// Package routewatch contains the ASN WHOIS fetcher for background updates.
|
|
package routewatch
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.eeqj.de/sneak/routewatch/internal/database"
|
|
"git.eeqj.de/sneak/routewatch/internal/server"
|
|
"git.eeqj.de/sneak/routewatch/internal/whois"
|
|
)
|
|
|
|
// ASN fetcher configuration constants.
|
|
const (
|
|
// baseInterval is the starting interval between fetch attempts.
|
|
baseInterval = 15 * time.Second
|
|
|
|
// minInterval is the minimum interval after successes (rate limit).
|
|
minInterval = 1 * time.Second
|
|
|
|
// maxInterval is the maximum interval after failures (backoff cap).
|
|
maxInterval = 5 * time.Minute
|
|
|
|
// backoffMultiplier is how much to multiply interval on failure.
|
|
backoffMultiplier = 2
|
|
|
|
// whoisStaleThreshold is how old WHOIS data can be before refresh.
|
|
whoisStaleThreshold = 30 * 24 * time.Hour // 30 days
|
|
|
|
// immediateQueueSize is the buffer size for immediate fetch requests.
|
|
immediateQueueSize = 100
|
|
|
|
// statsWindow is how long to keep stats for.
|
|
statsWindow = time.Hour
|
|
)
|
|
|
|
// ASNFetcher handles background WHOIS lookups for ASNs.
|
|
type ASNFetcher struct {
|
|
db database.Store
|
|
whoisClient *whois.Client
|
|
logger *slog.Logger
|
|
immediateQueue chan int
|
|
stopCh chan struct{}
|
|
wg sync.WaitGroup
|
|
|
|
// fetchMu ensures only one fetch runs at a time
|
|
fetchMu sync.Mutex
|
|
|
|
// interval tracking with mutex protection
|
|
intervalMu sync.Mutex
|
|
currentInterval time.Duration
|
|
consecutiveFails int
|
|
|
|
// hourly stats tracking
|
|
statsMu sync.Mutex
|
|
successTimes []time.Time
|
|
errorTimes []time.Time
|
|
}
|
|
|
|
// NewASNFetcher creates a new ASN fetcher.
|
|
func NewASNFetcher(db database.Store, logger *slog.Logger) *ASNFetcher {
|
|
return &ASNFetcher{
|
|
db: db,
|
|
whoisClient: whois.NewClient(),
|
|
logger: logger.With("component", "asn_fetcher"),
|
|
immediateQueue: make(chan int, immediateQueueSize),
|
|
stopCh: make(chan struct{}),
|
|
currentInterval: baseInterval,
|
|
successTimes: make([]time.Time, 0),
|
|
errorTimes: make([]time.Time, 0),
|
|
}
|
|
}
|
|
|
|
// Start begins the background ASN fetcher goroutine.
|
|
func (f *ASNFetcher) Start() {
|
|
f.wg.Add(1)
|
|
go f.run()
|
|
f.logger.Info("ASN fetcher started",
|
|
"base_interval", baseInterval,
|
|
"min_interval", minInterval,
|
|
"max_interval", maxInterval,
|
|
)
|
|
}
|
|
|
|
// Stop gracefully shuts down the fetcher.
|
|
func (f *ASNFetcher) Stop() {
|
|
close(f.stopCh)
|
|
f.wg.Wait()
|
|
f.logger.Info("ASN fetcher stopped")
|
|
}
|
|
|
|
// QueueImmediate queues an ASN for immediate WHOIS lookup.
|
|
// Non-blocking - if queue is full, the request is dropped.
|
|
func (f *ASNFetcher) QueueImmediate(asn int) {
|
|
select {
|
|
case f.immediateQueue <- asn:
|
|
f.logger.Debug("Queued immediate WHOIS lookup", "asn", asn)
|
|
default:
|
|
f.logger.Debug("Immediate queue full, dropping request", "asn", asn)
|
|
}
|
|
}
|
|
|
|
// GetStats returns statistics about fetcher activity.
|
|
func (f *ASNFetcher) GetStats() server.ASNFetcherStats {
|
|
f.statsMu.Lock()
|
|
defer f.statsMu.Unlock()
|
|
|
|
f.intervalMu.Lock()
|
|
interval := f.currentInterval
|
|
fails := f.consecutiveFails
|
|
f.intervalMu.Unlock()
|
|
|
|
// Prune old entries and count
|
|
cutoff := time.Now().Add(-statsWindow)
|
|
f.successTimes = pruneOldTimes(f.successTimes, cutoff)
|
|
f.errorTimes = pruneOldTimes(f.errorTimes, cutoff)
|
|
|
|
return server.ASNFetcherStats{
|
|
SuccessesLastHour: len(f.successTimes),
|
|
ErrorsLastHour: len(f.errorTimes),
|
|
CurrentInterval: interval,
|
|
ConsecutiveFails: fails,
|
|
}
|
|
}
|
|
|
|
// pruneOldTimes removes times older than cutoff and returns the pruned slice.
|
|
func pruneOldTimes(times []time.Time, cutoff time.Time) []time.Time {
|
|
result := make([]time.Time, 0, len(times))
|
|
for _, t := range times {
|
|
if t.After(cutoff) {
|
|
result = append(result, t)
|
|
}
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
// getInterval returns the current fetch interval.
|
|
func (f *ASNFetcher) getInterval() time.Duration {
|
|
f.intervalMu.Lock()
|
|
defer f.intervalMu.Unlock()
|
|
|
|
return f.currentInterval
|
|
}
|
|
|
|
// recordSuccess decreases the interval on successful fetch.
|
|
func (f *ASNFetcher) recordSuccess() {
|
|
f.intervalMu.Lock()
|
|
f.consecutiveFails = 0
|
|
|
|
// Decrease interval by half, but not below minimum
|
|
newInterval := f.currentInterval / backoffMultiplier
|
|
if newInterval < minInterval {
|
|
newInterval = minInterval
|
|
}
|
|
|
|
if newInterval != f.currentInterval {
|
|
f.logger.Debug("Decreased fetch interval",
|
|
"old_interval", f.currentInterval,
|
|
"new_interval", newInterval,
|
|
)
|
|
f.currentInterval = newInterval
|
|
}
|
|
f.intervalMu.Unlock()
|
|
|
|
// Record success time for stats
|
|
f.statsMu.Lock()
|
|
f.successTimes = append(f.successTimes, time.Now())
|
|
f.statsMu.Unlock()
|
|
}
|
|
|
|
// recordFailure increases the interval on failed fetch using exponential backoff.
|
|
func (f *ASNFetcher) recordFailure() {
|
|
f.intervalMu.Lock()
|
|
f.consecutiveFails++
|
|
|
|
// Exponential backoff: multiply by 2, capped at max
|
|
newInterval := f.currentInterval * backoffMultiplier
|
|
if newInterval > maxInterval {
|
|
newInterval = maxInterval
|
|
}
|
|
|
|
if newInterval != f.currentInterval {
|
|
f.logger.Debug("Increased fetch interval due to failure",
|
|
"old_interval", f.currentInterval,
|
|
"new_interval", newInterval,
|
|
"consecutive_failures", f.consecutiveFails,
|
|
)
|
|
f.currentInterval = newInterval
|
|
}
|
|
f.intervalMu.Unlock()
|
|
|
|
// Record error time for stats
|
|
f.statsMu.Lock()
|
|
f.errorTimes = append(f.errorTimes, time.Now())
|
|
f.statsMu.Unlock()
|
|
}
|
|
|
|
// run is the main background loop.
|
|
func (f *ASNFetcher) run() {
|
|
defer f.wg.Done()
|
|
|
|
timer := time.NewTimer(f.getInterval())
|
|
defer timer.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-f.stopCh:
|
|
return
|
|
|
|
case asn := <-f.immediateQueue:
|
|
// Process immediate request (respects lock)
|
|
f.tryFetch(asn)
|
|
// Reset timer after immediate fetch
|
|
timer.Reset(f.getInterval())
|
|
|
|
case <-timer.C:
|
|
// Background fetch of stale/missing ASN
|
|
f.fetchNextStale()
|
|
// Reset timer with potentially updated interval
|
|
timer.Reset(f.getInterval())
|
|
}
|
|
}
|
|
}
|
|
|
|
// tryFetch attempts to fetch and update an ASN, respecting the fetch lock.
|
|
// Returns true if fetch was successful.
|
|
func (f *ASNFetcher) tryFetch(asn int) bool {
|
|
// Try to acquire lock, skip if another fetch is running
|
|
if !f.fetchMu.TryLock() {
|
|
f.logger.Debug("Skipping fetch, another fetch in progress", "asn", asn)
|
|
|
|
return false
|
|
}
|
|
defer f.fetchMu.Unlock()
|
|
|
|
return f.fetchAndUpdate(asn)
|
|
}
|
|
|
|
// fetchNextStale finds and fetches the next ASN needing WHOIS data.
|
|
func (f *ASNFetcher) fetchNextStale() {
|
|
// Try to acquire lock, skip if another fetch is running
|
|
if !f.fetchMu.TryLock() {
|
|
f.logger.Debug("Skipping stale fetch, another fetch in progress")
|
|
|
|
return
|
|
}
|
|
defer f.fetchMu.Unlock()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
|
|
asn, err := f.db.GetNextStaleASN(ctx, whoisStaleThreshold)
|
|
if err != nil {
|
|
if err != database.ErrNoStaleASN {
|
|
f.logger.Error("Failed to get stale ASN", "error", err)
|
|
f.recordFailure()
|
|
}
|
|
// No stale ASN is not a failure, just nothing to do
|
|
|
|
return
|
|
}
|
|
|
|
f.fetchAndUpdate(asn)
|
|
}
|
|
|
|
// fetchAndUpdate performs a WHOIS lookup and updates the database.
|
|
// Returns true if successful.
|
|
func (f *ASNFetcher) fetchAndUpdate(asn int) bool {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
|
defer cancel()
|
|
|
|
f.logger.Info("Fetching WHOIS data", "asn", asn)
|
|
|
|
info, err := f.whoisClient.LookupASN(ctx, asn)
|
|
if err != nil {
|
|
f.logger.Error("WHOIS lookup failed", "asn", asn, "error", err)
|
|
f.recordFailure()
|
|
|
|
return false
|
|
}
|
|
|
|
// Update database with WHOIS data
|
|
err = f.db.UpdateASNWHOIS(ctx, &database.ASNWHOISUpdate{
|
|
ASN: asn,
|
|
ASName: info.ASName,
|
|
OrgName: info.OrgName,
|
|
OrgID: info.OrgID,
|
|
Address: info.Address,
|
|
CountryCode: info.CountryCode,
|
|
AbuseEmail: info.AbuseEmail,
|
|
AbusePhone: info.AbusePhone,
|
|
TechEmail: info.TechEmail,
|
|
TechPhone: info.TechPhone,
|
|
RIR: info.RIR,
|
|
RIRRegDate: info.RegDate,
|
|
RIRLastMod: info.LastMod,
|
|
WHOISRaw: info.RawResponse,
|
|
})
|
|
if err != nil {
|
|
f.logger.Error("Failed to update ASN WHOIS data", "asn", asn, "error", err)
|
|
f.recordFailure()
|
|
|
|
return false
|
|
}
|
|
|
|
f.recordSuccess()
|
|
f.logger.Info("Updated ASN WHOIS data",
|
|
"asn", asn,
|
|
"org_name", info.OrgName,
|
|
"country", info.CountryCode,
|
|
"rir", info.RIR,
|
|
"next_interval", f.getInterval(),
|
|
)
|
|
|
|
return true
|
|
}
|
|
|
|
// GetStaleThreshold returns the WHOIS stale threshold duration.
|
|
func GetStaleThreshold() time.Duration {
|
|
return whoisStaleThreshold
|
|
}
|