sneak/routewatch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: sneak:main
Branches Tags
sneak:main
sneak:fix-min-time-calculation
sneak:optimize-sqlite-settings
..
compare: sneak:fix-min-time-calculation
Branches Tags
sneak:main
sneak:fix-min-time-calculation
sneak:optimize-sqlite-settings

No commits in common. "main" and "fix-min-time-calculation" have entirely different histories.
main ... fix-min-ti

17 changed files with 178692 additions and 207617 deletions
Show Stats Expand all files Collapse all files

644
internal/database/database.go
View File

@ -22,9 +22,6 @@ import (
_ "github.com/mattn/go-sqlite3" // CGO SQLite driver _ "github.com/mattn/go-sqlite3" // CGO SQLite driver
) )
// IMPORTANT: NO schema changes are to be made outside of schema.sql
// We do NOT support migrations. All schema changes MUST be made in schema.sql only.
//
//go:embed schema.sql //go:embed schema.sql
var dbSchema string var dbSchema string
@ -194,42 +191,26 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error {
} }
}() }()
// Prepare statements for both IPv4 and IPv6 // Use prepared statement for better performance
queryV4 := ` query := `
INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, INSERT INTO live_routes (id, prefix, mask_length, ip_version, origin_asn, peer_ip, as_path, next_hop,
last_updated, ip_start, ip_end) last_updated, v4_ip_start, v4_ip_end)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET
mask_length = excluded.mask_length, mask_length = excluded.mask_length,
ip_version = excluded.ip_version,
as_path = excluded.as_path, as_path = excluded.as_path,
next_hop = excluded.next_hop, next_hop = excluded.next_hop,
last_updated = excluded.last_updated, last_updated = excluded.last_updated,
ip_start = excluded.ip_start, v4_ip_start = excluded.v4_ip_start,
ip_end = excluded.ip_end v4_ip_end = excluded.v4_ip_end
` `
queryV6 := ` stmt, err := tx.Prepare(query)
INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
last_updated)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET
mask_length = excluded.mask_length,
as_path = excluded.as_path,
next_hop = excluded.next_hop,
last_updated = excluded.last_updated
`
stmtV4, err := tx.Prepare(queryV4)
if err != nil { if err != nil {
return fmt.Errorf("failed to prepare IPv4 statement: %w", err) return fmt.Errorf("failed to prepare statement: %w", err)
} }
defer func() { _ = stmtV4.Close() }() defer func() { _ = stmt.Close() }()
stmtV6, err := tx.Prepare(queryV6)
if err != nil {
return fmt.Errorf("failed to prepare IPv6 statement: %w", err)
}
defer func() { _ = stmtV6.Close() }()
for _, route := range routes { for _, route := range routes {
// Encode AS path as JSON // Encode AS path as JSON
@ -238,38 +219,28 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error {
return fmt.Errorf("failed to encode AS path: %w", err) return fmt.Errorf("failed to encode AS path: %w", err)
} }
// Use appropriate statement based on IP version // Convert v4_ip_start and v4_ip_end to interface{} for SQL NULL handling
if route.IPVersion == ipVersionV4 { var v4Start, v4End interface{}
// IPv4 routes must have range values if route.V4IPStart != nil {
if route.V4IPStart == nil || route.V4IPEnd == nil { v4Start = *route.V4IPStart
return fmt.Errorf("IPv4 route %s missing range values", route.Prefix) }
if route.V4IPEnd != nil {
v4End = *route.V4IPEnd
} }
_, err = stmtV4.Exec( _, err = stmt.Exec(
route.ID.String(), route.ID.String(),
route.Prefix, route.Prefix,
route.MaskLength, route.MaskLength,
route.IPVersion,
route.OriginASN, route.OriginASN,
route.PeerIP, route.PeerIP,
string(pathJSON), string(pathJSON),
route.NextHop, route.NextHop,
route.LastUpdated, route.LastUpdated,
*route.V4IPStart, v4Start,
*route.V4IPEnd, v4End,
) )
} else {
// IPv6 routes
_, err = stmtV6.Exec(
route.ID.String(),
route.Prefix,
route.MaskLength,
route.OriginASN,
route.PeerIP,
string(pathJSON),
route.NextHop,
route.LastUpdated,
)
}
if err != nil { if err != nil {
return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err) return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
} }
@ -301,156 +272,47 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
} }
}() }()
// No longer need to separate deletions since we handle them in the loop below // Separate deletions by type and use prepared statements
var withOrigin []LiveRouteDeletion
var withoutOrigin []LiveRouteDeletion
// Prepare statements for both IPv4 and IPv6 tables
stmtV4WithOrigin, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`)
if err != nil {
return fmt.Errorf("failed to prepare IPv4 delete with origin statement: %w", err)
}
defer func() { _ = stmtV4WithOrigin.Close() }()
stmtV4WithoutOrigin, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`)
if err != nil {
return fmt.Errorf("failed to prepare IPv4 delete without origin statement: %w", err)
}
defer func() { _ = stmtV4WithoutOrigin.Close() }()
stmtV6WithOrigin, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`)
if err != nil {
return fmt.Errorf("failed to prepare IPv6 delete with origin statement: %w", err)
}
defer func() { _ = stmtV6WithOrigin.Close() }()
stmtV6WithoutOrigin, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`)
if err != nil {
return fmt.Errorf("failed to prepare IPv6 delete without origin statement: %w", err)
}
defer func() { _ = stmtV6WithoutOrigin.Close() }()
// Process deletions
for _, del := range deletions { for _, del := range deletions {
var stmt *sql.Stmt
// Select appropriate statement based on IP version and whether we have origin ASN
//nolint:nestif // Clear logic for selecting the right statement
if del.IPVersion == ipVersionV4 {
if del.OriginASN == 0 { if del.OriginASN == 0 {
stmt = stmtV4WithoutOrigin withoutOrigin = append(withoutOrigin, del)
} else { } else {
stmt = stmtV4WithOrigin withOrigin = append(withOrigin, del)
}
} else {
if del.OriginASN == 0 {
stmt = stmtV6WithoutOrigin
} else {
stmt = stmtV6WithOrigin
} }
} }
// Execute deletion // Process deletions with origin ASN
if del.OriginASN == 0 { if len(withOrigin) > 0 {
_, err = stmt.Exec(del.Prefix, del.PeerIP) stmt, err := tx.Prepare(`DELETE FROM live_routes WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`)
} else { if err != nil {
return fmt.Errorf("failed to prepare delete statement: %w", err)
}
defer func() { _ = stmt.Close() }()
for _, del := range withOrigin {
_, err = stmt.Exec(del.Prefix, del.OriginASN, del.PeerIP) _, err = stmt.Exec(del.Prefix, del.OriginASN, del.PeerIP)
}
if err != nil { if err != nil {
return fmt.Errorf("failed to delete route %s: %w", del.Prefix, err) return fmt.Errorf("failed to delete route %s: %w", del.Prefix, err)
} }
} }
if err = tx.Commit(); err != nil {
return fmt.Errorf("failed to commit transaction: %w", err)
} }
return nil // Process deletions without origin ASN
} if len(withoutOrigin) > 0 {
stmt, err := tx.Prepare(`DELETE FROM live_routes WHERE prefix = ? AND peer_ip = ?`)
// UpdatePrefixesBatch updates the last_seen time for multiple prefixes in a single transaction
func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
if len(prefixes) == 0 {
return nil
}
d.lock("UpdatePrefixesBatch")
defer d.unlock()
tx, err := d.beginTx()
if err != nil { if err != nil {
return fmt.Errorf("failed to begin transaction: %w", err) return fmt.Errorf("failed to prepare delete statement: %w", err)
} }
defer func() { defer func() { _ = stmt.Close() }()
if err := tx.Rollback(); err != nil && err != sql.ErrTxDone {
d.logger.Error("Failed to rollback transaction", "error", err)
}
}()
// Prepare statements for both IPv4 and IPv6 tables for _, del := range withoutOrigin {
selectV4Stmt, err := tx.Prepare("SELECT id FROM prefixes_v4 WHERE prefix = ?") _, err = stmt.Exec(del.Prefix, del.PeerIP)
if err != nil { if err != nil {
return fmt.Errorf("failed to prepare IPv4 select statement: %w", err) return fmt.Errorf("failed to delete route %s: %w", del.Prefix, err)
} }
defer func() { _ = selectV4Stmt.Close() }()
updateV4Stmt, err := tx.Prepare("UPDATE prefixes_v4 SET last_seen = ? WHERE prefix = ?")
if err != nil {
return fmt.Errorf("failed to prepare IPv4 update statement: %w", err)
}
defer func() { _ = updateV4Stmt.Close() }()
insertV4Stmt, err := tx.Prepare("INSERT INTO prefixes_v4 (id, prefix, first_seen, last_seen) VALUES (?, ?, ?, ?)")
if err != nil {
return fmt.Errorf("failed to prepare IPv4 insert statement: %w", err)
}
defer func() { _ = insertV4Stmt.Close() }()
selectV6Stmt, err := tx.Prepare("SELECT id FROM prefixes_v6 WHERE prefix = ?")
if err != nil {
return fmt.Errorf("failed to prepare IPv6 select statement: %w", err)
}
defer func() { _ = selectV6Stmt.Close() }()
updateV6Stmt, err := tx.Prepare("UPDATE prefixes_v6 SET last_seen = ? WHERE prefix = ?")
if err != nil {
return fmt.Errorf("failed to prepare IPv6 update statement: %w", err)
}
defer func() { _ = updateV6Stmt.Close() }()
insertV6Stmt, err := tx.Prepare("INSERT INTO prefixes_v6 (id, prefix, first_seen, last_seen) VALUES (?, ?, ?, ?)")
if err != nil {
return fmt.Errorf("failed to prepare IPv6 insert statement: %w", err)
}
defer func() { _ = insertV6Stmt.Close() }()
for prefix, timestamp := range prefixes {
ipVersion := detectIPVersion(prefix)
var selectStmt, updateStmt, insertStmt *sql.Stmt
if ipVersion == ipVersionV4 {
selectStmt, updateStmt, insertStmt = selectV4Stmt, updateV4Stmt, insertV4Stmt
} else {
selectStmt, updateStmt, insertStmt = selectV6Stmt, updateV6Stmt, insertV6Stmt
}
var id string
err = selectStmt.QueryRow(prefix).Scan(&id)
switch err {
case nil:
// Prefix exists, update last_seen
_, err = updateStmt.Exec(timestamp, prefix)
if err != nil {
return fmt.Errorf("failed to update prefix %s: %w", prefix, err)
}
case sql.ErrNoRows:
// Prefix doesn't exist, create it
_, err = insertStmt.Exec(generateUUID().String(), prefix, timestamp, timestamp)
if err != nil {
return fmt.Errorf("failed to insert prefix %s: %w", prefix, err)
}
default:
return fmt.Errorf("failed to query prefix %s: %w", prefix, err)
} }
} }
@ -482,20 +344,20 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
// Prepare statements // Prepare statements
selectStmt, err := tx.Prepare( selectStmt, err := tx.Prepare(
"SELECT asn, handle, description, first_seen, last_seen FROM asns WHERE asn = ?") "SELECT id, number, handle, description, first_seen, last_seen FROM asns WHERE number = ?")
if err != nil { if err != nil {
return fmt.Errorf("failed to prepare select statement: %w", err) return fmt.Errorf("failed to prepare select statement: %w", err)
} }
defer func() { _ = selectStmt.Close() }() defer func() { _ = selectStmt.Close() }()
updateStmt, err := tx.Prepare("UPDATE asns SET last_seen = ? WHERE asn = ?") updateStmt, err := tx.Prepare("UPDATE asns SET last_seen = ? WHERE id = ?")
if err != nil { if err != nil {
return fmt.Errorf("failed to prepare update statement: %w", err) return fmt.Errorf("failed to prepare update statement: %w", err)
} }
defer func() { _ = updateStmt.Close() }() defer func() { _ = updateStmt.Close() }()
insertStmt, err := tx.Prepare( insertStmt, err := tx.Prepare(
"INSERT INTO asns (asn, handle, description, first_seen, last_seen) VALUES (?, ?, ?, ?, ?)") "INSERT INTO asns (id, number, handle, description, first_seen, last_seen) VALUES (?, ?, ?, ?, ?, ?)")
if err != nil { if err != nil {
return fmt.Errorf("failed to prepare insert statement: %w", err) return fmt.Errorf("failed to prepare insert statement: %w", err)
} }
@ -503,13 +365,15 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
for number, timestamp := range asns { for number, timestamp := range asns {
var asn ASN var asn ASN
var idStr string
var handle, description sql.NullString var handle, description sql.NullString
err = selectStmt.QueryRow(number).Scan(&asn.ASN, &handle, &description, &asn.FirstSeen, &asn.LastSeen) err = selectStmt.QueryRow(number).Scan(&idStr, &asn.Number, &handle, &description, &asn.FirstSeen, &asn.LastSeen)
if err == nil { if err == nil {
// ASN exists, update last_seen // ASN exists, update last_seen
_, err = updateStmt.Exec(timestamp, number) asn.ID, _ = uuid.Parse(idStr)
_, err = updateStmt.Exec(timestamp, asn.ID.String())
if err != nil { if err != nil {
return fmt.Errorf("failed to update ASN %d: %w", number, err) return fmt.Errorf("failed to update ASN %d: %w", number, err)
} }
@ -520,7 +384,8 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
if err == sql.ErrNoRows { if err == sql.ErrNoRows {
// ASN doesn't exist, create it // ASN doesn't exist, create it
asn = ASN{ asn = ASN{
ASN: number, ID: generateUUID(),
Number: number,
FirstSeen: timestamp, FirstSeen: timestamp,
LastSeen: timestamp, LastSeen: timestamp,
} }
@ -531,7 +396,7 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
asn.Description = info.Description asn.Description = info.Description
} }
_, err = insertStmt.Exec(asn.ASN, asn.Handle, asn.Description, asn.FirstSeen, asn.LastSeen) _, err = insertStmt.Exec(asn.ID.String(), asn.Number, asn.Handle, asn.Description, asn.FirstSeen, asn.LastSeen)
if err != nil { if err != nil {
return fmt.Errorf("failed to insert ASN %d: %w", number, err) return fmt.Errorf("failed to insert ASN %d: %w", number, err)
} }
@ -567,15 +432,17 @@ func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error)
}() }()
var asn ASN var asn ASN
var idStr string
var handle, description sql.NullString var handle, description sql.NullString
err = tx.QueryRow("SELECT asn, handle, description, first_seen, last_seen FROM asns WHERE asn = ?", number). err = tx.QueryRow("SELECT id, number, handle, description, first_seen, last_seen FROM asns WHERE number = ?", number).
Scan(&asn.ASN, &handle, &description, &asn.FirstSeen, &asn.LastSeen) Scan(&idStr, &asn.Number, &handle, &description, &asn.FirstSeen, &asn.LastSeen)
if err == nil { if err == nil {
// ASN exists, update last_seen // ASN exists, update last_seen
asn.ID, _ = uuid.Parse(idStr)
asn.Handle = handle.String asn.Handle = handle.String
asn.Description = description.String asn.Description = description.String
_, err = tx.Exec("UPDATE asns SET last_seen = ? WHERE asn = ?", timestamp, number) _, err = tx.Exec("UPDATE asns SET last_seen = ? WHERE id = ?", timestamp, asn.ID.String())
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -596,7 +463,8 @@ 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 = ASN{ asn = ASN{
ASN: number, ID: generateUUID(),
Number: number,
FirstSeen: timestamp, FirstSeen: timestamp,
LastSeen: timestamp, LastSeen: timestamp,
} }
@ -607,8 +475,8 @@ func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error)
asn.Description = info.Description asn.Description = info.Description
} }
_, err = tx.Exec("INSERT INTO asns (asn, handle, description, first_seen, last_seen) VALUES (?, ?, ?, ?, ?)", _, err = tx.Exec("INSERT INTO asns (id, number, handle, description, first_seen, last_seen) VALUES (?, ?, ?, ?, ?, ?)",
asn.ASN, asn.Handle, asn.Description, asn.FirstSeen, asn.LastSeen) asn.ID.String(), asn.Number, asn.Handle, asn.Description, asn.FirstSeen, asn.LastSeen)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -637,25 +505,15 @@ func (d *Database) GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefi
} }
}() }()
// Determine table based on IP version
ipVersion := detectIPVersion(prefix)
tableName := "prefixes_v4"
if ipVersion == ipVersionV6 {
tableName = "prefixes_v6"
}
var p Prefix var p Prefix
var idStr string var idStr string
query := fmt.Sprintf("SELECT id, prefix, first_seen, last_seen FROM %s WHERE prefix = ?", tableName) err = tx.QueryRow("SELECT id, prefix, ip_version, first_seen, last_seen FROM prefixes WHERE prefix = ?", prefix).
err = tx.QueryRow(query, prefix). Scan(&idStr, &p.Prefix, &p.IPVersion, &p.FirstSeen, &p.LastSeen)
Scan(&idStr, &p.Prefix, &p.FirstSeen, &p.LastSeen)
if err == nil { if err == nil {
// Prefix exists, update last_seen // Prefix exists, update last_seen
p.ID, _ = uuid.Parse(idStr) p.ID, _ = uuid.Parse(idStr)
p.IPVersion = ipVersion _, err = tx.Exec("UPDATE prefixes SET last_seen = ? WHERE id = ?", timestamp, p.ID.String())
updateQuery := fmt.Sprintf("UPDATE %s SET last_seen = ? WHERE id = ?", tableName)
_, err = tx.Exec(updateQuery, timestamp, p.ID.String())
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -678,12 +536,12 @@ func (d *Database) GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefi
p = Prefix{ p = Prefix{
ID: generateUUID(), ID: generateUUID(),
Prefix: prefix, Prefix: prefix,
IPVersion: ipVersion, IPVersion: detectIPVersion(prefix),
FirstSeen: timestamp, FirstSeen: timestamp,
LastSeen: timestamp, LastSeen: timestamp,
} }
insertQuery := fmt.Sprintf("INSERT INTO %s (id, prefix, first_seen, last_seen) VALUES (?, ?, ?, ?)", tableName) _, err = tx.Exec("INSERT INTO prefixes (id, prefix, ip_version, first_seen, last_seen) VALUES (?, ?, ?, ?, ?)",
_, err = tx.Exec(insertQuery, p.ID.String(), p.Prefix, p.FirstSeen, p.LastSeen) p.ID.String(), p.Prefix, p.IPVersion, p.FirstSeen, p.LastSeen)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -703,10 +561,10 @@ func (d *Database) RecordAnnouncement(announcement *Announcement) error {
defer d.unlock() defer d.unlock()
err := d.exec(` err := d.exec(`
INSERT INTO announcements (id, prefix_id, peer_asn, origin_asn, path, next_hop, timestamp, is_withdrawal) INSERT INTO announcements (id, prefix_id, asn_id, origin_asn_id, path, next_hop, timestamp, is_withdrawal)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
announcement.ID.String(), announcement.PrefixID.String(), announcement.ID.String(), announcement.PrefixID.String(),
announcement.PeerASN, announcement.OriginASN, announcement.ASNID.String(), announcement.OriginASNID.String(),
announcement.Path, announcement.NextHop, announcement.Timestamp, announcement.IsWithdrawal) announcement.Path, announcement.NextHop, announcement.Timestamp, announcement.IsWithdrawal)
return err return err
@ -903,18 +761,26 @@ func (d *Database) GetStatsContext(ctx context.Context) (Stats, error) {
return stats, err return stats, err
} }
// Count prefixes from both tables // Count prefixes
err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM prefixes_v4").Scan(&stats.IPv4Prefixes) err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM prefixes").Scan(&stats.Prefixes)
if err != nil { if err != nil {
return stats, err return stats, err
} }
err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM prefixes_v6").Scan(&stats.IPv6Prefixes) // Count IPv4 and IPv6 prefixes
const ipVersionV4 = 4
err = d.db.QueryRowContext(ctx,
"SELECT COUNT(*) FROM prefixes WHERE ip_version = ?", ipVersionV4).Scan(&stats.IPv4Prefixes)
if err != nil { if err != nil {
return stats, err return stats, err
} }
stats.Prefixes = stats.IPv4Prefixes + stats.IPv6Prefixes const ipVersionV6 = 6
err = d.db.QueryRowContext(ctx,
"SELECT COUNT(*) FROM prefixes WHERE ip_version = ?", ipVersionV6).Scan(&stats.IPv6Prefixes)
if err != nil {
return stats, err
}
// Count peerings // Count peerings
err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM peerings").Scan(&stats.Peerings) err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM peerings").Scan(&stats.Peerings)
@ -937,17 +803,11 @@ func (d *Database) GetStatsContext(ctx context.Context) (Stats, error) {
stats.FileSizeBytes = fileInfo.Size() stats.FileSizeBytes = fileInfo.Size()
} }
// Get live routes count from both tables // Get live routes count
var v4Count, v6Count int err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes").Scan(&stats.LiveRoutes)
err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v4").Scan(&v4Count)
if err != nil { if err != nil {
return stats, fmt.Errorf("failed to count IPv4 routes: %w", err) return stats, fmt.Errorf("failed to count live routes: %w", err)
} }
err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v6").Scan(&v6Count)
if err != nil {
return stats, fmt.Errorf("failed to count IPv6 routes: %w", err)
}
stats.LiveRoutes = v4Count + v6Count
// Get prefix distribution // Get prefix distribution
stats.IPv4PrefixDistribution, stats.IPv6PrefixDistribution, err = d.GetPrefixDistributionContext(ctx) stats.IPv4PrefixDistribution, stats.IPv6PrefixDistribution, err = d.GetPrefixDistributionContext(ctx)
@ -964,38 +824,19 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
d.lock("UpsertLiveRoute") d.lock("UpsertLiveRoute")
defer d.unlock() defer d.unlock()
// Choose table based on IP version query := `
tableName := "live_routes_v4" INSERT INTO live_routes (id, prefix, mask_length, ip_version, origin_asn, peer_ip, as_path, next_hop,
if route.IPVersion == ipVersionV6 { last_updated, v4_ip_start, v4_ip_end)
tableName = "live_routes_v6" VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
}
var query string
if route.IPVersion == ipVersionV4 {
query = fmt.Sprintf(`
INSERT INTO %s (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
last_updated, ip_start, ip_end)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET
mask_length = excluded.mask_length, mask_length = excluded.mask_length,
ip_version = excluded.ip_version,
as_path = excluded.as_path, as_path = excluded.as_path,
next_hop = excluded.next_hop, next_hop = excluded.next_hop,
last_updated = excluded.last_updated, last_updated = excluded.last_updated,
ip_start = excluded.ip_start, v4_ip_start = excluded.v4_ip_start,
ip_end = excluded.ip_end v4_ip_end = excluded.v4_ip_end
`, tableName) `
} else {
query = fmt.Sprintf(`
INSERT INTO %s (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
last_updated)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET
mask_length = excluded.mask_length,
as_path = excluded.as_path,
next_hop = excluded.next_hop,
last_updated = excluded.last_updated
`, tableName)
}
// Encode AS path as JSON // Encode AS path as JSON
pathJSON, err := json.Marshal(route.ASPath) pathJSON, err := json.Marshal(route.ASPath)
@ -1003,7 +844,6 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
return fmt.Errorf("failed to encode AS path: %w", err) return fmt.Errorf("failed to encode AS path: %w", err)
} }
if route.IPVersion == ipVersionV4 {
// Convert v4_ip_start and v4_ip_end to interface{} for SQL NULL handling // Convert v4_ip_start and v4_ip_end to interface{} for SQL NULL handling
var v4Start, v4End interface{} var v4Start, v4End interface{}
if route.V4IPStart != nil { if route.V4IPStart != nil {
@ -1017,6 +857,7 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
route.ID.String(), route.ID.String(),
route.Prefix, route.Prefix,
route.MaskLength, route.MaskLength,
route.IPVersion,
route.OriginASN, route.OriginASN,
route.PeerIP, route.PeerIP,
string(pathJSON), string(pathJSON),
@ -1025,18 +866,6 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
v4Start, v4Start,
v4End, v4End,
) )
} else {
_, err = d.db.Exec(query,
route.ID.String(),
route.Prefix,
route.MaskLength,
route.OriginASN,
route.PeerIP,
string(pathJSON),
route.NextHop,
route.LastUpdated,
)
}
return err return err
} }
@ -1047,25 +876,16 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string)
d.lock("DeleteLiveRoute") d.lock("DeleteLiveRoute")
defer d.unlock() defer d.unlock()
// Determine table based on prefix IP version
_, ipnet, err := net.ParseCIDR(prefix)
if err != nil {
return fmt.Errorf("invalid prefix format: %w", err)
}
tableName := "live_routes_v4"
if ipnet.IP.To4() == nil {
tableName = "live_routes_v6"
}
var query string var query string
var err error
if originASN == 0 { if originASN == 0 {
// Delete all routes for this prefix from this peer // Delete all routes for this prefix from this peer
query = fmt.Sprintf(`DELETE FROM %s WHERE prefix = ? AND peer_ip = ?`, tableName) query = `DELETE FROM live_routes WHERE prefix = ? AND peer_ip = ?`
_, err = d.db.Exec(query, prefix, peerIP) _, err = d.db.Exec(query, prefix, peerIP)
} else { } else {
// Delete specific route // Delete specific route
query = fmt.Sprintf(`DELETE FROM %s WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`, tableName) query = `DELETE FROM live_routes WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
_, err = d.db.Exec(query, prefix, originASN, peerIP) _, err = d.db.Exec(query, prefix, originASN, peerIP)
} }
@ -1080,51 +900,45 @@ func (d *Database) GetPrefixDistribution() (ipv4 []PrefixDistribution, ipv6 []Pr
// GetPrefixDistributionContext returns the distribution of unique prefixes by mask length with context support // GetPrefixDistributionContext returns the distribution of unique prefixes by mask length with context support
func (d *Database) GetPrefixDistributionContext(ctx context.Context) ( func (d *Database) GetPrefixDistributionContext(ctx context.Context) (
ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error) { ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error) {
// IPv4 distribution - count unique prefixes from v4 table // IPv4 distribution - count unique prefixes, not routes
query := ` query := `
SELECT mask_length, COUNT(DISTINCT prefix) as count SELECT mask_length, COUNT(DISTINCT prefix) as count
FROM live_routes_v4 FROM live_routes
WHERE ip_version = 4
GROUP BY mask_length GROUP BY mask_length
ORDER BY mask_length ORDER BY mask_length
` `
rows4, err := d.db.QueryContext(ctx, query) rows, err := d.db.QueryContext(ctx, query)
if err != nil { if err != nil {
return nil, nil, fmt.Errorf("failed to query IPv4 distribution: %w", err) return nil, nil, fmt.Errorf("failed to query IPv4 distribution: %w", err)
} }
defer func() { defer func() { _ = rows.Close() }()
if rows4 != nil {
_ = rows4.Close()
}
}()
for rows4.Next() { for rows.Next() {
var dist PrefixDistribution var dist PrefixDistribution
if err := rows4.Scan(&dist.MaskLength, &dist.Count); err != nil { if err := rows.Scan(&dist.MaskLength, &dist.Count); err != nil {
return nil, nil, fmt.Errorf("failed to scan IPv4 distribution: %w", err) return nil, nil, fmt.Errorf("failed to scan IPv4 distribution: %w", err)
} }
ipv4 = append(ipv4, dist) ipv4 = append(ipv4, dist)
} }
// IPv6 distribution - count unique prefixes from v6 table // IPv6 distribution - count unique prefixes, not routes
query = ` query = `
SELECT mask_length, COUNT(DISTINCT prefix) as count SELECT mask_length, COUNT(DISTINCT prefix) as count
FROM live_routes_v6 FROM live_routes
WHERE ip_version = 6
GROUP BY mask_length GROUP BY mask_length
ORDER BY mask_length ORDER BY mask_length
` `
rows6, err := d.db.QueryContext(ctx, query) rows, err = d.db.QueryContext(ctx, query)
if err != nil { if err != nil {
return nil, nil, fmt.Errorf("failed to query IPv6 distribution: %w", err) return nil, nil, fmt.Errorf("failed to query IPv6 distribution: %w", err)
} }
defer func() { defer func() { _ = rows.Close() }()
if rows6 != nil {
_ = rows6.Close()
}
}()
for rows6.Next() { for rows.Next() {
var dist PrefixDistribution var dist PrefixDistribution
if err := rows6.Scan(&dist.MaskLength, &dist.Count); err != nil { if err := rows.Scan(&dist.MaskLength, &dist.Count); err != nil {
return nil, nil, fmt.Errorf("failed to scan IPv6 distribution: %w", err) return nil, nil, fmt.Errorf("failed to scan IPv6 distribution: %w", err)
} }
ipv6 = append(ipv6, dist) ipv6 = append(ipv6, dist)
@ -1140,14 +954,14 @@ func (d *Database) GetLiveRouteCounts() (ipv4Count, ipv6Count int, err error) {
// GetLiveRouteCountsContext returns the count of IPv4 and IPv6 routes with context support // GetLiveRouteCountsContext returns the count of IPv4 and IPv6 routes with context support
func (d *Database) GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, ipv6Count int, err error) { func (d *Database) GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, ipv6Count int, err error) {
// Get IPv4 count from dedicated table // Get IPv4 count
err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v4").Scan(&ipv4Count) err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes WHERE ip_version = 4").Scan(&ipv4Count)
if err != nil { if err != nil {
return 0, 0, fmt.Errorf("failed to count IPv4 routes: %w", err) return 0, 0, fmt.Errorf("failed to count IPv4 routes: %w", err)
} }
// Get IPv6 count from dedicated table // Get IPv6 count
err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v6").Scan(&ipv6Count) err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes WHERE ip_version = 6").Scan(&ipv6Count)
if err != nil { if err != nil {
return 0, 0, fmt.Errorf("failed to count IPv6 routes: %w", err) return 0, 0, fmt.Errorf("failed to count IPv6 routes: %w", err)
} }
@ -1182,9 +996,9 @@ func (d *Database) GetASInfoForIPContext(ctx context.Context, ip string) (*ASInf
query := ` query := `
SELECT DISTINCT lr.prefix, lr.mask_length, lr.origin_asn, lr.last_updated, a.handle, a.description SELECT DISTINCT lr.prefix, lr.mask_length, lr.origin_asn, lr.last_updated, a.handle, a.description
FROM live_routes_v4 lr FROM live_routes lr
LEFT JOIN asns a ON a.asn = lr.origin_asn LEFT JOIN asns a ON a.number = lr.origin_asn
WHERE lr.ip_start <= ? AND lr.ip_end >= ? WHERE lr.ip_version = ? AND lr.v4_ip_start <= ? AND lr.v4_ip_end >= ?
ORDER BY lr.mask_length DESC ORDER BY lr.mask_length DESC
LIMIT 1 LIMIT 1
` `
@ -1194,7 +1008,7 @@ func (d *Database) GetASInfoForIPContext(ctx context.Context, ip string) (*ASInf
var lastUpdated time.Time var lastUpdated time.Time
var handle, description sql.NullString var handle, description sql.NullString
err := d.db.QueryRowContext(ctx, query, ipUint, ipUint).Scan( err := d.db.QueryRowContext(ctx, query, ipVersionV4, ipUint, ipUint).Scan(
&prefix, &maskLength, &originASN, &lastUpdated, &handle, &description) &prefix, &maskLength, &originASN, &lastUpdated, &handle, &description)
if err != nil { if err != nil {
if err == sql.ErrNoRows { if err == sql.ErrNoRows {
@ -1219,12 +1033,13 @@ func (d *Database) GetASInfoForIPContext(ctx context.Context, ip string) (*ASInf
// For IPv6, use the original method since we don't have range optimization // For IPv6, use the original method since we don't have range optimization
query := ` query := `
SELECT DISTINCT lr.prefix, lr.mask_length, lr.origin_asn, lr.last_updated, a.handle, a.description SELECT DISTINCT lr.prefix, lr.mask_length, lr.origin_asn, lr.last_updated, a.handle, a.description
FROM live_routes_v6 lr FROM live_routes lr
LEFT JOIN asns a ON a.asn = lr.origin_asn LEFT JOIN asns a ON a.number = lr.origin_asn
WHERE lr.ip_version = ?
ORDER BY lr.mask_length DESC ORDER BY lr.mask_length DESC
` `
rows, err := d.db.QueryContext(ctx, query) rows, err := d.db.QueryContext(ctx, query, ipVersionV6)
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to query routes: %w", err) return nil, fmt.Errorf("failed to query routes: %w", err)
} }
@ -1334,11 +1149,12 @@ func (d *Database) GetASDetails(asn int) (*ASN, []LiveRoute, error) {
func (d *Database) GetASDetailsContext(ctx context.Context, asn int) (*ASN, []LiveRoute, error) { func (d *Database) GetASDetailsContext(ctx context.Context, asn int) (*ASN, []LiveRoute, error) {
// Get AS information // Get AS information
var asnInfo ASN var asnInfo ASN
var idStr string
var handle, description sql.NullString var handle, description sql.NullString
err := d.db.QueryRowContext(ctx, err := d.db.QueryRowContext(ctx,
"SELECT asn, handle, description, first_seen, last_seen FROM asns WHERE asn = ?", "SELECT id, number, handle, description, first_seen, last_seen FROM asns WHERE number = ?",
asn, asn,
).Scan(&asnInfo.ASN, &handle, &description, &asnInfo.FirstSeen, &asnInfo.LastSeen) ).Scan(&idStr, &asnInfo.Number, &handle, &description, &asnInfo.FirstSeen, &asnInfo.LastSeen)
if err != nil { if err != nil {
if err == sql.ErrNoRows { if err == sql.ErrNoRows {
@ -1348,151 +1164,50 @@ func (d *Database) GetASDetailsContext(ctx context.Context, asn int) (*ASN, []Li
return nil, nil, fmt.Errorf("failed to query AS: %w", err) return nil, nil, fmt.Errorf("failed to query AS: %w", err)
} }
asnInfo.ID, _ = uuid.Parse(idStr)
asnInfo.Handle = handle.String asnInfo.Handle = handle.String
asnInfo.Description = description.String asnInfo.Description = description.String
// Get prefixes announced by this AS from both tables // Get prefixes announced by this AS (unique prefixes with most recent update)
var allPrefixes []LiveRoute
// Query IPv4 prefixes
queryV4 := `
SELECT prefix, mask_length, MAX(last_updated) as last_updated
FROM live_routes_v4
WHERE origin_asn = ?
GROUP BY prefix, mask_length
`
rows4, err := d.db.QueryContext(ctx, queryV4, asn)
if err != nil {
return &asnInfo, nil, fmt.Errorf("failed to query IPv4 prefixes: %w", err)
}
defer func() { _ = rows4.Close() }()
for rows4.Next() {
var route LiveRoute
var lastUpdatedStr string
err := rows4.Scan(&route.Prefix, &route.MaskLength, &lastUpdatedStr)
if err != nil {
d.logger.Error("Failed to scan IPv4 prefix row", "error", err, "asn", asn)
continue
}
// Parse the timestamp string
route.LastUpdated, err = time.Parse("2006-01-02 15:04:05-07:00", lastUpdatedStr)
if err != nil {
// Try without timezone
route.LastUpdated, err = time.Parse("2006-01-02 15:04:05", lastUpdatedStr)
if err != nil {
d.logger.Error("Failed to parse timestamp", "error", err, "timestamp", lastUpdatedStr)
continue
}
}
route.OriginASN = asn
route.IPVersion = ipVersionV4
allPrefixes = append(allPrefixes, route)
}
// Query IPv6 prefixes
queryV6 := `
SELECT prefix, mask_length, MAX(last_updated) as last_updated
FROM live_routes_v6
WHERE origin_asn = ?
GROUP BY prefix, mask_length
`
rows6, err := d.db.QueryContext(ctx, queryV6, asn)
if err != nil {
return &asnInfo, allPrefixes, fmt.Errorf("failed to query IPv6 prefixes: %w", err)
}
defer func() { _ = rows6.Close() }()
for rows6.Next() {
var route LiveRoute
var lastUpdatedStr string
err := rows6.Scan(&route.Prefix, &route.MaskLength, &lastUpdatedStr)
if err != nil {
d.logger.Error("Failed to scan IPv6 prefix row", "error", err, "asn", asn)
continue
}
// Parse the timestamp string
route.LastUpdated, err = time.Parse("2006-01-02 15:04:05-07:00", lastUpdatedStr)
if err != nil {
// Try without timezone
route.LastUpdated, err = time.Parse("2006-01-02 15:04:05", lastUpdatedStr)
if err != nil {
d.logger.Error("Failed to parse timestamp", "error", err, "timestamp", lastUpdatedStr)
continue
}
}
route.OriginASN = asn
route.IPVersion = ipVersionV6
allPrefixes = append(allPrefixes, route)
}
return &asnInfo, allPrefixes, nil
}
// ASPeer represents a peering relationship with another AS
type ASPeer struct {
ASN int `json:"asn"`
Handle string `json:"handle"`
Description string `json:"description"`
FirstSeen time.Time `json:"first_seen"`
LastSeen time.Time `json:"last_seen"`
}
// GetASPeers returns all ASes that peer with the given AS
func (d *Database) GetASPeers(asn int) ([]ASPeer, error) {
return d.GetASPeersContext(context.Background(), asn)
}
// GetASPeersContext returns all ASes that peer with the given AS with context support
func (d *Database) GetASPeersContext(ctx context.Context, asn int) ([]ASPeer, error) {
query := ` query := `
SELECT SELECT prefix, mask_length, ip_version, MAX(last_updated) as last_updated
CASE FROM live_routes
WHEN p.as_a = ? THEN p.as_b WHERE origin_asn = ?
ELSE p.as_a GROUP BY prefix, mask_length, ip_version
END as peer_asn,
COALESCE(a.handle, '') as handle,
COALESCE(a.description, '') as description,
p.first_seen,
p.last_seen
FROM peerings p
LEFT JOIN asns a ON a.asn = CASE
WHEN p.as_a = ? THEN p.as_b
ELSE p.as_a
END
WHERE p.as_a = ? OR p.as_b = ?
ORDER BY peer_asn
` `
rows, err := d.db.QueryContext(ctx, query, asn, asn, asn, asn) rows, err := d.db.QueryContext(ctx, query, asn)
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to query AS peers: %w", err) return &asnInfo, nil, fmt.Errorf("failed to query prefixes: %w", err)
} }
defer func() { _ = rows.Close() }() defer func() { _ = rows.Close() }()
var peers []ASPeer var prefixes []LiveRoute
for rows.Next() { for rows.Next() {
var peer ASPeer var route LiveRoute
err := rows.Scan(&peer.ASN, &peer.Handle, &peer.Description, &peer.FirstSeen, &peer.LastSeen) var lastUpdatedStr string
err := rows.Scan(&route.Prefix, &route.MaskLength, &route.IPVersion, &lastUpdatedStr)
if err != nil { if err != nil {
d.logger.Error("Failed to scan peer row", "error", err, "asn", asn) d.logger.Error("Failed to scan prefix row", "error", err, "asn", asn)
continue continue
} }
peers = append(peers, peer) // Parse the timestamp string
route.LastUpdated, err = time.Parse("2006-01-02 15:04:05-07:00", lastUpdatedStr)
if err != nil {
// Try without timezone
route.LastUpdated, err = time.Parse("2006-01-02 15:04:05", lastUpdatedStr)
if err != nil {
d.logger.Error("Failed to parse timestamp", "error", err, "timestamp", lastUpdatedStr)
continue
}
}
route.OriginASN = asn
prefixes = append(prefixes, route)
} }
if err := rows.Err(); err != nil { return &asnInfo, prefixes, nil
return nil, fmt.Errorf("error iterating AS peers: %w", err)
}
return peers, nil
} }
// GetPrefixDetails returns detailed information about a prefix // GetPrefixDetails returns detailed information about a prefix
@ -1502,28 +1217,14 @@ func (d *Database) GetPrefixDetails(prefix string) ([]LiveRoute, error) {
// GetPrefixDetailsContext returns detailed information about a prefix with context support // GetPrefixDetailsContext returns detailed information about a prefix with context support
func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, error) { func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, error) {
// Determine if it's IPv4 or IPv6 by parsing the prefix query := `
_, ipnet, err := net.ParseCIDR(prefix)
if err != nil {
return nil, fmt.Errorf("invalid prefix format: %w", err)
}
tableName := "live_routes_v4"
ipVersion := ipVersionV4
if ipnet.IP.To4() == nil {
tableName = "live_routes_v6"
ipVersion = ipVersionV6
}
//nolint:gosec // Table name is hardcoded based on IP version
query := fmt.Sprintf(`
SELECT lr.origin_asn, lr.peer_ip, lr.as_path, lr.next_hop, lr.last_updated, SELECT lr.origin_asn, lr.peer_ip, lr.as_path, lr.next_hop, lr.last_updated,
a.handle, a.description a.handle, a.description
FROM %s lr FROM live_routes lr
LEFT JOIN asns a ON a.asn = lr.origin_asn LEFT JOIN asns a ON a.number = lr.origin_asn
WHERE lr.prefix = ? WHERE lr.prefix = ?
ORDER BY lr.origin_asn, lr.peer_ip ORDER BY lr.origin_asn, lr.peer_ip
`, tableName) `
rows, err := d.db.QueryContext(ctx, query, prefix) rows, err := d.db.QueryContext(ctx, query, prefix)
if err != nil { if err != nil {
@ -1551,8 +1252,6 @@ func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) (
} }
route.Prefix = prefix route.Prefix = prefix
route.IPVersion = ipVersion
route.MaskLength, _ = ipnet.Mask.Size()
routes = append(routes, route) routes = append(routes, route)
} }
@ -1572,29 +1271,23 @@ func (d *Database) GetRandomPrefixesByLength(maskLength, ipVersion, limit int) (
func (d *Database) GetRandomPrefixesByLengthContext( func (d *Database) GetRandomPrefixesByLengthContext(
ctx context.Context, maskLength, ipVersion, limit int) ([]LiveRoute, error) { ctx context.Context, maskLength, ipVersion, limit int) ([]LiveRoute, error) {
// Select unique prefixes with their most recent route information // Select unique prefixes with their most recent route information
tableName := "live_routes_v4" query := `
if ipVersion == ipVersionV6 {
tableName = "live_routes_v6"
}
//nolint:gosec // Table name is hardcoded based on IP version
query := fmt.Sprintf(`
WITH unique_prefixes AS ( WITH unique_prefixes AS (
SELECT prefix, MAX(last_updated) as max_updated SELECT prefix, MAX(last_updated) as max_updated
FROM %s FROM live_routes
WHERE mask_length = ? WHERE mask_length = ? AND ip_version = ?
GROUP BY prefix GROUP BY prefix
ORDER BY RANDOM() ORDER BY RANDOM()
LIMIT ? LIMIT ?
) )
SELECT lr.prefix, lr.mask_length, lr.origin_asn, lr.as_path, SELECT lr.prefix, lr.mask_length, lr.ip_version, lr.origin_asn, lr.as_path,
lr.peer_ip, lr.last_updated lr.peer_ip, lr.last_updated
FROM %s lr FROM live_routes lr
INNER JOIN unique_prefixes up ON lr.prefix = up.prefix AND lr.last_updated = up.max_updated INNER JOIN unique_prefixes up ON lr.prefix = up.prefix AND lr.last_updated = up.max_updated
WHERE lr.mask_length = ? WHERE lr.mask_length = ? AND lr.ip_version = ?
`, tableName, tableName) `
rows, err := d.db.QueryContext(ctx, query, maskLength, limit, maskLength) rows, err := d.db.QueryContext(ctx, query, maskLength, ipVersion, limit, maskLength, ipVersion)
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to query random prefixes: %w", err) return nil, fmt.Errorf("failed to query random prefixes: %w", err)
} }
@ -1609,13 +1302,12 @@ func (d *Database) GetRandomPrefixesByLengthContext(
err := rows.Scan( err := rows.Scan(
&route.Prefix, &route.Prefix,
&route.MaskLength, &route.MaskLength,
&route.IPVersion,
&route.OriginASN, &route.OriginASN,
&pathJSON, &pathJSON,
&route.PeerIP, &route.PeerIP,
&route.LastUpdated, &route.LastUpdated,
) )
// Set IP version based on which table we queried
route.IPVersion = ipVersion
if err != nil { if err != nil {
continue continue
} }

3
internal/database/interface.go
View File

@ -27,7 +27,6 @@ type Store interface {
// Prefix operations // Prefix operations
GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefix, error) GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefix, error)
UpdatePrefixesBatch(prefixes map[string]time.Time) error
// Announcement operations // Announcement operations
RecordAnnouncement(announcement *Announcement) error RecordAnnouncement(announcement *Announcement) error
@ -60,8 +59,6 @@ type Store interface {
// AS and prefix detail operations // AS and prefix detail operations
GetASDetails(asn int) (*ASN, []LiveRoute, error) GetASDetails(asn int) (*ASN, []LiveRoute, error)
GetASDetailsContext(ctx context.Context, asn int) (*ASN, []LiveRoute, error) GetASDetailsContext(ctx context.Context, asn int) (*ASN, []LiveRoute, error)
GetASPeers(asn int) ([]ASPeer, error)
GetASPeersContext(ctx context.Context, asn int) ([]ASPeer, error)
GetPrefixDetails(prefix string) ([]LiveRoute, error) GetPrefixDetails(prefix string) ([]LiveRoute, error)
GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, error) GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, error)
GetRandomPrefixesByLength(maskLength, ipVersion, limit int) ([]LiveRoute, error) GetRandomPrefixesByLength(maskLength, ipVersion, limit int) ([]LiveRoute, error)

12
internal/database/models.go
View File

@ -8,7 +8,8 @@ import (
// ASN represents an Autonomous System Number // ASN represents an Autonomous System Number
type ASN struct { type ASN struct {
ASN int `json:"asn"` ID uuid.UUID `json:"id"`
Number int `json:"number"`
Handle string `json:"handle"` Handle string `json:"handle"`
Description string `json:"description"` Description string `json:"description"`
FirstSeen time.Time `json:"first_seen"` FirstSeen time.Time `json:"first_seen"`
@ -28,8 +29,8 @@ type Prefix struct {
type Announcement struct { type Announcement struct {
ID uuid.UUID `json:"id"` ID uuid.UUID `json:"id"`
PrefixID uuid.UUID `json:"prefix_id"` PrefixID uuid.UUID `json:"prefix_id"`
PeerASN int `json:"peer_asn"` ASNID uuid.UUID `json:"asn_id"`
OriginASN int `json:"origin_asn"` OriginASNID uuid.UUID `json:"origin_asn_id"`
Path string `json:"path"` // JSON-encoded AS path Path string `json:"path"` // JSON-encoded AS path
NextHop string `json:"next_hop"` NextHop string `json:"next_hop"`
Timestamp time.Time `json:"timestamp"` Timestamp time.Time `json:"timestamp"`
@ -39,8 +40,8 @@ type Announcement struct {
// ASNPeering represents a peering relationship between two ASNs // ASNPeering represents a peering relationship between two ASNs
type ASNPeering struct { type ASNPeering struct {
ID uuid.UUID `json:"id"` ID uuid.UUID `json:"id"`
ASA int `json:"as_a"` FromASNID uuid.UUID `json:"from_asn_id"`
ASB int `json:"as_b"` ToASNID uuid.UUID `json:"to_asn_id"`
FirstSeen time.Time `json:"first_seen"` FirstSeen time.Time `json:"first_seen"`
LastSeen time.Time `json:"last_seen"` LastSeen time.Time `json:"last_seen"`
} }
@ -82,7 +83,6 @@ type LiveRouteDeletion struct {
Prefix string Prefix string
OriginASN int OriginASN int
PeerIP string PeerIP string
IPVersion int
} }
// PeerUpdate represents parameters for updating a peer // PeerUpdate represents parameters for updating a peer

94
internal/database/schema.sql
View File

@ -1,27 +1,16 @@
-- IMPORTANT: This is the ONLY place where schema changes should be made.
-- We do NOT support migrations. All schema changes MUST be in this file.
-- DO NOT make schema changes anywhere else in the codebase.
CREATE TABLE IF NOT EXISTS asns ( CREATE TABLE IF NOT EXISTS asns (
asn INTEGER PRIMARY KEY, id TEXT PRIMARY KEY,
number INTEGER UNIQUE NOT NULL,
handle TEXT, handle TEXT,
description TEXT, description TEXT,
first_seen DATETIME NOT NULL, first_seen DATETIME NOT NULL,
last_seen DATETIME NOT NULL last_seen DATETIME NOT NULL
); );
-- IPv4 prefixes table CREATE TABLE IF NOT EXISTS prefixes (
CREATE TABLE IF NOT EXISTS prefixes_v4 (
id TEXT PRIMARY KEY,
prefix TEXT UNIQUE NOT NULL,
first_seen DATETIME NOT NULL,
last_seen DATETIME NOT NULL
);
-- IPv6 prefixes table
CREATE TABLE IF NOT EXISTS prefixes_v6 (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
prefix TEXT UNIQUE NOT NULL, prefix TEXT UNIQUE NOT NULL,
ip_version INTEGER NOT NULL, -- 4 for IPv4, 6 for IPv6
first_seen DATETIME NOT NULL, first_seen DATETIME NOT NULL,
last_seen DATETIME NOT NULL last_seen DATETIME NOT NULL
); );
@ -29,14 +18,15 @@ CREATE TABLE IF NOT EXISTS prefixes_v6 (
CREATE TABLE IF NOT EXISTS announcements ( CREATE TABLE IF NOT EXISTS announcements (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
prefix_id TEXT NOT NULL, prefix_id TEXT NOT NULL,
peer_asn INTEGER NOT NULL, asn_id TEXT NOT NULL,
origin_asn INTEGER NOT NULL, origin_asn_id TEXT NOT NULL,
path TEXT NOT NULL, path TEXT NOT NULL,
next_hop TEXT, next_hop TEXT,
timestamp DATETIME NOT NULL, timestamp DATETIME NOT NULL,
is_withdrawal BOOLEAN NOT NULL DEFAULT 0, is_withdrawal BOOLEAN NOT NULL DEFAULT 0,
FOREIGN KEY (peer_asn) REFERENCES asns(asn), FOREIGN KEY (prefix_id) REFERENCES prefixes(id),
FOREIGN KEY (origin_asn) REFERENCES asns(asn) FOREIGN KEY (asn_id) REFERENCES asns(id),
FOREIGN KEY (origin_asn_id) REFERENCES asns(id)
); );
CREATE TABLE IF NOT EXISTS peerings ( CREATE TABLE IF NOT EXISTS peerings (
@ -58,71 +48,49 @@ CREATE TABLE IF NOT EXISTS bgp_peers (
last_message_type TEXT last_message_type TEXT
); );
-- Indexes for prefixes_v4 table CREATE INDEX IF NOT EXISTS idx_prefixes_ip_version ON prefixes(ip_version);
CREATE INDEX IF NOT EXISTS idx_prefixes_v4_prefix ON prefixes_v4(prefix); CREATE INDEX IF NOT EXISTS idx_prefixes_version_prefix ON prefixes(ip_version, prefix);
-- Indexes for prefixes_v6 table
CREATE INDEX IF NOT EXISTS idx_prefixes_v6_prefix ON prefixes_v6(prefix);
CREATE INDEX IF NOT EXISTS idx_announcements_timestamp ON announcements(timestamp); CREATE INDEX IF NOT EXISTS idx_announcements_timestamp ON announcements(timestamp);
CREATE INDEX IF NOT EXISTS idx_announcements_prefix_id ON announcements(prefix_id); CREATE INDEX IF NOT EXISTS idx_announcements_prefix_id ON announcements(prefix_id);
CREATE INDEX IF NOT EXISTS idx_announcements_peer_asn ON announcements(peer_asn); CREATE INDEX IF NOT EXISTS idx_announcements_asn_id ON announcements(asn_id);
CREATE INDEX IF NOT EXISTS idx_announcements_origin_asn ON announcements(origin_asn);
CREATE INDEX IF NOT EXISTS idx_peerings_as_a ON peerings(as_a); CREATE INDEX IF NOT EXISTS idx_peerings_as_a ON peerings(as_a);
CREATE INDEX IF NOT EXISTS idx_peerings_as_b ON peerings(as_b); CREATE INDEX IF NOT EXISTS idx_peerings_as_b ON peerings(as_b);
CREATE INDEX IF NOT EXISTS idx_peerings_lookup ON peerings(as_a, as_b); CREATE INDEX IF NOT EXISTS idx_peerings_lookup ON peerings(as_a, as_b);
-- Additional indexes for prefixes table
CREATE INDEX IF NOT EXISTS idx_prefixes_prefix ON prefixes(prefix);
-- Indexes for asns table -- Indexes for asns table
CREATE INDEX IF NOT EXISTS idx_asns_asn ON asns(asn); CREATE INDEX IF NOT EXISTS idx_asns_number ON asns(number);
-- Indexes for bgp_peers table -- Indexes for bgp_peers table
CREATE INDEX IF NOT EXISTS idx_bgp_peers_asn ON bgp_peers(peer_asn); CREATE INDEX IF NOT EXISTS idx_bgp_peers_asn ON bgp_peers(peer_asn);
CREATE INDEX IF NOT EXISTS idx_bgp_peers_last_seen ON bgp_peers(last_seen); CREATE INDEX IF NOT EXISTS idx_bgp_peers_last_seen ON bgp_peers(last_seen);
CREATE INDEX IF NOT EXISTS idx_bgp_peers_ip ON bgp_peers(peer_ip); CREATE INDEX IF NOT EXISTS idx_bgp_peers_ip ON bgp_peers(peer_ip);
-- IPv4 routing table maintained by PrefixHandler -- Live routing table maintained by PrefixHandler
CREATE TABLE IF NOT EXISTS live_routes_v4 ( CREATE TABLE IF NOT EXISTS live_routes (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
prefix TEXT NOT NULL, prefix TEXT NOT NULL,
mask_length INTEGER NOT NULL, -- CIDR mask length (0-32) mask_length INTEGER NOT NULL, -- CIDR mask length (0-32 for IPv4, 0-128 for IPv6)
ip_version INTEGER NOT NULL, -- 4 or 6
origin_asn INTEGER NOT NULL, origin_asn INTEGER NOT NULL,
peer_ip TEXT NOT NULL, peer_ip TEXT NOT NULL,
as_path TEXT NOT NULL, -- JSON array as_path TEXT NOT NULL, -- JSON array
next_hop TEXT NOT NULL, next_hop TEXT NOT NULL,
last_updated DATETIME NOT NULL, last_updated DATETIME NOT NULL,
-- IPv4 range columns for fast lookups -- IPv4 range columns for fast lookups (NULL for IPv6)
ip_start INTEGER NOT NULL, -- Start of IPv4 range as 32-bit unsigned int v4_ip_start INTEGER, -- Start of IPv4 range as 32-bit unsigned int
ip_end INTEGER NOT NULL, -- End of IPv4 range as 32-bit unsigned int v4_ip_end INTEGER, -- End of IPv4 range as 32-bit unsigned int
UNIQUE(prefix, origin_asn, peer_ip) UNIQUE(prefix, origin_asn, peer_ip)
); );
-- IPv6 routing table maintained by PrefixHandler -- Indexes for live_routes table
CREATE TABLE IF NOT EXISTS live_routes_v6 ( CREATE INDEX IF NOT EXISTS idx_live_routes_prefix ON live_routes(prefix);
id TEXT PRIMARY KEY, CREATE INDEX IF NOT EXISTS idx_live_routes_mask_length ON live_routes(mask_length);
prefix TEXT NOT NULL, CREATE INDEX IF NOT EXISTS idx_live_routes_ip_version_mask ON live_routes(ip_version, mask_length);
mask_length INTEGER NOT NULL, -- CIDR mask length (0-128) CREATE INDEX IF NOT EXISTS idx_live_routes_last_updated ON live_routes(last_updated);
origin_asn INTEGER NOT NULL,
peer_ip TEXT NOT NULL,
as_path TEXT NOT NULL, -- JSON array
next_hop TEXT NOT NULL,
last_updated DATETIME NOT NULL,
-- Note: IPv6 doesn't use integer range columns
UNIQUE(prefix, origin_asn, peer_ip)
);
-- Indexes for live_routes_v4 table
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_prefix ON live_routes_v4(prefix);
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_length ON live_routes_v4(mask_length);
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_origin_asn ON live_routes_v4(origin_asn);
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_last_updated ON live_routes_v4(last_updated);
-- Indexes for IPv4 range queries -- Indexes for IPv4 range queries
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_ip_range ON live_routes_v4(ip_start, ip_end); CREATE INDEX IF NOT EXISTS idx_live_routes_ipv4_range ON live_routes(v4_ip_start, v4_ip_end) WHERE ip_version = 4;
-- Index to optimize prefix distribution queries -- Index to optimize COUNT(DISTINCT prefix) queries
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_prefix ON live_routes_v4(mask_length, prefix); CREATE INDEX IF NOT EXISTS idx_live_routes_ip_mask_prefix ON live_routes(ip_version, mask_length, prefix);
-- Indexes for live_routes_v6 table
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_prefix ON live_routes_v6(prefix);
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_length ON live_routes_v6(mask_length);
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_origin_asn ON live_routes_v6(origin_asn);
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_last_updated ON live_routes_v6(last_updated);
-- Index to optimize prefix distribution queries
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_prefix ON live_routes_v6(mask_length, prefix);

2
internal/database/slowquery.go
View File

@ -8,7 +8,7 @@ import (
"git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/logger"
) )
const slowQueryThreshold = 25 * time.Millisecond const slowQueryThreshold = 50 * time.Millisecond
// logSlowQuery logs queries that take longer than slowQueryThreshold // logSlowQuery logs queries that take longer than slowQueryThreshold
func logSlowQuery(logger *logger.Logger, query string, start time.Time) { func logSlowQuery(logger *logger.Logger, query string, start time.Time) {

48
internal/routewatch/app_integration_test.go
View File

@ -61,7 +61,8 @@ func (m *mockStore) GetOrCreateASN(number int, timestamp time.Time) (*database.A
} }
asn := &database.ASN{ asn := &database.ASN{
ASN: number, ID: uuid.New(),
Number: number,
FirstSeen: timestamp, FirstSeen: timestamp,
LastSeen: timestamp, LastSeen: timestamp,
} }
@ -71,37 +72,6 @@ func (m *mockStore) GetOrCreateASN(number int, timestamp time.Time) (*database.A
return asn, nil return asn, nil
} }
// UpdatePrefixesBatch mock implementation
func (m *mockStore) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
m.mu.Lock()
defer m.mu.Unlock()
for prefix, timestamp := range prefixes {
if p, exists := m.Prefixes[prefix]; exists {
p.LastSeen = timestamp
} else {
const (
ipVersionV4 = 4
ipVersionV6 = 6
)
ipVersion := ipVersionV4
if strings.Contains(prefix, ":") {
ipVersion = ipVersionV6
}
m.Prefixes[prefix] = &database.Prefix{
ID: uuid.New(),
Prefix: prefix,
IPVersion: ipVersion,
FirstSeen: timestamp,
LastSeen: timestamp,
}
}
}
return nil
}
// GetOrCreatePrefix mock implementation // GetOrCreatePrefix mock implementation
func (m *mockStore) GetOrCreatePrefix(prefix string, timestamp time.Time) (*database.Prefix, error) { func (m *mockStore) GetOrCreatePrefix(prefix string, timestamp time.Time) (*database.Prefix, error) {
m.mu.Lock() m.mu.Lock()
@ -291,17 +261,6 @@ func (m *mockStore) GetRandomPrefixesByLengthContext(ctx context.Context, maskLe
return m.GetRandomPrefixesByLength(maskLength, ipVersion, limit) return m.GetRandomPrefixesByLength(maskLength, ipVersion, limit)
} }
// GetASPeers mock implementation
func (m *mockStore) GetASPeers(asn int) ([]database.ASPeer, error) {
// Return empty peers for now
return []database.ASPeer{}, nil
}
// GetASPeersContext mock implementation with context support
func (m *mockStore) GetASPeersContext(ctx context.Context, asn int) ([]database.ASPeer, error) {
return m.GetASPeers(asn)
}
// UpsertLiveRouteBatch mock implementation // UpsertLiveRouteBatch mock implementation
func (m *mockStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error { func (m *mockStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error {
m.mu.Lock() m.mu.Lock()
@ -343,7 +302,8 @@ func (m *mockStore) GetOrCreateASNBatch(asns map[int]time.Time) error {
for number, timestamp := range asns { for number, timestamp := 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, ID: uuid.New(),
Number: number,
FirstSeen: timestamp, FirstSeen: timestamp,
LastSeen: timestamp, LastSeen: timestamp,
} }

2
internal/routewatch/peeringhandler.go
View File

@ -21,7 +21,7 @@ const (
pathExpirationTime = 30 * time.Minute pathExpirationTime = 30 * time.Minute
// peeringProcessInterval is how often to process AS paths into peerings // peeringProcessInterval is how often to process AS paths into peerings
peeringProcessInterval = 30 * time.Second peeringProcessInterval = 2 * time.Minute
// pathPruneInterval is how often to prune old AS paths // pathPruneInterval is how often to prune old AS paths
pathPruneInterval = 5 * time.Minute pathPruneInterval = 5 * time.Minute

28
internal/routewatch/prefixhandler.go
View File

@ -19,7 +19,7 @@ const (
prefixHandlerQueueSize = 100000 prefixHandlerQueueSize = 100000
// prefixBatchSize is the number of prefix updates to batch together // prefixBatchSize is the number of prefix updates to batch together
prefixBatchSize = 25000 prefixBatchSize = 20000
// prefixBatchTimeout is the maximum time to wait before flushing a batch // prefixBatchTimeout is the maximum time to wait before flushing a batch
// DO NOT reduce this timeout - larger batches are more efficient // DO NOT reduce this timeout - larger batches are more efficient
@ -182,15 +182,9 @@ func (h *PrefixHandler) flushBatchLocked() {
var routesToUpsert []*database.LiveRoute var routesToUpsert []*database.LiveRoute
var routesToDelete []database.LiveRouteDeletion var routesToDelete []database.LiveRouteDeletion
// Collect unique prefixes to update // Skip the prefix table updates entirely - just update live_routes
prefixesToUpdate := make(map[string]time.Time) // The prefix table is not critical for routing lookups
for _, update := range prefixMap { for _, update := range prefixMap {
// 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 { if update.messageType == "announcement" && update.originASN > 0 {
// Create live route for batch upsert // Create live route for batch upsert
route := h.createLiveRoute(update) route := h.createLiveRoute(update)
@ -198,20 +192,11 @@ func (h *PrefixHandler) flushBatchLocked() {
routesToUpsert = append(routesToUpsert, route) routesToUpsert = append(routesToUpsert, route)
} }
} else if update.messageType == "withdrawal" { } 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 // Create deletion record for batch delete
routesToDelete = append(routesToDelete, database.LiveRouteDeletion{ routesToDelete = append(routesToDelete, database.LiveRouteDeletion{
Prefix: update.prefix, Prefix: update.prefix,
OriginASN: update.originASN, OriginASN: update.originASN,
PeerIP: update.peer, PeerIP: update.peer,
IPVersion: ipVersion,
}) })
} }
} }
@ -234,13 +219,6 @@ func (h *PrefixHandler) flushBatchLocked() {
} }
} }
// 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) elapsed := time.Since(startTime)
h.logger.Debug("Flushed prefix batch", h.logger.Debug("Flushed prefix batch",
"batch_size", batchSize, "batch_size", batchSize,

32
internal/server/handlers.go
View File

@ -80,8 +80,8 @@ func (s *Server) handleStatusJSON() http.HandlerFunc {
} }
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
// Create a 4 second timeout context for this request // Create a 1 second timeout context for this request
ctx, cancel := context.WithTimeout(r.Context(), 4*time.Second) ctx, cancel := context.WithTimeout(r.Context(), 1*time.Second)
defer cancel() defer cancel()
metrics := s.streamer.GetMetrics() metrics := s.streamer.GetMetrics()
@ -177,7 +177,6 @@ func (s *Server) handleStats() http.HandlerFunc {
Name string `json:"name"` Name string `json:"name"`
QueueLength int `json:"queue_length"` QueueLength int `json:"queue_length"`
QueueCapacity int `json:"queue_capacity"` QueueCapacity int `json:"queue_capacity"`
QueueHighWaterMark int `json:"queue_high_water_mark"`
ProcessedCount uint64 `json:"processed_count"` ProcessedCount uint64 `json:"processed_count"`
DroppedCount uint64 `json:"dropped_count"` DroppedCount uint64 `json:"dropped_count"`
AvgProcessTimeMs float64 `json:"avg_process_time_ms"` AvgProcessTimeMs float64 `json:"avg_process_time_ms"`
@ -214,8 +213,8 @@ func (s *Server) handleStats() http.HandlerFunc {
} }
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
// Create a 4 second timeout context for this request // Create a 1 second timeout context for this request
ctx, cancel := context.WithTimeout(r.Context(), 4*time.Second) ctx, cancel := context.WithTimeout(r.Context(), 1*time.Second)
defer cancel() defer cancel()
// Check if context is already cancelled // Check if context is already cancelled
@ -252,7 +251,7 @@ func (s *Server) handleStats() http.HandlerFunc {
return return
case err := <-errChan: case err := <-errChan:
s.logger.Error("Failed to get database stats", "error", err) s.logger.Error("Failed to get database stats", "error", err)
writeJSONError(w, http.StatusInternalServerError, err.Error()) http.Error(w, err.Error(), http.StatusInternalServerError)
return return
case dbStats = <-statsChan: case dbStats = <-statsChan:
@ -285,7 +284,6 @@ func (s *Server) handleStats() http.HandlerFunc {
Name: hs.Name, Name: hs.Name,
QueueLength: hs.QueueLength, QueueLength: hs.QueueLength,
QueueCapacity: hs.QueueCapacity, QueueCapacity: hs.QueueCapacity,
QueueHighWaterMark: hs.QueueHighWaterMark,
ProcessedCount: hs.ProcessedCount, ProcessedCount: hs.ProcessedCount,
DroppedCount: hs.DroppedCount, DroppedCount: hs.DroppedCount,
AvgProcessTimeMs: float64(hs.AvgProcessTime.Microseconds()) / microsecondsPerMillisecond, AvgProcessTimeMs: float64(hs.AvgProcessTime.Microseconds()) / microsecondsPerMillisecond,
@ -493,14 +491,6 @@ func (s *Server) handleASDetail() http.HandlerFunc {
return return
} }
// Get peers
peers, err := s.db.GetASPeersContext(r.Context(), asn)
if err != nil {
s.logger.Error("Failed to get AS peers", "error", err)
// Continue without peers rather than failing the whole request
peers = []database.ASPeer{}
}
// Group prefixes by IP version // Group prefixes by IP version
const ipVersionV4 = 4 const ipVersionV4 = 4
var ipv4Prefixes, ipv6Prefixes []database.LiveRoute var ipv4Prefixes, ipv6Prefixes []database.LiveRoute
@ -557,8 +547,6 @@ func (s *Server) handleASDetail() http.HandlerFunc {
TotalCount int TotalCount int
IPv4Count int IPv4Count int
IPv6Count int IPv6Count int
Peers []database.ASPeer
PeerCount int
}{ }{
ASN: asInfo, ASN: asInfo,
IPv4Prefixes: ipv4Prefixes, IPv4Prefixes: ipv4Prefixes,
@ -566,8 +554,6 @@ func (s *Server) handleASDetail() http.HandlerFunc {
TotalCount: len(prefixes), TotalCount: len(prefixes),
IPv4Count: len(ipv4Prefixes), IPv4Count: len(ipv4Prefixes),
IPv6Count: len(ipv6Prefixes), IPv6Count: len(ipv6Prefixes),
Peers: peers,
PeerCount: len(peers),
} }
// Check if context is still valid before writing response // Check if context is still valid before writing response
@ -619,7 +605,7 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
// Group by origin AS and collect unique AS info // Group by origin AS and collect unique AS info
type ASNInfo struct { type ASNInfo struct {
ASN int Number int
Handle string Handle string
Description string Description string
PeerCount int PeerCount int
@ -636,7 +622,7 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
description = asInfo.Description description = asInfo.Description
} }
originMap[route.OriginASN] = &ASNInfo{ originMap[route.OriginASN] = &ASNInfo{
ASN: route.OriginASN, Number: route.OriginASN,
Handle: handle, Handle: handle,
Description: description, Description: description,
PeerCount: 0, PeerCount: 0,
@ -669,7 +655,7 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
// Create enhanced routes with AS path handles // Create enhanced routes with AS path handles
type ASPathEntry struct { type ASPathEntry struct {
ASN int Number int
Handle string Handle string
} }
type EnhancedRoute struct { type EnhancedRoute struct {
@ -688,7 +674,7 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
for j, asn := range route.ASPath { for j, asn := range route.ASPath {
handle := asinfo.GetHandle(asn) handle := asinfo.GetHandle(asn)
enhancedRoute.ASPathWithHandle[j] = ASPathEntry{ enhancedRoute.ASPathWithHandle[j] = ASPathEntry{
ASN: asn, Number: asn,
Handle: handle, Handle: handle,
} }
} }

65
internal/server/middleware.go
View File

@ -217,68 +217,3 @@ func TimeoutMiddleware(timeout time.Duration) func(http.Handler) http.Handler {
}) })
} }
} }
// JSONValidationMiddleware ensures all JSON API responses are valid JSON
func JSONValidationMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// Create a custom response writer to capture the response
rw := &responseWriter{
ResponseWriter: w,
body: &bytes.Buffer{},
statusCode: http.StatusOK,
}
// Serve the request
next.ServeHTTP(rw, r)
// Check if it's meant to be a JSON response
contentType := rw.Header().Get("Content-Type")
isJSON := contentType == "application/json" || contentType == ""
// If it's not JSON or has content, pass through
if !isJSON && rw.body.Len() > 0 {
w.WriteHeader(rw.statusCode)
_, _ = w.Write(rw.body.Bytes())
return
}
// For JSON responses, validate the JSON
if rw.body.Len() > 0 {
var testParse interface{}
if err := json.Unmarshal(rw.body.Bytes(), &testParse); err == nil {
// Valid JSON, write it out
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(rw.statusCode)
_, _ = w.Write(rw.body.Bytes())
return
}
}
// If we get here, either there's no body or invalid JSON
// Write a proper error response
w.Header().Set("Content-Type", "application/json")
// Determine appropriate status code
statusCode := rw.statusCode
if statusCode == http.StatusOK {
statusCode = http.StatusInternalServerError
}
w.WriteHeader(statusCode)
errorMsg := "Internal server error"
if statusCode == http.StatusRequestTimeout {
errorMsg = "Request timeout"
}
response := map[string]interface{}{
"status": "error",
"error": map[string]interface{}{
"msg": errorMsg,
"code": statusCode,
},
}
_ = json.NewEncoder(w).Encode(response)
})
}

5
internal/server/routes.go
View File

@ -16,14 +16,14 @@ func (s *Server) setupRoutes() {
r.Use(middleware.RealIP) r.Use(middleware.RealIP)
r.Use(middleware.Logger) r.Use(middleware.Logger)
r.Use(middleware.Recoverer) r.Use(middleware.Recoverer)
const requestTimeout = 8 * time.Second const requestTimeout = 2 * time.Second
r.Use(TimeoutMiddleware(requestTimeout)) r.Use(TimeoutMiddleware(requestTimeout))
r.Use(JSONResponseMiddleware) r.Use(JSONResponseMiddleware)
// Routes // Routes
r.Get("/", s.handleRoot()) r.Get("/", s.handleRoot())
r.Get("/status", s.handleStatusHTML()) r.Get("/status", s.handleStatusHTML())
r.Get("/status.json", JSONValidationMiddleware(s.handleStatusJSON()).ServeHTTP) r.Get("/status.json", s.handleStatusJSON())
// AS and prefix detail pages // AS and prefix detail pages
r.Get("/as/{asn}", s.handleASDetail()) r.Get("/as/{asn}", s.handleASDetail())
@ -33,7 +33,6 @@ func (s *Server) setupRoutes() {
// API routes // API routes
r.Route("/api/v1", func(r chi.Router) { r.Route("/api/v1", func(r chi.Router) {
r.Use(JSONValidationMiddleware)
r.Get("/stats", s.handleStats()) r.Get("/stats", s.handleStats())
r.Get("/ip/{ip}", s.handleIPLookup()) r.Get("/ip/{ip}", s.handleIPLookup())
r.Get("/as/{asn}", s.handleASDetailJSON()) r.Get("/as/{asn}", s.handleASDetailJSON())

2
internal/server/server.go
View File

@ -42,7 +42,7 @@ func (s *Server) Start() error {
port = "8080" port = "8080"
} }
const readHeaderTimeout = 40 * time.Second const readHeaderTimeout = 10 * time.Second
s.srv = &http.Server{ s.srv = &http.Server{
Addr: ":" + port, Addr: ":" + port,
Handler: s.router, Handler: s.router,

55
internal/streamer/streamer.go
View File

@ -8,7 +8,6 @@ import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"math" "math"
"math/rand"
"net/http" "net/http"
"sync" "sync"
"sync/atomic" "sync/atomic"
@ -30,10 +29,6 @@ const (
bytesPerKB = 1024 bytesPerKB = 1024
bytesPerMB = 1024 * 1024 bytesPerMB = 1024 * 1024
maxConcurrentHandlers = 800 // Maximum number of concurrent message handlers maxConcurrentHandlers = 800 // Maximum number of concurrent message handlers
// Backpressure constants
backpressureThreshold = 0.5 // Start dropping at 50% queue utilization
backpressureSlope = 2.0 // Slope for linear drop probability increase
) )
// MessageHandler is an interface for handling RIS messages // MessageHandler is an interface for handling RIS messages
@ -59,7 +54,6 @@ type handlerMetrics struct {
totalTime time.Duration // Total processing time (for average calculation) totalTime time.Duration // Total processing time (for average calculation)
minTime time.Duration // Minimum processing time minTime time.Duration // Minimum processing time
maxTime time.Duration // Maximum processing time maxTime time.Duration // Maximum processing time
queueHighWaterMark int // Maximum queue length seen
mu sync.Mutex // Protects the metrics mu sync.Mutex // Protects the metrics
} }
@ -81,7 +75,6 @@ type Streamer struct {
running bool running bool
metrics *metrics.Tracker metrics *metrics.Tracker
totalDropped uint64 // Total dropped messages across all handlers totalDropped uint64 // Total dropped messages across all handlers
random *rand.Rand // Random number generator for backpressure drops
} }
// New creates a new RIS streamer // New creates a new RIS streamer
@ -93,8 +86,6 @@ func New(logger *logger.Logger, metrics *metrics.Tracker) *Streamer {
}, },
handlers: make([]*handlerInfo, 0), handlers: make([]*handlerInfo, 0),
metrics: metrics, metrics: metrics,
//nolint:gosec // Non-cryptographic randomness is fine for backpressure
random: rand.New(rand.NewSource(time.Now().UnixNano())),
} }
} }
@ -218,7 +209,6 @@ type HandlerStats struct {
Name string Name string
QueueLength int QueueLength int
QueueCapacity int QueueCapacity int
QueueHighWaterMark int
ProcessedCount uint64 ProcessedCount uint64
DroppedCount uint64 DroppedCount uint64
AvgProcessTime time.Duration AvgProcessTime time.Duration
@ -240,7 +230,6 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
Name: fmt.Sprintf("%T", info.handler), Name: fmt.Sprintf("%T", info.handler),
QueueLength: len(info.queue), QueueLength: len(info.queue),
QueueCapacity: cap(info.queue), QueueCapacity: cap(info.queue),
QueueHighWaterMark: info.metrics.queueHighWaterMark,
ProcessedCount: info.metrics.processedCount, ProcessedCount: info.metrics.processedCount,
DroppedCount: info.metrics.droppedCount, DroppedCount: info.metrics.droppedCount,
MinProcessTime: info.metrics.minTime, MinProcessTime: info.metrics.minTime,
@ -537,35 +526,17 @@ func (s *Streamer) stream(ctx context.Context) error {
// Dispatch to interested handlers // Dispatch to interested handlers
s.mu.RLock() s.mu.RLock()
for _, info := range s.handlers { for _, info := range s.handlers {
if !info.handler.WantsMessage(msg.Type) { if info.handler.WantsMessage(msg.Type) {
continue
}
// Check if we should drop due to backpressure
if s.shouldDropForBackpressure(info) {
atomic.AddUint64(&info.metrics.droppedCount, 1)
atomic.AddUint64(&s.totalDropped, 1)
continue
}
// Try to queue the message
select { select {
case info.queue <- &msg: case info.queue <- &msg:
// Message queued successfully // Message queued successfully
// Update high water mark if needed
queueLen := len(info.queue)
info.metrics.mu.Lock()
if queueLen > info.metrics.queueHighWaterMark {
info.metrics.queueHighWaterMark = queueLen
}
info.metrics.mu.Unlock()
default: default:
// Queue is full, drop the message // Queue is full, drop the message
atomic.AddUint64(&info.metrics.droppedCount, 1) atomic.AddUint64(&info.metrics.droppedCount, 1)
atomic.AddUint64(&s.totalDropped, 1) atomic.AddUint64(&s.totalDropped, 1)
} }
} }
}
s.mu.RUnlock() s.mu.RUnlock()
} }
@ -575,25 +546,3 @@ func (s *Streamer) stream(ctx context.Context) error {
return nil return nil
} }
// shouldDropForBackpressure determines if a message should be dropped based on queue utilization
func (s *Streamer) shouldDropForBackpressure(info *handlerInfo) bool {
// Calculate queue utilization
queueLen := len(info.queue)
queueCap := cap(info.queue)
utilization := float64(queueLen) / float64(queueCap)
// No drops below threshold
if utilization < backpressureThreshold {
return false
}
// Calculate drop probability (0.0 at threshold, 1.0 at 100% full)
dropProbability := (utilization - backpressureThreshold) * backpressureSlope
if dropProbability > 1.0 {
dropProbability = 1.0
}
// Random drop based on probability
return s.random.Float64() < dropProbability
}

46
internal/templates/as_detail.html
View File

@ -3,7 +3,7 @@
<head> <head>
<meta charset="UTF-8"> <meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0"> <meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>AS{{.ASN.ASN}} - {{.ASN.Handle}} - RouteWatch</title> <title>AS{{.ASN.Number}} - {{.ASN.Handle}} - RouteWatch</title>
<style> <style>
body { body {
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
@ -136,7 +136,7 @@
<div class="container"> <div class="container">
<a href="/status" class="nav-link">← Back to Status</a> <a href="/status" class="nav-link">← Back to Status</a>
<h1>AS{{.ASN.ASN}}{{if .ASN.Handle}} - {{.ASN.Handle}}{{end}}</h1> <h1>AS{{.ASN.Number}}{{if .ASN.Handle}} - {{.ASN.Handle}}{{end}}</h1>
{{if .ASN.Description}} {{if .ASN.Description}}
<p class="subtitle">{{.ASN.Description}}</p> <p class="subtitle">{{.ASN.Description}}</p>
{{end}} {{end}}
@ -154,10 +154,6 @@
<div class="info-label">IPv6 Prefixes</div> <div class="info-label">IPv6 Prefixes</div>
<div class="info-value">{{.IPv6Count}}</div> <div class="info-value">{{.IPv6Count}}</div>
</div> </div>
<div class="info-card">
<div class="info-label">Peer ASNs</div>
<div class="info-value">{{.PeerCount}}</div>
</div>
<div class="info-card"> <div class="info-card">
<div class="info-label">First Seen</div> <div class="info-label">First Seen</div>
<div class="info-value">{{.ASN.FirstSeen.Format "2006-01-02"}}</div> <div class="info-value">{{.ASN.FirstSeen.Format "2006-01-02"}}</div>
@ -227,44 +223,6 @@
<p>No prefixes announced by this AS</p> <p>No prefixes announced by this AS</p>
</div> </div>
{{end}} {{end}}
{{if .Peers}}
<div class="prefix-section">
<div class="prefix-header">
<h2>Peer ASNs</h2>
<span class="prefix-count">{{.PeerCount}}</span>
</div>
<table class="prefix-table">
<thead>
<tr>
<th>ASN</th>
<th>Handle</th>
<th>Description</th>
<th>First Seen</th>
<th>Last Seen</th>
</tr>
</thead>
<tbody>
{{range .Peers}}
<tr>
<td><a href="/as/{{.ASN}}" class="prefix-link">AS{{.ASN}}</a></td>
<td>{{if .Handle}}{{.Handle}}{{else}}-{{end}}</td>
<td>{{if .Description}}{{.Description}}{{else}}-{{end}}</td>
<td>{{.FirstSeen.Format "2006-01-02"}}</td>
<td>{{.LastSeen.Format "2006-01-02"}}</td>
</tr>
{{end}}
</tbody>
</table>
</div>
{{else}}
<div class="prefix-section">
<h2>Peer ASNs</h2>
<div class="empty-state">
<p>No peering relationships found for this AS</p>
</div>
</div>
{{end}}
</div> </div>
</body> </body>
</html> </html>

4
internal/templates/prefix_detail.html
View File

@ -207,7 +207,7 @@
<div class="origin-list"> <div class="origin-list">
{{range .Origins}} {{range .Origins}}
<div class="origin-item"> <div class="origin-item">
<a href="/as/{{.ASN}}" class="as-link">AS{{.ASN}}</a> <a href="/as/{{.Number}}" class="as-link">AS{{.Number}}</a>
{{if .Handle}} ({{.Handle}}){{end}} {{if .Handle}} ({{.Handle}}){{end}}
<span style="color: #7f8c8d; margin-left: 10px;">{{.PeerCount}} peer{{if ne .PeerCount 1}}s{{end}}</span> <span style="color: #7f8c8d; margin-left: 10px;">{{.PeerCount}} peer{{if ne .PeerCount 1}}s{{end}}</span>
</div> </div>
@ -240,7 +240,7 @@
<a href="/as/{{.OriginASN}}" class="as-link">AS{{.OriginASN}}</a> <a href="/as/{{.OriginASN}}" class="as-link">AS{{.OriginASN}}</a>
</td> </td>
<td class="peer-ip">{{.PeerIP}}</td> <td class="peer-ip">{{.PeerIP}}</td>
<td class="as-path">{{range $i, $as := .ASPathWithHandle}}{{if $i}} → {{end}}<a href="/as/{{$as.ASN}}" class="as-link">{{if $as.Handle}}{{$as.Handle}}{{else}}AS{{$as.ASN}}{{end}}</a>{{end}}</td> <td class="as-path">{{range $i, $as := .ASPathWithHandle}}{{if $i}} → {{end}}<a href="/as/{{$as.Number}}" class="as-link">{{if $as.Handle}}{{$as.Handle}}{{else}}AS{{$as.Number}}{{end}}</a>{{end}}</td>
<td class="peer-ip">{{.NextHop}}</td> <td class="peer-ip">{{.NextHop}}</td>
<td>{{.LastUpdated.Format "2006-01-02 15:04:05"}}</td> <td>{{.LastUpdated.Format "2006-01-02 15:04:05"}}</td>
<td class="age">{{.LastUpdated | timeSince}}</td> <td class="age">{{.LastUpdated | timeSince}}</td>

43
internal/templates/status.html
View File

@ -264,10 +264,6 @@
<span class="metric-label">Queue</span> <span class="metric-label">Queue</span>
<span class="metric-value">${handler.queue_length}/${handler.queue_capacity}</span> <span class="metric-value">${handler.queue_length}/${handler.queue_capacity}</span>
</div> </div>
<div class="metric">
<span class="metric-label">High Water Mark</span>
<span class="metric-value">${handler.queue_high_water_mark}/${handler.queue_capacity} (${Math.round(handler.queue_high_water_mark * 100 / handler.queue_capacity)}%)</span>
</div>
<div class="metric"> <div class="metric">
<span class="metric-label">Processed</span> <span class="metric-label">Processed</span>
<span class="metric-value">${formatNumber(handler.processed_count)}</span> <span class="metric-value">${formatNumber(handler.processed_count)}</span>
@ -290,39 +286,6 @@
}); });
} }
function resetAllFields() {
// Reset all metric fields to '-'
document.getElementById('connected').textContent = '-';
document.getElementById('connected').className = 'metric-value';
document.getElementById('uptime').textContent = '-';
document.getElementById('go_version').textContent = '-';
document.getElementById('goroutines').textContent = '-';
document.getElementById('memory_usage').textContent = '-';
document.getElementById('total_messages').textContent = '-';
document.getElementById('messages_per_sec').textContent = '-';
document.getElementById('total_bytes').textContent = '-';
document.getElementById('mbits_per_sec').textContent = '-';
document.getElementById('asns').textContent = '-';
document.getElementById('prefixes').textContent = '-';
document.getElementById('ipv4_prefixes').textContent = '-';
document.getElementById('ipv6_prefixes').textContent = '-';
document.getElementById('peerings').textContent = '-';
document.getElementById('peers').textContent = '-';
document.getElementById('database_size').textContent = '-';
document.getElementById('live_routes').textContent = '-';
document.getElementById('ipv4_routes').textContent = '-';
document.getElementById('ipv6_routes').textContent = '-';
document.getElementById('ipv4_updates_per_sec').textContent = '-';
document.getElementById('ipv6_updates_per_sec').textContent = '-';
// Clear handler stats
document.getElementById('handler-stats-container').innerHTML = '';
// Clear prefix distributions
document.getElementById('ipv4-prefix-distribution').innerHTML = '<div class="metric"><span class="metric-label">No data</span></div>';
document.getElementById('ipv6-prefix-distribution').innerHTML = '<div class="metric"><span class="metric-label">No data</span></div>';
}
function updateStatus() { function updateStatus() {
fetch('/api/v1/stats') fetch('/api/v1/stats')
.then(response => response.json()) .then(response => response.json())
@ -331,7 +294,6 @@
if (response.status === 'error') { if (response.status === 'error') {
document.getElementById('error').textContent = 'Error: ' + response.error.msg; document.getElementById('error').textContent = 'Error: ' + response.error.msg;
document.getElementById('error').style.display = 'block'; document.getElementById('error').style.display = 'block';
resetAllFields();
return; return;
} }
@ -378,13 +340,12 @@
.catch(error => { .catch(error => {
document.getElementById('error').textContent = 'Error fetching status: ' + error; document.getElementById('error').textContent = 'Error fetching status: ' + error;
document.getElementById('error').style.display = 'block'; document.getElementById('error').style.display = 'block';
resetAllFields();
}); });
} }
// Update immediately and then every 2 seconds // Update immediately and then every 500ms
updateStatus(); updateStatus();
setInterval(updateStatus, 2000); setInterval(updateStatus, 500);
</script> </script>
</body> </body>
</html> </html>

385054
log.txt
View File

File diff suppressed because it is too large Load Diff