Store a new prefix's earliest timestamp as first_seen (closes #112)
check / check (push) Waiting to run
check / check (push) Waiting to run
Each handler took the latest time per key in a batch and inserted it as a new row's first_seen, so a key updated twice in one batch, seconds apart, was stored first seen at the later time, and the IP lookup could answer with a prefix first_seen after the route's last_seen. The prefix handler now keeps the earliest time of all of the batch's updates for each prefix. The AS, peer and peering handlers keep the earliest and the latest time per key in database.SeenTimes, and GetOrCreateASNBatch, UpdatePeerBatch and RecordPeering store a new row first seen at the earliest and last seen at the latest. A stored row still has only its last_seen set, to the latest. Model: opus-5-5
This commit is contained in:
@@ -148,7 +148,9 @@ func TestLiveCountsTrackWritesInRealtime(t *testing.T) {
|
|||||||
func writeASNsPrefixesAndPeers(t *testing.T, db *Database, ts time.Time) {
|
func writeASNsPrefixesAndPeers(t *testing.T, db *Database, ts time.Time) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
err := db.GetOrCreateASNBatch(map[int]time.Time{64500: ts, 64501: ts})
|
seen := SeenTimes{First: ts, Last: ts}
|
||||||
|
|
||||||
|
err := db.GetOrCreateASNBatch(map[int]SeenTimes{64500: seen, 64501: seen})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("GetOrCreateASNBatch: %v", err)
|
t.Fatalf("GetOrCreateASNBatch: %v", err)
|
||||||
}
|
}
|
||||||
@@ -164,14 +166,14 @@ func writeASNsPrefixesAndPeers(t *testing.T, db *Database, ts time.Time) {
|
|||||||
err = db.UpdatePeerBatch(map[string]PeerUpdate{
|
err = db.UpdatePeerBatch(map[string]PeerUpdate{
|
||||||
testV4PeerIP: {
|
testV4PeerIP: {
|
||||||
PeerIP: testV4PeerIP, PeerASN: 64500,
|
PeerIP: testV4PeerIP, PeerASN: 64500,
|
||||||
MessageType: "UPDATE", Timestamp: ts,
|
MessageType: "UPDATE", Seen: seen,
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("UpdatePeerBatch: %v", err)
|
t.Fatalf("UpdatePeerBatch: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
err = db.RecordPeering(64500, 64501, ts)
|
err = db.RecordPeering(64500, 64501, seen)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("RecordPeering: %v", err)
|
t.Fatalf("RecordPeering: %v", err)
|
||||||
}
|
}
|
||||||
@@ -192,7 +194,10 @@ func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) {
|
|||||||
|
|
||||||
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
|
||||||
err = db.GetOrCreateASNBatch(map[int]time.Time{64500: ts, 64501: ts, 64502: ts})
|
seen := SeenTimes{First: ts, Last: ts}
|
||||||
|
|
||||||
|
err = db.GetOrCreateASNBatch(
|
||||||
|
map[int]SeenTimes{64500: seen, 64501: seen, 64502: seen})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("GetOrCreateASNBatch: %v", err)
|
t.Fatalf("GetOrCreateASNBatch: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -903,13 +903,14 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// newASN returns a new ASN record first and last seen at timestamp, with the
|
// newASN returns a new ASN record first seen at seen.First and last seen at
|
||||||
// handle and description from the AS info lookup when it knows the number.
|
// seen.Last, with the handle and description from the AS info lookup when it
|
||||||
func newASN(number int, timestamp time.Time) ASN {
|
// knows the number.
|
||||||
|
func newASN(number int, seen SeenTimes) ASN {
|
||||||
asn := ASN{
|
asn := ASN{
|
||||||
ASN: number,
|
ASN: number,
|
||||||
FirstSeen: timestamp,
|
FirstSeen: seen.First,
|
||||||
LastSeen: timestamp,
|
LastSeen: seen.Last,
|
||||||
}
|
}
|
||||||
|
|
||||||
// Look up ASN info
|
// Look up ASN info
|
||||||
@@ -921,12 +922,12 @@ func newASN(number int, timestamp time.Time) ASN {
|
|||||||
return asn
|
return asn
|
||||||
}
|
}
|
||||||
|
|
||||||
// upsertASN updates the last_seen time of ASN number with the given
|
// upsertASN sets the last_seen time of ASN number to seen.Last with the given
|
||||||
// statements, inserting the ASN when it is new. It reports whether it inserted
|
// statements, inserting the ASN when it is new. It reports whether it inserted
|
||||||
// it.
|
// it.
|
||||||
func upsertASN(
|
func upsertASN(
|
||||||
ctx context.Context, selectStmt, updateStmt, insertStmt *sql.Stmt,
|
ctx context.Context, selectStmt, updateStmt, insertStmt *sql.Stmt,
|
||||||
number int, timestamp time.Time,
|
number int, seen SeenTimes,
|
||||||
) (bool, error) {
|
) (bool, error) {
|
||||||
var (
|
var (
|
||||||
asn ASN
|
asn ASN
|
||||||
@@ -937,7 +938,7 @@ func upsertASN(
|
|||||||
&description, &asn.FirstSeen, &asn.LastSeen)
|
&description, &asn.FirstSeen, &asn.LastSeen)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
// ASN exists, update last_seen
|
// ASN exists, update last_seen
|
||||||
_, err = updateStmt.ExecContext(ctx, timestamp, number)
|
_, err = updateStmt.ExecContext(ctx, seen.Last, number)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, fmt.Errorf("failed to update ASN %d: %w", number, err)
|
return false, fmt.Errorf("failed to update ASN %d: %w", number, err)
|
||||||
}
|
}
|
||||||
@@ -950,7 +951,7 @@ func upsertASN(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// ASN doesn't exist, create it
|
// ASN doesn't exist, create it
|
||||||
asn = newASN(number, timestamp)
|
asn = newASN(number, seen)
|
||||||
|
|
||||||
_, err = insertStmt.ExecContext(ctx, asn.ASN, asn.Handle,
|
_, err = insertStmt.ExecContext(ctx, asn.ASN, asn.Handle,
|
||||||
asn.Description, asn.FirstSeen, asn.LastSeen)
|
asn.Description, asn.FirstSeen, asn.LastSeen)
|
||||||
@@ -962,7 +963,7 @@ func upsertASN(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// GetOrCreateASNBatch creates or updates multiple ASNs in a single transaction
|
// GetOrCreateASNBatch creates or updates multiple ASNs in a single transaction
|
||||||
func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
|
func (d *Database) GetOrCreateASNBatch(asns map[int]SeenTimes) error {
|
||||||
if len(asns) == 0 {
|
if len(asns) == 0 {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -1004,9 +1005,9 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
|
|||||||
|
|
||||||
var newASNs int
|
var newASNs int
|
||||||
|
|
||||||
for number, timestamp := range asns {
|
for number, seen := range asns {
|
||||||
inserted, err := upsertASN(ctx, selectStmt, updateStmt, insertStmt,
|
inserted, err := upsertASN(ctx, selectStmt, updateStmt, insertStmt,
|
||||||
number, timestamp)
|
number, seen)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -1075,7 +1076,7 @@ func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error)
|
|||||||
}
|
}
|
||||||
|
|
||||||
// ASN doesn't exist, create it with ASN info lookup
|
// ASN doesn't exist, create it with ASN info lookup
|
||||||
asn = newASN(number, timestamp)
|
asn = newASN(number, SeenTimes{First: timestamp, Last: timestamp})
|
||||||
|
|
||||||
_, err = tx.Exec("INSERT INTO asns "+
|
_, err = tx.Exec("INSERT INTO asns "+
|
||||||
"(asn, handle, description, first_seen, last_seen) "+
|
"(asn, handle, description, first_seen, last_seen) "+
|
||||||
@@ -1182,7 +1183,7 @@ func (d *Database) RecordAnnouncement(announcement *Announcement) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// RecordPeering records a peering relationship between two ASNs.
|
// RecordPeering records a peering relationship between two ASNs.
|
||||||
func (d *Database) RecordPeering(asA, asB int, timestamp time.Time) error {
|
func (d *Database) RecordPeering(asA, asB int, seen SeenTimes) error {
|
||||||
// Validate ASNs
|
// Validate ASNs
|
||||||
if asA <= 0 || asB <= 0 {
|
if asA <= 0 || asB <= 0 {
|
||||||
return fmt.Errorf("%w: asA=%d, asB=%d", errInvalidASN, asA, asB)
|
return fmt.Errorf("%w: asA=%d, asB=%d", errInvalidASN, asA, asB)
|
||||||
@@ -1218,12 +1219,12 @@ func (d *Database) RecordPeering(asA, asB int, timestamp time.Time) error {
|
|||||||
if exists {
|
if exists {
|
||||||
_, err = tx.Exec(
|
_, err = tx.Exec(
|
||||||
"UPDATE peerings SET last_seen = ? WHERE as_a = ? AND as_b = ?",
|
"UPDATE peerings SET last_seen = ? WHERE as_a = ? AND as_b = ?",
|
||||||
timestamp, asA, asB)
|
seen.Last, asA, asB)
|
||||||
} else {
|
} else {
|
||||||
_, err = tx.Exec(`
|
_, err = tx.Exec(`
|
||||||
INSERT INTO peerings (id, as_a, as_b, first_seen, last_seen)
|
INSERT INTO peerings (id, as_a, as_b, first_seen, last_seen)
|
||||||
VALUES (?, ?, ?, ?, ?)`,
|
VALUES (?, ?, ?, ?, ?)`,
|
||||||
generateUUID().String(), asA, asB, timestamp, timestamp)
|
generateUUID().String(), asA, asB, seen.First, seen.Last)
|
||||||
}
|
}
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -1301,11 +1302,11 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error {
|
|||||||
|
|
||||||
if exists {
|
if exists {
|
||||||
_, err = updateStmt.ExecContext(ctx, update.PeerASN,
|
_, err = updateStmt.ExecContext(ctx, update.PeerASN,
|
||||||
update.Timestamp, update.MessageType, update.PeerIP)
|
update.Seen.Last, update.MessageType, update.PeerIP)
|
||||||
} else {
|
} else {
|
||||||
_, err = insertStmt.ExecContext(ctx,
|
_, err = insertStmt.ExecContext(ctx,
|
||||||
generateUUID().String(), update.PeerIP, update.PeerASN,
|
generateUUID().String(), update.PeerIP, update.PeerASN,
|
||||||
update.Timestamp, update.Timestamp, update.MessageType)
|
update.Seen.First, update.Seen.Last, update.MessageType)
|
||||||
}
|
}
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -615,10 +615,12 @@ func TestBatchWriteDuringCheckpoint(t *testing.T) {
|
|||||||
})
|
})
|
||||||
|
|
||||||
ts := time.Now().UTC()
|
ts := time.Now().UTC()
|
||||||
|
seen := SeenTimes{First: ts, Last: ts}
|
||||||
|
|
||||||
for i := range contentionIterations {
|
for i := range contentionIterations {
|
||||||
asns := map[int]time.Time{
|
asns := map[int]SeenTimes{
|
||||||
i % contendedASNCount: ts,
|
i % contendedASNCount: seen,
|
||||||
(i % contendedASNCount) + asnSecondBand: ts,
|
(i % contendedASNCount) + asnSecondBand: seen,
|
||||||
}
|
}
|
||||||
|
|
||||||
err := db.GetOrCreateASNBatch(asns)
|
err := db.GetOrCreateASNBatch(asns)
|
||||||
@@ -747,7 +749,8 @@ func TestBatchWriteWaitsForAnalyze(t *testing.T) {
|
|||||||
|
|
||||||
db, messages := newDatabaseWithLockMessages(t)
|
db, messages := newDatabaseWithLockMessages(t)
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
asns := map[int]time.Time{lockTestASN: time.Now().UTC()}
|
now := time.Now().UTC()
|
||||||
|
asns := map[int]SeenTimes{lockTestASN: {First: now, Last: now}}
|
||||||
|
|
||||||
// A stored AS gives ANALYZE a table to analyze and statistics to write.
|
// A stored AS gives ANALYZE a table to analyze and statistics to write.
|
||||||
err := db.GetOrCreateASNBatch(asns)
|
err := db.GetOrCreateASNBatch(asns)
|
||||||
@@ -1663,6 +1666,48 @@ func TestUpdatePrefixesBatchKeepsFirstSeen(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestUpdatePeerBatchStoresTheEarliestAndLatestTimesOfANewPeer writes a new
|
||||||
|
// peer whose updates span 5 seconds. Its row is first seen at the earliest time
|
||||||
|
// and last seen at the latest.
|
||||||
|
func TestUpdatePeerBatchStoresTheEarliestAndLatestTimesOfANewPeer(
|
||||||
|
t *testing.T,
|
||||||
|
) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to create database: %v", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = db.Close() }()
|
||||||
|
|
||||||
|
first := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
last := first.Add(5 * time.Second)
|
||||||
|
|
||||||
|
err = db.UpdatePeerBatch(map[string]PeerUpdate{
|
||||||
|
testPeerIP: {
|
||||||
|
PeerIP: testPeerIP, PeerASN: 64500, MessageType: "UPDATE",
|
||||||
|
Seen: SeenTimes{First: first, Last: last},
|
||||||
|
},
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("UpdatePeerBatch: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
var firstSeen, lastSeen time.Time
|
||||||
|
|
||||||
|
err = db.db.QueryRowContext(t.Context(),
|
||||||
|
"SELECT first_seen, last_seen FROM bgp_peers WHERE peer_ip = ?",
|
||||||
|
testPeerIP).Scan(&firstSeen, &lastSeen)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("reading peer %s: %v", testPeerIP, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if !firstSeen.Equal(first) || !lastSeen.Equal(last) {
|
||||||
|
t.Errorf("peer %s: first seen %s, last seen %s; want %s, %s",
|
||||||
|
testPeerIP, firstSeen, lastSeen, first, last)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// TestUpsertLiveRouteBatchWritesInKeyOrder writes new routes of two peers out
|
// TestUpsertLiveRouteBatchWritesInKeyOrder writes new routes of two peers out
|
||||||
// of key order, the last one a second announcement of the first, and checks
|
// of key order, the last one a second announcement of the first, and checks
|
||||||
// that the rows were inserted in key order, peer then prefix (each new row's id
|
// that the rows were inserted in key order, peer then prefix (each new row's id
|
||||||
|
|||||||
@@ -34,7 +34,7 @@ type Stats struct {
|
|||||||
type Store interface {
|
type Store interface {
|
||||||
// ASN operations
|
// ASN operations
|
||||||
GetOrCreateASN(number int, timestamp time.Time) (*ASN, error)
|
GetOrCreateASN(number int, timestamp time.Time) (*ASN, error)
|
||||||
GetOrCreateASNBatch(asns map[int]time.Time) error
|
GetOrCreateASNBatch(asns map[int]SeenTimes) error
|
||||||
|
|
||||||
// Prefix operations
|
// Prefix operations
|
||||||
GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefix, error)
|
GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefix, error)
|
||||||
@@ -44,7 +44,7 @@ type Store interface {
|
|||||||
RecordAnnouncement(announcement *Announcement) error
|
RecordAnnouncement(announcement *Announcement) error
|
||||||
|
|
||||||
// Peering operations
|
// Peering operations
|
||||||
RecordPeering(asA, asB int, timestamp time.Time) error
|
RecordPeering(asA, asB int, seen SeenTimes) error
|
||||||
|
|
||||||
// Statistics
|
// Statistics
|
||||||
GetStats() (Stats, error)
|
GetStats() (Stats, error)
|
||||||
|
|||||||
@@ -122,12 +122,35 @@ type LiveRouteDeletion struct {
|
|||||||
IPVersion int
|
IPVersion int
|
||||||
}
|
}
|
||||||
|
|
||||||
// PeerUpdate represents parameters for updating a peer
|
// SeenTimes is the earliest and the latest time one batch of updates saw an AS,
|
||||||
|
// a peer or a peering. A row the batch adds is stored first seen at First and
|
||||||
|
// last seen at Last; a row already stored keeps its first_seen and has its
|
||||||
|
// last_seen set to Last.
|
||||||
|
type SeenTimes struct {
|
||||||
|
First time.Time
|
||||||
|
Last time.Time
|
||||||
|
}
|
||||||
|
|
||||||
|
// Add makes t the earliest time of s when it is earlier than First, and the
|
||||||
|
// latest when it is later than Last. The zero SeenTimes holds no time yet, so
|
||||||
|
// the first Add sets both.
|
||||||
|
func (s *SeenTimes) Add(t time.Time) {
|
||||||
|
if s.First.IsZero() || t.Before(s.First) {
|
||||||
|
s.First = t
|
||||||
|
}
|
||||||
|
|
||||||
|
if s.Last.IsZero() || t.After(s.Last) {
|
||||||
|
s.Last = t
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// PeerUpdate represents parameters for updating a peer. PeerASN and
|
||||||
|
// MessageType are those of the peer's latest update.
|
||||||
type PeerUpdate struct {
|
type PeerUpdate struct {
|
||||||
PeerIP string
|
PeerIP string
|
||||||
PeerASN int
|
PeerASN int
|
||||||
MessageType string
|
MessageType string
|
||||||
Timestamp time.Time
|
Seen SeenTimes
|
||||||
}
|
}
|
||||||
|
|
||||||
// ASNWHOISUpdate contains WHOIS data for updating an ASN record.
|
// ASNWHOISUpdate contains WHOIS data for updating an ASN record.
|
||||||
|
|||||||
@@ -152,7 +152,7 @@ func (m *mockStore) RecordAnnouncement(_ *database.Announcement) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// RecordPeering mock implementation
|
// RecordPeering mock implementation
|
||||||
func (m *mockStore) RecordPeering(asA, asB int, _ time.Time) error {
|
func (m *mockStore) RecordPeering(asA, asB int, _ database.SeenTimes) error {
|
||||||
m.mu.Lock()
|
m.mu.Lock()
|
||||||
defer m.mu.Unlock()
|
defer m.mu.Unlock()
|
||||||
|
|
||||||
@@ -396,16 +396,16 @@ func (m *mockStore) DeleteLiveRoutesOfPeer(_ string) (int, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// GetOrCreateASNBatch mock implementation
|
// GetOrCreateASNBatch mock implementation
|
||||||
func (m *mockStore) GetOrCreateASNBatch(asns map[int]time.Time) error {
|
func (m *mockStore) GetOrCreateASNBatch(asns map[int]database.SeenTimes) error {
|
||||||
m.mu.Lock()
|
m.mu.Lock()
|
||||||
defer m.mu.Unlock()
|
defer m.mu.Unlock()
|
||||||
|
|
||||||
for number, timestamp := range asns {
|
for number, seen := range asns {
|
||||||
if _, exists := m.ASNs[number]; !exists {
|
if _, exists := m.ASNs[number]; !exists {
|
||||||
m.ASNs[number] = &database.ASN{
|
m.ASNs[number] = &database.ASN{
|
||||||
ASN: number,
|
ASN: number,
|
||||||
FirstSeen: timestamp,
|
FirstSeen: seen.First,
|
||||||
LastSeen: timestamp,
|
LastSeen: seen.Last,
|
||||||
}
|
}
|
||||||
m.ASNCount++
|
m.ASNCount++
|
||||||
}
|
}
|
||||||
@@ -766,7 +766,9 @@ const (
|
|||||||
callClose = "close"
|
callClose = "close"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (s *closeOrderStore) GetOrCreateASNBatch(asns map[int]time.Time) error {
|
func (s *closeOrderStore) GetOrCreateASNBatch(
|
||||||
|
asns map[int]database.SeenTimes,
|
||||||
|
) error {
|
||||||
s.record(callWrite)
|
s.record(callWrite)
|
||||||
|
|
||||||
return s.Store.GetOrCreateASNBatch(asns)
|
return s.Store.GetOrCreateASNBatch(asns)
|
||||||
|
|||||||
@@ -229,19 +229,21 @@ func (h *ASHandler) flushBatch() {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Process ASNs first (deduped)
|
// The earliest and the latest time the batch saw each AS. The batch holds
|
||||||
asnMap := make(map[int]time.Time)
|
// several updates for an AS, from many peers, so a new AS is stored first
|
||||||
|
// seen at the earliest of them and last seen at the latest.
|
||||||
|
asns := make(map[int]database.SeenTimes)
|
||||||
for _, op := range batch {
|
for _, op := range batch {
|
||||||
if existing, ok := asnMap[op.number]; !ok || op.timestamp.After(existing) {
|
seen := asns[op.number]
|
||||||
asnMap[op.number] = op.timestamp
|
seen.Add(op.timestamp)
|
||||||
}
|
asns[op.number] = seen
|
||||||
}
|
}
|
||||||
|
|
||||||
// Process all ASNs in a single batch transaction
|
// Process all ASNs in a single batch transaction
|
||||||
err := h.db.GetOrCreateASNBatch(asnMap)
|
err := h.db.GetOrCreateASNBatch(asns)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
h.logger.Error("Failed to process ASN batch",
|
h.logger.Error("Failed to process ASN batch",
|
||||||
"error", err, "count", len(asnMap))
|
"error", err, "count", len(asns))
|
||||||
}
|
}
|
||||||
|
|
||||||
h.mu.Lock()
|
h.mu.Lock()
|
||||||
|
|||||||
@@ -0,0 +1,44 @@
|
|||||||
|
package routewatch
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"git.eeqj.de/sneak/routewatch/internal/logger"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS handles two updates
|
||||||
|
// whose path carries an AS the database does not hold yet, the second 5
|
||||||
|
// seconds after the first, in one batch into a real database. The AS is stored
|
||||||
|
// first seen at the time of the first update and last seen at the time of the
|
||||||
|
// second.
|
||||||
|
func TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
db := newTestDatabase(t)
|
||||||
|
|
||||||
|
// Built without NewASHandler's flush goroutine, so the test flushes itself.
|
||||||
|
h := &ASHandler{db: db, logger: logger.New()}
|
||||||
|
|
||||||
|
first := peerAnnouncement(testPeerIP, testV4Prefix)
|
||||||
|
second := peerAnnouncement(testPeerIP, testV4Prefix)
|
||||||
|
second.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter)
|
||||||
|
|
||||||
|
h.HandleMessage(first)
|
||||||
|
h.HandleMessage(second)
|
||||||
|
h.flushBatch()
|
||||||
|
|
||||||
|
asn, _, err := db.GetASDetails(testASNA)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetASDetails(%d): %v", testASNA, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if !asn.FirstSeen.Equal(first.ParsedTimestamp) {
|
||||||
|
t.Errorf("first_seen of AS%d = %s, want %s", testASNA,
|
||||||
|
asn.FirstSeen, first.ParsedTimestamp)
|
||||||
|
}
|
||||||
|
|
||||||
|
if !asn.LastSeen.Equal(second.ParsedTimestamp) {
|
||||||
|
t.Errorf("last_seen of AS%d = %s, want %s", testASNA,
|
||||||
|
asn.LastSeen, second.ParsedTimestamp)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -66,7 +66,7 @@ func (s *blockingStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *blockingStore) GetOrCreateASNBatch(asns map[int]time.Time) error {
|
func (s *blockingStore) GetOrCreateASNBatch(asns map[int]database.SeenTimes) error {
|
||||||
s.waitForRelease()
|
s.waitForRelease()
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
defer s.mu.Unlock()
|
defer s.mu.Unlock()
|
||||||
@@ -96,10 +96,12 @@ func (s *blockingStore) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
|
|||||||
return s.mockStore.DeleteLiveRoutesOfPeer(peerIP)
|
return s.mockStore.DeleteLiveRoutesOfPeer(peerIP)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *blockingStore) RecordPeering(asA, asB int, ts time.Time) error {
|
func (s *blockingStore) RecordPeering(
|
||||||
|
asA, asB int, seen database.SeenTimes,
|
||||||
|
) error {
|
||||||
s.waitForRelease()
|
s.waitForRelease()
|
||||||
|
|
||||||
return s.mockStore.RecordPeering(asA, asB, ts)
|
return s.mockStore.RecordPeering(asA, asB, seen)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *blockingStore) waitForRelease() {
|
func (s *blockingStore) waitForRelease() {
|
||||||
|
|||||||
@@ -37,8 +37,8 @@ const (
|
|||||||
|
|
||||||
// PeerHandler tracks BGP peers from UPDATE messages using batched operations.
|
// PeerHandler tracks BGP peers from UPDATE messages using batched operations.
|
||||||
// It maintains a queue of peer updates and periodically flushes them to the database
|
// It maintains a queue of peer updates and periodically flushes them to the database
|
||||||
// in batches to improve performance. The handler deduplicates peer updates within
|
// in batches to improve performance. The handler writes one row per peer IP
|
||||||
// each batch, keeping only the most recent update for each peer IP address.
|
// address for each batch, with the earliest and the latest time of its updates.
|
||||||
type PeerHandler struct {
|
type PeerHandler struct {
|
||||||
db database.Store
|
db database.Store
|
||||||
logger *logger.Logger
|
logger *logger.Logger
|
||||||
@@ -227,31 +227,28 @@ func (h *PeerHandler) flushBatch() {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Deduplicate by peer IP, keeping the latest update for each peer
|
// The earliest and the latest time the batch saw each peer, with the peer
|
||||||
peerMap := make(map[string]peerUpdate)
|
// AS and message type of its latest update. The batch holds several
|
||||||
|
// updates for a peer, so a new peer is stored first seen at the earliest of
|
||||||
|
// them and last seen at the latest.
|
||||||
|
peers := make(map[string]database.PeerUpdate)
|
||||||
for _, update := range batch {
|
for _, update := range batch {
|
||||||
existing, ok := peerMap[update.peerIP]
|
peer, ok := peers[update.peerIP]
|
||||||
if !ok || update.timestamp.After(existing.timestamp) {
|
if !ok || update.timestamp.After(peer.Seen.Last) {
|
||||||
peerMap[update.peerIP] = update
|
peer.PeerIP = update.peerIP
|
||||||
|
peer.PeerASN = update.peerASN
|
||||||
|
peer.MessageType = update.messageType
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
// Convert to database format
|
peer.Seen.Add(update.timestamp)
|
||||||
dbPeerMap := make(map[string]database.PeerUpdate)
|
peers[update.peerIP] = peer
|
||||||
for peerIP, update := range peerMap {
|
|
||||||
dbPeerMap[peerIP] = database.PeerUpdate{
|
|
||||||
PeerIP: update.peerIP,
|
|
||||||
PeerASN: update.peerASN,
|
|
||||||
MessageType: update.messageType,
|
|
||||||
Timestamp: update.timestamp,
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Process all peers in a single batch transaction
|
// Process all peers in a single batch transaction
|
||||||
err := h.db.UpdatePeerBatch(dbPeerMap)
|
err := h.db.UpdatePeerBatch(peers)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
h.logger.Error("Failed to process peer batch",
|
h.logger.Error("Failed to process peer batch",
|
||||||
"error", err, "count", len(dbPeerMap))
|
"error", err, "count", len(peers))
|
||||||
}
|
}
|
||||||
|
|
||||||
h.mu.Lock()
|
h.mu.Lock()
|
||||||
|
|||||||
@@ -0,0 +1,59 @@
|
|||||||
|
package routewatch
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"git.eeqj.de/sneak/routewatch/internal/database"
|
||||||
|
"git.eeqj.de/sneak/routewatch/internal/logger"
|
||||||
|
)
|
||||||
|
|
||||||
|
// peerBatchStore keeps the peers of the last UpdatePeerBatch call. The
|
||||||
|
// database has no read of the peer table, so the peer handler's test checks
|
||||||
|
// what the handler writes.
|
||||||
|
type peerBatchStore struct {
|
||||||
|
*mockStore
|
||||||
|
|
||||||
|
peers map[string]database.PeerUpdate
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *peerBatchStore) UpdatePeerBatch(
|
||||||
|
peers map[string]database.PeerUpdate,
|
||||||
|
) error {
|
||||||
|
s.peers = peers
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestPeerHandlerWritesTheEarliestAndLatestTimesOfANewPeer handles two updates
|
||||||
|
// from one peer, the second 5 seconds after the first, in one batch. The peer
|
||||||
|
// handler writes the peer first seen at the time of the first update and last
|
||||||
|
// seen at the time of the second.
|
||||||
|
func TestPeerHandlerWritesTheEarliestAndLatestTimesOfANewPeer(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
store := &peerBatchStore{mockStore: newMockStore()}
|
||||||
|
|
||||||
|
// Built without NewPeerHandler's flush goroutine, so the test flushes
|
||||||
|
// itself.
|
||||||
|
h := &PeerHandler{db: store, logger: logger.New()}
|
||||||
|
|
||||||
|
first := peerAnnouncement(testPeerIP, testV4Prefix)
|
||||||
|
second := peerAnnouncement(testPeerIP, testV4Prefix)
|
||||||
|
second.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter)
|
||||||
|
|
||||||
|
h.HandleMessage(first)
|
||||||
|
h.HandleMessage(second)
|
||||||
|
h.flushBatch()
|
||||||
|
|
||||||
|
seen := store.peers[testPeerIP].Seen
|
||||||
|
|
||||||
|
if !seen.First.Equal(first.ParsedTimestamp) {
|
||||||
|
t.Errorf("first_seen of peer %s = %s, want %s", testPeerIP,
|
||||||
|
seen.First, first.ParsedTimestamp)
|
||||||
|
}
|
||||||
|
|
||||||
|
if !seen.Last.Equal(second.ParsedTimestamp) {
|
||||||
|
t.Errorf("last_seen of peer %s = %s, want %s", testPeerIP,
|
||||||
|
seen.Last, second.ParsedTimestamp)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -42,10 +42,11 @@ type PeeringHandler struct {
|
|||||||
db database.Store
|
db database.Store
|
||||||
logger *logger.Logger
|
logger *logger.Logger
|
||||||
|
|
||||||
// In-memory AS path tracking
|
// In-memory AS path tracking: each path, JSON-encoded, with the earliest
|
||||||
|
// and the latest time an update carried it.
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
asPaths map[string]time.Time // key is JSON-encoded AS path
|
asPaths map[string]database.SeenTimes
|
||||||
droppedPaths int // paths dropped because the map was full
|
droppedPaths int // paths dropped because the map was full
|
||||||
// writing is how many paths processPeerings took from asPaths and has
|
// writing is how many paths processPeerings took from asPaths and has
|
||||||
// not finished writing the peerings of.
|
// not finished writing the peerings of.
|
||||||
writing int
|
writing int
|
||||||
@@ -64,7 +65,7 @@ func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler
|
|||||||
h := &PeeringHandler{
|
h := &PeeringHandler{
|
||||||
db: db,
|
db: db,
|
||||||
logger: logger,
|
logger: logger,
|
||||||
asPaths: make(map[string]time.Time),
|
asPaths: make(map[string]database.SeenTimes),
|
||||||
stopCh: make(chan context.Context, 1),
|
stopCh: make(chan context.Context, 1),
|
||||||
done: make(chan struct{}),
|
done: make(chan struct{}),
|
||||||
}
|
}
|
||||||
@@ -118,16 +119,18 @@ func (h *PeeringHandler) HandleMessage(msg *ristypes.RISMessage) {
|
|||||||
key := string(pathJSON)
|
key := string(pathJSON)
|
||||||
|
|
||||||
h.mu.Lock()
|
h.mu.Lock()
|
||||||
if _, exists := h.asPaths[key]; exists {
|
defer h.mu.Unlock()
|
||||||
// Already tracked: refresh its timestamp.
|
|
||||||
h.asPaths[key] = timestamp
|
seen, exists := h.asPaths[key]
|
||||||
} else if len(h.asPaths) >= maxTrackedPaths {
|
if !exists && len(h.asPaths) >= maxTrackedPaths {
|
||||||
// Map is full; drop this new path and count it.
|
// Map is full; drop this new path and count it.
|
||||||
h.droppedPaths++
|
h.droppedPaths++
|
||||||
} else {
|
|
||||||
h.asPaths[key] = timestamp
|
return
|
||||||
}
|
}
|
||||||
h.mu.Unlock()
|
|
||||||
|
seen.Add(timestamp)
|
||||||
|
h.asPaths[key] = seen
|
||||||
}
|
}
|
||||||
|
|
||||||
// ProcessPeeringsNow triggers immediate processing of all accumulated AS
|
// ProcessPeeringsNow triggers immediate processing of all accumulated AS
|
||||||
@@ -190,7 +193,7 @@ func (h *PeeringHandler) processPeerings() {
|
|||||||
// released, so the map never grows past a single interval's traffic.
|
// released, so the map never grows past a single interval's traffic.
|
||||||
h.mu.Lock()
|
h.mu.Lock()
|
||||||
paths := h.asPaths
|
paths := h.asPaths
|
||||||
h.asPaths = make(map[string]time.Time)
|
h.asPaths = make(map[string]database.SeenTimes)
|
||||||
h.writing = len(paths)
|
h.writing = len(paths)
|
||||||
dropped := h.droppedPaths
|
dropped := h.droppedPaths
|
||||||
h.mu.Unlock()
|
h.mu.Unlock()
|
||||||
@@ -204,9 +207,13 @@ func (h *PeeringHandler) processPeerings() {
|
|||||||
low, high int
|
low, high int
|
||||||
}
|
}
|
||||||
|
|
||||||
peerings := make(map[peeringKey]time.Time)
|
// The earliest and the latest time an update carried each peering. Several
|
||||||
|
// paths, each carried at several times, can hold a peering, so a new
|
||||||
|
// peering is stored first seen at the earliest of them and last seen at the
|
||||||
|
// latest.
|
||||||
|
peerings := make(map[peeringKey]database.SeenTimes)
|
||||||
|
|
||||||
for pathJSON, timestamp := range paths {
|
for pathJSON, pathSeen := range paths {
|
||||||
var path []int
|
var path []int
|
||||||
|
|
||||||
err := json.Unmarshal([]byte(pathJSON), &path)
|
err := json.Unmarshal([]byte(pathJSON), &path)
|
||||||
@@ -233,10 +240,11 @@ func (h *PeeringHandler) processPeerings() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
key := peeringKey{low: low, high: high}
|
key := peeringKey{low: low, high: high}
|
||||||
// Update timestamp if this is newer
|
|
||||||
if existing, ok := peerings[key]; !ok || timestamp.After(existing) {
|
seen := peerings[key]
|
||||||
peerings[key] = timestamp
|
seen.Add(pathSeen.First)
|
||||||
}
|
seen.Add(pathSeen.Last)
|
||||||
|
peerings[key] = seen
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -244,8 +252,8 @@ func (h *PeeringHandler) processPeerings() {
|
|||||||
start := time.Now()
|
start := time.Now()
|
||||||
successCount := 0
|
successCount := 0
|
||||||
|
|
||||||
for key, ts := range peerings {
|
for key, seen := range peerings {
|
||||||
err := h.db.RecordPeering(key.low, key.high, ts)
|
err := h.db.RecordPeering(key.low, key.high, seen)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
h.logger.Error("Failed to record peering",
|
h.logger.Error("Failed to record peering",
|
||||||
"as_a", key.low,
|
"as_a", key.low,
|
||||||
|
|||||||
@@ -31,12 +31,14 @@ type recordingStore struct {
|
|||||||
calls int
|
calls int
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *recordingStore) RecordPeering(asA, asB int, ts time.Time) error {
|
func (r *recordingStore) RecordPeering(
|
||||||
|
asA, asB int, seen database.SeenTimes,
|
||||||
|
) error {
|
||||||
r.mu.Lock()
|
r.mu.Lock()
|
||||||
r.calls++
|
r.calls++
|
||||||
r.mu.Unlock()
|
r.mu.Unlock()
|
||||||
|
|
||||||
return r.mockStore.RecordPeering(asA, asB, ts)
|
return r.mockStore.RecordPeering(asA, asB, seen)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *recordingStore) callCount() int {
|
func (r *recordingStore) callCount() int {
|
||||||
@@ -52,7 +54,7 @@ func newTestHandler(db database.Store) *PeeringHandler {
|
|||||||
return &PeeringHandler{
|
return &PeeringHandler{
|
||||||
db: db,
|
db: db,
|
||||||
logger: logger.New(),
|
logger: logger.New(),
|
||||||
asPaths: make(map[string]time.Time),
|
asPaths: make(map[string]database.SeenTimes),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -220,10 +222,10 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
|
|||||||
existing := pathKey(t, testASNA, testASNB)
|
existing := pathKey(t, testASNA, testASNB)
|
||||||
|
|
||||||
h.mu.Lock()
|
h.mu.Lock()
|
||||||
h.asPaths[existing] = now
|
h.asPaths[existing] = database.SeenTimes{First: now, Last: now}
|
||||||
|
|
||||||
for i := 0; len(h.asPaths) < maxTrackedPaths; i++ {
|
for i := 0; len(h.asPaths) < maxTrackedPaths; i++ {
|
||||||
h.asPaths[strconv.Itoa(i)] = now
|
h.asPaths[strconv.Itoa(i)] = database.SeenTimes{First: now, Last: now}
|
||||||
}
|
}
|
||||||
h.mu.Unlock()
|
h.mu.Unlock()
|
||||||
|
|
||||||
@@ -253,11 +255,53 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
h.mu.Lock()
|
h.mu.Lock()
|
||||||
gotTS := h.asPaths[existing]
|
gotSeen := h.asPaths[existing]
|
||||||
h.mu.Unlock()
|
h.mu.Unlock()
|
||||||
|
|
||||||
if !gotTS.Equal(refreshed) {
|
if !gotSeen.Last.Equal(refreshed) {
|
||||||
t.Fatalf("existing path timestamp not refreshed: want %v, got %v",
|
t.Fatalf("existing path last seen not refreshed: want %v, got %v",
|
||||||
refreshed, gotTS)
|
refreshed, gotSeen.Last)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestPeeringHandlerStoresTheEarliestAndLatestTimesOfANewPeering handles two
|
||||||
|
// updates whose path carries a peering the database does not hold yet, the
|
||||||
|
// second 5 seconds after the first, in one run into a real database. The
|
||||||
|
// peering is stored first seen at the time of the first update and last seen
|
||||||
|
// at the time of the second.
|
||||||
|
func TestPeeringHandlerStoresTheEarliestAndLatestTimesOfANewPeering(
|
||||||
|
t *testing.T,
|
||||||
|
) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
db := newTestDatabase(t)
|
||||||
|
h := newTestHandler(db)
|
||||||
|
|
||||||
|
first := peerAnnouncement(testPeerIP, testV4Prefix)
|
||||||
|
second := peerAnnouncement(testPeerIP, testV4Prefix)
|
||||||
|
second.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter)
|
||||||
|
|
||||||
|
h.HandleMessage(first)
|
||||||
|
h.HandleMessage(second)
|
||||||
|
h.ProcessPeeringsNow()
|
||||||
|
|
||||||
|
peers, err := db.GetASPeers(testASNA)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetASPeers(%d): %v", testASNA, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(peers) != 1 || peers[0].ASN != testASNB {
|
||||||
|
t.Fatalf("peers of AS%d = %+v, want AS%d alone", testASNA, peers,
|
||||||
|
testASNB)
|
||||||
|
}
|
||||||
|
|
||||||
|
if !peers[0].FirstSeen.Equal(first.ParsedTimestamp) {
|
||||||
|
t.Errorf("first_seen of peering AS%d-AS%d = %s, want %s", testASNA,
|
||||||
|
testASNB, peers[0].FirstSeen, first.ParsedTimestamp)
|
||||||
|
}
|
||||||
|
|
||||||
|
if !peers[0].LastSeen.Equal(second.ParsedTimestamp) {
|
||||||
|
t.Errorf("last_seen of peering AS%d-AS%d = %s, want %s", testASNA,
|
||||||
|
testASNB, peers[0].LastSeen, second.ParsedTimestamp)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -347,8 +347,19 @@ func (h *PrefixHandler) flushBatch() {
|
|||||||
|
|
||||||
lastUpdates := make(map[prefixAndPeer]prefixUpdate)
|
lastUpdates := make(map[prefixAndPeer]prefixUpdate)
|
||||||
|
|
||||||
|
// Each prefix the batch names, announced or withdrawn, with the earliest
|
||||||
|
// timestamp among all of its updates, not only each peer's last one: the
|
||||||
|
// prefix tables store only first_seen, written when the prefix is new and
|
||||||
|
// never changed after.
|
||||||
|
prefixesToUpdate := make(map[string]time.Time)
|
||||||
|
|
||||||
for _, update := range batch {
|
for _, update := range batch {
|
||||||
lastUpdates[prefixAndPeer{update.prefix, update.peer}] = update
|
lastUpdates[prefixAndPeer{update.prefix, update.peer}] = update
|
||||||
|
|
||||||
|
seen, exists := prefixesToUpdate[update.prefix]
|
||||||
|
if !exists || update.timestamp.Before(seen) {
|
||||||
|
prefixesToUpdate[update.prefix] = update.timestamp
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Collect routes to upsert and delete
|
// Collect routes to upsert and delete
|
||||||
@@ -357,16 +368,7 @@ func (h *PrefixHandler) flushBatch() {
|
|||||||
routesToDelete []database.LiveRouteDeletion
|
routesToDelete []database.LiveRouteDeletion
|
||||||
)
|
)
|
||||||
|
|
||||||
// Collect unique prefixes to update
|
|
||||||
prefixesToUpdate := make(map[string]time.Time)
|
|
||||||
|
|
||||||
for _, update := range lastUpdates {
|
for _, update := range lastUpdates {
|
||||||
// Track prefix for both announcements and withdrawals
|
|
||||||
seen, exists := prefixesToUpdate[update.prefix]
|
|
||||||
if !exists || update.timestamp.After(seen) {
|
|
||||||
prefixesToUpdate[update.prefix] = update.timestamp
|
|
||||||
}
|
|
||||||
|
|
||||||
// An announcement is stored when its path gives an origin, or when it
|
// An announcement is stored when its path gives an origin, or when it
|
||||||
// ends in an AS set of several members: that route is stored with
|
// ends in an AS set of several members: that route is stored with
|
||||||
// origin 0 (none). An announcement with an empty path is not stored.
|
// origin 0 (none). An announcement with an empty path is not stored.
|
||||||
|
|||||||
@@ -26,6 +26,10 @@ const (
|
|||||||
testV6Prefix = "2001:db8:1::/48"
|
testV6Prefix = "2001:db8:1::/48"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// secondUpdateAfter is how long after the first update the tests of first_seen
|
||||||
|
// and last_seen send the second, in the same batch.
|
||||||
|
const secondUpdateAfter = 5 * time.Second
|
||||||
|
|
||||||
// TestPrefixHandlerStoresPrefixesTheIPLookupFinds runs announcements and
|
// TestPrefixHandlerStoresPrefixesTheIPLookupFinds runs announcements and
|
||||||
// withdrawals through the prefix handler into a real database and looks the
|
// withdrawals through the prefix handler into a real database and looks the
|
||||||
// addresses up. The prefixes are written the way the feed sends them: IPv6
|
// addresses up. The prefixes are written the way the feed sends them: IPv6
|
||||||
@@ -83,6 +87,49 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestPrefixHandlerStoresANewPrefixFirstSeenAtItsEarliestUpdate announces a
|
||||||
|
// prefix the database does not hold yet twice in one batch, the second time 5
|
||||||
|
// seconds after the first, through the prefix handler into a real database. The
|
||||||
|
// prefix is stored first seen at the time of the first announcement, whether
|
||||||
|
// the two come from two peers or from one, and the IP lookup answers with that
|
||||||
|
// time.
|
||||||
|
func TestPrefixHandlerStoresANewPrefixFirstSeenAtItsEarliestUpdate(
|
||||||
|
t *testing.T,
|
||||||
|
) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
const otherPeerIP = "2001:db8:ffff::2"
|
||||||
|
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
firstPeer, secondPeer string
|
||||||
|
}{
|
||||||
|
{name: "two peers", firstPeer: testPeerIP, secondPeer: otherPeerIP},
|
||||||
|
{name: "one peer", firstPeer: testPeerIP, secondPeer: testPeerIP},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
first := peerAnnouncement(tt.firstPeer, testV4Prefix)
|
||||||
|
second := peerAnnouncement(tt.secondPeer, testV4Prefix)
|
||||||
|
second.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter)
|
||||||
|
db := handleBatches(t, [][]*ristypes.RISMessage{{first, second}})
|
||||||
|
|
||||||
|
info, err := db.GetIPInfo("192.0.2.1")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetIPInfo(192.0.2.1): %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if !info.FirstSeen.Equal(first.ParsedTimestamp) {
|
||||||
|
t.Errorf("first_seen of %s = %s, want %s", testV4Prefix,
|
||||||
|
info.FirstSeen, first.ParsedTimestamp)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// TestPrefixHandlerAppliesEachPeersLastUpdateInABatch runs updates for one
|
// TestPrefixHandlerAppliesEachPeersLastUpdateInABatch runs updates for one
|
||||||
// prefix through the prefix handler into a real database. All updates carry the
|
// prefix through the prefix handler into a real database. All updates carry the
|
||||||
// same timestamp, as updates within one second of the feed do. Within a batch,
|
// same timestamp, as updates within one second of the feed do. Within a batch,
|
||||||
|
|||||||
Reference in New Issue
Block a user