diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index e38aa90..2bd6e95 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -58,6 +58,7 @@ var ( udpPacketSizeBytes observability.Histogram holePunchEventsTotal observability.Counter proxyMappingActive observability.UpDownCounter + relayUDPConnectionsActive observability.UpDownCounter sessionRebuiltTotal observability.Counter commPatternActive observability.UpDownCounter proxyCleanupRemovedTotal observability.Counter @@ -342,6 +343,11 @@ func createInstruments() error { if err != nil { return err } + relayUDPConnectionsActive, err = newUpDownCounter("gerbil_relay_udp_connections_active", + "Number of open per-peer outbound UDP sockets held by the relay connection pool", "ifname") + if err != nil { + return err + } sessionRebuiltTotal, err = newCounter("gerbil_session_rebuilt_total", "Count of sessions rebuilt from communication patterns", "ifname") if err != nil { @@ -711,6 +717,13 @@ func RecordSession(ifname string, delta int64) { activeSessions.Add(context.Background(), delta, observability.Labels{"ifname": ifname}) } +func RecordUDPConnection(ifname string, delta int64) { + if relayUDPConnectionsActive == nil { + return + } + relayUDPConnectionsActive.Add(context.Background(), delta, observability.Labels{"ifname": ifname}) +} + func RecordSessionRebuilt(ifname string) { if sessionRebuiltTotal == nil { return diff --git a/main.go b/main.go index 2a31c2b..0e8cca7 100644 --- a/main.go +++ b/main.go @@ -46,6 +46,7 @@ var ( doTrafficShaping bool bandwidthLimit string ifbName string // IFB device name for ingress traffic shaping + disableFirewall bool ) type WgConfig struct { @@ -196,6 +197,7 @@ func main() { proxyProtocolStr := os.Getenv("PROXY_PROTOCOL") doTrafficShapingStr := os.Getenv("DO_TRAFFIC_SHAPING") bandwidthLimitStr := os.Getenv("BANDWIDTH_LIMIT") + disableFirewallStr := os.Getenv("DISABLE_FIREWALL") // Read metrics env vars (defaults applied by DefaultMetricsConfig; these override defaults). metricsEnabled = true // default @@ -316,6 +318,13 @@ func main() { flag.BoolVar(&doTrafficShaping, "do-traffic-shaping", false, "Whether to set up traffic shaping rules for peers (requires tc command and root privileges)") } + if disableFirewallStr != "" { + disableFirewall = strings.ToLower(disableFirewallStr) == "true" + } + if disableFirewallStr == "" { + flag.BoolVar(&disableFirewall, "disable-firewall", false, "Disable WireGuard firewall rules to allow all inbound traffic on the interface") + } + if bandwidthLimitStr != "" { bandwidthLimit = bandwidthLimitStr } @@ -732,7 +741,9 @@ func ensureWireguardInterface(wgconfig WgConfig) error { logger.Warn("Failed to ensure MSS clamping: %v", err) } - if err := ensureWireguardFirewall(); err != nil { + if disableFirewall { + logger.Warn("Firewall disabled: all inbound traffic on %s will be allowed", interfaceName) + } else if err := ensureWireguardFirewall(); err != nil { logger.Warn("Failed to ensure WireGuard firewall rules: %v", err) } @@ -1493,6 +1504,24 @@ func calculatePeerBandwidth() ([]PeerBandwidth, error) { return peerBandwidths, nil } +// defaultBandwidthReportBatchSize caps how many peer bandwidth readings are +// sent in a single POST. Reporting every peer in one request grows unbounded +// with fleet size and can exceed the remote server's request body limit, +// which surfaces as "413 Payload Too Large" and silently drops that whole +// report cycle. Overridable via GERBIL_BANDWIDTH_BATCH_SIZE. +const defaultBandwidthReportBatchSize = 250 + +var bandwidthReportBatchSize = loadBandwidthReportBatchSize() + +func loadBandwidthReportBatchSize() int { + if v := os.Getenv("GERBIL_BANDWIDTH_BATCH_SIZE"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n + } + } + return defaultBandwidthReportBatchSize +} + func reportPeerBandwidth(apiURL string) error { bandwidths, err := calculatePeerBandwidth() if err != nil { @@ -1501,26 +1530,47 @@ func reportPeerBandwidth(apiURL string) error { return fmt.Errorf("failed to calculate peer bandwidth: %v", err) } - jsonData, err := json.Marshal(bandwidths) + if len(bandwidths) == 0 { + return nil + } + + var batchErrs []error + for start := 0; start < len(bandwidths); start += bandwidthReportBatchSize { + end := start + bandwidthReportBatchSize + if end > len(bandwidths) { + end = len(bandwidths) + } + + if err := sendPeerBandwidthBatch(apiURL, bandwidths[start:end]); err != nil { + metrics.RecordBandwidthReport("error") + batchErrs = append(batchErrs, err) + continue + } + metrics.RecordBandwidthReport("success") + } + + if len(batchErrs) > 0 { + return fmt.Errorf("failed to report %d bandwidth batch(es): %w", len(batchErrs), errors.Join(batchErrs...)) + } + return nil +} + +func sendPeerBandwidthBatch(apiURL string, batch []PeerBandwidth) error { + jsonData, err := json.Marshal(batch) if err != nil { - metrics.RecordBandwidthReport("error") return fmt.Errorf("failed to marshal bandwidth data: %v", err) } resp, err := http.Post(apiURL, "application/json", bytes.NewBuffer(jsonData)) if err != nil { - metrics.RecordBandwidthReport("error") return fmt.Errorf("failed to send bandwidth data: %v", err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - metrics.RecordBandwidthReport("error") return fmt.Errorf("API returned non-OK status: %s", resp.Status) } - // Record successful bandwidth report - metrics.RecordBandwidthReport("success") return nil } diff --git a/relay/relay.go b/relay/relay.go index 12b5e60..4d0d4df 100644 --- a/relay/relay.go +++ b/relay/relay.go @@ -10,8 +10,11 @@ import ( "io" "net" "net/http" + "os" "runtime" + "strconv" "sync" + "sync/atomic" "time" "github.com/fosrl/gerbil/internal/metrics" @@ -60,8 +63,31 @@ type PeerDestination struct { } type DestinationConn struct { - conn *net.UDPConn - lastUsed time.Time + conn *net.UDPConn + // lastUsed is unix nanoseconds, read/written via atomic ops since it's + // touched from packet workers and the response goroutine concurrently + // with no lock, and is also scanned for LRU eviction below. + lastUsed atomic.Int64 +} + +// defaultMaxUDPConnections caps the number of concurrent per-peer outbound +// UDP sockets the relay will keep open in s.connections. Without a cap, a +// burst of peer churn creates sockets faster than the 5-minute idle cleanup +// can reap them, exhausting the host's ephemeral port range or fd ulimit. +// That surfaces as "dial udp ...: resource temporarily unavailable" on every +// subsequent packet and pegs the CPU logging the flood (outage 2026-07-03, +// recurrence 2026-07-05). Overridable via GERBIL_MAX_UDP_CONNECTIONS. +const defaultMaxUDPConnections = 8192 + +var maxUDPConnections = loadMaxUDPConnections() + +func loadMaxUDPConnections() int64 { + if v := os.Getenv("GERBIL_MAX_UDP_CONNECTIONS"); v != "" { + if n, err := strconv.ParseInt(v, 10, 64); err == nil && n > 0 { + return n + } + } + return defaultMaxUDPConnections } // Type for storing WireGuard handshake information @@ -172,10 +198,13 @@ type UDPProxyServer struct { conn *net.UDPConn proxyMappings sync.Map // map[string]ProxyMapping where key is "ip:port" connections sync.Map // map[string]*DestinationConn where key is destination "ip:port" - privateKey wgtypes.Key - packetChan chan Packet - ctx context.Context - cancel context.CancelFunc + // connectionCount mirrors len(connections) without an O(n) sync.Map walk, + // so the cap check in getOrCreateConnection is cheap on the hot path. + connectionCount atomic.Int64 + privateKey wgtypes.Key + packetChan chan Packet + ctx context.Context + cancel context.CancelFunc // Session tracking for WireGuard peers // Key format: "senderIndex:receiverIndex" @@ -862,10 +891,17 @@ func (s *UDPProxyServer) getOrCreateConnection(destAddr *net.UDPAddr, remoteAddr // Check if we have an existing connection if conn, ok := s.connections.Load(key); ok { destConn := conn.(*DestinationConn) - destConn.lastUsed = time.Now() + destConn.lastUsed.Store(time.Now().UnixNano()) return destConn.conn, nil } + // Enforce a hard cap on concurrent sockets so a burst of peer churn can't + // exhaust the host's ephemeral ports/fds. Evict the least-recently-used + // connection to make room instead of growing unbounded. + if s.connectionCount.Load() >= maxUDPConnections { + s.evictLRUConnection() + } + // Create new connection newConn, err := net.DialUDP("udp", nil, destAddr) if err != nil { @@ -873,11 +909,17 @@ func (s *UDPProxyServer) getOrCreateConnection(destAddr *net.UDPAddr, remoteAddr return nil, fmt.Errorf("failed to create UDP connection: %v", err) } - // Store the new connection - s.connections.Store(key, &DestinationConn{ - conn: newConn, - lastUsed: time.Now(), - }) + destConn := &DestinationConn{conn: newConn} + destConn.lastUsed.Store(time.Now().UnixNano()) + + // Store the new connection. If another goroutine raced us and already + // created one for this key, close ours and use theirs instead. + if existing, loaded := s.connections.LoadOrStore(key, destConn); loaded { + newConn.Close() + return existing.(*DestinationConn).conn, nil + } + s.connectionCount.Add(1) + metrics.RecordUDPConnection(relayIfname, 1) // Start a goroutine to handle responses go s.handleResponses(newConn, destAddr, remoteAddr) @@ -885,6 +927,33 @@ func (s *UDPProxyServer) getOrCreateConnection(destAddr *net.UDPAddr, remoteAddr return newConn, nil } +// evictLRUConnection closes and removes the least-recently-used destination +// connection so a new one can be created under the concurrent connection cap. +func (s *UDPProxyServer) evictLRUConnection() { + var oldestKey interface{} + var oldestConn *DestinationConn + var oldestTime int64 + + s.connections.Range(func(key, value interface{}) bool { + destConn := value.(*DestinationConn) + lu := destConn.lastUsed.Load() + if oldestKey == nil || lu < oldestTime { + oldestKey = key + oldestConn = destConn + oldestTime = lu + } + return true + }) + + if oldestKey != nil { + s.connections.Delete(oldestKey) + oldestConn.conn.Close() + s.connectionCount.Add(-1) + metrics.RecordUDPConnection(relayIfname, -1) + metrics.RecordProxyCleanupRemoved(relayIfname, "conn_evicted", 1) + } +} + func (s *UDPProxyServer) handleResponses(conn *net.UDPConn, destAddr *net.UDPAddr, remoteAddr *net.UDPAddr) { buffer := make([]byte, 1500) for { @@ -936,22 +1005,32 @@ func (s *UDPProxyServer) handleResponses(conn *net.UDPConn, destAddr *net.UDPAdd // Add a cleanup method to periodically remove idle connections func (s *UDPProxyServer) cleanupIdleConnections() { - ticker := time.NewTicker(5 * time.Minute) + // Ticker interval and idle threshold were previously 5min/10min, meaning + // a socket could sit open for up to 15 minutes after going idle. Under a + // reconnect/churn burst that lag is enough to exhaust ephemeral ports + // before cleanup catches up, so both are tightened here. + ticker := time.NewTicker(1 * time.Minute) defer ticker.Stop() for { select { case <-ticker.C: cleanupStart := time.Now() - now := time.Now() + now := time.Now().UnixNano() + removed := int64(0) s.connections.Range(func(key, value interface{}) bool { destConn := value.(*DestinationConn) - if now.Sub(destConn.lastUsed) > 10*time.Minute { + if now-destConn.lastUsed.Load() > int64(5*time.Minute) { destConn.conn.Close() s.connections.Delete(key) - metrics.RecordProxyCleanupRemoved(relayIfname, "conn", 1) + removed++ } return true }) + if removed > 0 { + s.connectionCount.Add(-removed) + metrics.RecordUDPConnection(relayIfname, -removed) + metrics.RecordProxyCleanupRemoved(relayIfname, "conn", removed) + } metrics.RecordProxyIdleCleanupDuration(relayIfname, "conn", time.Since(cleanupStart).Seconds()) case <-s.ctx.Done(): return @@ -1113,6 +1192,10 @@ func (s *UDPProxyServer) clearConnectionsForWGIP(wgIP string) { for _, key := range keysToDelete { s.connections.Delete(key) } + if len(keysToDelete) > 0 { + s.connectionCount.Add(-int64(len(keysToDelete))) + metrics.RecordUDPConnection(relayIfname, -int64(len(keysToDelete))) + } logger.Info("Cleared %d connections for WG IP: %s", len(keysToDelete), wgIP) }