diff --git a/cmd/asinfo-gen/main.go b/cmd/asinfo-gen/main.go index 1d895b9..fbc0a50 100644 --- a/cmd/asinfo-gen/main.go +++ b/cmd/asinfo-gen/main.go @@ -2,8 +2,10 @@ package main import ( + "context" "encoding/csv" "encoding/json" + "errors" "io" "log" "net/http" @@ -27,13 +29,17 @@ func main() { // Fetch the 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 { log.Fatalf("Failed to fetch CSV: %v", err) } - defer func() { - _ = resp.Body.Close() - }() if resp.StatusCode != http.StatusOK { 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) } - 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) } // Read all records var asInfos []ASInfo + for { record, err := reader.Read() - if err == io.EOF { + if errors.Is(err, io.EOF) { break } + if err != nil { 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)) // Convert to JSON and output to stdout encoder := json.NewEncoder(os.Stdout) 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) } } diff --git a/cmd/streamdumper/main.go b/cmd/streamdumper/main.go index 7754d7a..883c703 100644 --- a/cmd/streamdumper/main.go +++ b/cmd/streamdumper/main.go @@ -26,9 +26,17 @@ func main() { // Register raw message handler that prints to stdout 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 ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -36,18 +44,13 @@ func main() { // Handle signals sigChan := make(chan os.Signal, 1) signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM) + go func() { <-sigChan log.Println("Received shutdown signal") cancel() }() - // Start streaming - if err := s.Start(); err != nil { - log.Fatal("Failed to start streamer:", err) - } - defer s.Stop() - // Wait for context cancellation <-ctx.Done() } diff --git a/internal/config/config.go b/internal/config/config.go index 87a6ce3..efb290f 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -2,6 +2,7 @@ package config import ( + "errors" "fmt" "os" "path/filepath" @@ -27,6 +28,12 @@ const ( 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 type Config struct { // 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 time.Duration - // EnableBatchedDatabaseWrites enables batched database operations for better performance + // EnableBatchedDatabaseWrites enables batched database operations for + // better performance EnableBatchedDatabaseWrites bool - // RouteExpirationTimeout is how long a route can go without being refreshed before expiring - // Default is 2 hours which is conservative for BGP (typical BGP hold time is 90-180 seconds) + // RouteExpirationTimeout is how long a route can go without being + // refreshed before expiring + // Default is 2 hours which is conservative for BGP (typical BGP hold + // time is 90-180 seconds) RouteExpirationTimeout time.Duration } @@ -59,11 +69,14 @@ func New() (*Config, error) { } return &Config{ - StateDir: stateDir, - Port: port, - MaxRuntime: 0, // Run forever by default - EnableBatchedDatabaseWrites: true, // Enable batching by default - RouteExpirationTimeout: defaultRouteExpirationMinutes * time.Minute, // For active route monitoring + StateDir: stateDir, + Port: port, + // Run forever by default + MaxRuntime: 0, + // Enable batching by default + EnableBatchedDatabaseWrites: true, + // For active route monitoring + RouteExpirationTimeout: defaultRouteExpirationMinutes * time.Minute, }, nil } @@ -89,7 +102,7 @@ func getStateDirectory() (string, error) { // the database somewhere unexpected. xdgData := os.Getenv("XDG_DATA_HOME") 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 @@ -111,7 +124,7 @@ func getStateDirectory() (string, error) { return filepath.Join(home, ".local", "share", AppIdentifier), nil 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" port, err := strconv.ParseUint(value, 10, 0) 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 @@ -134,7 +148,8 @@ func getPort() (int, error) { // EnsureDirectories creates all necessary directories if they don't exist func (c *Config) EnsureDirectories() error { // 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) } diff --git a/internal/config/config_test.go b/internal/config/config_internal_test.go similarity index 91% rename from internal/config/config_test.go rename to internal/config/config_internal_test.go index 84f9414..c9f10bd 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_internal_test.go @@ -36,7 +36,8 @@ func TestNewRefusesInvalidPort(t *testing.T) { t.Setenv("PORT", value) 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) } }) @@ -51,12 +52,16 @@ func TestNewRefusesRelativeXDGDataHome(t *testing.T) { t.Setenv("PORT", "") 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.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) } } diff --git a/internal/logger/logger.go b/internal/logger/logger.go index 3e90ad8..d46d523 100644 --- a/internal/logger/logger.go +++ b/internal/logger/logger.go @@ -74,6 +74,7 @@ func getSourceAttrs() []slog.Attr { // Get the function name fn := runtime.FuncForPC(pc) + var funcName string if fn != nil { funcName = filepath.Base(fn.Name()) diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 86f2057..f05bc07 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -115,9 +115,11 @@ func (t *Tracker) GetStreamMetrics() StreamMetrics { if msgCount >= 0 { totalMessages = uint64(msgCount) } + if byteCount >= 0 { totalBytes = uint64(byteCount) } + if wireByteCount >= 0 { totalWireBytes = uint64(wireByteCount) } @@ -161,7 +163,7 @@ func (t *Tracker) RecordWithdrawal() { func (t *Tracker) SetBGPPeerCount(count int) { // BGP peer count is always small (< 1000), so int32 is safe 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 // TotalBytes is the total number of decompressed bytes received since startup 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 // ConnectedSince is the time when the current connection was established ConnectedSince time.Time @@ -217,9 +220,11 @@ type StreamMetrics struct { Connected bool // MessagesPerSec is the rate of messages received per second (1-minute average) 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 - // 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 // ReconnectCount is the number of reconnections since startup ReconnectCount uint64 diff --git a/internal/ristypes/ris.go b/internal/ristypes/ris.go index a6831a4..1ad2072 100644 --- a/internal/ristypes/ris.go +++ b/internal/ristypes/ris.go @@ -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 import ( @@ -17,25 +18,30 @@ type ASPath []int func (p *ASPath) UnmarshalJSON(data []byte) error { // First try to unmarshal as a simple array of integers var simple []int - if err := json.Unmarshal(data, &simple); err == nil { + + err := json.Unmarshal(data, &simple) + if err == nil { *p = ASPath(simple) return nil } // If that fails, unmarshal as array of interfaces and flatten - var raw []interface{} - if err := json.Unmarshal(data, &raw); err != nil { + var raw []any + + err = json.Unmarshal(data, &raw) + if err != nil { return err } // Flatten the array result := make([]int, 0) + for _, item := range raw { switch v := item.(type) { case float64: result = append(result, int(v)) - case []interface{}: + case []any: // Nested array - flatten it for _, nested := range v { 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 // the actual BGP update data including AS path, communities, announcements, // and withdrawals. +// +//nolint:tagliatelle // field names are the RIS Live feed's type RISMessage struct { Type string `json:"type"` Timestamp float64 `json:"timestamp"` @@ -90,6 +98,8 @@ type RISMessage struct { // RISAnnouncement represents a BGP route announcement within a RIS message. // It contains the next hop IP address and the list of prefixes being announced // via that next hop. +// +//nolint:tagliatelle // field names are the RIS Live feed's type RISAnnouncement struct { NextHop string `json:"next_hop"` Prefixes []string `json:"prefixes"` diff --git a/internal/ristypes/ris_test.go b/internal/ristypes/ris_test.go index 6a3b7b0..2b3f7d2 100644 --- a/internal/ristypes/ris_test.go +++ b/internal/ristypes/ris_test.go @@ -1,10 +1,12 @@ -package ristypes +package ristypes_test import ( "bufio" "encoding/json" "os" "testing" + + "git.eeqj.de/sneak/routewatch/internal/ristypes" ) // 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) // stay empty while the fields handlers use (Path, Announcements) still decode. func TestDecodeDropsCommunityAndRaw(t *testing.T) { + t.Parallel() + f, err := os.Open(messageExamplesPath) if err != nil { t.Fatalf("open fixtures: %v", err) } - defer f.Close() + + defer func() { + _ = f.Close() + }() var messages, withPath, withAnnouncements int scanner := bufio.NewScanner(f) scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for scanner.Scan() { line := scanner.Bytes() if len(line) == 0 { continue } - var wrapper RISLiveMessage - if err := json.Unmarshal(line, &wrapper); err != nil { + var wrapper ristypes.RISLiveMessage + + err := json.Unmarshal(line, &wrapper) + if err != nil { t.Fatalf("unmarshal message %d: %v", messages+1, err) } + messages++ msg := wrapper.Data 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 != "" { t.Errorf("message %d: Raw decoded, want empty", messages) } @@ -48,11 +61,14 @@ func TestDecodeDropsCommunityAndRaw(t *testing.T) { if len(msg.Path) > 0 { withPath++ } + if len(msg.Announcements) > 0 { withAnnouncements++ } } - if err := scanner.Err(); err != nil { + + err = scanner.Err() + if err != nil { 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, // otherwise an empty Community/Raw would prove nothing. 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) } } diff --git a/internal/streamer/streamer.go b/internal/streamer/streamer.go index b8b75a7..169ebfd 100644 --- a/internal/streamer/streamer.go +++ b/internal/streamer/streamer.go @@ -1,5 +1,6 @@ -// Package streamer implements an HTTP client that connects to the RIPE RIS Live streaming API, -// parses BGP UPDATE messages from the JSON stream, and dispatches them to registered handlers. +// Package streamer implements an HTTP client that connects to the RIPE RIS Live +// streaming API, parses BGP UPDATE messages from the JSON stream, and +// dispatches them to registered handlers. package streamer import ( @@ -7,6 +8,7 @@ import ( "compress/gzip" "context" "encoding/json" + "errors" "fmt" "io" "math" @@ -24,20 +26,20 @@ import ( // countingReader wraps an io.Reader and counts bytes read type countingReader struct { reader io.Reader - count int64 + count atomic.Int64 } // Read implements io.Reader and counts bytes func (c *countingReader) Read(p []byte) (int, error) { n, err := c.reader.Read(p) - atomic.AddInt64(&c.count, int64(n)) + c.count.Add(int64(n)) return n, err } // Count returns the total bytes read func (c *countingReader) Count() int64 { - return atomic.LoadInt64(&c.count) + return c.count.Load() } // Configuration constants for the RIS Live streamer. @@ -58,11 +60,17 @@ const ( 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. // Implementations must specify which message types they want to receive, // how to process messages, and their desired queue capacity. 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 // HandleMessage processes a RIS message. This method is called from a dedicated @@ -113,7 +121,7 @@ type Streamer struct { cancel context.CancelFunc running bool 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 bgpPeers map[string]bool // Track active BGP peers by peer IP 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. // 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 { return &Streamer{ logger: logger, @@ -153,7 +162,8 @@ func (s *Streamer) RegisterHandler(handler MessageHandler) { handler: handler, queue: make(chan *ristypes.RISMessage, handler.QueueCapacity()), 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) { s.mu.Lock() defer s.mu.Unlock() + s.rawHandler = handler } @@ -183,7 +194,7 @@ func (s *Streamer) Start() error { defer s.mu.Unlock() if s.running { - return fmt.Errorf("streamer already running") + return errAlreadyRunning } ctx, cancel := context.WithCancel(context.Background()) @@ -216,44 +227,21 @@ func (s *Streamer) Stop() { return } + s.cancel() s.cancel = nil // Close all handler queues to signal workers to stop for _, info := range s.handlers { close(info.queue) } + s.running = false s.mu.Unlock() s.metrics.SetConnected(false) } -// 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() - } -} - -// IsRunning reports whether the streamer is currently connected and processing messages. -// This is safe to call concurrently from multiple goroutines. +// 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 { s.mu.RLock() defer s.mu.RUnlock() @@ -314,12 +302,15 @@ func (s *Streamer) GetHandlerStats() []HandlerStats { // Calculate average time if info.metrics.processedCount > 0 { processedCount := info.metrics.processedCount + const maxInt64 = 1<<63 - 1 + if 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() @@ -330,11 +321,38 @@ func (s *Streamer) GetHandlerStats() []HandlerStats { return stats } -// GetDroppedMessages returns the total number of messages dropped across all handlers -// due to queue overflow or backpressure. This counter is monotonically increasing -// and is safe to call concurrently. +// GetDroppedMessages returns the total number of messages dropped across all +// handlers due to queue overflow or backpressure. This counter is monotonically +// increasing and is safe to call concurrently. 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 @@ -343,7 +361,8 @@ func (s *Streamer) logMetrics() { uptime := time.Since(metrics.ConnectedSince) const bitsPerMegabit = 1000000 - totalDropped := atomic.LoadUint64(&s.totalDropped) + + totalDropped := s.totalDropped.Load() s.logger.Info( "Stream statistics", @@ -369,16 +388,19 @@ func (s *Streamer) logMetrics() { // Log per-handler statistics s.mu.RLock() + for i, info := range s.handlers { info.metrics.mu.Lock() if info.metrics.processedCount > 0 { // Safe conversion: processedCount is bounded by maxInt64 processedCount := info.metrics.processedCount + const maxInt64 = 1<<63 - 1 + if processedCount > maxInt64 { processedCount = maxInt64 } - //nolint:gosec // processedCount is explicitly bounded above + avgTime := info.metrics.totalTime / time.Duration(processedCount) s.logger.Info( "Handler statistics", @@ -395,6 +417,7 @@ func (s *Streamer) logMetrics() { } info.metrics.mu.Unlock() } + s.mu.RUnlock() } @@ -403,7 +426,8 @@ func (s *Streamer) updateMetrics(messageBytes int) { 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) { backoffDelay := minBackoffDelay consecutiveFailures := 0 @@ -444,6 +468,7 @@ func (s *Streamer) streamWithReconnect(ctx context.Context) { if streamDuration > 30*time.Second { s.logger.Info("Resetting backoff delay due to successful connection", "stream_duration", streamDuration) + backoffDelay = minBackoffDelay consecutiveFailures = 0 } else { @@ -477,7 +502,7 @@ func (s *Streamer) stream(ctx context.Context) error { connCtx, connCancel := context.WithCancel(ctx) defer connCancel() - req, err := http.NewRequestWithContext(ctx, "GET", s.url, nil) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.url, nil) if err != nil { 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) } 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) } }() 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 @@ -510,7 +536,9 @@ func (s *Streamer) stream(ctx context.Context) error { return fmt.Errorf("failed to create gzip reader: %w", err) } defer func() { _ = gzReader.Close() }() + reader = gzReader + s.logger.Info("Connected to RIS Live stream", "compression", "gzip") } else { 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) - // Track wire bytes for metrics updates - var lastWireBytes int64 - // Start metrics logging goroutine metricsTicker := time.NewTicker(metricsLogInterval) 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 + 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) defer wireUpdateTicker.Stop() - go func() { - for { - select { - case <-wireUpdateTicker.C: - currentBytes := wireCounter.Count() - delta := currentBytes - lastWireBytes - if delta > 0 { - s.metrics.RecordWireBytes(delta) - lastWireBytes = currentBytes - } - case <-connCtx.Done(): - return - } - } - }() + var lastWireBytes int64 + 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) for scanner.Scan() { @@ -572,149 +613,176 @@ func (s *Streamer) stream(ctx context.Context) error { continue } - // 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)) + err := s.processLine(ctx, line) + if err != nil { + return err } - - // 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 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 { // Calculate queue utilization queueLen := len(info.queue) diff --git a/internal/streamer/streamer_test.go b/internal/streamer/streamer_internal_test.go similarity index 80% rename from internal/streamer/streamer_test.go rename to internal/streamer/streamer_internal_test.go index 5d9582b..e080cb4 100644 --- a/internal/streamer/streamer_test.go +++ b/internal/streamer/streamer_internal_test.go @@ -16,6 +16,8 @@ import ( ) func TestNewStreamer(t *testing.T) { + t.Parallel() + logger := logger.New() metricsTracker := metrics.New() s := New(logger, metricsTracker) @@ -46,10 +48,13 @@ func TestNewStreamer(t *testing.T) { // goroutine count returns to its starting value. Each connection starts two // ticker goroutines; before the fix they lived until the streamer's lifetime // context was cancelled, so every reconnect leaked two. +// +//nolint:paralleltest // counts goroutines, so no other test may run alongside func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) { // The handler returns immediately, so the response body is empty and each // 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() 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 // 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) } + s.client.CloseIdleConnections() baseline := settledGoroutineCount() const reconnects = 20 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) } } + s.client.CloseIdleConnections() if !waitForGoroutines(baseline) { @@ -81,9 +90,13 @@ func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) { // updateHandler wants UPDATE messages and does nothing with them. type updateHandler struct{} -func (updateHandler) WantsMessage(messageType string) bool { return messageType == "UPDATE" } -func (updateHandler) HandleMessage(*ristypes.RISMessage) {} -func (updateHandler) QueueCapacity() int { return 10 } +func (updateHandler) WantsMessage(messageType string) bool { + return messageType == "UPDATE" +} + +func (updateHandler) HandleMessage(*ristypes.RISMessage) {} + +func (updateHandler) QueueCapacity() int { return 10 } // TestStopBeforeMessageReachesQueues stops the streamer after the read loop // 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 // the read loop itself, so calling Stop from it hits the gap every time. func TestStopBeforeMessageReachesQueues(t *testing.T) { - const line = `{"type":"ris_message","data":{"type":"UPDATE","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) - })) + t.Parallel() + + const line = `{"type":"ris_message","data":{"type":"UPDATE",` + + `"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() s := New(logger.New(), metrics.New()) @@ -109,7 +126,8 @@ func TestStopBeforeMessageReachesQueues(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) 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) } @@ -121,12 +139,15 @@ func TestStopBeforeMessageReachesQueues(t *testing.T) { // current count. func settledGoroutineCount() int { prev := runtime.NumGoroutine() + for range 20 { time.Sleep(10 * time.Millisecond) + cur := runtime.NumGoroutine() if cur == prev { return cur } + prev = cur } @@ -139,6 +160,7 @@ func waitForGoroutines(target int) bool { if runtime.NumGoroutine() <= target { return true } + time.Sleep(10 * time.Millisecond) } diff --git a/internal/templates/templates.go b/internal/templates/templates.go index b388343..16d6737 100644 --- a/internal/templates/templates.go +++ b/internal/templates/templates.go @@ -60,6 +60,7 @@ func timeSince(t time.Time) string { if duration < time.Minute { return "just now" } + if duration < time.Hour { minutes := int(duration.Minutes()) if minutes == 1 { @@ -68,6 +69,7 @@ func timeSince(t time.Time) string { return duration.Truncate(time.Minute).String() + " ago" } + if duration < hoursPerDay*time.Hour { hours := int(duration.Hours()) if hours == 1 { @@ -76,10 +78,12 @@ func timeSince(t time.Time) string { return duration.Truncate(time.Hour).String() + " ago" } + days := int(duration.Hours() / hoursPerDay) if days == 1 { return "1 day ago" } + if days < daysPerMonth { return duration.Truncate(hoursPerDay*time.Hour).String() + " ago" } @@ -118,7 +122,7 @@ func initTemplates() { "appLicense": func() string { return version.License }, "appRepoURL": func() string { return version.RepoURL }, "appGitRevision": func() string { return version.GitRevisionShort }, - "appGitCommitURL": func() string { return version.CommitURL() }, + "appGitCommitURL": version.CommitURL, } // Parse index template @@ -134,19 +138,22 @@ func initTemplates() { } // 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 { panic("failed to parse AS detail template: " + err.Error()) } // 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 { panic("failed to parse prefix detail template: " + err.Error()) } // 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 { panic("failed to parse prefix length template: " + err.Error()) } diff --git a/internal/version/version.go b/internal/version/version.go index b32b2d8..20c44d5 100644 --- a/internal/version/version.go +++ b/internal/version/version.go @@ -3,7 +3,7 @@ package version // 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 ( // GitRevision is the git commit hash GitRevision = "unknown" diff --git a/internal/whois/whois.go b/internal/whois/whois.go index 1c4e55b..3713c9f 100644 --- a/internal/whois/whois.go +++ b/internal/whois/whois.go @@ -98,7 +98,8 @@ func (c *Client) LookupASN(ctx context.Context, asn int) (*ASNInfo, error) { if err != nil { // Return partial data from IANA if RIR query fails 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 } @@ -120,28 +121,33 @@ func (c *Client) query(ctx context.Context, server, query string) (string, error defer func() { _ = conn.Close() }() // 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) } // 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) } // 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) } var sb strings.Builder + scanner := bufio.NewScanner(conn) for scanner.Scan() { sb.WriteString(scanner.Text()) sb.WriteString("\n") } - if err := scanner.Err(); err != nil { + err = scanner.Err() + if err != nil { 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. -func (c *Client) parseIANAReferral(response string) (rir, whoisServer string) { - lines := strings.Split(response, "\n") +func (c *Client) parseIANAReferral(response string) (string, string) { + var rir string - for _, line := range lines { + for line := range strings.SplitSeq(response, "\n") { line = strings.TrimSpace(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 if strings.HasPrefix(strings.ToLower(line), "organisation:") { - org := strings.ToLower(line) - switch { - case strings.Contains(org, "arin"): - 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 + orgRIR := rirInOrganisationLine(strings.ToLower(line)) + if orgRIR != "" { + rir = orgRIR } } } @@ -202,6 +199,25 @@ func (c *Client) parseIANAReferral(response string) (rir, whoisServer string) { 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. func (c *Client) parseResponse(asn int, rir, response string) *ASNInfo { info := &ASNInfo{ @@ -211,11 +227,13 @@ func (c *Client) parseResponse(asn int, rir, response string) *ASNInfo { } lines := strings.Split(response, "\n") + var addressLines []string for _, line := range lines { line = strings.TrimSpace(line) - if line == "" || strings.HasPrefix(line, "%") || strings.HasPrefix(line, "#") { + if line == "" || strings.HasPrefix(line, "%") || + strings.HasPrefix(line, "#") { continue } @@ -232,63 +250,14 @@ func (c *Client) parseResponse(asn int, rir, response string) *ASNInfo { 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) - case "address": + if key == "address" { addressLines = append(addressLines, value) - // Country - 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) - } + continue } + + c.setResponseField(info, key, value) } // Combine address lines @@ -304,6 +273,65 @@ func (c *Client) parseResponse(asn int, rir, response string) *ASNInfo { 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. func (c *Client) parseDate(value string) *time.Time { // Common formats @@ -319,15 +347,15 @@ func (c *Client) parseDate(value string) *time.Time { value = strings.TrimSpace(value) // Handle "YYYYMMDD" format from LACNIC if len(value) == lacnicDateFormatLen { - if _, err := time.Parse("20060102", value); err == nil { - t, _ := time.Parse("20060102", value) - + t, err := time.Parse("20060102", value) + if err == nil { return &t } } for _, format := range formats { - if t, err := time.Parse(format, value); err == nil { + t, err := time.Parse(format, value) + if err == nil { return &t } } diff --git a/pkg/asinfo/asinfo.go b/pkg/asinfo/asinfo.go index 500ec20..372f2e1 100644 --- a/pkg/asinfo/asinfo.go +++ b/pkg/asinfo/asinfo.go @@ -1,4 +1,3 @@ -// Package asinfo provides information about Autonomous Systems (AS) package asinfo import ( @@ -54,7 +53,9 @@ func initRegistry() { } 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()) } @@ -137,6 +138,7 @@ func Search(query string) []Info { defer defaultRegistry.mu.RUnlock() var results []Info + for _, info := range defaultRegistry.byASN { if contains(info.Handle, query) || contains(info.Description, query) { results = append(results, *info) diff --git a/pkg/asinfo/asinfo_test.go b/pkg/asinfo/asinfo_test.go index 5d8c205..3dabfa6 100644 --- a/pkg/asinfo/asinfo_test.go +++ b/pkg/asinfo/asinfo_test.go @@ -1,10 +1,15 @@ -package asinfo +package asinfo_test import ( + "strings" "testing" + + "git.eeqj.de/sneak/routewatch/pkg/asinfo" ) func TestGet(t *testing.T) { + t.Parallel() + tests := []struct { name string asn int @@ -38,20 +43,26 @@ func TestGet(t *testing.T) { for _, tt := range tests { 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 { t.Errorf("Get(%d) ok = %v, want %v", tt.asn, ok, tt.wantOK) + return } 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) { + t.Parallel() + tests := []struct { asn int want string @@ -62,7 +73,10 @@ func TestGetDescription(t *testing.T) { for _, tt := range tests { 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) } }) @@ -70,6 +84,8 @@ func TestGetDescription(t *testing.T) { } func TestGetHandle(t *testing.T) { + t.Parallel() + tests := []struct { asn int want string @@ -81,7 +97,10 @@ func TestGetHandle(t *testing.T) { for _, tt := range tests { 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) } }) @@ -89,18 +108,23 @@ func TestGetHandle(t *testing.T) { } func TestTotal(t *testing.T) { - total := Total() + t.Parallel() + + total := asinfo.Total() if total < 100000 { t.Errorf("Total() = %d, expected > 100000", total) } // 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) } } func TestSearch(t *testing.T) { + t.Parallel() + tests := []struct { name string query string @@ -130,16 +154,21 @@ func TestSearch(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - results := Search(tt.query) + t.Parallel() + + results := asinfo.Search(tt.query) 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 if tt.query != "" { for _, r := range results { - if !contains(r.Handle, tt.query) && !contains(r.Description, tt.query) { - t.Errorf("Search result ASN %d doesn't contain query %q", r.ASN, tt.query) + if !strings.Contains(r.Handle, 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) { + t.Parallel() + // Verify no duplicate ASNs - all := All() + all := asinfo.All() seen := make(map[int]bool) + for _, info := range all { if seen[info.ASN] { t.Errorf("Duplicate ASN found: %d", info.ASN) } + seen[info.ASN] = true } @@ -163,6 +196,7 @@ func TestDataIntegrity(t *testing.T) { if info.Handle == "" && info.ASN != 0 { t.Errorf("ASN %d has empty handle", info.ASN) } + if info.Description == "" { 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} 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"} 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)]) } }