WIP: lint fixes in the smaller packages
Model: opus-5-5
This commit is contained in:
+22
-7
@@ -2,8 +2,10 @@
|
|||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"encoding/csv"
|
"encoding/csv"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"io"
|
"io"
|
||||||
"log"
|
"log"
|
||||||
"net/http"
|
"net/http"
|
||||||
@@ -27,13 +29,17 @@ func main() {
|
|||||||
|
|
||||||
// Fetch the CSV data
|
// Fetch the CSV data
|
||||||
log.Println("Fetching AS CSV data...")
|
log.Println("Fetching AS CSV data...")
|
||||||
resp, err := http.Get(asCsvURL)
|
|
||||||
|
req, err := http.NewRequestWithContext(
|
||||||
|
context.Background(), http.MethodGet, asCsvURL, nil)
|
||||||
|
if err != nil {
|
||||||
|
log.Fatalf("Failed to fetch CSV: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
resp, err := http.DefaultClient.Do(req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatalf("Failed to fetch CSV: %v", err)
|
log.Fatalf("Failed to fetch CSV: %v", err)
|
||||||
}
|
}
|
||||||
defer func() {
|
|
||||||
_ = resp.Body.Close()
|
|
||||||
}()
|
|
||||||
|
|
||||||
if resp.StatusCode != http.StatusOK {
|
if resp.StatusCode != http.StatusOK {
|
||||||
log.Fatalf("Failed to fetch CSV: HTTP %d", resp.StatusCode)
|
log.Fatalf("Failed to fetch CSV: HTTP %d", resp.StatusCode)
|
||||||
@@ -48,17 +54,20 @@ func main() {
|
|||||||
log.Fatalf("Failed to read CSV header: %v", err)
|
log.Fatalf("Failed to read CSV header: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
if len(header) != 3 || header[0] != "asn" || header[1] != "handle" || header[2] != "description" {
|
if len(header) != 3 || header[0] != "asn" || header[1] != "handle" ||
|
||||||
|
header[2] != "description" {
|
||||||
log.Fatalf("Unexpected CSV header: %v", header)
|
log.Fatalf("Unexpected CSV header: %v", header)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Read all records
|
// Read all records
|
||||||
var asInfos []ASInfo
|
var asInfos []ASInfo
|
||||||
|
|
||||||
for {
|
for {
|
||||||
record, err := reader.Read()
|
record, err := reader.Read()
|
||||||
if err == io.EOF {
|
if errors.Is(err, io.EOF) {
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("Error reading CSV record: %v", err)
|
log.Printf("Error reading CSV record: %v", err)
|
||||||
|
|
||||||
@@ -86,12 +95,18 @@ func main() {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Closed here rather than deferred: log.Fatalf exits without running
|
||||||
|
// deferred calls.
|
||||||
|
_ = resp.Body.Close()
|
||||||
|
|
||||||
log.Printf("Parsed %d AS records", len(asInfos))
|
log.Printf("Parsed %d AS records", len(asInfos))
|
||||||
|
|
||||||
// Convert to JSON and output to stdout
|
// Convert to JSON and output to stdout
|
||||||
encoder := json.NewEncoder(os.Stdout)
|
encoder := json.NewEncoder(os.Stdout)
|
||||||
encoder.SetIndent("", "\t")
|
encoder.SetIndent("", "\t")
|
||||||
if err := encoder.Encode(asInfos); err != nil {
|
|
||||||
|
err = encoder.Encode(asInfos)
|
||||||
|
if err != nil {
|
||||||
log.Fatalf("Failed to encode JSON: %v", err)
|
log.Fatalf("Failed to encode JSON: %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -26,9 +26,17 @@ func main() {
|
|||||||
|
|
||||||
// Register raw message handler that prints to stdout
|
// Register raw message handler that prints to stdout
|
||||||
s.RegisterRawHandler(func(line string) {
|
s.RegisterRawHandler(func(line string) {
|
||||||
fmt.Println(line)
|
_, _ = fmt.Fprintln(os.Stdout, line)
|
||||||
})
|
})
|
||||||
|
|
||||||
|
// Start streaming before any defer: log.Fatal exits without running
|
||||||
|
// deferred calls.
|
||||||
|
err := s.Start()
|
||||||
|
if err != nil {
|
||||||
|
log.Fatal("Failed to start streamer:", err)
|
||||||
|
}
|
||||||
|
defer s.Stop()
|
||||||
|
|
||||||
// Set up context with cancellation
|
// Set up context with cancellation
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
defer cancel()
|
defer cancel()
|
||||||
@@ -36,18 +44,13 @@ func main() {
|
|||||||
// Handle signals
|
// Handle signals
|
||||||
sigChan := make(chan os.Signal, 1)
|
sigChan := make(chan os.Signal, 1)
|
||||||
signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
|
signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
<-sigChan
|
<-sigChan
|
||||||
log.Println("Received shutdown signal")
|
log.Println("Received shutdown signal")
|
||||||
cancel()
|
cancel()
|
||||||
}()
|
}()
|
||||||
|
|
||||||
// Start streaming
|
|
||||||
if err := s.Start(); err != nil {
|
|
||||||
log.Fatal("Failed to start streamer:", err)
|
|
||||||
}
|
|
||||||
defer s.Stop()
|
|
||||||
|
|
||||||
// Wait for context cancellation
|
// Wait for context cancellation
|
||||||
<-ctx.Done()
|
<-ctx.Done()
|
||||||
}
|
}
|
||||||
|
|||||||
+27
-12
@@ -2,6 +2,7 @@
|
|||||||
package config
|
package config
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
@@ -27,6 +28,12 @@ const (
|
|||||||
maxPort = 65535
|
maxPort = 65535
|
||||||
)
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
errRelativeXDGDataHome = errors.New("XDG_DATA_HOME must be an absolute path")
|
||||||
|
errUnsupportedOS = errors.New("unsupported operating system")
|
||||||
|
errInvalidPort = errors.New("PORT must be a whole number")
|
||||||
|
)
|
||||||
|
|
||||||
// Config holds configuration for the entire application
|
// Config holds configuration for the entire application
|
||||||
type Config struct {
|
type Config struct {
|
||||||
// StateDir is the directory for all application state (database, snapshots)
|
// StateDir is the directory for all application state (database, snapshots)
|
||||||
@@ -38,11 +45,14 @@ type Config struct {
|
|||||||
// MaxRuntime is the maximum runtime (0 = run forever)
|
// MaxRuntime is the maximum runtime (0 = run forever)
|
||||||
MaxRuntime time.Duration
|
MaxRuntime time.Duration
|
||||||
|
|
||||||
// EnableBatchedDatabaseWrites enables batched database operations for better performance
|
// EnableBatchedDatabaseWrites enables batched database operations for
|
||||||
|
// better performance
|
||||||
EnableBatchedDatabaseWrites bool
|
EnableBatchedDatabaseWrites bool
|
||||||
|
|
||||||
// RouteExpirationTimeout is how long a route can go without being refreshed before expiring
|
// RouteExpirationTimeout is how long a route can go without being
|
||||||
// Default is 2 hours which is conservative for BGP (typical BGP hold time is 90-180 seconds)
|
// refreshed before expiring
|
||||||
|
// Default is 2 hours which is conservative for BGP (typical BGP hold
|
||||||
|
// time is 90-180 seconds)
|
||||||
RouteExpirationTimeout time.Duration
|
RouteExpirationTimeout time.Duration
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -59,11 +69,14 @@ func New() (*Config, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
return &Config{
|
return &Config{
|
||||||
StateDir: stateDir,
|
StateDir: stateDir,
|
||||||
Port: port,
|
Port: port,
|
||||||
MaxRuntime: 0, // Run forever by default
|
// Run forever by default
|
||||||
EnableBatchedDatabaseWrites: true, // Enable batching by default
|
MaxRuntime: 0,
|
||||||
RouteExpirationTimeout: defaultRouteExpirationMinutes * time.Minute, // For active route monitoring
|
// Enable batching by default
|
||||||
|
EnableBatchedDatabaseWrites: true,
|
||||||
|
// For active route monitoring
|
||||||
|
RouteExpirationTimeout: defaultRouteExpirationMinutes * time.Minute,
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -89,7 +102,7 @@ func getStateDirectory() (string, error) {
|
|||||||
// the database somewhere unexpected.
|
// the database somewhere unexpected.
|
||||||
xdgData := os.Getenv("XDG_DATA_HOME")
|
xdgData := os.Getenv("XDG_DATA_HOME")
|
||||||
if xdgData != "" && !filepath.IsAbs(xdgData) {
|
if xdgData != "" && !filepath.IsAbs(xdgData) {
|
||||||
return "", fmt.Errorf("XDG_DATA_HOME must be an absolute path, got %q", xdgData)
|
return "", fmt.Errorf("%w, got %q", errRelativeXDGDataHome, xdgData)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Unix-like: /var/lib/berlin.sneak.app.routewatch if root, else XDG_DATA_HOME
|
// Unix-like: /var/lib/berlin.sneak.app.routewatch if root, else XDG_DATA_HOME
|
||||||
@@ -111,7 +124,7 @@ func getStateDirectory() (string, error) {
|
|||||||
return filepath.Join(home, ".local", "share", AppIdentifier), nil
|
return filepath.Join(home, ".local", "share", AppIdentifier), nil
|
||||||
|
|
||||||
default:
|
default:
|
||||||
return "", fmt.Errorf("unsupported operating system: %s", runtime.GOOS)
|
return "", fmt.Errorf("%w: %s", errUnsupportedOS, runtime.GOOS)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -125,7 +138,8 @@ func getPort() (int, error) {
|
|||||||
// ParseUint, unlike Atoi, refuses a sign: the health check URL cannot use "+9090"
|
// ParseUint, unlike Atoi, refuses a sign: the health check URL cannot use "+9090"
|
||||||
port, err := strconv.ParseUint(value, 10, 0)
|
port, err := strconv.ParseUint(value, 10, 0)
|
||||||
if err != nil || port < 1 || port > maxPort {
|
if err != nil || port < 1 || port > maxPort {
|
||||||
return 0, fmt.Errorf("PORT must be a whole number from 1 to %d, got %q", maxPort, value)
|
return 0, fmt.Errorf("%w from 1 to %d, got %q",
|
||||||
|
errInvalidPort, maxPort, value)
|
||||||
}
|
}
|
||||||
|
|
||||||
return int(port), nil
|
return int(port), nil
|
||||||
@@ -134,7 +148,8 @@ func getPort() (int, error) {
|
|||||||
// EnsureDirectories creates all necessary directories if they don't exist
|
// EnsureDirectories creates all necessary directories if they don't exist
|
||||||
func (c *Config) EnsureDirectories() error {
|
func (c *Config) EnsureDirectories() error {
|
||||||
// Ensure state directory exists
|
// Ensure state directory exists
|
||||||
if err := os.MkdirAll(c.StateDir, dirPermissions); err != nil {
|
err := os.MkdirAll(c.StateDir, dirPermissions)
|
||||||
|
if err != nil {
|
||||||
return fmt.Errorf("failed to create state directory: %w", err)
|
return fmt.Errorf("failed to create state directory: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -36,7 +36,8 @@ func TestNewRefusesInvalidPort(t *testing.T) {
|
|||||||
t.Setenv("PORT", value)
|
t.Setenv("PORT", value)
|
||||||
t.Setenv("XDG_DATA_HOME", "")
|
t.Setenv("XDG_DATA_HOME", "")
|
||||||
|
|
||||||
if _, err := New(); err == nil {
|
_, err := New()
|
||||||
|
if err == nil {
|
||||||
t.Errorf("New() with PORT=%q returned no error", value)
|
t.Errorf("New() with PORT=%q returned no error", value)
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
@@ -51,12 +52,16 @@ func TestNewRefusesRelativeXDGDataHome(t *testing.T) {
|
|||||||
t.Setenv("PORT", "")
|
t.Setenv("PORT", "")
|
||||||
|
|
||||||
t.Setenv("XDG_DATA_HOME", "relative/path")
|
t.Setenv("XDG_DATA_HOME", "relative/path")
|
||||||
if _, err := New(); err == nil {
|
|
||||||
|
_, err := New()
|
||||||
|
if err == nil {
|
||||||
t.Error("New() with a relative XDG_DATA_HOME returned no error")
|
t.Error("New() with a relative XDG_DATA_HOME returned no error")
|
||||||
}
|
}
|
||||||
|
|
||||||
t.Setenv("XDG_DATA_HOME", "/var/lib")
|
t.Setenv("XDG_DATA_HOME", "/var/lib")
|
||||||
if _, err := New(); err != nil {
|
|
||||||
|
_, err = New()
|
||||||
|
if err != nil {
|
||||||
t.Errorf("New() with XDG_DATA_HOME=/var/lib: %v", err)
|
t.Errorf("New() with XDG_DATA_HOME=/var/lib: %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -74,6 +74,7 @@ func getSourceAttrs() []slog.Attr {
|
|||||||
|
|
||||||
// Get the function name
|
// Get the function name
|
||||||
fn := runtime.FuncForPC(pc)
|
fn := runtime.FuncForPC(pc)
|
||||||
|
|
||||||
var funcName string
|
var funcName string
|
||||||
if fn != nil {
|
if fn != nil {
|
||||||
funcName = filepath.Base(fn.Name())
|
funcName = filepath.Base(fn.Name())
|
||||||
|
|||||||
@@ -115,9 +115,11 @@ func (t *Tracker) GetStreamMetrics() StreamMetrics {
|
|||||||
if msgCount >= 0 {
|
if msgCount >= 0 {
|
||||||
totalMessages = uint64(msgCount)
|
totalMessages = uint64(msgCount)
|
||||||
}
|
}
|
||||||
|
|
||||||
if byteCount >= 0 {
|
if byteCount >= 0 {
|
||||||
totalBytes = uint64(byteCount)
|
totalBytes = uint64(byteCount)
|
||||||
}
|
}
|
||||||
|
|
||||||
if wireByteCount >= 0 {
|
if wireByteCount >= 0 {
|
||||||
totalWireBytes = uint64(wireByteCount)
|
totalWireBytes = uint64(wireByteCount)
|
||||||
}
|
}
|
||||||
@@ -161,7 +163,7 @@ func (t *Tracker) RecordWithdrawal() {
|
|||||||
func (t *Tracker) SetBGPPeerCount(count int) {
|
func (t *Tracker) SetBGPPeerCount(count int) {
|
||||||
// BGP peer count is always small (< 1000), so int32 is safe
|
// BGP peer count is always small (< 1000), so int32 is safe
|
||||||
if count > 0 && count < 1<<31 {
|
if count > 0 && count < 1<<31 {
|
||||||
t.bgpPeerCount.Store(int32(count)) //nolint:gosec // count is validated
|
t.bgpPeerCount.Store(int32(count))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -209,7 +211,8 @@ type StreamMetrics struct {
|
|||||||
TotalMessages uint64
|
TotalMessages uint64
|
||||||
// TotalBytes is the total number of decompressed bytes received since startup
|
// TotalBytes is the total number of decompressed bytes received since startup
|
||||||
TotalBytes uint64
|
TotalBytes uint64
|
||||||
// TotalWireBytes is the total number of bytes received on the wire (before decompression)
|
// TotalWireBytes is the total number of bytes received on the wire (before
|
||||||
|
// decompression)
|
||||||
TotalWireBytes uint64
|
TotalWireBytes uint64
|
||||||
// ConnectedSince is the time when the current connection was established
|
// ConnectedSince is the time when the current connection was established
|
||||||
ConnectedSince time.Time
|
ConnectedSince time.Time
|
||||||
@@ -217,9 +220,11 @@ type StreamMetrics struct {
|
|||||||
Connected bool
|
Connected bool
|
||||||
// MessagesPerSec is the rate of messages received per second (1-minute average)
|
// MessagesPerSec is the rate of messages received per second (1-minute average)
|
||||||
MessagesPerSec float64
|
MessagesPerSec float64
|
||||||
// BitsPerSec is the rate of decompressed bits received per second (1-minute average)
|
// BitsPerSec is the rate of decompressed bits received per second (1-minute
|
||||||
|
// average)
|
||||||
BitsPerSec float64
|
BitsPerSec float64
|
||||||
// WireBitsPerSec is the rate of bits received on the wire per second (1-minute average)
|
// WireBitsPerSec is the rate of bits received on the wire per second
|
||||||
|
// (1-minute average)
|
||||||
WireBitsPerSec float64
|
WireBitsPerSec float64
|
||||||
// ReconnectCount is the number of reconnections since startup
|
// ReconnectCount is the number of reconnections since startup
|
||||||
ReconnectCount uint64
|
ReconnectCount uint64
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
// Package ristypes defines the data structures for RIS Live BGP messages and announcements.
|
// Package ristypes defines the data structures for RIS Live BGP messages and
|
||||||
|
// announcements.
|
||||||
package ristypes
|
package ristypes
|
||||||
|
|
||||||
import (
|
import (
|
||||||
@@ -17,25 +18,30 @@ type ASPath []int
|
|||||||
func (p *ASPath) UnmarshalJSON(data []byte) error {
|
func (p *ASPath) UnmarshalJSON(data []byte) error {
|
||||||
// First try to unmarshal as a simple array of integers
|
// First try to unmarshal as a simple array of integers
|
||||||
var simple []int
|
var simple []int
|
||||||
if err := json.Unmarshal(data, &simple); err == nil {
|
|
||||||
|
err := json.Unmarshal(data, &simple)
|
||||||
|
if err == nil {
|
||||||
*p = ASPath(simple)
|
*p = ASPath(simple)
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// If that fails, unmarshal as array of interfaces and flatten
|
// If that fails, unmarshal as array of interfaces and flatten
|
||||||
var raw []interface{}
|
var raw []any
|
||||||
if err := json.Unmarshal(data, &raw); err != nil {
|
|
||||||
|
err = json.Unmarshal(data, &raw)
|
||||||
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
// Flatten the array
|
// Flatten the array
|
||||||
result := make([]int, 0)
|
result := make([]int, 0)
|
||||||
|
|
||||||
for _, item := range raw {
|
for _, item := range raw {
|
||||||
switch v := item.(type) {
|
switch v := item.(type) {
|
||||||
case float64:
|
case float64:
|
||||||
result = append(result, int(v))
|
result = append(result, int(v))
|
||||||
case []interface{}:
|
case []any:
|
||||||
// Nested array - flatten it
|
// Nested array - flatten it
|
||||||
for _, nested := range v {
|
for _, nested := range v {
|
||||||
if num, ok := nested.(float64); ok {
|
if num, ok := nested.(float64); ok {
|
||||||
@@ -62,6 +68,8 @@ type RISLiveMessage struct {
|
|||||||
// It contains metadata about the BGP session (peer, ASN, host) along with
|
// It contains metadata about the BGP session (peer, ASN, host) along with
|
||||||
// the actual BGP update data including AS path, communities, announcements,
|
// the actual BGP update data including AS path, communities, announcements,
|
||||||
// and withdrawals.
|
// and withdrawals.
|
||||||
|
//
|
||||||
|
//nolint:tagliatelle // field names are the RIS Live feed's
|
||||||
type RISMessage struct {
|
type RISMessage struct {
|
||||||
Type string `json:"type"`
|
Type string `json:"type"`
|
||||||
Timestamp float64 `json:"timestamp"`
|
Timestamp float64 `json:"timestamp"`
|
||||||
@@ -90,6 +98,8 @@ type RISMessage struct {
|
|||||||
// RISAnnouncement represents a BGP route announcement within a RIS message.
|
// RISAnnouncement represents a BGP route announcement within a RIS message.
|
||||||
// It contains the next hop IP address and the list of prefixes being announced
|
// It contains the next hop IP address and the list of prefixes being announced
|
||||||
// via that next hop.
|
// via that next hop.
|
||||||
|
//
|
||||||
|
//nolint:tagliatelle // field names are the RIS Live feed's
|
||||||
type RISAnnouncement struct {
|
type RISAnnouncement struct {
|
||||||
NextHop string `json:"next_hop"`
|
NextHop string `json:"next_hop"`
|
||||||
Prefixes []string `json:"prefixes"`
|
Prefixes []string `json:"prefixes"`
|
||||||
|
|||||||
@@ -1,10 +1,12 @@
|
|||||||
package ristypes
|
package ristypes_test
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bufio"
|
"bufio"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"os"
|
"os"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
"git.eeqj.de/sneak/routewatch/internal/ristypes"
|
||||||
)
|
)
|
||||||
|
|
||||||
// messageExamplesPath is the captured RIS Live feed used as decode fixtures,
|
// messageExamplesPath is the captured RIS Live feed used as decode fixtures,
|
||||||
@@ -15,32 +17,43 @@ const messageExamplesPath = "../../docs/message-examples.json"
|
|||||||
// streamer does and checks that the fields no handler reads (Community, Raw)
|
// streamer does and checks that the fields no handler reads (Community, Raw)
|
||||||
// stay empty while the fields handlers use (Path, Announcements) still decode.
|
// stay empty while the fields handlers use (Path, Announcements) still decode.
|
||||||
func TestDecodeDropsCommunityAndRaw(t *testing.T) {
|
func TestDecodeDropsCommunityAndRaw(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
f, err := os.Open(messageExamplesPath)
|
f, err := os.Open(messageExamplesPath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("open fixtures: %v", err)
|
t.Fatalf("open fixtures: %v", err)
|
||||||
}
|
}
|
||||||
defer f.Close()
|
|
||||||
|
defer func() {
|
||||||
|
_ = f.Close()
|
||||||
|
}()
|
||||||
|
|
||||||
var messages, withPath, withAnnouncements int
|
var messages, withPath, withAnnouncements int
|
||||||
|
|
||||||
scanner := bufio.NewScanner(f)
|
scanner := bufio.NewScanner(f)
|
||||||
scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
||||||
|
|
||||||
for scanner.Scan() {
|
for scanner.Scan() {
|
||||||
line := scanner.Bytes()
|
line := scanner.Bytes()
|
||||||
if len(line) == 0 {
|
if len(line) == 0 {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
var wrapper RISLiveMessage
|
var wrapper ristypes.RISLiveMessage
|
||||||
if err := json.Unmarshal(line, &wrapper); err != nil {
|
|
||||||
|
err := json.Unmarshal(line, &wrapper)
|
||||||
|
if err != nil {
|
||||||
t.Fatalf("unmarshal message %d: %v", messages+1, err)
|
t.Fatalf("unmarshal message %d: %v", messages+1, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
messages++
|
messages++
|
||||||
|
|
||||||
msg := wrapper.Data
|
msg := wrapper.Data
|
||||||
if msg.Community != nil {
|
if msg.Community != nil {
|
||||||
t.Errorf("message %d: Community decoded, want empty: %v", messages, msg.Community)
|
t.Errorf("message %d: Community decoded, want empty: %v",
|
||||||
|
messages, msg.Community)
|
||||||
}
|
}
|
||||||
|
|
||||||
if msg.Raw != "" {
|
if msg.Raw != "" {
|
||||||
t.Errorf("message %d: Raw decoded, want empty", messages)
|
t.Errorf("message %d: Raw decoded, want empty", messages)
|
||||||
}
|
}
|
||||||
@@ -48,11 +61,14 @@ func TestDecodeDropsCommunityAndRaw(t *testing.T) {
|
|||||||
if len(msg.Path) > 0 {
|
if len(msg.Path) > 0 {
|
||||||
withPath++
|
withPath++
|
||||||
}
|
}
|
||||||
|
|
||||||
if len(msg.Announcements) > 0 {
|
if len(msg.Announcements) > 0 {
|
||||||
withAnnouncements++
|
withAnnouncements++
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if err := scanner.Err(); err != nil {
|
|
||||||
|
err = scanner.Err()
|
||||||
|
if err != nil {
|
||||||
t.Fatalf("scan fixtures: %v", err)
|
t.Fatalf("scan fixtures: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -62,6 +78,7 @@ func TestDecodeDropsCommunityAndRaw(t *testing.T) {
|
|||||||
// The fixtures include announcement messages; a used field must still decode,
|
// The fixtures include announcement messages; a used field must still decode,
|
||||||
// otherwise an empty Community/Raw would prove nothing.
|
// otherwise an empty Community/Raw would prove nothing.
|
||||||
if withPath == 0 || withAnnouncements == 0 {
|
if withPath == 0 || withAnnouncements == 0 {
|
||||||
t.Fatalf("used fields did not decode: withPath=%d withAnnouncements=%d", withPath, withAnnouncements)
|
t.Fatalf("used fields did not decode: withPath=%d withAnnouncements=%d",
|
||||||
|
withPath, withAnnouncements)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+269
-201
@@ -1,5 +1,6 @@
|
|||||||
// Package streamer implements an HTTP client that connects to the RIPE RIS Live streaming API,
|
// Package streamer implements an HTTP client that connects to the RIPE RIS Live
|
||||||
// parses BGP UPDATE messages from the JSON stream, and dispatches them to registered handlers.
|
// streaming API, parses BGP UPDATE messages from the JSON stream, and
|
||||||
|
// dispatches them to registered handlers.
|
||||||
package streamer
|
package streamer
|
||||||
|
|
||||||
import (
|
import (
|
||||||
@@ -7,6 +8,7 @@ import (
|
|||||||
"compress/gzip"
|
"compress/gzip"
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"math"
|
"math"
|
||||||
@@ -24,20 +26,20 @@ import (
|
|||||||
// countingReader wraps an io.Reader and counts bytes read
|
// countingReader wraps an io.Reader and counts bytes read
|
||||||
type countingReader struct {
|
type countingReader struct {
|
||||||
reader io.Reader
|
reader io.Reader
|
||||||
count int64
|
count atomic.Int64
|
||||||
}
|
}
|
||||||
|
|
||||||
// Read implements io.Reader and counts bytes
|
// Read implements io.Reader and counts bytes
|
||||||
func (c *countingReader) Read(p []byte) (int, error) {
|
func (c *countingReader) Read(p []byte) (int, error) {
|
||||||
n, err := c.reader.Read(p)
|
n, err := c.reader.Read(p)
|
||||||
atomic.AddInt64(&c.count, int64(n))
|
c.count.Add(int64(n))
|
||||||
|
|
||||||
return n, err
|
return n, err
|
||||||
}
|
}
|
||||||
|
|
||||||
// Count returns the total bytes read
|
// Count returns the total bytes read
|
||||||
func (c *countingReader) Count() int64 {
|
func (c *countingReader) Count() int64 {
|
||||||
return atomic.LoadInt64(&c.count)
|
return c.count.Load()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Configuration constants for the RIS Live streamer.
|
// Configuration constants for the RIS Live streamer.
|
||||||
@@ -58,11 +60,17 @@ const (
|
|||||||
backpressureSlope = 2.0 // Slope for linear drop probability increase
|
backpressureSlope = 2.0 // Slope for linear drop probability increase
|
||||||
)
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
errAlreadyRunning = errors.New("streamer already running")
|
||||||
|
errUnexpectedStatus = errors.New("unexpected status code")
|
||||||
|
)
|
||||||
|
|
||||||
// MessageHandler defines the interface for processing RIS messages.
|
// MessageHandler defines the interface for processing RIS messages.
|
||||||
// Implementations must specify which message types they want to receive,
|
// Implementations must specify which message types they want to receive,
|
||||||
// how to process messages, and their desired queue capacity.
|
// how to process messages, and their desired queue capacity.
|
||||||
type MessageHandler interface {
|
type MessageHandler interface {
|
||||||
// WantsMessage returns true if this handler wants to process messages of the given type.
|
// WantsMessage returns true if this handler wants to process messages of
|
||||||
|
// the given type.
|
||||||
WantsMessage(messageType string) bool
|
WantsMessage(messageType string) bool
|
||||||
|
|
||||||
// HandleMessage processes a RIS message. This method is called from a dedicated
|
// HandleMessage processes a RIS message. This method is called from a dedicated
|
||||||
@@ -113,7 +121,7 @@ type Streamer struct {
|
|||||||
cancel context.CancelFunc
|
cancel context.CancelFunc
|
||||||
running bool
|
running bool
|
||||||
metrics *metrics.Tracker
|
metrics *metrics.Tracker
|
||||||
totalDropped uint64 // Total dropped messages across all handlers
|
totalDropped atomic.Uint64 // Total dropped messages across all handlers
|
||||||
random *rand.Rand // Random number generator for backpressure drops
|
random *rand.Rand // Random number generator for backpressure drops
|
||||||
bgpPeers map[string]bool // Track active BGP peers by peer IP
|
bgpPeers map[string]bool // Track active BGP peers by peer IP
|
||||||
bgpPeersMu sync.RWMutex // Protects bgpPeers map
|
bgpPeersMu sync.RWMutex // Protects bgpPeers map
|
||||||
@@ -121,7 +129,8 @@ type Streamer struct {
|
|||||||
|
|
||||||
// New creates a new Streamer instance configured to connect to the RIS Live API.
|
// New creates a new Streamer instance configured to connect to the RIS Live API.
|
||||||
// The logger is used for structured logging of connection events and errors.
|
// The logger is used for structured logging of connection events and errors.
|
||||||
// The metrics tracker is used to record message counts, bytes received, and connection status.
|
// The metrics tracker is used to record message counts, bytes received, and
|
||||||
|
// connection status.
|
||||||
func New(logger *logger.Logger, metrics *metrics.Tracker) *Streamer {
|
func New(logger *logger.Logger, metrics *metrics.Tracker) *Streamer {
|
||||||
return &Streamer{
|
return &Streamer{
|
||||||
logger: logger,
|
logger: logger,
|
||||||
@@ -153,7 +162,8 @@ func (s *Streamer) RegisterHandler(handler MessageHandler) {
|
|||||||
handler: handler,
|
handler: handler,
|
||||||
queue: make(chan *ristypes.RISMessage, handler.QueueCapacity()),
|
queue: make(chan *ristypes.RISMessage, handler.QueueCapacity()),
|
||||||
metrics: handlerMetrics{
|
metrics: handlerMetrics{
|
||||||
minTime: time.Duration(math.MaxInt64), // Initialize to max so first value sets the floor
|
// Initialize to max so first value sets the floor
|
||||||
|
minTime: time.Duration(math.MaxInt64),
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -171,6 +181,7 @@ func (s *Streamer) RegisterHandler(handler MessageHandler) {
|
|||||||
func (s *Streamer) RegisterRawHandler(handler RawMessageHandler) {
|
func (s *Streamer) RegisterRawHandler(handler RawMessageHandler) {
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
defer s.mu.Unlock()
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
s.rawHandler = handler
|
s.rawHandler = handler
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -183,7 +194,7 @@ func (s *Streamer) Start() error {
|
|||||||
defer s.mu.Unlock()
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
if s.running {
|
if s.running {
|
||||||
return fmt.Errorf("streamer already running")
|
return errAlreadyRunning
|
||||||
}
|
}
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
@@ -216,44 +227,21 @@ func (s *Streamer) Stop() {
|
|||||||
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
s.cancel()
|
s.cancel()
|
||||||
s.cancel = nil
|
s.cancel = nil
|
||||||
// Close all handler queues to signal workers to stop
|
// Close all handler queues to signal workers to stop
|
||||||
for _, info := range s.handlers {
|
for _, info := range s.handlers {
|
||||||
close(info.queue)
|
close(info.queue)
|
||||||
}
|
}
|
||||||
|
|
||||||
s.running = false
|
s.running = false
|
||||||
s.mu.Unlock()
|
s.mu.Unlock()
|
||||||
s.metrics.SetConnected(false)
|
s.metrics.SetConnected(false)
|
||||||
}
|
}
|
||||||
|
|
||||||
// runHandlerWorker processes messages for a specific handler
|
// IsRunning reports whether the streamer is currently connected and processing
|
||||||
func (s *Streamer) runHandlerWorker(info *handlerInfo) {
|
// messages. This is safe to call concurrently from multiple goroutines.
|
||||||
for msg := range info.queue {
|
|
||||||
start := time.Now()
|
|
||||||
info.handler.HandleMessage(msg)
|
|
||||||
elapsed := time.Since(start)
|
|
||||||
|
|
||||||
// Update metrics
|
|
||||||
info.metrics.mu.Lock()
|
|
||||||
info.metrics.processedCount++
|
|
||||||
info.metrics.totalTime += elapsed
|
|
||||||
|
|
||||||
// Update min time
|
|
||||||
if elapsed < info.metrics.minTime {
|
|
||||||
info.metrics.minTime = elapsed
|
|
||||||
}
|
|
||||||
|
|
||||||
// Update max time
|
|
||||||
if elapsed > info.metrics.maxTime {
|
|
||||||
info.metrics.maxTime = elapsed
|
|
||||||
}
|
|
||||||
info.metrics.mu.Unlock()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// IsRunning reports whether the streamer is currently connected and processing messages.
|
|
||||||
// This is safe to call concurrently from multiple goroutines.
|
|
||||||
func (s *Streamer) IsRunning() bool {
|
func (s *Streamer) IsRunning() bool {
|
||||||
s.mu.RLock()
|
s.mu.RLock()
|
||||||
defer s.mu.RUnlock()
|
defer s.mu.RUnlock()
|
||||||
@@ -314,12 +302,15 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
|
|||||||
// Calculate average time
|
// Calculate average time
|
||||||
if info.metrics.processedCount > 0 {
|
if info.metrics.processedCount > 0 {
|
||||||
processedCount := info.metrics.processedCount
|
processedCount := info.metrics.processedCount
|
||||||
|
|
||||||
const maxInt64 = 1<<63 - 1
|
const maxInt64 = 1<<63 - 1
|
||||||
|
|
||||||
if processedCount > maxInt64 {
|
if processedCount > maxInt64 {
|
||||||
processedCount = maxInt64
|
processedCount = maxInt64
|
||||||
}
|
}
|
||||||
//nolint:gosec // processedCount is explicitly bounded above
|
|
||||||
hs.AvgProcessTime = info.metrics.totalTime / time.Duration(processedCount)
|
hs.AvgProcessTime = info.metrics.totalTime /
|
||||||
|
time.Duration(processedCount)
|
||||||
}
|
}
|
||||||
|
|
||||||
info.metrics.mu.Unlock()
|
info.metrics.mu.Unlock()
|
||||||
@@ -330,11 +321,38 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
|
|||||||
return stats
|
return stats
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetDroppedMessages returns the total number of messages dropped across all handlers
|
// GetDroppedMessages returns the total number of messages dropped across all
|
||||||
// due to queue overflow or backpressure. This counter is monotonically increasing
|
// handlers due to queue overflow or backpressure. This counter is monotonically
|
||||||
// and is safe to call concurrently.
|
// increasing and is safe to call concurrently.
|
||||||
func (s *Streamer) GetDroppedMessages() uint64 {
|
func (s *Streamer) GetDroppedMessages() uint64 {
|
||||||
return atomic.LoadUint64(&s.totalDropped)
|
return s.totalDropped.Load()
|
||||||
|
}
|
||||||
|
|
||||||
|
// runHandlerWorker processes messages for a specific handler
|
||||||
|
func (s *Streamer) runHandlerWorker(info *handlerInfo) {
|
||||||
|
for msg := range info.queue {
|
||||||
|
start := time.Now()
|
||||||
|
|
||||||
|
info.handler.HandleMessage(msg)
|
||||||
|
|
||||||
|
elapsed := time.Since(start)
|
||||||
|
|
||||||
|
// Update metrics
|
||||||
|
info.metrics.mu.Lock()
|
||||||
|
info.metrics.processedCount++
|
||||||
|
info.metrics.totalTime += elapsed
|
||||||
|
|
||||||
|
// Update min time
|
||||||
|
if elapsed < info.metrics.minTime {
|
||||||
|
info.metrics.minTime = elapsed
|
||||||
|
}
|
||||||
|
|
||||||
|
// Update max time
|
||||||
|
if elapsed > info.metrics.maxTime {
|
||||||
|
info.metrics.maxTime = elapsed
|
||||||
|
}
|
||||||
|
info.metrics.mu.Unlock()
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// logMetrics logs the current streaming statistics
|
// logMetrics logs the current streaming statistics
|
||||||
@@ -343,7 +361,8 @@ func (s *Streamer) logMetrics() {
|
|||||||
uptime := time.Since(metrics.ConnectedSince)
|
uptime := time.Since(metrics.ConnectedSince)
|
||||||
|
|
||||||
const bitsPerMegabit = 1000000
|
const bitsPerMegabit = 1000000
|
||||||
totalDropped := atomic.LoadUint64(&s.totalDropped)
|
|
||||||
|
totalDropped := s.totalDropped.Load()
|
||||||
|
|
||||||
s.logger.Info(
|
s.logger.Info(
|
||||||
"Stream statistics",
|
"Stream statistics",
|
||||||
@@ -369,16 +388,19 @@ func (s *Streamer) logMetrics() {
|
|||||||
|
|
||||||
// Log per-handler statistics
|
// Log per-handler statistics
|
||||||
s.mu.RLock()
|
s.mu.RLock()
|
||||||
|
|
||||||
for i, info := range s.handlers {
|
for i, info := range s.handlers {
|
||||||
info.metrics.mu.Lock()
|
info.metrics.mu.Lock()
|
||||||
if info.metrics.processedCount > 0 {
|
if info.metrics.processedCount > 0 {
|
||||||
// Safe conversion: processedCount is bounded by maxInt64
|
// Safe conversion: processedCount is bounded by maxInt64
|
||||||
processedCount := info.metrics.processedCount
|
processedCount := info.metrics.processedCount
|
||||||
|
|
||||||
const maxInt64 = 1<<63 - 1
|
const maxInt64 = 1<<63 - 1
|
||||||
|
|
||||||
if processedCount > maxInt64 {
|
if processedCount > maxInt64 {
|
||||||
processedCount = maxInt64
|
processedCount = maxInt64
|
||||||
}
|
}
|
||||||
//nolint:gosec // processedCount is explicitly bounded above
|
|
||||||
avgTime := info.metrics.totalTime / time.Duration(processedCount)
|
avgTime := info.metrics.totalTime / time.Duration(processedCount)
|
||||||
s.logger.Info(
|
s.logger.Info(
|
||||||
"Handler statistics",
|
"Handler statistics",
|
||||||
@@ -395,6 +417,7 @@ func (s *Streamer) logMetrics() {
|
|||||||
}
|
}
|
||||||
info.metrics.mu.Unlock()
|
info.metrics.mu.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
s.mu.RUnlock()
|
s.mu.RUnlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -403,7 +426,8 @@ func (s *Streamer) updateMetrics(messageBytes int) {
|
|||||||
s.metrics.RecordMessage(int64(messageBytes))
|
s.metrics.RecordMessage(int64(messageBytes))
|
||||||
}
|
}
|
||||||
|
|
||||||
// streamWithReconnect handles streaming with automatic reconnection and exponential backoff
|
// streamWithReconnect handles streaming with automatic reconnection and
|
||||||
|
// exponential backoff
|
||||||
func (s *Streamer) streamWithReconnect(ctx context.Context) {
|
func (s *Streamer) streamWithReconnect(ctx context.Context) {
|
||||||
backoffDelay := minBackoffDelay
|
backoffDelay := minBackoffDelay
|
||||||
consecutiveFailures := 0
|
consecutiveFailures := 0
|
||||||
@@ -444,6 +468,7 @@ func (s *Streamer) streamWithReconnect(ctx context.Context) {
|
|||||||
if streamDuration > 30*time.Second {
|
if streamDuration > 30*time.Second {
|
||||||
s.logger.Info("Resetting backoff delay due to successful connection",
|
s.logger.Info("Resetting backoff delay due to successful connection",
|
||||||
"stream_duration", streamDuration)
|
"stream_duration", streamDuration)
|
||||||
|
|
||||||
backoffDelay = minBackoffDelay
|
backoffDelay = minBackoffDelay
|
||||||
consecutiveFailures = 0
|
consecutiveFailures = 0
|
||||||
} else {
|
} else {
|
||||||
@@ -477,7 +502,7 @@ func (s *Streamer) stream(ctx context.Context) error {
|
|||||||
connCtx, connCancel := context.WithCancel(ctx)
|
connCtx, connCancel := context.WithCancel(ctx)
|
||||||
defer connCancel()
|
defer connCancel()
|
||||||
|
|
||||||
req, err := http.NewRequestWithContext(ctx, "GET", s.url, nil)
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.url, nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to create request: %w", err)
|
return fmt.Errorf("failed to create request: %w", err)
|
||||||
}
|
}
|
||||||
@@ -490,13 +515,14 @@ func (s *Streamer) stream(ctx context.Context) error {
|
|||||||
return fmt.Errorf("failed to connect to RIS Live: %w", err)
|
return fmt.Errorf("failed to connect to RIS Live: %w", err)
|
||||||
}
|
}
|
||||||
defer func() {
|
defer func() {
|
||||||
if err := resp.Body.Close(); err != nil {
|
err := resp.Body.Close()
|
||||||
|
if err != nil {
|
||||||
s.logger.Error("Failed to close response body", "error", err)
|
s.logger.Error("Failed to close response body", "error", err)
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
if resp.StatusCode != http.StatusOK {
|
if resp.StatusCode != http.StatusOK {
|
||||||
return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
|
return fmt.Errorf("%w: %d", errUnexpectedStatus, resp.StatusCode)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Wrap body with counting reader to track actual wire bytes
|
// Wrap body with counting reader to track actual wire bytes
|
||||||
@@ -510,7 +536,9 @@ func (s *Streamer) stream(ctx context.Context) error {
|
|||||||
return fmt.Errorf("failed to create gzip reader: %w", err)
|
return fmt.Errorf("failed to create gzip reader: %w", err)
|
||||||
}
|
}
|
||||||
defer func() { _ = gzReader.Close() }()
|
defer func() { _ = gzReader.Close() }()
|
||||||
|
|
||||||
reader = gzReader
|
reader = gzReader
|
||||||
|
|
||||||
s.logger.Info("Connected to RIS Live stream", "compression", "gzip")
|
s.logger.Info("Connected to RIS Live stream", "compression", "gzip")
|
||||||
} else {
|
} else {
|
||||||
s.logger.Info("Connected to RIS Live stream", "compression", "none")
|
s.logger.Info("Connected to RIS Live stream", "compression", "none")
|
||||||
@@ -518,9 +546,6 @@ func (s *Streamer) stream(ctx context.Context) error {
|
|||||||
|
|
||||||
s.metrics.SetConnected(true)
|
s.metrics.SetConnected(true)
|
||||||
|
|
||||||
// Track wire bytes for metrics updates
|
|
||||||
var lastWireBytes int64
|
|
||||||
|
|
||||||
// Start metrics logging goroutine
|
// Start metrics logging goroutine
|
||||||
metricsTicker := time.NewTicker(metricsLogInterval)
|
metricsTicker := time.NewTicker(metricsLogInterval)
|
||||||
defer metricsTicker.Stop()
|
defer metricsTicker.Stop()
|
||||||
@@ -537,25 +562,41 @@ func (s *Streamer) stream(ctx context.Context) error {
|
|||||||
}()
|
}()
|
||||||
|
|
||||||
// Wire byte update ticker - update metrics with actual wire bytes periodically
|
// Wire byte update ticker - update metrics with actual wire bytes periodically
|
||||||
|
go s.recordWireBytes(connCtx, wireCounter)
|
||||||
|
|
||||||
|
return s.readStream(ctx, reader)
|
||||||
|
}
|
||||||
|
|
||||||
|
// recordWireBytes adds the bytes wireCounter has counted to the metrics once a
|
||||||
|
// second until ctx is cancelled.
|
||||||
|
func (s *Streamer) recordWireBytes(
|
||||||
|
ctx context.Context, wireCounter *countingReader,
|
||||||
|
) {
|
||||||
wireUpdateTicker := time.NewTicker(time.Second)
|
wireUpdateTicker := time.NewTicker(time.Second)
|
||||||
defer wireUpdateTicker.Stop()
|
defer wireUpdateTicker.Stop()
|
||||||
|
|
||||||
go func() {
|
var lastWireBytes int64
|
||||||
for {
|
|
||||||
select {
|
|
||||||
case <-wireUpdateTicker.C:
|
|
||||||
currentBytes := wireCounter.Count()
|
|
||||||
delta := currentBytes - lastWireBytes
|
|
||||||
if delta > 0 {
|
|
||||||
s.metrics.RecordWireBytes(delta)
|
|
||||||
lastWireBytes = currentBytes
|
|
||||||
}
|
|
||||||
case <-connCtx.Done():
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-wireUpdateTicker.C:
|
||||||
|
currentBytes := wireCounter.Count()
|
||||||
|
|
||||||
|
delta := currentBytes - lastWireBytes
|
||||||
|
if delta > 0 {
|
||||||
|
s.metrics.RecordWireBytes(delta)
|
||||||
|
|
||||||
|
lastWireBytes = currentBytes
|
||||||
|
}
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// readStream hands each line of the stream to processLine until the stream
|
||||||
|
// ends, a line fails, or ctx is cancelled.
|
||||||
|
func (s *Streamer) readStream(ctx context.Context, reader io.Reader) error {
|
||||||
scanner := bufio.NewScanner(reader)
|
scanner := bufio.NewScanner(reader)
|
||||||
|
|
||||||
for scanner.Scan() {
|
for scanner.Scan() {
|
||||||
@@ -572,149 +613,176 @@ func (s *Streamer) stream(ctx context.Context) error {
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
// Update metrics with decompressed message size
|
err := s.processLine(ctx, line)
|
||||||
s.updateMetrics(len(line))
|
if err != nil {
|
||||||
|
return err
|
||||||
// Call raw handler if registered
|
|
||||||
s.mu.RLock()
|
|
||||||
rawHandler := s.rawHandler
|
|
||||||
s.mu.RUnlock()
|
|
||||||
|
|
||||||
if rawHandler != nil {
|
|
||||||
// Call raw handler synchronously to preserve order
|
|
||||||
rawHandler(string(line))
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Parse the message first
|
|
||||||
var wrapper ristypes.RISLiveMessage
|
|
||||||
if err := json.Unmarshal(line, &wrapper); err != nil {
|
|
||||||
// Log the error and return to trigger reconnection
|
|
||||||
s.logger.Error("Failed to parse JSON",
|
|
||||||
"error", err,
|
|
||||||
"line", string(line),
|
|
||||||
"line_length", len(line))
|
|
||||||
|
|
||||||
return fmt.Errorf("JSON parse error: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Check if it's a ris_message wrapper
|
|
||||||
if wrapper.Type != "ris_message" {
|
|
||||||
s.logger.Error("Unexpected wrapper type",
|
|
||||||
"type", wrapper.Type,
|
|
||||||
"line", string(line),
|
|
||||||
)
|
|
||||||
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
// Get the actual message
|
|
||||||
msg := wrapper.Data
|
|
||||||
|
|
||||||
// Parse the timestamp
|
|
||||||
msg.ParsedTimestamp = time.Unix(int64(msg.Timestamp), 0).UTC()
|
|
||||||
|
|
||||||
// Process based on message type
|
|
||||||
switch msg.Type {
|
|
||||||
case "UPDATE":
|
|
||||||
// Process BGP UPDATE messages
|
|
||||||
// Will be dispatched to handlers
|
|
||||||
case "RIS_PEER_STATE":
|
|
||||||
// RIS peer state messages - silently ignore
|
|
||||||
continue
|
|
||||||
case "KEEPALIVE":
|
|
||||||
// BGP keepalive messages - silently process
|
|
||||||
continue
|
|
||||||
case "OPEN":
|
|
||||||
// BGP open messages - track peer as active
|
|
||||||
s.bgpPeersMu.Lock()
|
|
||||||
s.bgpPeers[msg.Peer] = true
|
|
||||||
peerCount := len(s.bgpPeers)
|
|
||||||
s.bgpPeersMu.Unlock()
|
|
||||||
s.metrics.SetBGPPeerCount(peerCount)
|
|
||||||
|
|
||||||
s.logger.Info("BGP session opened",
|
|
||||||
"peer", msg.Peer,
|
|
||||||
"peer_asn", msg.PeerASN,
|
|
||||||
"total_peers", peerCount,
|
|
||||||
)
|
|
||||||
|
|
||||||
continue
|
|
||||||
case "NOTIFICATION":
|
|
||||||
// BGP notification messages (session closed)
|
|
||||||
s.bgpPeersMu.Lock()
|
|
||||||
delete(s.bgpPeers, msg.Peer)
|
|
||||||
peerCount := len(s.bgpPeers)
|
|
||||||
s.bgpPeersMu.Unlock()
|
|
||||||
s.metrics.SetBGPPeerCount(peerCount)
|
|
||||||
|
|
||||||
s.logger.Warn("BGP notification",
|
|
||||||
"peer", msg.Peer,
|
|
||||||
"peer_asn", msg.PeerASN,
|
|
||||||
"total_peers", peerCount,
|
|
||||||
)
|
|
||||||
|
|
||||||
continue
|
|
||||||
case "STATE":
|
|
||||||
// Peer state changes - silently ignore
|
|
||||||
continue
|
|
||||||
default:
|
|
||||||
s.logger.Warn("Unknown message type, skipping",
|
|
||||||
"type", msg.Type,
|
|
||||||
)
|
|
||||||
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
// Dispatch to interested handlers. Stop cancels ctx and closes the
|
|
||||||
// queues under the write lock, so if ctx is cancelled here, under the
|
|
||||||
// read lock, the queues are closed and must not be sent to.
|
|
||||||
s.mu.RLock()
|
|
||||||
if ctx.Err() != nil {
|
|
||||||
s.mu.RUnlock()
|
|
||||||
|
|
||||||
return ctx.Err()
|
|
||||||
}
|
|
||||||
for _, info := range s.handlers {
|
|
||||||
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 {
|
|
||||||
case info.queue <- &msg:
|
|
||||||
// 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:
|
|
||||||
// Queue is full, drop the message
|
|
||||||
atomic.AddUint64(&info.metrics.droppedCount, 1)
|
|
||||||
atomic.AddUint64(&s.totalDropped, 1)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
s.mu.RUnlock()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := scanner.Err(); err != nil {
|
err := scanner.Err()
|
||||||
|
if err != nil {
|
||||||
return fmt.Errorf("scanner error: %w", err)
|
return fmt.Errorf("scanner error: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// shouldDropForBackpressure determines if a message should be dropped based on queue utilization
|
// processLine passes one line of the stream to the raw handler, parses it,
|
||||||
|
// and dispatches an UPDATE message to the handlers. An error ends the
|
||||||
|
// connection.
|
||||||
|
func (s *Streamer) processLine(ctx context.Context, line []byte) error {
|
||||||
|
// Update metrics with decompressed message size
|
||||||
|
s.updateMetrics(len(line))
|
||||||
|
|
||||||
|
// Call raw handler if registered
|
||||||
|
s.mu.RLock()
|
||||||
|
rawHandler := s.rawHandler
|
||||||
|
s.mu.RUnlock()
|
||||||
|
|
||||||
|
if rawHandler != nil {
|
||||||
|
// Call raw handler synchronously to preserve order
|
||||||
|
rawHandler(string(line))
|
||||||
|
}
|
||||||
|
|
||||||
|
// Parse the message first
|
||||||
|
var wrapper ristypes.RISLiveMessage
|
||||||
|
|
||||||
|
err := json.Unmarshal(line, &wrapper)
|
||||||
|
if err != nil {
|
||||||
|
// Log the error and return to trigger reconnection
|
||||||
|
s.logger.Error("Failed to parse JSON",
|
||||||
|
"error", err,
|
||||||
|
"line", string(line),
|
||||||
|
"line_length", len(line))
|
||||||
|
|
||||||
|
return fmt.Errorf("JSON parse error: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check if it's a ris_message wrapper
|
||||||
|
if wrapper.Type != "ris_message" {
|
||||||
|
s.logger.Error("Unexpected wrapper type",
|
||||||
|
"type", wrapper.Type,
|
||||||
|
"line", string(line),
|
||||||
|
)
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Get the actual message
|
||||||
|
msg := wrapper.Data
|
||||||
|
|
||||||
|
// Parse the timestamp
|
||||||
|
msg.ParsedTimestamp = time.Unix(int64(msg.Timestamp), 0).UTC()
|
||||||
|
|
||||||
|
// Process based on message type
|
||||||
|
switch msg.Type {
|
||||||
|
case "UPDATE":
|
||||||
|
// Process BGP UPDATE messages
|
||||||
|
// Will be dispatched to handlers
|
||||||
|
case "RIS_PEER_STATE":
|
||||||
|
// RIS peer state messages - silently ignore
|
||||||
|
return nil
|
||||||
|
case "KEEPALIVE":
|
||||||
|
// BGP keepalive messages - silently process
|
||||||
|
return nil
|
||||||
|
case "OPEN":
|
||||||
|
// BGP open messages - track peer as active
|
||||||
|
s.bgpPeersMu.Lock()
|
||||||
|
s.bgpPeers[msg.Peer] = true
|
||||||
|
peerCount := len(s.bgpPeers)
|
||||||
|
s.bgpPeersMu.Unlock()
|
||||||
|
s.metrics.SetBGPPeerCount(peerCount)
|
||||||
|
|
||||||
|
s.logger.Info("BGP session opened",
|
||||||
|
"peer", msg.Peer,
|
||||||
|
"peer_asn", msg.PeerASN,
|
||||||
|
"total_peers", peerCount,
|
||||||
|
)
|
||||||
|
|
||||||
|
return nil
|
||||||
|
case "NOTIFICATION":
|
||||||
|
// BGP notification messages (session closed)
|
||||||
|
s.bgpPeersMu.Lock()
|
||||||
|
delete(s.bgpPeers, msg.Peer)
|
||||||
|
peerCount := len(s.bgpPeers)
|
||||||
|
s.bgpPeersMu.Unlock()
|
||||||
|
s.metrics.SetBGPPeerCount(peerCount)
|
||||||
|
|
||||||
|
s.logger.Warn("BGP notification",
|
||||||
|
"peer", msg.Peer,
|
||||||
|
"peer_asn", msg.PeerASN,
|
||||||
|
"total_peers", peerCount,
|
||||||
|
)
|
||||||
|
|
||||||
|
return nil
|
||||||
|
case "STATE":
|
||||||
|
// Peer state changes - silently ignore
|
||||||
|
return nil
|
||||||
|
default:
|
||||||
|
s.logger.Warn("Unknown message type, skipping",
|
||||||
|
"type", msg.Type,
|
||||||
|
)
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
return s.dispatchMessage(ctx, &msg)
|
||||||
|
}
|
||||||
|
|
||||||
|
// dispatchMessage queues msg for each handler that wants it, dropping it for
|
||||||
|
// a handler whose queue is full or under backpressure.
|
||||||
|
func (s *Streamer) dispatchMessage(
|
||||||
|
ctx context.Context, msg *ristypes.RISMessage,
|
||||||
|
) error {
|
||||||
|
// Stop cancels ctx and closes the queues under the write lock, so if ctx
|
||||||
|
// is cancelled here, under the read lock, the queues are closed and must
|
||||||
|
// not be sent to.
|
||||||
|
s.mu.RLock()
|
||||||
|
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
s.mu.RUnlock()
|
||||||
|
|
||||||
|
return ctx.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, info := range s.handlers {
|
||||||
|
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)
|
||||||
|
s.totalDropped.Add(1)
|
||||||
|
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
// Try to queue the message
|
||||||
|
select {
|
||||||
|
case info.queue <- msg:
|
||||||
|
// 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:
|
||||||
|
// Queue is full, drop the message
|
||||||
|
atomic.AddUint64(&info.metrics.droppedCount, 1)
|
||||||
|
s.totalDropped.Add(1)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
s.mu.RUnlock()
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// shouldDropForBackpressure determines if a message should be dropped based on
|
||||||
|
// queue utilization
|
||||||
func (s *Streamer) shouldDropForBackpressure(info *handlerInfo) bool {
|
func (s *Streamer) shouldDropForBackpressure(info *handlerInfo) bool {
|
||||||
// Calculate queue utilization
|
// Calculate queue utilization
|
||||||
queueLen := len(info.queue)
|
queueLen := len(info.queue)
|
||||||
|
|||||||
@@ -16,6 +16,8 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func TestNewStreamer(t *testing.T) {
|
func TestNewStreamer(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
logger := logger.New()
|
logger := logger.New()
|
||||||
metricsTracker := metrics.New()
|
metricsTracker := metrics.New()
|
||||||
s := New(logger, metricsTracker)
|
s := New(logger, metricsTracker)
|
||||||
@@ -46,10 +48,13 @@ func TestNewStreamer(t *testing.T) {
|
|||||||
// goroutine count returns to its starting value. Each connection starts two
|
// goroutine count returns to its starting value. Each connection starts two
|
||||||
// ticker goroutines; before the fix they lived until the streamer's lifetime
|
// ticker goroutines; before the fix they lived until the streamer's lifetime
|
||||||
// context was cancelled, so every reconnect leaked two.
|
// context was cancelled, so every reconnect leaked two.
|
||||||
|
//
|
||||||
|
//nolint:paralleltest // counts goroutines, so no other test may run alongside
|
||||||
func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) {
|
func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) {
|
||||||
// The handler returns immediately, so the response body is empty and each
|
// The handler returns immediately, so the response body is empty and each
|
||||||
// stream call ends at once, standing in for a dropped connection.
|
// stream call ends at once, standing in for a dropped connection.
|
||||||
srv := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) {}))
|
srv := httptest.NewServer(http.HandlerFunc(
|
||||||
|
func(_ http.ResponseWriter, _ *http.Request) {}))
|
||||||
defer srv.Close()
|
defer srv.Close()
|
||||||
|
|
||||||
s := New(logger.New(), metrics.New())
|
s := New(logger.New(), metrics.New())
|
||||||
@@ -57,19 +62,23 @@ func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) {
|
|||||||
|
|
||||||
// One warm-up connection so any persistent HTTP transport goroutine exists
|
// One warm-up connection so any persistent HTTP transport goroutine exists
|
||||||
// before we take the baseline.
|
// before we take the baseline.
|
||||||
if err := s.stream(context.Background()); err != nil {
|
err := s.stream(context.Background())
|
||||||
|
if err != nil {
|
||||||
t.Fatalf("warm-up stream returned error: %v", err)
|
t.Fatalf("warm-up stream returned error: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
s.client.CloseIdleConnections()
|
s.client.CloseIdleConnections()
|
||||||
|
|
||||||
baseline := settledGoroutineCount()
|
baseline := settledGoroutineCount()
|
||||||
|
|
||||||
const reconnects = 20
|
const reconnects = 20
|
||||||
for range reconnects {
|
for range reconnects {
|
||||||
if err := s.stream(context.Background()); err != nil {
|
err := s.stream(context.Background())
|
||||||
|
if err != nil {
|
||||||
t.Fatalf("stream returned error: %v", err)
|
t.Fatalf("stream returned error: %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
s.client.CloseIdleConnections()
|
s.client.CloseIdleConnections()
|
||||||
|
|
||||||
if !waitForGoroutines(baseline) {
|
if !waitForGoroutines(baseline) {
|
||||||
@@ -81,9 +90,13 @@ func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) {
|
|||||||
// updateHandler wants UPDATE messages and does nothing with them.
|
// updateHandler wants UPDATE messages and does nothing with them.
|
||||||
type updateHandler struct{}
|
type updateHandler struct{}
|
||||||
|
|
||||||
func (updateHandler) WantsMessage(messageType string) bool { return messageType == "UPDATE" }
|
func (updateHandler) WantsMessage(messageType string) bool {
|
||||||
func (updateHandler) HandleMessage(*ristypes.RISMessage) {}
|
return messageType == "UPDATE"
|
||||||
func (updateHandler) QueueCapacity() int { return 10 }
|
}
|
||||||
|
|
||||||
|
func (updateHandler) HandleMessage(*ristypes.RISMessage) {}
|
||||||
|
|
||||||
|
func (updateHandler) QueueCapacity() int { return 10 }
|
||||||
|
|
||||||
// TestStopBeforeMessageReachesQueues stops the streamer after the read loop
|
// TestStopBeforeMessageReachesQueues stops the streamer after the read loop
|
||||||
// has checked for cancellation but before it hands the message to the handler
|
// has checked for cancellation but before it hands the message to the handler
|
||||||
@@ -91,11 +104,15 @@ func (updateHandler) QueueCapacity() int { return 10 }
|
|||||||
// used to end in "send on closed channel". The raw handler runs in that gap on
|
// used to end in "send on closed channel". The raw handler runs in that gap on
|
||||||
// the read loop itself, so calling Stop from it hits the gap every time.
|
// the read loop itself, so calling Stop from it hits the gap every time.
|
||||||
func TestStopBeforeMessageReachesQueues(t *testing.T) {
|
func TestStopBeforeMessageReachesQueues(t *testing.T) {
|
||||||
const line = `{"type":"ris_message","data":{"type":"UPDATE","peer":"192.0.2.1",` +
|
t.Parallel()
|
||||||
`"peer_asn":"64496","timestamp":1700000000}}` + "\n"
|
|
||||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
const line = `{"type":"ris_message","data":{"type":"UPDATE",` +
|
||||||
_, _ = io.WriteString(w, line)
|
`"peer":"192.0.2.1","peer_asn":"64496","timestamp":1700000000}}` + "\n"
|
||||||
}))
|
|
||||||
|
srv := httptest.NewServer(http.HandlerFunc(
|
||||||
|
func(w http.ResponseWriter, _ *http.Request) {
|
||||||
|
_, _ = io.WriteString(w, line)
|
||||||
|
}))
|
||||||
defer srv.Close()
|
defer srv.Close()
|
||||||
|
|
||||||
s := New(logger.New(), metrics.New())
|
s := New(logger.New(), metrics.New())
|
||||||
@@ -109,7 +126,8 @@ func TestStopBeforeMessageReachesQueues(t *testing.T) {
|
|||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
s.cancel = cancel
|
s.cancel = cancel
|
||||||
|
|
||||||
if err := s.stream(ctx); !errors.Is(err, context.Canceled) {
|
err := s.stream(ctx)
|
||||||
|
if !errors.Is(err, context.Canceled) {
|
||||||
t.Fatalf("stream returned %v, want %v", err, context.Canceled)
|
t.Fatalf("stream returned %v, want %v", err, context.Canceled)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -121,12 +139,15 @@ func TestStopBeforeMessageReachesQueues(t *testing.T) {
|
|||||||
// current count.
|
// current count.
|
||||||
func settledGoroutineCount() int {
|
func settledGoroutineCount() int {
|
||||||
prev := runtime.NumGoroutine()
|
prev := runtime.NumGoroutine()
|
||||||
|
|
||||||
for range 20 {
|
for range 20 {
|
||||||
time.Sleep(10 * time.Millisecond)
|
time.Sleep(10 * time.Millisecond)
|
||||||
|
|
||||||
cur := runtime.NumGoroutine()
|
cur := runtime.NumGoroutine()
|
||||||
if cur == prev {
|
if cur == prev {
|
||||||
return cur
|
return cur
|
||||||
}
|
}
|
||||||
|
|
||||||
prev = cur
|
prev = cur
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -139,6 +160,7 @@ func waitForGoroutines(target int) bool {
|
|||||||
if runtime.NumGoroutine() <= target {
|
if runtime.NumGoroutine() <= target {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
time.Sleep(10 * time.Millisecond)
|
time.Sleep(10 * time.Millisecond)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -60,6 +60,7 @@ func timeSince(t time.Time) string {
|
|||||||
if duration < time.Minute {
|
if duration < time.Minute {
|
||||||
return "just now"
|
return "just now"
|
||||||
}
|
}
|
||||||
|
|
||||||
if duration < time.Hour {
|
if duration < time.Hour {
|
||||||
minutes := int(duration.Minutes())
|
minutes := int(duration.Minutes())
|
||||||
if minutes == 1 {
|
if minutes == 1 {
|
||||||
@@ -68,6 +69,7 @@ func timeSince(t time.Time) string {
|
|||||||
|
|
||||||
return duration.Truncate(time.Minute).String() + " ago"
|
return duration.Truncate(time.Minute).String() + " ago"
|
||||||
}
|
}
|
||||||
|
|
||||||
if duration < hoursPerDay*time.Hour {
|
if duration < hoursPerDay*time.Hour {
|
||||||
hours := int(duration.Hours())
|
hours := int(duration.Hours())
|
||||||
if hours == 1 {
|
if hours == 1 {
|
||||||
@@ -76,10 +78,12 @@ func timeSince(t time.Time) string {
|
|||||||
|
|
||||||
return duration.Truncate(time.Hour).String() + " ago"
|
return duration.Truncate(time.Hour).String() + " ago"
|
||||||
}
|
}
|
||||||
|
|
||||||
days := int(duration.Hours() / hoursPerDay)
|
days := int(duration.Hours() / hoursPerDay)
|
||||||
if days == 1 {
|
if days == 1 {
|
||||||
return "1 day ago"
|
return "1 day ago"
|
||||||
}
|
}
|
||||||
|
|
||||||
if days < daysPerMonth {
|
if days < daysPerMonth {
|
||||||
return duration.Truncate(hoursPerDay*time.Hour).String() + " ago"
|
return duration.Truncate(hoursPerDay*time.Hour).String() + " ago"
|
||||||
}
|
}
|
||||||
@@ -118,7 +122,7 @@ func initTemplates() {
|
|||||||
"appLicense": func() string { return version.License },
|
"appLicense": func() string { return version.License },
|
||||||
"appRepoURL": func() string { return version.RepoURL },
|
"appRepoURL": func() string { return version.RepoURL },
|
||||||
"appGitRevision": func() string { return version.GitRevisionShort },
|
"appGitRevision": func() string { return version.GitRevisionShort },
|
||||||
"appGitCommitURL": func() string { return version.CommitURL() },
|
"appGitCommitURL": version.CommitURL,
|
||||||
}
|
}
|
||||||
|
|
||||||
// Parse index template
|
// Parse index template
|
||||||
@@ -134,19 +138,22 @@ func initTemplates() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Parse AS detail template
|
// Parse AS detail template
|
||||||
defaultTemplates.ASDetail, err = template.New("asDetail").Funcs(funcs).Parse(asDetailHTML)
|
defaultTemplates.ASDetail, err = template.New("asDetail").
|
||||||
|
Funcs(funcs).Parse(asDetailHTML)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
panic("failed to parse AS detail template: " + err.Error())
|
panic("failed to parse AS detail template: " + err.Error())
|
||||||
}
|
}
|
||||||
|
|
||||||
// Parse prefix detail template
|
// Parse prefix detail template
|
||||||
defaultTemplates.PrefixDetail, err = template.New("prefixDetail").Funcs(funcs).Parse(prefixDetailHTML)
|
defaultTemplates.PrefixDetail, err = template.New("prefixDetail").
|
||||||
|
Funcs(funcs).Parse(prefixDetailHTML)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
panic("failed to parse prefix detail template: " + err.Error())
|
panic("failed to parse prefix detail template: " + err.Error())
|
||||||
}
|
}
|
||||||
|
|
||||||
// Parse prefix length template
|
// Parse prefix length template
|
||||||
defaultTemplates.PrefixLength, err = template.New("prefixLength").Funcs(funcs).Parse(prefixLengthHTML)
|
defaultTemplates.PrefixLength, err = template.New("prefixLength").
|
||||||
|
Funcs(funcs).Parse(prefixLengthHTML)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
panic("failed to parse prefix length template: " + err.Error())
|
panic("failed to parse prefix length template: " + err.Error())
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3,7 +3,7 @@ package version
|
|||||||
|
|
||||||
// Build-time variables set via ldflags
|
// Build-time variables set via ldflags
|
||||||
//
|
//
|
||||||
//nolint:gochecknoglobals // These must be variables to allow ldflags injection at build time
|
//nolint:gochecknoglobals // variables so that ldflags can set them at build time
|
||||||
var (
|
var (
|
||||||
// GitRevision is the git commit hash
|
// GitRevision is the git commit hash
|
||||||
GitRevision = "unknown"
|
GitRevision = "unknown"
|
||||||
|
|||||||
+106
-78
@@ -98,7 +98,8 @@ func (c *Client) LookupASN(ctx context.Context, asn int) (*ASNInfo, error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
// Return partial data from IANA if RIR query fails
|
// Return partial data from IANA if RIR query fails
|
||||||
info := c.parseResponse(asn, rir, ianaResp)
|
info := c.parseResponse(asn, rir, ianaResp)
|
||||||
info.RawResponse = ianaResp + "\n--- RIR query failed: " + err.Error() + " ---\n"
|
info.RawResponse = fmt.Sprintf("%s\n--- RIR query failed: %v ---\n",
|
||||||
|
ianaResp, err)
|
||||||
|
|
||||||
return info, nil
|
return info, nil
|
||||||
}
|
}
|
||||||
@@ -120,28 +121,33 @@ func (c *Client) query(ctx context.Context, server, query string) (string, error
|
|||||||
defer func() { _ = conn.Close() }()
|
defer func() { _ = conn.Close() }()
|
||||||
|
|
||||||
// Set deadlines
|
// Set deadlines
|
||||||
if err := conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil {
|
err = conn.SetWriteDeadline(time.Now().Add(writeTimeout))
|
||||||
|
if err != nil {
|
||||||
return "", fmt.Errorf("set write deadline: %w", err)
|
return "", fmt.Errorf("set write deadline: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Send query
|
// Send query
|
||||||
if _, err := fmt.Fprintf(conn, "%s\r\n", query); err != nil {
|
_, err = fmt.Fprintf(conn, "%s\r\n", query)
|
||||||
|
if err != nil {
|
||||||
return "", fmt.Errorf("write query: %w", err)
|
return "", fmt.Errorf("write query: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Read response
|
// Read response
|
||||||
if err := conn.SetReadDeadline(time.Now().Add(readTimeout)); err != nil {
|
err = conn.SetReadDeadline(time.Now().Add(readTimeout))
|
||||||
|
if err != nil {
|
||||||
return "", fmt.Errorf("set read deadline: %w", err)
|
return "", fmt.Errorf("set read deadline: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
var sb strings.Builder
|
var sb strings.Builder
|
||||||
|
|
||||||
scanner := bufio.NewScanner(conn)
|
scanner := bufio.NewScanner(conn)
|
||||||
for scanner.Scan() {
|
for scanner.Scan() {
|
||||||
sb.WriteString(scanner.Text())
|
sb.WriteString(scanner.Text())
|
||||||
sb.WriteString("\n")
|
sb.WriteString("\n")
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := scanner.Err(); err != nil {
|
err = scanner.Err()
|
||||||
|
if err != nil {
|
||||||
return sb.String(), fmt.Errorf("read response: %w", err)
|
return sb.String(), fmt.Errorf("read response: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -149,10 +155,10 @@ func (c *Client) query(ctx context.Context, server, query string) (string, error
|
|||||||
}
|
}
|
||||||
|
|
||||||
// parseIANAReferral extracts the RIR and WHOIS server from an IANA response.
|
// parseIANAReferral extracts the RIR and WHOIS server from an IANA response.
|
||||||
func (c *Client) parseIANAReferral(response string) (rir, whoisServer string) {
|
func (c *Client) parseIANAReferral(response string) (string, string) {
|
||||||
lines := strings.Split(response, "\n")
|
var rir string
|
||||||
|
|
||||||
for _, line := range lines {
|
for line := range strings.SplitSeq(response, "\n") {
|
||||||
line = strings.TrimSpace(line)
|
line = strings.TrimSpace(line)
|
||||||
|
|
||||||
// Look for "refer:" line
|
// Look for "refer:" line
|
||||||
@@ -183,18 +189,9 @@ func (c *Client) parseIANAReferral(response string) (rir, whoisServer string) {
|
|||||||
|
|
||||||
// Also check organisation line for RIR hints
|
// Also check organisation line for RIR hints
|
||||||
if strings.HasPrefix(strings.ToLower(line), "organisation:") {
|
if strings.HasPrefix(strings.ToLower(line), "organisation:") {
|
||||||
org := strings.ToLower(line)
|
orgRIR := rirInOrganisationLine(strings.ToLower(line))
|
||||||
switch {
|
if orgRIR != "" {
|
||||||
case strings.Contains(org, "arin"):
|
rir = orgRIR
|
||||||
rir = RIRARIN
|
|
||||||
case strings.Contains(org, "ripe"):
|
|
||||||
rir = RIRRIPE
|
|
||||||
case strings.Contains(org, "apnic"):
|
|
||||||
rir = RIRAPNIC
|
|
||||||
case strings.Contains(org, "lacnic"):
|
|
||||||
rir = RIRLACNIC
|
|
||||||
case strings.Contains(org, "afrinic"):
|
|
||||||
rir = RIRAFRNIC
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -202,6 +199,25 @@ func (c *Client) parseIANAReferral(response string) (rir, whoisServer string) {
|
|||||||
return rir, ""
|
return rir, ""
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// rirInOrganisationLine returns the RIR a lowercased IANA "organisation:"
|
||||||
|
// line names, or "" when it names none.
|
||||||
|
func rirInOrganisationLine(org string) string {
|
||||||
|
switch {
|
||||||
|
case strings.Contains(org, "arin"):
|
||||||
|
return RIRARIN
|
||||||
|
case strings.Contains(org, "ripe"):
|
||||||
|
return RIRRIPE
|
||||||
|
case strings.Contains(org, "apnic"):
|
||||||
|
return RIRAPNIC
|
||||||
|
case strings.Contains(org, "lacnic"):
|
||||||
|
return RIRLACNIC
|
||||||
|
case strings.Contains(org, "afrinic"):
|
||||||
|
return RIRAFRNIC
|
||||||
|
default:
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// parseResponse extracts ASN information from a WHOIS response.
|
// parseResponse extracts ASN information from a WHOIS response.
|
||||||
func (c *Client) parseResponse(asn int, rir, response string) *ASNInfo {
|
func (c *Client) parseResponse(asn int, rir, response string) *ASNInfo {
|
||||||
info := &ASNInfo{
|
info := &ASNInfo{
|
||||||
@@ -211,11 +227,13 @@ func (c *Client) parseResponse(asn int, rir, response string) *ASNInfo {
|
|||||||
}
|
}
|
||||||
|
|
||||||
lines := strings.Split(response, "\n")
|
lines := strings.Split(response, "\n")
|
||||||
|
|
||||||
var addressLines []string
|
var addressLines []string
|
||||||
|
|
||||||
for _, line := range lines {
|
for _, line := range lines {
|
||||||
line = strings.TrimSpace(line)
|
line = strings.TrimSpace(line)
|
||||||
if line == "" || strings.HasPrefix(line, "%") || strings.HasPrefix(line, "#") {
|
if line == "" || strings.HasPrefix(line, "%") ||
|
||||||
|
strings.HasPrefix(line, "#") {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -232,63 +250,14 @@ func (c *Client) parseResponse(asn int, rir, response string) *ASNInfo {
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
switch key {
|
|
||||||
// AS Name (varies by RIR)
|
|
||||||
case "asname", "as-name":
|
|
||||||
if info.ASName == "" {
|
|
||||||
info.ASName = value
|
|
||||||
}
|
|
||||||
|
|
||||||
// Organization
|
|
||||||
case "orgname", "org-name", "owner":
|
|
||||||
if info.OrgName == "" {
|
|
||||||
info.OrgName = value
|
|
||||||
}
|
|
||||||
case "orgid", "org-id", "org":
|
|
||||||
if info.OrgID == "" {
|
|
||||||
info.OrgID = value
|
|
||||||
}
|
|
||||||
|
|
||||||
// Address (collect multiple lines)
|
// Address (collect multiple lines)
|
||||||
case "address":
|
if key == "address" {
|
||||||
addressLines = append(addressLines, value)
|
addressLines = append(addressLines, value)
|
||||||
|
|
||||||
// Country
|
continue
|
||||||
case "country":
|
|
||||||
if info.CountryCode == "" && len(value) == 2 {
|
|
||||||
info.CountryCode = strings.ToUpper(value)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Abuse contact
|
|
||||||
case "orgabuseemail", "abuse-mailbox":
|
|
||||||
if info.AbuseEmail == "" {
|
|
||||||
info.AbuseEmail = value
|
|
||||||
}
|
|
||||||
case "orgabusephone":
|
|
||||||
if info.AbusePhone == "" {
|
|
||||||
info.AbusePhone = value
|
|
||||||
}
|
|
||||||
|
|
||||||
// Tech contact
|
|
||||||
case "orgtechemail":
|
|
||||||
if info.TechEmail == "" {
|
|
||||||
info.TechEmail = value
|
|
||||||
}
|
|
||||||
case "orgtechphone":
|
|
||||||
if info.TechPhone == "" {
|
|
||||||
info.TechPhone = value
|
|
||||||
}
|
|
||||||
|
|
||||||
// Registration dates
|
|
||||||
case "regdate", "created":
|
|
||||||
if info.RegDate == nil {
|
|
||||||
info.RegDate = c.parseDate(value)
|
|
||||||
}
|
|
||||||
case "updated", "last-modified", "changed":
|
|
||||||
if info.LastMod == nil {
|
|
||||||
info.LastMod = c.parseDate(value)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
c.setResponseField(info, key, value)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Combine address lines
|
// Combine address lines
|
||||||
@@ -304,6 +273,65 @@ func (c *Client) parseResponse(asn int, rir, response string) *ASNInfo {
|
|||||||
return info
|
return info
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// setResponseField stores the value of one "key: value" line of a WHOIS
|
||||||
|
// response in info, keeping the first value seen for each field.
|
||||||
|
func (c *Client) setResponseField(info *ASNInfo, key, value string) {
|
||||||
|
switch key {
|
||||||
|
// Country
|
||||||
|
case "country":
|
||||||
|
if info.CountryCode == "" && len(value) == 2 {
|
||||||
|
info.CountryCode = strings.ToUpper(value)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Registration dates
|
||||||
|
case "regdate", "created":
|
||||||
|
if info.RegDate == nil {
|
||||||
|
info.RegDate = c.parseDate(value)
|
||||||
|
}
|
||||||
|
case "updated", "last-modified", "changed":
|
||||||
|
if info.LastMod == nil {
|
||||||
|
info.LastMod = c.parseDate(value)
|
||||||
|
}
|
||||||
|
|
||||||
|
default:
|
||||||
|
field := textFieldForKey(info, key)
|
||||||
|
if field != nil && *field == "" {
|
||||||
|
*field = value
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// textFieldForKey returns the text field of info that a WHOIS response key
|
||||||
|
// fills, or nil when the key fills none.
|
||||||
|
func textFieldForKey(info *ASNInfo, key string) *string {
|
||||||
|
switch key {
|
||||||
|
// AS Name (varies by RIR)
|
||||||
|
case "asname", "as-name":
|
||||||
|
return &info.ASName
|
||||||
|
|
||||||
|
// Organization
|
||||||
|
case "orgname", "org-name", "owner":
|
||||||
|
return &info.OrgName
|
||||||
|
case "orgid", "org-id", "org":
|
||||||
|
return &info.OrgID
|
||||||
|
|
||||||
|
// Abuse contact
|
||||||
|
case "orgabuseemail", "abuse-mailbox":
|
||||||
|
return &info.AbuseEmail
|
||||||
|
case "orgabusephone":
|
||||||
|
return &info.AbusePhone
|
||||||
|
|
||||||
|
// Tech contact
|
||||||
|
case "orgtechemail":
|
||||||
|
return &info.TechEmail
|
||||||
|
case "orgtechphone":
|
||||||
|
return &info.TechPhone
|
||||||
|
|
||||||
|
default:
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// parseDate attempts to parse various date formats used in WHOIS responses.
|
// parseDate attempts to parse various date formats used in WHOIS responses.
|
||||||
func (c *Client) parseDate(value string) *time.Time {
|
func (c *Client) parseDate(value string) *time.Time {
|
||||||
// Common formats
|
// Common formats
|
||||||
@@ -319,15 +347,15 @@ func (c *Client) parseDate(value string) *time.Time {
|
|||||||
value = strings.TrimSpace(value)
|
value = strings.TrimSpace(value)
|
||||||
// Handle "YYYYMMDD" format from LACNIC
|
// Handle "YYYYMMDD" format from LACNIC
|
||||||
if len(value) == lacnicDateFormatLen {
|
if len(value) == lacnicDateFormatLen {
|
||||||
if _, err := time.Parse("20060102", value); err == nil {
|
t, err := time.Parse("20060102", value)
|
||||||
t, _ := time.Parse("20060102", value)
|
if err == nil {
|
||||||
|
|
||||||
return &t
|
return &t
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, format := range formats {
|
for _, format := range formats {
|
||||||
if t, err := time.Parse(format, value); err == nil {
|
t, err := time.Parse(format, value)
|
||||||
|
if err == nil {
|
||||||
return &t
|
return &t
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,4 +1,3 @@
|
|||||||
// Package asinfo provides information about Autonomous Systems (AS)
|
|
||||||
package asinfo
|
package asinfo
|
||||||
|
|
||||||
import (
|
import (
|
||||||
@@ -54,7 +53,9 @@ func initRegistry() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
var asInfos []Info
|
var asInfos []Info
|
||||||
if err := json.Unmarshal(jsonData, &asInfos); err != nil {
|
|
||||||
|
err = json.Unmarshal(jsonData, &asInfos)
|
||||||
|
if err != nil {
|
||||||
panic("failed to unmarshal embedded AS data: " + err.Error())
|
panic("failed to unmarshal embedded AS data: " + err.Error())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -137,6 +138,7 @@ func Search(query string) []Info {
|
|||||||
defer defaultRegistry.mu.RUnlock()
|
defer defaultRegistry.mu.RUnlock()
|
||||||
|
|
||||||
var results []Info
|
var results []Info
|
||||||
|
|
||||||
for _, info := range defaultRegistry.byASN {
|
for _, info := range defaultRegistry.byASN {
|
||||||
if contains(info.Handle, query) || contains(info.Description, query) {
|
if contains(info.Handle, query) || contains(info.Description, query) {
|
||||||
results = append(results, *info)
|
results = append(results, *info)
|
||||||
|
|||||||
+52
-16
@@ -1,10 +1,15 @@
|
|||||||
package asinfo
|
package asinfo_test
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
"git.eeqj.de/sneak/routewatch/pkg/asinfo"
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestGet(t *testing.T) {
|
func TestGet(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
tests := []struct {
|
tests := []struct {
|
||||||
name string
|
name string
|
||||||
asn int
|
asn int
|
||||||
@@ -38,20 +43,26 @@ func TestGet(t *testing.T) {
|
|||||||
|
|
||||||
for _, tt := range tests {
|
for _, tt := range tests {
|
||||||
t.Run(tt.name, func(t *testing.T) {
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
info, ok := Get(tt.asn)
|
t.Parallel()
|
||||||
|
|
||||||
|
info, ok := asinfo.Get(tt.asn)
|
||||||
if ok != tt.wantOK {
|
if ok != tt.wantOK {
|
||||||
t.Errorf("Get(%d) ok = %v, want %v", tt.asn, ok, tt.wantOK)
|
t.Errorf("Get(%d) ok = %v, want %v", tt.asn, ok, tt.wantOK)
|
||||||
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
if ok && info.Description != tt.wantDesc {
|
if ok && info.Description != tt.wantDesc {
|
||||||
t.Errorf("Get(%d) description = %q, want %q", tt.asn, info.Description, tt.wantDesc)
|
t.Errorf("Get(%d) description = %q, want %q",
|
||||||
|
tt.asn, info.Description, tt.wantDesc)
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestGetDescription(t *testing.T) {
|
func TestGetDescription(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
tests := []struct {
|
tests := []struct {
|
||||||
asn int
|
asn int
|
||||||
want string
|
want string
|
||||||
@@ -62,7 +73,10 @@ func TestGetDescription(t *testing.T) {
|
|||||||
|
|
||||||
for _, tt := range tests {
|
for _, tt := range tests {
|
||||||
t.Run("", func(t *testing.T) {
|
t.Run("", func(t *testing.T) {
|
||||||
if got := GetDescription(tt.asn); got != tt.want {
|
t.Parallel()
|
||||||
|
|
||||||
|
got := asinfo.GetDescription(tt.asn)
|
||||||
|
if got != tt.want {
|
||||||
t.Errorf("GetDescription(%d) = %q, want %q", tt.asn, got, tt.want)
|
t.Errorf("GetDescription(%d) = %q, want %q", tt.asn, got, tt.want)
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
@@ -70,6 +84,8 @@ func TestGetDescription(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestGetHandle(t *testing.T) {
|
func TestGetHandle(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
tests := []struct {
|
tests := []struct {
|
||||||
asn int
|
asn int
|
||||||
want string
|
want string
|
||||||
@@ -81,7 +97,10 @@ func TestGetHandle(t *testing.T) {
|
|||||||
|
|
||||||
for _, tt := range tests {
|
for _, tt := range tests {
|
||||||
t.Run("", func(t *testing.T) {
|
t.Run("", func(t *testing.T) {
|
||||||
if got := GetHandle(tt.asn); got != tt.want {
|
t.Parallel()
|
||||||
|
|
||||||
|
got := asinfo.GetHandle(tt.asn)
|
||||||
|
if got != tt.want {
|
||||||
t.Errorf("GetHandle(%d) = %q, want %q", tt.asn, got, tt.want)
|
t.Errorf("GetHandle(%d) = %q, want %q", tt.asn, got, tt.want)
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
@@ -89,18 +108,23 @@ func TestGetHandle(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestTotal(t *testing.T) {
|
func TestTotal(t *testing.T) {
|
||||||
total := Total()
|
t.Parallel()
|
||||||
|
|
||||||
|
total := asinfo.Total()
|
||||||
if total < 100000 {
|
if total < 100000 {
|
||||||
t.Errorf("Total() = %d, expected > 100000", total)
|
t.Errorf("Total() = %d, expected > 100000", total)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Verify it's consistent
|
// Verify it's consistent
|
||||||
if total2 := Total(); total2 != total {
|
total2 := asinfo.Total()
|
||||||
|
if total2 != total {
|
||||||
t.Errorf("Total() returned different values: %d vs %d", total, total2)
|
t.Errorf("Total() returned different values: %d vs %d", total, total2)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestSearch(t *testing.T) {
|
func TestSearch(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
tests := []struct {
|
tests := []struct {
|
||||||
name string
|
name string
|
||||||
query string
|
query string
|
||||||
@@ -130,16 +154,21 @@ func TestSearch(t *testing.T) {
|
|||||||
|
|
||||||
for _, tt := range tests {
|
for _, tt := range tests {
|
||||||
t.Run(tt.name, func(t *testing.T) {
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
results := Search(tt.query)
|
t.Parallel()
|
||||||
|
|
||||||
|
results := asinfo.Search(tt.query)
|
||||||
if len(results) < tt.wantMin {
|
if len(results) < tt.wantMin {
|
||||||
t.Errorf("Search(%q) returned %d results, want at least %d", tt.query, len(results), tt.wantMin)
|
t.Errorf("Search(%q) returned %d results, want at least %d",
|
||||||
|
tt.query, len(results), tt.wantMin)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Verify all results contain the query
|
// Verify all results contain the query
|
||||||
if tt.query != "" {
|
if tt.query != "" {
|
||||||
for _, r := range results {
|
for _, r := range results {
|
||||||
if !contains(r.Handle, tt.query) && !contains(r.Description, tt.query) {
|
if !strings.Contains(r.Handle, tt.query) &&
|
||||||
t.Errorf("Search result ASN %d doesn't contain query %q", r.ASN, tt.query)
|
!strings.Contains(r.Description, tt.query) {
|
||||||
|
t.Errorf("Search result ASN %d doesn't contain query %q",
|
||||||
|
r.ASN, tt.query)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -148,13 +177,17 @@ func TestSearch(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestDataIntegrity(t *testing.T) {
|
func TestDataIntegrity(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
// Verify no duplicate ASNs
|
// Verify no duplicate ASNs
|
||||||
all := All()
|
all := asinfo.All()
|
||||||
seen := make(map[int]bool)
|
seen := make(map[int]bool)
|
||||||
|
|
||||||
for _, info := range all {
|
for _, info := range all {
|
||||||
if seen[info.ASN] {
|
if seen[info.ASN] {
|
||||||
t.Errorf("Duplicate ASN found: %d", info.ASN)
|
t.Errorf("Duplicate ASN found: %d", info.ASN)
|
||||||
}
|
}
|
||||||
|
|
||||||
seen[info.ASN] = true
|
seen[info.ASN] = true
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -163,6 +196,7 @@ func TestDataIntegrity(t *testing.T) {
|
|||||||
if info.Handle == "" && info.ASN != 0 {
|
if info.Handle == "" && info.ASN != 0 {
|
||||||
t.Errorf("ASN %d has empty handle", info.ASN)
|
t.Errorf("ASN %d has empty handle", info.ASN)
|
||||||
}
|
}
|
||||||
|
|
||||||
if info.Description == "" {
|
if info.Description == "" {
|
||||||
t.Errorf("ASN %d has empty description", info.ASN)
|
t.Errorf("ASN %d has empty description", info.ASN)
|
||||||
}
|
}
|
||||||
@@ -174,8 +208,9 @@ func BenchmarkGet(b *testing.B) {
|
|||||||
asns := []int{1, 15169, 13335, 32934, 8075, 16509}
|
asns := []int{1, 15169, 13335, 32934, 8075, 16509}
|
||||||
|
|
||||||
b.ResetTimer()
|
b.ResetTimer()
|
||||||
for i := 0; i < b.N; i++ {
|
|
||||||
Get(asns[i%len(asns)])
|
for i := range b.N {
|
||||||
|
asinfo.Get(asns[i%len(asns)])
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -183,7 +218,8 @@ func BenchmarkSearch(b *testing.B) {
|
|||||||
queries := []string{"Google", "Amazon", "Microsoft", "University"}
|
queries := []string{"Google", "Amazon", "Microsoft", "University"}
|
||||||
|
|
||||||
b.ResetTimer()
|
b.ResetTimer()
|
||||||
for i := 0; i < b.N; i++ {
|
|
||||||
Search(queries[i%len(queries)])
|
for i := range b.N {
|
||||||
|
asinfo.Search(queries[i%len(queries)])
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user