fix(ingressd): harden accept loop and report downstream health

cmd/ingressd/main.go

@@ -1,13 +1,15 @@
 package main
 
 import (
+	"errors"
 	"flag"
 	"fmt"
 	"io"
 	"log"
 	"log/slog"
 	"net"
 	"sync"
+	"time"
 
 	proxyproto "github.com/pires/go-proxyproto"
 	"google.golang.org/grpc"
@@ -36,72 +38,117 @@ func main() {
 
 	grpcLn, err := net.Listen("tcp", fmt.Sprintf(":%d", *grpcPort))
 	if err != nil {
-		log.Fatalf("can't listen to gRPC port %d: %v", *httpPort, err)
-		return
+		log.Fatalf("can't listen to gRPC port %d: %v", *grpcPort, err)
 	}
 
 	httpLn, err := net.Listen("tcp", fmt.Sprintf(":%d", *httpPort))
 	if err != nil {
 		log.Fatalf("can't listen to HTTP port %d: %v", *httpPort, err)
-		return
 	}
 
 	httpsLn, err := net.Listen("tcp", fmt.Sprintf(":%d", *httpsPort))
 	if err != nil {
 		log.Fatalf("can't listen to HTTPS port %d: %v", *httpsPort, err)
-		return
 	}
 
 	srv := grpc.NewServer()
 	healthpb.RegisterHealthServer(srv, healthSrv)
 	reflection.Register(srv)
 
-	go s.Handle(httpsLn, *httpsTarget, "https")
-	go s.Handle(httpLn, *httpTarget, "http")
-
 	healthSrv.SetServingStatus("grpc", healthpb.HealthCheckResponse_SERVING)
+	healthSrv.SetServingStatus("http", healthpb.HealthCheckResponse_NOT_SERVING)
+	healthSrv.SetServingStatus("https", healthpb.HealthCheckResponse_NOT_SERVING)
+
+	errCh := make(chan error, 3)
+	go func() { errCh <- s.Handle(httpsLn, *httpsTarget, "https") }()
+	go func() { errCh <- s.Handle(httpLn, *httpTarget, "http") }()
+	go func() { errCh <- srv.Serve(grpcLn) }()
+
+	go s.Probe(*httpTarget, "http")
+	go s.Probe(*httpsTarget, "https")
 
-	log.Fatal(srv.Serve(grpcLn))
+	if err := <-errCh; err != nil {
+		log.Fatal(err)
+	}
 }
 
 type Server struct{}
 
-func (s *Server) Handle(l net.Listener, dest, service string) {
-	defer l.Close()
+const (
+	probeInterval = 10 * time.Second
+	probeTimeout  = 3 * time.Second
+)
 
-	healthSrv.SetServingStatus(service, healthpb.HealthCheckResponse_SERVING)
+func (s *Server) Handle(l net.Listener, dest, service string) error {
+	defer l.Close()
 
+	backoff := 100 * time.Millisecond
 	for {
 		conn, err := l.Accept()
 		if err != nil {
-			slog.Error("can't accept connection", "localAddr", l.Addr().String(), "err", err)
-			return
+			if errors.Is(err, net.ErrClosed) {
+				return err
+			}
+
+			slog.Error("can't accept connection; retrying",
+				"localAddr", l.Addr().String(),
+				"service", service,
+				"err", err,
+				"retryIn", backoff)
+			time.Sleep(backoff)
+			backoff = min(backoff*2, 5*time.Second)
+			continue
 		}
+		backoff = 100 * time.Millisecond
 
 		slog.Debug("accepted connection", "service", service, "addr", conn.RemoteAddr().String())
+		go s.HandleConn(conn, dest)
+	}
+}
+
+// Probe dials dest every probeInterval and mirrors its reachability into the
+// gRPC health status for service. It logs only on status transitions.
+func (s *Server) Probe(dest, service string) {
+	ticker := time.NewTicker(probeInterval)
+	defer ticker.Stop()
+
+	status := healthpb.HealthCheckResponse_NOT_SERVING
+	for {
+		conn, err := net.DialTimeout("tcp", dest, probeTimeout)
+		if err != nil {
+			if status == healthpb.HealthCheckResponse_SERVING {
+				slog.Error("downstream probe failed", "service", service, "dest", dest, "err", err)
+			}
+			status = healthpb.HealthCheckResponse_NOT_SERVING
+		} else {
+			conn.Close()
+			if status != healthpb.HealthCheckResponse_SERVING {
+				slog.Info("downstream probe recovered", "service", service, "dest", dest)
+			}
+			status = healthpb.HealthCheckResponse_SERVING
+		}
+		healthSrv.SetServingStatus(service, status)
 
-		go s.HandleConn(conn, dest, service)
+		<-ticker.C
 	}
 }
 
-func (s *Server) HandleConn(conn net.Conn, dest, service string) {
+func (s *Server) HandleConn(conn net.Conn, dest string) {
 	defer conn.Close()
 
 	slog.Debug("dialing remote host", "remoteHost", dest)
 
 	destConn, err := net.Dial("tcp", dest)
 	if err != nil {
-		healthSrv.SetServingStatus(service, healthpb.HealthCheckResponse_SERVICE_UNKNOWN)
 		slog.Error("can't dial downstream", "err", err)
 		return
 	}
 	defer destConn.Close()
 
 	slog.Debug("dialed remote host", "remoteAddr", conn.RemoteAddr().String(), "destAddr", destConn.RemoteAddr().String())
 
-	header := proxyproto.HeaderProxyFromAddrs(2, conn.RemoteAddr(), destConn.RemoteAddr())
+	header := proxyproto.HeaderProxyFromAddrs(2, conn.RemoteAddr(), conn.LocalAddr())
 	if _, err := header.WriteTo(destConn); err != nil {
-		healthSrv.SetServingStatus(service, healthpb.HealthCheckResponse_SERVICE_UNKNOWN)
 		slog.Error("can't write haproxy header to downstream", "header", header, "err", err)
 		return
 	}
@@ -116,18 +163,30 @@ func (s *Server) HandleConn(conn net.Conn, dest, service string) {
 		defer wg.Done()
 		io.Copy(conn, destConn)
 		// Signal peer that no more data is coming.
-		conn.(*net.TCPConn).CloseWrite()
+		closeWrite(conn)
 		slog.Debug("done copying from dest to src")
 	}()
 	go func() {
 		slog.Debug("copying from src to dest")
 		defer wg.Done()
 		io.Copy(destConn, conn)
 		// Signal peer that no more data is coming.
-		destConn.(*net.TCPConn).CloseWrite()
+		closeWrite(destConn)
 		slog.Debug("done copying from src to dest")
 	}()
 
 	wg.Wait()
 	slog.Debug("done")
 }
+
+// closeWrite half-closes conn when it supports it. net.TCPConn does; test
+// doubles and future wrappers may not.
+func closeWrite(conn net.Conn) {
+	cw, ok := conn.(interface{ CloseWrite() error })
+	if !ok {
+		return
+	}
+	if err := cw.CloseWrite(); err != nil {
+		slog.Debug("can't half-close connection", "err", err)
+	}
+}

cmd/ingressd/main_test.go

@@ -0,0 +1,69 @@
+package main
+
+import (
+	"errors"
+	"net"
+	"sync"
+	"testing"
+)
+
+// scriptedListener returns a transient Accept error, then conn, then
+// net.ErrClosed.
+type scriptedListener struct {
+	mu      sync.Mutex
+	conn    net.Conn
+	accepts int
+}
+
+func (l *scriptedListener) Accept() (net.Conn, error) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	l.accepts++
+	switch l.accepts {
+	case 1:
+		return nil, errors.New("accept: too many open files")
+	case 2:
+		conn := l.conn
+		l.conn = nil
+		return conn, nil
+	default:
+		return nil, net.ErrClosed
+	}
+}
+
+func (l *scriptedListener) Close() error   { return nil }
+func (l *scriptedListener) Addr() net.Addr { return &net.TCPAddr{IP: net.IPv4zero} }
+
+func TestHandleRetriesAcceptErrors(t *testing.T) {
+	client, server := net.Pipe()
+	defer client.Close()
+
+	l := &scriptedListener{conn: server}
+	s := &Server{}
+
+	errCh := make(chan error, 1)
+	// "invalid" makes net.Dial fail immediately during address parsing, so the
+	// handler goroutine cannot linger on a real connection attempt.
+	go func() { errCh <- s.Handle(l, "invalid", "http") }()
+
+	err := <-errCh
+	if !errors.Is(err, net.ErrClosed) {
+		t.Fatalf("Handle returned %v, want net.ErrClosed", err)
+	}
+
+	l.mu.Lock()
+	accepts := l.accepts
+	l.mu.Unlock()
+	if accepts != 3 {
+		t.Fatalf("Accept called %d times, want 3", accepts)
+	}
+}
+
+func TestCloseWriteNonTCPConn(t *testing.T) {
+	a, b := net.Pipe()
+	defer a.Close()
+	defer b.Close()
+
+	closeWrite(a)
+}