Never move last seen back or first seen forward (closes #162)
check / check (push) Canceled after 0s
check / check (push) Canceled after 0s
Some RIS collectors send updates days old while they catch up on a backlog. Every update to a stored AS, peering or peer set last_seen to the update's time, even an older one, and never lowered first_seen. An AS page could then list a peer first seen after it was last seen. The update statements for asns, peerings and bgp_peers now keep the earlier of the stored and new first_seen and the later of the stored and new last_seen, with MIN and MAX in the SQL. The batch and the single-row writes share each statement. GetOrCreateASN returns the times the row now holds. Route times are unchanged. Model: opus-5-5
This commit was merged in pull request #170.
This commit is contained in:
@@ -945,6 +945,26 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// These statements update a stored AS, peering or peer. Each keeps the earlier
|
||||
// of the stored and the new first_seen, and the later of the stored and the new
|
||||
// last_seen. Some collectors send updates days old while they catch up on a
|
||||
// backlog. MIN and MAX compare the times as text, which is time order here: the
|
||||
// driver writes every time in one format, and every time is UTC.
|
||||
const (
|
||||
updateASNSQL = `UPDATE asns
|
||||
SET first_seen = MIN(first_seen, ?), last_seen = MAX(last_seen, ?)
|
||||
WHERE asn = ?`
|
||||
|
||||
updatePeeringSQL = `UPDATE peerings
|
||||
SET first_seen = MIN(first_seen, ?), last_seen = MAX(last_seen, ?)
|
||||
WHERE as_a = ? AND as_b = ?`
|
||||
|
||||
updatePeerSQL = `UPDATE bgp_peers
|
||||
SET peer_asn = ?, last_message_type = ?,
|
||||
first_seen = MIN(first_seen, ?), last_seen = MAX(last_seen, ?)
|
||||
WHERE peer_ip = ?`
|
||||
)
|
||||
|
||||
// newASN returns a new ASN record first seen at seen.First and last seen at
|
||||
// seen.Last, with the handle and description from the AS info lookup when it
|
||||
// knows the number.
|
||||
@@ -964,9 +984,8 @@ func newASN(number int, seen SeenTimes) ASN {
|
||||
return asn
|
||||
}
|
||||
|
||||
// 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
|
||||
// it.
|
||||
// upsertASN stores the seen times of ASN number with the given statements,
|
||||
// inserting the ASN when it is new. It reports whether it inserted it.
|
||||
func upsertASN(
|
||||
ctx context.Context, selectStmt, updateStmt, insertStmt *sql.Stmt,
|
||||
number int, seen SeenTimes,
|
||||
@@ -979,8 +998,8 @@ func upsertASN(
|
||||
err := selectStmt.QueryRowContext(ctx, number).Scan(&asn.ASN, &handle,
|
||||
&description, &asn.FirstSeen, &asn.LastSeen)
|
||||
if err == nil {
|
||||
// ASN exists, update last_seen
|
||||
_, err = updateStmt.ExecContext(ctx, seen.Last, number)
|
||||
// ASN exists, update its seen times
|
||||
_, err = updateStmt.ExecContext(ctx, seen.First, seen.Last, number)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("failed to update ASN %d: %w", number, err)
|
||||
}
|
||||
@@ -1030,8 +1049,7 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]SeenTimes) error {
|
||||
}
|
||||
defer func() { _ = selectStmt.Close() }()
|
||||
|
||||
updateStmt, err := tx.PrepareContext(ctx,
|
||||
"UPDATE asns SET last_seen = ? WHERE asn = ?")
|
||||
updateStmt, err := tx.PrepareContext(ctx, updateASNSQL)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to prepare update statement: %w", err)
|
||||
}
|
||||
@@ -1090,17 +1108,19 @@ func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error)
|
||||
"FROM asns WHERE asn = ?", number).
|
||||
Scan(&asn.ASN, &handle, &description, &asn.FirstSeen, &asn.LastSeen)
|
||||
if err == nil {
|
||||
// ASN exists, update last_seen
|
||||
// ASN exists, update its seen times
|
||||
asn.Handle = handle.String
|
||||
asn.Description = description.String
|
||||
|
||||
_, err = tx.Exec("UPDATE asns SET last_seen = ? WHERE asn = ?",
|
||||
timestamp, number)
|
||||
_, err = tx.Exec(updateASNSQL, timestamp, timestamp, number)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
asn.LastSeen = timestamp
|
||||
// Return the seen times the update stored.
|
||||
seen := SeenTimes{First: asn.FirstSeen, Last: asn.LastSeen}
|
||||
seen.Add(timestamp)
|
||||
asn.FirstSeen, asn.LastSeen = seen.First, seen.Last
|
||||
|
||||
err = tx.Commit()
|
||||
if err != nil {
|
||||
@@ -1259,9 +1279,7 @@ func (d *Database) RecordPeering(asA, asB int, seen SeenTimes) error {
|
||||
}
|
||||
|
||||
if exists {
|
||||
_, err = tx.Exec(
|
||||
"UPDATE peerings SET last_seen = ? WHERE as_a = ? AND as_b = ?",
|
||||
seen.Last, asA, asB)
|
||||
_, err = tx.Exec(updatePeeringSQL, seen.First, seen.Last, asA, asB)
|
||||
} else {
|
||||
_, err = tx.Exec(`
|
||||
INSERT INTO peerings (id, as_a, as_b, first_seen, last_seen)
|
||||
@@ -1316,9 +1334,7 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error {
|
||||
}
|
||||
defer func() { _ = checkStmt.Close() }()
|
||||
|
||||
updateStmt, err := tx.PrepareContext(ctx,
|
||||
"UPDATE bgp_peers SET peer_asn = ?, last_seen = ?, "+
|
||||
"last_message_type = ? WHERE peer_ip = ?")
|
||||
updateStmt, err := tx.PrepareContext(ctx, updatePeerSQL)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to prepare update statement: %w", err)
|
||||
}
|
||||
@@ -1344,7 +1360,8 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error {
|
||||
|
||||
if exists {
|
||||
_, err = updateStmt.ExecContext(ctx, update.PeerASN,
|
||||
update.Seen.Last, update.MessageType, update.PeerIP)
|
||||
update.MessageType, update.Seen.First, update.Seen.Last,
|
||||
update.PeerIP)
|
||||
} else {
|
||||
_, err = insertStmt.ExecContext(ctx,
|
||||
generateUUID().String(), update.PeerIP, update.PeerASN,
|
||||
@@ -1393,11 +1410,8 @@ func (d *Database) UpdatePeer(
|
||||
}
|
||||
|
||||
if exists {
|
||||
_, err = tx.Exec(
|
||||
"UPDATE bgp_peers SET peer_asn = ?, last_seen = ?, "+
|
||||
"last_message_type = ? WHERE peer_ip = ?",
|
||||
peerASN, timestamp, messageType, peerIP,
|
||||
)
|
||||
_, err = tx.Exec(updatePeerSQL,
|
||||
peerASN, messageType, timestamp, timestamp, peerIP)
|
||||
} else {
|
||||
_, err = tx.Exec(
|
||||
"INSERT INTO bgp_peers (id, peer_ip, peer_asn, first_seen, "+
|
||||
|
||||
@@ -20,6 +20,7 @@ import (
|
||||
|
||||
"git.eeqj.de/sneak/routewatch/internal/config"
|
||||
"git.eeqj.de/sneak/routewatch/internal/logger"
|
||||
"git.eeqj.de/sneak/routewatch/internal/ristypes"
|
||||
"git.eeqj.de/sneak/routewatch/pkg/asinfo"
|
||||
"github.com/mattn/go-sqlite3"
|
||||
)
|
||||
@@ -1960,6 +1961,158 @@ func TestUpdatePeerBatchStoresTheEarliestAndLatestTimesOfANewPeer(
|
||||
}
|
||||
}
|
||||
|
||||
// backlogAge is how much older one write is than the other in the tests of
|
||||
// first_seen and last_seen, as when a collector catches up on a backlog.
|
||||
const backlogAge = 24 * time.Hour
|
||||
|
||||
// batchSpan is how much earlier a batch's First is than its Last in the tests
|
||||
// of first_seen and last_seen.
|
||||
const batchSpan = time.Hour
|
||||
|
||||
// TestWritesKeepEarliestFirstSeenAndLatestLastSeen writes a row of each of
|
||||
// asns, peerings and bgp_peers through each function that writes one, at a
|
||||
// newer time and at an older one. Each row takes the older write's First as
|
||||
// first_seen and the newer write's Last as last_seen. The test writes in both
|
||||
// orders, each in a database of its own. Newer then older is a collector
|
||||
// catching up on a backlog; older then newer is a feed that arrives in order.
|
||||
// Each function writes a row of its own. A batch write's First is batchSpan
|
||||
// before its Last, so the test catches the two swapped. A single-row write
|
||||
// takes one time.
|
||||
func TestWritesKeepEarliestFirstSeenAndLatestLastSeen(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const (
|
||||
batchASN, singleASN = 64500, 64501
|
||||
peerASN = 64502
|
||||
asnQuery = "SELECT first_seen, last_seen FROM asns WHERE asn = ?"
|
||||
peeringQuery = "SELECT first_seen, last_seen FROM peerings " +
|
||||
"WHERE as_a = ? AND as_b = ?"
|
||||
peerQuery = "SELECT first_seen, last_seen FROM bgp_peers " +
|
||||
"WHERE peer_ip = ?"
|
||||
)
|
||||
|
||||
for name, olderFirst := range map[string]bool{
|
||||
"newer then older": false,
|
||||
"older then newer": true,
|
||||
} {
|
||||
t.Run(name, func(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() }()
|
||||
|
||||
checkSeenTimes(t, db, olderFirst, func(seen SeenTimes) error {
|
||||
return db.GetOrCreateASNBatch(map[int]SeenTimes{batchASN: seen})
|
||||
}, batchSpan, asnQuery, batchASN)
|
||||
|
||||
checkSeenTimes(t, db, olderFirst, func(seen SeenTimes) error {
|
||||
_, err := db.GetOrCreateASN(singleASN, seen.Last)
|
||||
|
||||
return err
|
||||
}, 0, asnQuery, singleASN)
|
||||
|
||||
checkSeenTimes(t, db, olderFirst, func(seen SeenTimes) error {
|
||||
return db.RecordPeering(batchASN, singleASN, seen)
|
||||
}, batchSpan, peeringQuery, batchASN, singleASN)
|
||||
|
||||
checkSeenTimes(t, db, olderFirst, func(seen SeenTimes) error {
|
||||
return db.UpdatePeerBatch(map[string]PeerUpdate{
|
||||
testV4PeerIP: {
|
||||
PeerIP: testV4PeerIP, PeerASN: peerASN,
|
||||
MessageType: ristypes.MessageTypeUpdate,
|
||||
Seen: seen,
|
||||
},
|
||||
})
|
||||
}, batchSpan, peerQuery, testV4PeerIP)
|
||||
|
||||
checkSeenTimes(t, db, olderFirst, func(seen SeenTimes) error {
|
||||
return db.UpdatePeer(testPeerIP, peerASN,
|
||||
ristypes.MessageTypeUpdate, seen.Last)
|
||||
}, 0, peerQuery, testPeerIP)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// checkSeenTimes calls write twice, each time with First span before Last: at
|
||||
// newer times and at times backlogAge older. It writes the older times first
|
||||
// when olderFirst is true. It checks that query, given args, reads the older
|
||||
// write's First as first_seen and the newer write's Last as last_seen.
|
||||
func checkSeenTimes(
|
||||
t *testing.T, db *Database, olderFirst bool,
|
||||
write func(seen SeenTimes) error,
|
||||
span time.Duration, query string, args ...any,
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
newerLast := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||
newer := SeenTimes{First: newerLast.Add(-span), Last: newerLast}
|
||||
older := SeenTimes{
|
||||
First: newer.First.Add(-backlogAge),
|
||||
Last: newer.Last.Add(-backlogAge),
|
||||
}
|
||||
|
||||
writes := []SeenTimes{newer, older}
|
||||
if olderFirst {
|
||||
writes = []SeenTimes{older, newer}
|
||||
}
|
||||
|
||||
for _, seen := range writes {
|
||||
err := write(seen)
|
||||
if err != nil {
|
||||
t.Fatalf("writing first %s, last %s: %v", seen.First, seen.Last, err)
|
||||
}
|
||||
}
|
||||
|
||||
var firstSeen, lastSeen time.Time
|
||||
|
||||
err := db.db.QueryRowContext(t.Context(), query, args...).
|
||||
Scan(&firstSeen, &lastSeen)
|
||||
if err != nil {
|
||||
t.Fatalf("reading the row: %v", err)
|
||||
}
|
||||
|
||||
if !firstSeen.Equal(older.First) || !lastSeen.Equal(newer.Last) {
|
||||
t.Errorf("first seen %s, last seen %s; want %s, %s",
|
||||
firstSeen, lastSeen, older.First, newer.Last)
|
||||
}
|
||||
}
|
||||
|
||||
// TestGetOrCreateASNReturnsStoredSeenTimes stores an AS at a newer time, then
|
||||
// at an older one. The second call returns the times the row now holds: the
|
||||
// older first_seen and the newer last_seen.
|
||||
func TestGetOrCreateASNReturnsStoredSeenTimes(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() }()
|
||||
|
||||
const asn = 64500
|
||||
|
||||
newer := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||
older := newer.Add(-backlogAge)
|
||||
|
||||
_, err = db.GetOrCreateASN(asn, newer)
|
||||
if err != nil {
|
||||
t.Fatalf("GetOrCreateASN at %s: %v", newer, err)
|
||||
}
|
||||
|
||||
got, err := db.GetOrCreateASN(asn, older)
|
||||
if err != nil {
|
||||
t.Fatalf("GetOrCreateASN at %s: %v", older, err)
|
||||
}
|
||||
|
||||
if !got.FirstSeen.Equal(older) || !got.LastSeen.Equal(newer) {
|
||||
t.Errorf("first seen %s, last seen %s; want %s, %s",
|
||||
got.FirstSeen, got.LastSeen, older, newer)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUpsertLiveRouteBatchWritesInKeyOrder writes new routes of two peers out
|
||||
// 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
|
||||
|
||||
@@ -125,8 +125,8 @@ type LiveRouteDeletion struct {
|
||||
|
||||
// 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.
|
||||
// last seen at Last. A row already stored takes First as its first_seen only
|
||||
// when First is earlier, and Last as its last_seen only when Last is later.
|
||||
type SeenTimes struct {
|
||||
First time.Time
|
||||
Last time.Time
|
||||
|
||||
Reference in New Issue
Block a user