diff --git a/cache_helpers.go b/cache_helpers.go index c5c812b..33aeebb 100644 --- a/cache_helpers.go +++ b/cache_helpers.go @@ -786,6 +786,13 @@ func cacheDump(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/octet-stream") w.Header().Set("Content-Disposition", "attachment; filename=cache.db") + // The server sets a 60s WriteTimeout to bound normal request lifetimes, but + // this streams the whole multi-GB BoltDB and can legitimately run longer. + // Clear the write deadline for this response so a slow consumer isn't cut off. + if err := http.NewResponseController(w).SetWriteDeadline(time.Time{}); err != nil { + log.Warnf("%s Could not clear write deadline for cache dump: %v", logcolors.LogCache, err) + } + n, err := persistentCache.WriteTo(w) if err != nil { log.Errorf("%s Failed to stream cache dump: %v", logcolors.LogCache, err) diff --git a/fd_monitor.go b/fd_monitor.go new file mode 100644 index 0000000..f0ff8ba --- /dev/null +++ b/fd_monitor.go @@ -0,0 +1,171 @@ +package main + +import ( + "bufio" + "io" + "lyrics-api-go/logcolors" + "lyrics-api-go/services/notifier" + "os" + "runtime" + "strings" + "sync/atomic" + "syscall" + "time" + + log "github.com/sirupsen/logrus" +) + +const ( + fdMonitorInterval = 2 * time.Minute + fdAlertThresholdRatio = 0.8 // alert once open fds reach 80% of the limit +) + +var fdAlertNotified atomic.Bool + +// startFDMonitor launches a background goroutine that watches the process's +// open file-descriptor count against RLIMIT_NOFILE. +// +// This exists because a leak of idle/stuck connections once exhausted the fd +// limit and took the API fully down with "accept: too many open files". A slow +// climb like that is invisible until it hits the ceiling, so this fires a +// one-time notification at 80% (with a socket-state breakdown) while there is +// still headroom to react. +func startFDMonitor() { + limit := getFDLimit() + if limit == 0 { + log.Warnf("%s FD monitor disabled: could not read RLIMIT_NOFILE", logcolors.LogMemory) + return + } + + go func() { + for { + open := getOpenFDCount() + if open < 0 { + // /proc/self/fd unavailable (non-Linux dev host): nothing to watch. + time.Sleep(fdMonitorInterval) + continue + } + + pct := float64(open) / float64(limit) * 100 + + if fdThresholdExceeded(open, limit) { + states := getSocketStates() + log.Warnf("%s FD usage high: %d/%d (%.0f%%) | sockets: %v | goroutines: %d", + logcolors.LogMemoryAlert, open, limit, pct, states, runtime.NumGoroutine()) + + if fdAlertNotified.CompareAndSwap(false, true) { + notifier.PublishFDThresholdExceeded(open, limit, map[string]interface{}{ + "percent": int(pct), + "socket_states": states, + "goroutines": runtime.NumGoroutine(), + }) + } + } else { + fdAlertNotified.Store(false) + log.Infof("%s FDs: %d/%d (%.0f%%)", logcolors.LogMemory, open, limit, pct) + } + + time.Sleep(fdMonitorInterval) + } + }() + + log.Infof("%s FD monitor started (limit: %d, alert: %.0f%%, interval: %v)", + logcolors.LogMemory, limit, fdAlertThresholdRatio*100, fdMonitorInterval) +} + +// getOpenFDCount returns the number of open file descriptors for this process, +// or -1 when /proc is unavailable (e.g. macOS development). +func getOpenFDCount() int { + return countDirEntries("/proc/self/fd") +} + +// countDirEntries returns the number of entries in dir, or -1 on error. +func countDirEntries(dir string) int { + entries, err := os.ReadDir(dir) + if err != nil { + return -1 + } + return len(entries) +} + +// getFDLimit returns the soft RLIMIT_NOFILE for this process, or 0 on error. +func getFDLimit() uint64 { + var rl syscall.Rlimit + if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &rl); err != nil { + return 0 + } + return rl.Cur +} + +// fdThresholdExceeded reports whether open fds have reached the alert ratio of limit. +func fdThresholdExceeded(open int, limit uint64) bool { + if open < 0 || limit == 0 { + return false + } + return float64(open) >= float64(limit)*fdAlertThresholdRatio +} + +// getSocketStates returns a count of TCP sockets by connection state, merged +// across IPv4 and IPv6. Empty when /proc is unavailable. +// +// Note: /proc/self/net/tcp reflects the whole network namespace, not only fds +// this process holds. In production (one process per container/namespace) that +// is effectively per-process; on a shared-netns host the counts include peers. +func getSocketStates() map[string]int { + merged := make(map[string]int) + for _, path := range []string{"/proc/self/net/tcp", "/proc/self/net/tcp6"} { + f, err := os.Open(path) + if err != nil { + continue + } + for state, n := range parseSocketStates(f) { + merged[state] += n + } + f.Close() + } + return merged +} + +// parseSocketStates parses the /proc/net/tcp format and counts rows by TCP state. +func parseSocketStates(r io.Reader) map[string]int { + states := make(map[string]int) + scanner := bufio.NewScanner(r) + for scanner.Scan() { + fields := strings.Fields(scanner.Text()) + if len(fields) < 4 || fields[0] == "sl" { + continue // too short, or the header row + } + states[tcpStateName(fields[3])]++ + } + return states +} + +// tcpStateName maps the hex state from /proc/net/tcp to its human name. +func tcpStateName(hex string) string { + switch strings.ToUpper(hex) { + case "01": + return "ESTABLISHED" + case "02": + return "SYN_SENT" + case "03": + return "SYN_RECV" + case "04": + return "FIN_WAIT1" + case "05": + return "FIN_WAIT2" + case "06": + return "TIME_WAIT" + case "07": + return "CLOSE" + case "08": + return "CLOSE_WAIT" + case "09": + return "LAST_ACK" + case "0A": + return "LISTEN" + case "0B": + return "CLOSING" + default: + return "STATE_" + hex + } +} diff --git a/fd_monitor_test.go b/fd_monitor_test.go new file mode 100644 index 0000000..1ca4da4 --- /dev/null +++ b/fd_monitor_test.go @@ -0,0 +1,73 @@ +package main + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +func TestCountDirEntries(t *testing.T) { + dir := t.TempDir() + for _, name := range []string{"a", "b", "c"} { + if err := os.WriteFile(filepath.Join(dir, name), []byte("x"), 0644); err != nil { + t.Fatal(err) + } + } + if got := countDirEntries(dir); got != 3 { + t.Errorf("countDirEntries = %d, want 3", got) + } + if got := countDirEntries(filepath.Join(dir, "does-not-exist")); got != -1 { + t.Errorf("countDirEntries(missing) = %d, want -1", got) + } +} + +func TestFDThresholdExceeded(t *testing.T) { + // fdAlertThresholdRatio is 0.8; 65536 * 0.8 = 52428.8 + tests := []struct { + name string + open int + limit uint64 + want bool + }{ + {"at threshold", 52429, 65536, true}, + {"just below threshold", 52428, 65536, false}, + {"well under", 100, 65536, false}, + {"maxed out", 65536, 65536, true}, + {"unknown limit", 1000, 0, false}, + {"unknown open count", -1, 65536, false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := fdThresholdExceeded(tc.open, tc.limit); got != tc.want { + t.Errorf("fdThresholdExceeded(%d, %d) = %v, want %v", tc.open, tc.limit, got, tc.want) + } + }) + } +} + +func TestParseSocketStates(t *testing.T) { + // Minimal /proc/net/tcp sample: header + LISTEN, ESTABLISHED, ESTABLISHED, CLOSE_WAIT + sample := ` sl local_address rem_address st tx_queue rx_queue tr tm->when retrnsmt uid timeout inode + 0: 0100007F:1F90 00000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 12345 1 0000 100 0 0 10 0 + 1: 0100007F:1F90 0100007F:C1B2 01 00000000:00000000 00:00000000 00000000 1000 0 12346 1 0000 20 4 30 10 -1 + 2: 0100007F:1F90 0100007F:C1B3 01 00000000:00000000 00:00000000 00000000 1000 0 12347 1 0000 20 4 30 10 -1 + 3: 0100007F:1F90 0100007F:C1B4 08 00000000:00000000 00:00000000 00000000 1000 0 12348 1 0000 20 4 30 10 -1 +` + got := parseSocketStates(strings.NewReader(sample)) + if got["ESTABLISHED"] != 2 { + t.Errorf("ESTABLISHED = %d, want 2", got["ESTABLISHED"]) + } + if got["LISTEN"] != 1 { + t.Errorf("LISTEN = %d, want 1", got["LISTEN"]) + } + if got["CLOSE_WAIT"] != 1 { + t.Errorf("CLOSE_WAIT = %d, want 1", got["CLOSE_WAIT"]) + } +} + +func TestGetFDLimit(t *testing.T) { + if got := getFDLimit(); got == 0 { + t.Error("getFDLimit() returned 0, expected the process RLIMIT_NOFILE soft limit") + } +} diff --git a/main.go b/main.go index aeba8b9..a00e7b0 100644 --- a/main.go +++ b/main.go @@ -8,7 +8,6 @@ import ( "lyrics-api-go/services/notifier" "lyrics-api-go/services/providers/ttml" "lyrics-api-go/stats" - "net/http" "os" "sync" "time" @@ -111,6 +110,9 @@ func main() { // Start memory monitor (logs RSS, alerts at threshold) startMemoryMonitor(cachePath) + // Start file-descriptor monitor (logs open fds, alerts before the limit is hit) + startFDMonitor() + router := mux.NewRouter() setupRoutes(router) @@ -181,5 +183,6 @@ func main() { // Publish server started event notifier.PublishServerStarted(port, len(activeAccounts), outOfServiceNames) - log.Fatal(http.ListenAndServe(":"+port, handler)) + srv := newHTTPServer(":"+port, handler) + log.Fatal(srv.ListenAndServe()) } diff --git a/services/notifier/alerts.go b/services/notifier/alerts.go index adabaa2..d95020f 100644 --- a/services/notifier/alerts.go +++ b/services/notifier/alerts.go @@ -145,6 +145,18 @@ func (h *AlertHandler) formatAlert(event *Event) (subject, message string) { } message += "\nThe process may be OOM-killed soon. Check Railway metrics." + case EventFDThresholdExceeded: + openFDs := event.Data["open_fds"].(int) + limitFDs := event.Data["limit_fds"].(uint64) + subject = "File Descriptor Threshold Exceeded" + message = fmt.Sprintf("Open file descriptors have reached %d of %d.\n\n", openFDs, limitFDs) + if details, ok := event.Data["details"].(map[string]interface{}); ok { + for k, v := range details { + message += fmt.Sprintf(" • %s: %v\n", k, v) + } + } + message += "\nA connection or fd leak takes the API fully down when it hits the limit. Investigate now." + case EventServerStartupFailed: component := event.Data["component"].(string) errMsg := event.Data["error"].(string) diff --git a/services/notifier/alerts_test.go b/services/notifier/alerts_test.go new file mode 100644 index 0000000..754031d --- /dev/null +++ b/services/notifier/alerts_test.go @@ -0,0 +1,27 @@ +package notifier + +import ( + "strings" + "testing" +) + +// TestFormatAlertFDThresholdExceeded guards delivery of the fd-exhaustion alert. +// formatAlert returns ("", "") for event types it does not handle, and +// handleEvent drops those silently. If the EventFDThresholdExceeded case is +// missing, the alert is published but never reaches any notifier. +func TestFormatAlertFDThresholdExceeded(t *testing.T) { + h := NewAlertHandler(AlertConfig{}) + ev := NewEvent(EventFDThresholdExceeded, SeverityCritical, "test"). + WithData("open_fds", 52429). + WithData("limit_fds", uint64(65536)). + WithData("details", map[string]interface{}{"percent": 80, "socket_states": map[string]int{"CLOSE_WAIT": 40000}}) + + subject, message := h.formatAlert(ev) + + if subject == "" { + t.Fatal("EventFDThresholdExceeded produced empty subject: it would be dropped and never delivered") + } + if !strings.Contains(message, "52429") || !strings.Contains(message, "65536") { + t.Errorf("alert message missing fd counts, got: %q", message) + } +} diff --git a/services/notifier/events.go b/services/notifier/events.go index b0590c6..e41f481 100644 --- a/services/notifier/events.go +++ b/services/notifier/events.go @@ -17,6 +17,7 @@ const ( EventMUTHealthCheckFailed EventType = "mut_health_check_failed" EventMemoryThresholdExceeded EventType = "memory_threshold_exceeded" + EventFDThresholdExceeded EventType = "fd_threshold_exceeded" // Warning events EventHighFailureRate EventType = "high_failure_rate" @@ -235,6 +236,16 @@ func PublishMemoryThresholdExceeded(rssMB uint64, details map[string]interface{} GetEventBus().Publish(event) } +// PublishFDThresholdExceeded publishes when open file descriptors approach the process limit +func PublishFDThresholdExceeded(openFDs int, limitFDs uint64, details map[string]interface{}) { + event := NewEvent(EventFDThresholdExceeded, SeverityCritical, + "Open file descriptors approaching the process limit"). + WithData("open_fds", openFDs). + WithData("limit_fds", limitFDs). + WithData("details", details) + GetEventBus().Publish(event) +} + // PublishMUTHealthCheckFailed publishes when MUT health check detects unhealthy accounts func PublishMUTHealthCheckFailed(unhealthyAccounts interface{}) { event := NewEvent(EventMUTHealthCheckFailed, SeverityCritical, diff --git a/startup.go b/startup.go index cd29350..e83d071 100644 --- a/startup.go +++ b/startup.go @@ -9,10 +9,29 @@ import ( "lyrics-api-go/stats" "net/http" "os" + "time" log "github.com/sirupsen/logrus" ) +// newHTTPServer builds the API's HTTP server with explicit timeouts. +// +// http.ListenAndServe uses a zero-value http.Server with no timeouts, so idle +// or stuck keep-alive connections (dead peers, slowloris, clients that stop +// reading) are never reaped. On a long-lived, public-facing process they leak +// file descriptors until accept() fails with "too many open files" and the +// server stops answering. These timeouts bound every connection's lifetime. +func newHTTPServer(addr string, handler http.Handler) *http.Server { + return &http.Server{ + Addr: addr, + Handler: handler, + ReadHeaderTimeout: 10 * time.Second, // slowloris guard: header must arrive quickly + ReadTimeout: 30 * time.Second, // whole request (headers + body) + WriteTimeout: 60 * time.Second, // response write ceiling (lyrics JSON, not streaming) + IdleTimeout: 90 * time.Second, // reap idle keep-alive connections + } +} + func getEnvOrDefault(key, defaultValue string) string { if value := os.Getenv(key); value != "" { return value diff --git a/startup_test.go b/startup_test.go new file mode 100644 index 0000000..622614b --- /dev/null +++ b/startup_test.go @@ -0,0 +1,83 @@ +package main + +import ( + "bufio" + "fmt" + "io" + "net" + "net/http" + "testing" + "time" +) + +// TestNewHTTPServerSetsTimeouts guards against the fd-exhaustion outage: +// a server built with http.ListenAndServe has zero timeouts, so idle or +// stuck keep-alive connections are never reaped and accumulate as open +// file descriptors until accept() fails with EMFILE. Every timeout below +// must stay non-zero. +func TestNewHTTPServerSetsTimeouts(t *testing.T) { + srv := newHTTPServer(":8080", http.NewServeMux()) + + if srv.Addr != ":8080" { + t.Errorf("Addr = %q, want %q", srv.Addr, ":8080") + } + if srv.Handler == nil { + t.Error("Handler must not be nil") + } + + timeouts := []struct { + name string + got time.Duration + }{ + {"ReadHeaderTimeout", srv.ReadHeaderTimeout}, + {"ReadTimeout", srv.ReadTimeout}, + {"WriteTimeout", srv.WriteTimeout}, + {"IdleTimeout", srv.IdleTimeout}, + } + for _, tc := range timeouts { + if tc.got <= 0 { + t.Errorf("%s = %v, must be > 0 to prevent connection/fd leak", tc.name, tc.got) + } + } +} + +// TestNewHTTPServerReapsIdleConnections reproduces the leak scenario end to end: +// an idle keep-alive connection must be closed by the server. With IdleTimeout +// unset (the old bug), the read below would block until the test deadline +// instead of returning EOF. +func TestNewHTTPServerReapsIdleConnections(t *testing.T) { + srv := newHTTPServer("127.0.0.1:0", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusNoContent) + })) + srv.IdleTimeout = 150 * time.Millisecond // shorten so the test is fast + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + go srv.Serve(ln) + defer srv.Close() + + conn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + + if _, err := fmt.Fprintf(conn, "GET / HTTP/1.1\r\nHost: test\r\nConnection: keep-alive\r\n\r\n"); err != nil { + t.Fatalf("write request: %v", err) + } + + br := bufio.NewReader(conn) + resp, err := http.ReadResponse(br, nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + resp.Body.Close() + + // Connection is now idle. The server must close it after IdleTimeout. + conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + if _, err := br.ReadByte(); err != io.EOF { + t.Fatalf("expected server to close idle connection (EOF), got: %v", err) + } +}