diff --git a/controller/rpc/keepalive/keepalive_test.go b/controller/rpc/keepalive/keepalive_test.go new file mode 100644 index 00000000..a6021830 --- /dev/null +++ b/controller/rpc/keepalive/keepalive_test.go @@ -0,0 +1,255 @@ +package keepalive_test + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "errors" + "fmt" + "io" + "math/big" + "net" + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/google/uuid" + "google.golang.org/grpc" + "google.golang.org/grpc/connectivity" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/keepalive" + "google.golang.org/grpc/metadata" + + "github.com/pasarguard/node/common" + "github.com/pasarguard/node/config" + "github.com/pasarguard/node/controller/rpc" + "github.com/pasarguard/node/pkg/tlsutil" +) + +func selfSignedPair(t *testing.T) (string, string) { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "keepalive-node"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + DNSNames: []string{"localhost"}, + IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}, + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + if err != nil { + t.Fatal(err) + } + keyDER, err := x509.MarshalECPrivateKey(key) + if err != nil { + t.Fatal(err) + } + dir := t.TempDir() + certPath := filepath.Join(dir, "cert.pem") + keyPath := filepath.Join(dir, "key.pem") + if err := os.WriteFile(certPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(keyPath, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), 0o600); err != nil { + t.Fatal(err) + } + return certPath, keyPath +} + +func unusedAddr(t *testing.T) string { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer l.Close() + return l.Addr().String() +} + +func startNode(t *testing.T) (addr, certPath string, key uuid.UUID) { + t.Helper() + certPath, keyPath := selfSignedPair(t) + tlsConfig, err := tlsutil.LoadTLSCredentials(certPath, keyPath) + if err != nil { + t.Fatal(err) + } + key = uuid.New() + cfg := config.NewTestConfig(t.TempDir(), key) + addr = unusedAddr(t) + shutdown, service, err := rpc.StartGRPCListener(tlsConfig, addr, cfg) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + service.Disconnect() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = shutdown(ctx) + }) + return addr, certPath, key +} + +func dialNode(t *testing.T, addr, certPath string, opts ...grpc.DialOption) *grpc.ClientConn { + t.Helper() + pool, err := tlsutil.LoadClientPool(certPath) + if err != nil { + t.Fatal(err) + } + creds := credentials.NewTLS(&tls.Config{RootCAs: pool, ServerName: "localhost"}) + conn, err := grpc.NewClient(addr, append([]grpc.DialOption{grpc.WithTransportCredentials(creds)}, opts...)...) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { conn.Close() }) + return conn +} + +func baseInfo(t *testing.T, conn *grpc.ClientConn, key uuid.UUID) { + t.Helper() + ctx, cancel := context.WithTimeout(metadata.NewOutgoingContext(context.Background(), metadata.Pairs("x-api-key", key.String())), 5*time.Second) + defer cancel() + if _, err := common.NewNodeServiceClient(conn).GetBaseInfo(ctx, &common.Empty{}); err != nil { + t.Fatalf("GetBaseInfo: %v", err) + } +} + +func TestNodeKeepsAnIdlePanelConnectionThatPingsEveryTenSeconds(t *testing.T) { + t.Parallel() + addr, certPath, key := startNode(t) + conn := dialNode(t, addr, certPath, grpc.WithKeepaliveParams(keepalive.ClientParameters{ + Time: 10 * time.Second, + Timeout: 5 * time.Second, + PermitWithoutStream: true, + })) + baseInfo(t, conn, key) + if state := conn.GetState(); state != connectivity.Ready { + t.Fatalf("connection not ready after the first call: %v", state) + } + + started := time.Now() + ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second) + defer cancel() + if conn.WaitForStateChange(ctx, connectivity.Ready) { + t.Fatalf("the node dropped an idle panel connection that pings every 10 s after %.0fs; state now %v", time.Since(started).Seconds(), conn.GetState()) + } +} + +type freezableProxy struct { + listener net.Listener + target string + frozen atomic.Bool + serverClosed chan struct{} + failed chan error + once sync.Once + failOnce sync.Once +} + +func startProxy(t *testing.T, target string) *freezableProxy { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + p := &freezableProxy{ + listener: l, + target: target, + serverClosed: make(chan struct{}), + failed: make(chan error, 1), + } + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(func() { + cancel() + l.Close() + }) + go p.serve(ctx) + return p +} + +func (p *freezableProxy) fail(err error) { + p.failOnce.Do(func() { p.failed <- err }) +} + +func (p *freezableProxy) serve(ctx context.Context) { + var dialer net.Dialer + for { + client, err := p.listener.Accept() + if err != nil { + return + } + server, err := dialer.DialContext(ctx, "tcp", p.target) + if err != nil { + p.fail(fmt.Errorf("the proxy could not reach the node at %s: %w", p.target, err)) + client.Close() + continue + } + go func() { + buf := make([]byte, 32*1024) + for { + n, err := client.Read(buf) + if n > 0 && !p.frozen.Load() { + if _, werr := server.Write(buf[:n]); werr != nil { + return + } + } + if err != nil { + return + } + } + }() + go func() { + buf := make([]byte, 32*1024) + for { + n, err := server.Read(buf) + if n > 0 && !p.frozen.Load() { + _, _ = client.Write(buf[:n]) + } + if err != nil { + if errors.Is(err, io.EOF) || p.frozen.Load() { + p.once.Do(func() { close(p.serverClosed) }) + } else { + p.fail(fmt.Errorf("the proxy lost its connection to the node before the freeze: %w", err)) + } + client.Close() + return + } + } + }() + } +} + +func TestNodeClosesAConnectionWhosePanelStoppedAnswering(t *testing.T) { + t.Parallel() + addr, certPath, key := startNode(t) + proxy := startProxy(t, addr) + conn := dialNode(t, proxy.listener.Addr().String(), certPath) + baseInfo(t, conn, key) + + proxy.frozen.Store(true) + started := time.Now() + select { + case <-proxy.serverClosed: + elapsed := time.Since(started) + if elapsed < 30*time.Second { + t.Fatalf("the node closed a merely idle connection after %.0fs, before its keepalive ping was due", elapsed.Seconds()) + } + t.Logf("the node closed the connection %.1fs after the panel stopped answering", elapsed.Seconds()) + case err := <-proxy.failed: + t.Fatalf("the proxy's own link to the node broke, so this run proves nothing about the node: %v", err) + case <-time.After(70 * time.Second): + t.Fatal("the node kept a connection to a panel that stopped answering for 70 s; it must ping an idle panel and close the connection when the ping is not acknowledged") + } +} diff --git a/controller/rpc/keepalive/tcp_user_timeout_linux_test.go b/controller/rpc/keepalive/tcp_user_timeout_linux_test.go new file mode 100644 index 00000000..3d067563 --- /dev/null +++ b/controller/rpc/keepalive/tcp_user_timeout_linux_test.go @@ -0,0 +1,77 @@ +//go:build linux + +package keepalive_test + +import ( + "context" + "net" + "os" + "strconv" + "sync/atomic" + "testing" + "time" + + "golang.org/x/sys/unix" + "google.golang.org/grpc" +) + +func sockaddrIs(sa unix.Sockaddr, want *net.TCPAddr) bool { + in4, ok := sa.(*unix.SockaddrInet4) + if !ok { + return false + } + return in4.Port == want.Port && net.IP(in4.Addr[:]).Equal(want.IP) +} + +func acceptedSocketUserTimeout(t *testing.T, node, client *net.TCPAddr) time.Duration { + t.Helper() + entries, err := os.ReadDir("/proc/self/fd") + if err != nil { + t.Fatal(err) + } + for _, entry := range entries { + fd, err := strconv.Atoi(entry.Name()) + if err != nil { + continue + } + local, err := unix.Getsockname(fd) + if err != nil || !sockaddrIs(local, node) { + continue + } + peer, err := unix.Getpeername(fd) + if err != nil || !sockaddrIs(peer, client) { + continue + } + ms, err := unix.GetsockoptInt(fd, unix.IPPROTO_TCP, unix.TCP_USER_TIMEOUT) + if err != nil { + t.Fatalf("getsockopt TCP_USER_TIMEOUT: %v", err) + } + return time.Duration(ms) * time.Millisecond + } + t.Fatalf("no socket in this process was accepted on %v from %v", node, client) + return 0 +} + +func TestNodeKeepsGrpcDefaultTCPUserTimeoutOnPanelConnections(t *testing.T) { + t.Parallel() + addr, certPath, key := startNode(t) + var clientAddr atomic.Pointer[net.TCPAddr] + conn := dialNode(t, addr, certPath, grpc.WithContextDialer(func(ctx context.Context, target string) (net.Conn, error) { + c, err := (&net.Dialer{}).DialContext(ctx, "tcp", target) + if err == nil { + clientAddr.Store(c.LocalAddr().(*net.TCPAddr)) + } + return c, err + })) + baseInfo(t, conn, key) + + nodeAddr, err := net.ResolveTCPAddr("tcp", addr) + if err != nil { + t.Fatal(err) + } + got := acceptedSocketUserTimeout(t, nodeAddr, clientAddr.Load()) + t.Logf("TCP_USER_TIMEOUT on the accepted panel connection: %s", got) + if got != 20*time.Second { + t.Fatalf("TCP_USER_TIMEOUT on the accepted panel connection is %s; grpc-go's default keepalive Timeout gives 20s, and a lower value aborts a stalling slow link sooner", got) + } +} diff --git a/controller/rpc/service.go b/controller/rpc/service.go index d633c412..fef923e9 100644 --- a/controller/rpc/service.go +++ b/controller/rpc/service.go @@ -6,12 +6,19 @@ import ( "fmt" "log" "net" + "time" "github.com/pasarguard/node/common" "github.com/pasarguard/node/config" "github.com/pasarguard/node/controller" "google.golang.org/grpc" "google.golang.org/grpc/credentials" + "google.golang.org/grpc/keepalive" +) + +var ( + panelKeepaliveEnforcement = keepalive.EnforcementPolicy{MinTime: 5 * time.Second, PermitWithoutStream: true} + panelKeepalive = keepalive.ServerParameters{Time: 30 * time.Second, Timeout: 20 * time.Second} ) type Service struct { @@ -37,6 +44,8 @@ func StartGRPCListener(tlsConfig *tls.Config, addr string, cfg *config.Config) ( grpc.Creds(creds), grpc.MaxRecvMsgSize(maxMsgSize), grpc.MaxSendMsgSize(maxMsgSize), + grpc.KeepaliveEnforcementPolicy(panelKeepaliveEnforcement), + grpc.KeepaliveParams(panelKeepalive), grpc.UnaryInterceptor(ConditionalMiddleware(s)), grpc.StreamInterceptor(ConditionalStreamMiddleware(s)), )