check / check (push) Waiting to run
The read loop added to a handler's drop count atomically while the statistics endpoint and the metrics log read it as a plain field, a data race whenever messages were dropped. The count is now an atomic.Uint64, read with Load. A test reads the handler statistics while the read loop drops messages for a handler that never finishes one. Run created the WHOIS fetcher and gave it to the HTTP server after the server had started, while request handlers could read it. The fetcher is now set before the server starts. Model: opus-5-5
147 lines
3.7 KiB
Go
147 lines
3.7 KiB
Go
// Package server provides HTTP endpoints for status monitoring and statistics
|
|
package server
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"net"
|
|
"net/http"
|
|
"net/netip"
|
|
"strconv"
|
|
"time"
|
|
|
|
"git.eeqj.de/sneak/routewatch/internal/config"
|
|
"git.eeqj.de/sneak/routewatch/internal/database"
|
|
"git.eeqj.de/sneak/routewatch/internal/logger"
|
|
"git.eeqj.de/sneak/routewatch/internal/streamer"
|
|
"github.com/go-chi/chi/v5"
|
|
)
|
|
|
|
// ASNFetcherStats contains WHOIS fetcher statistics.
|
|
type ASNFetcherStats struct {
|
|
SuccessesLastHour int
|
|
ErrorsLastHour int
|
|
CurrentInterval time.Duration
|
|
ConsecutiveFails int
|
|
}
|
|
|
|
// ASNFetcher is an interface for queuing ASN WHOIS lookups.
|
|
type ASNFetcher interface {
|
|
QueueImmediate(asn int)
|
|
GetStats() ASNFetcherStats
|
|
}
|
|
|
|
// dnsResolver looks up the addresses of host names and the names of addresses
|
|
// in DNS, as *net.Resolver does.
|
|
type dnsResolver interface {
|
|
LookupHost(ctx context.Context, host string) ([]string, error)
|
|
LookupAddr(ctx context.Context, addr string) ([]string, error)
|
|
}
|
|
|
|
// Server provides HTTP endpoints for status monitoring
|
|
type Server struct {
|
|
router *chi.Mux
|
|
db database.Store
|
|
streamer *streamer.Streamer
|
|
logger *logger.Logger
|
|
port int
|
|
srv *http.Server
|
|
asnFetcher ASNFetcher
|
|
|
|
// trustedProxies are the reverse proxies whose X-Forwarded-For header
|
|
// RealIPMiddleware believes
|
|
trustedProxies []netip.Prefix
|
|
|
|
// resolver makes the IP lookup's DNS lookups: net.DefaultResolver, or a
|
|
// fake in tests
|
|
resolver dnsResolver
|
|
}
|
|
|
|
// New creates a new HTTP server
|
|
func New(
|
|
db database.Store,
|
|
streamer *streamer.Streamer,
|
|
logger *logger.Logger,
|
|
cfg *config.Config,
|
|
) *Server {
|
|
s := &Server{
|
|
db: db,
|
|
streamer: streamer,
|
|
logger: logger,
|
|
port: cfg.Port,
|
|
trustedProxies: cfg.TrustedProxies,
|
|
resolver: net.DefaultResolver,
|
|
}
|
|
|
|
s.setupRoutes()
|
|
|
|
return s
|
|
}
|
|
|
|
// Start listens on the configured port and serves HTTP in the background.
|
|
// It returns an error if it cannot listen, for example because the port is
|
|
// already in use. ctx bounds only opening the port, not serving.
|
|
func (s *Server) Start(ctx context.Context) error {
|
|
const (
|
|
network = "tcp"
|
|
readHeaderTimeout = 40 * time.Second
|
|
readTimeout = 60 * time.Second
|
|
writeTimeout = 60 * time.Second
|
|
idleTimeout = 120 * time.Second
|
|
)
|
|
|
|
s.srv = &http.Server{
|
|
Addr: ":" + strconv.Itoa(s.port),
|
|
Handler: s.router,
|
|
ReadHeaderTimeout: readHeaderTimeout,
|
|
ReadTimeout: readTimeout,
|
|
WriteTimeout: writeTimeout,
|
|
IdleTimeout: idleTimeout,
|
|
}
|
|
|
|
s.logger.Info("Starting HTTP server", "port", s.port, "addr", s.srv.Addr)
|
|
|
|
var listenConfig net.ListenConfig
|
|
|
|
listener, err := listenConfig.Listen(ctx, network, s.srv.Addr)
|
|
if err != nil {
|
|
return fmt.Errorf("starting HTTP server: %w", err)
|
|
}
|
|
|
|
s.logger.Info("HTTP server listening", "addr", listener.Addr().String())
|
|
|
|
go func() {
|
|
err := s.srv.Serve(listener)
|
|
if err != nil && !errors.Is(err, http.ErrServerClosed) {
|
|
s.logger.Error("HTTP server error", "error", err)
|
|
}
|
|
}()
|
|
|
|
return nil
|
|
}
|
|
|
|
// Stop stops the HTTP server. It waits, until ctx ends, for the requests in
|
|
// progress to finish, and then closes the connections of those still running
|
|
// and returns ctx's error.
|
|
func (s *Server) Stop(ctx context.Context) error {
|
|
if s.srv == nil {
|
|
return nil
|
|
}
|
|
|
|
s.logger.Info("Stopping HTTP server")
|
|
|
|
err := s.srv.Shutdown(ctx)
|
|
if err != nil {
|
|
return errors.Join(err, s.srv.Close())
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// SetASNFetcher sets the ASN WHOIS fetcher for on-demand lookups. Call it
|
|
// before Start: the handlers read the fetcher without a lock.
|
|
func (s *Server) SetASNFetcher(fetcher ASNFetcher) {
|
|
s.asnFetcher = fetcher
|
|
}
|