Never move last seen back or first seen forward (closes #162)
check / check (push) Waiting to run
check / check (push) Waiting to run
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 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, "+
|
||||
|
||||
Reference in New Issue
Block a user