diff --git a/cmd/downloader/main.go b/cmd/downloader/main.go index 2698afa2436..dbea52a28b3 100644 --- a/cmd/downloader/main.go +++ b/cmd/downloader/main.go @@ -22,29 +22,22 @@ import ( "errors" "fmt" "io/fs" - "net" "net/url" "os" "path/filepath" "runtime" "slices" "strings" - "time" g "github.com/anacrolix/generics" "github.com/anacrolix/missinggo/v2/panicif" "github.com/anacrolix/torrent/metainfo" "github.com/go-viper/mapstructure/v2" - "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/recovery" "github.com/pelletier/go-toml/v2" "github.com/spf13/cobra" "github.com/urfave/cli/v2" "google.golang.org/grpc" "google.golang.org/grpc/credentials" - "google.golang.org/grpc/health" - "google.golang.org/grpc/health/grpc_health_v1" - "google.golang.org/grpc/keepalive" - "google.golang.org/grpc/reflection" "github.com/erigontech/erigon/cmd/utils/app" "github.com/erigontech/erigon/db/downloader/webseeds" @@ -67,6 +60,7 @@ import ( chainspec "github.com/erigontech/erigon/execution/chain/spec" "github.com/erigontech/erigon/node/debug" "github.com/erigontech/erigon/node/gointerfaces/downloaderproto" + "github.com/erigontech/erigon/node/gointerfaces/grpcutil" "github.com/erigontech/erigon/node/logging" "github.com/erigontech/erigon/node/paths" "github.com/erigontech/erigon/p2p/nat" @@ -761,57 +755,15 @@ func doDiffTorrentHashes(ctx context.Context, local map[string]string) error { return nil } -func StartGrpc(snServer *downloader.GrpcServer, addr string, creds *credentials.TransportCredentials, logger log.Logger) (*grpc.Server, error) { - lis, err := net.Listen("tcp", addr) - if err != nil { - return nil, fmt.Errorf("could not create listener: %w, addr=%s", err, addr) - } - - var ( - streamInterceptors = make([]grpc.StreamServerInterceptor, 0, 1) - unaryInterceptors = make([]grpc.UnaryServerInterceptor, 0, 1) - ) - streamInterceptors = append(streamInterceptors, recovery.StreamServerInterceptor()) - unaryInterceptors = append(unaryInterceptors, recovery.UnaryServerInterceptor()) - - //if metrics.Enabled { - // streamInterceptors = append(streamInterceptors, grpc_prometheus.StreamServerInterceptor) - // unaryInterceptors = append(unaryInterceptors, grpc_prometheus.UnaryServerInterceptor) - //} - - opts := []grpc.ServerOption{ - // https://github.com/grpc/grpc-go/issues/3171#issuecomment-552796779 - grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ - MinTime: 10 * time.Second, - PermitWithoutStream: true, - }), - grpc.ChainStreamInterceptor(streamInterceptors...), - grpc.ChainUnaryInterceptor(unaryInterceptors...), - } - if creds == nil { - // no specific opts - } else { - opts = append(opts, grpc.Creds(*creds)) - } - grpcServer := grpc.NewServer(opts...) - reflection.Register(grpcServer) // Register reflection service on gRPC server. +func StartGrpc(snServer *downloader.GrpcServer, addr string, creds credentials.TransportCredentials, logger log.Logger) (*grpc.Server, error) { + grpcServer := grpcutil.NewServerWithOpts(creds) if snServer != nil { downloaderproto.RegisterDownloaderServer(grpcServer, snServer) } - //if metrics.Enabled { - // grpc_prometheus.Register(grpcServer) - //} - - healthServer := health.NewServer() - grpc_health_v1.RegisterHealthServer(grpcServer, healthServer) - - go func() { - defer healthServer.Shutdown() - if err := grpcServer.Serve(lis); err != nil { - logger.Error("gRPC server stop", "err", err) - } - }() + if err := grpcutil.StartServer(grpcServer, addr, true, logger, "gRPC server stop"); err != nil { + return nil, err + } logger.Info("Started gRPC server", "on", addr) return grpcServer, nil } diff --git a/node/gointerfaces/grpcutil/utils.go b/node/gointerfaces/grpcutil/utils.go index a68038de631..80470508cfd 100644 --- a/node/gointerfaces/grpcutil/utils.go +++ b/node/gointerfaces/grpcutil/utils.go @@ -23,6 +23,7 @@ import ( "errors" "fmt" "io" + "net" "os" "strings" "time" @@ -33,9 +34,13 @@ import ( "google.golang.org/grpc/backoff" "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials" + "google.golang.org/grpc/health" + "google.golang.org/grpc/health/grpc_health_v1" "google.golang.org/grpc/keepalive" "google.golang.org/grpc/reflection" "google.golang.org/grpc/status" + + "github.com/erigontech/erigon/common/log/v3" ) func TLS(tlsCACert, tlsCertFile, tlsKeyFile string) (credentials.TransportCredentials, error) { @@ -68,40 +73,53 @@ func TLS(tlsCACert, tlsCertFile, tlsKeyFile string) (credentials.TransportCreden } func NewServer(rateLimit uint32, creds credentials.TransportCredentials) *grpc.Server { - var ( - streamInterceptors = make([]grpc.StreamServerInterceptor, 0, 1) - unaryInterceptors = make([]grpc.UnaryServerInterceptor, 0, 1) - ) - streamInterceptors = append(streamInterceptors, recovery.StreamServerInterceptor()) - unaryInterceptors = append(unaryInterceptors, recovery.UnaryServerInterceptor()) - - //if metrics.Enabled { - // streamInterceptors = append(streamInterceptors, grpc_prometheus.StreamServerInterceptor) - // unaryInterceptors = append(unaryInterceptors, grpc_prometheus.UnaryServerInterceptor) - //} + return NewServerWithOpts(creds, grpc.MaxConcurrentStreams(rateLimit)) // to force clients to reduce concurrency level +} - //cpus := uint32(runtime.GOMAXPROCS(-1)) +func NewServerWithOpts(creds credentials.TransportCredentials, extraOpts ...grpc.ServerOption) *grpc.Server { opts := []grpc.ServerOption{ - //grpc.NumStreamWorkers(cpus), // reduce amount of goroutines - grpc.MaxConcurrentStreams(rateLimit), // to force clients reduce concurrency level - // Don't drop the connection, settings accordign to this comment on GitHub + // Don't drop the connection, settings according to this comment on GitHub // https://github.com/grpc/grpc-go/issues/3171#issuecomment-552796779 grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ MinTime: 10 * time.Second, PermitWithoutStream: true, }), - grpc.ChainStreamInterceptor(streamInterceptors...), - grpc.ChainUnaryInterceptor(unaryInterceptors...), - grpc.Creds(creds), + grpc.ChainStreamInterceptor(recovery.StreamServerInterceptor()), + grpc.ChainUnaryInterceptor(recovery.UnaryServerInterceptor()), + } + if creds != nil { + opts = append(opts, grpc.Creds(creds)) } + opts = append(opts, extraOpts...) + grpcServer := grpc.NewServer(opts...) reflection.Register(grpcServer) + return grpcServer +} - //if metrics.Enabled { - // grpc_prometheus.Register(grpcServer) - //} +func StartServer(srv *grpc.Server, addr string, healthCheck bool, logger log.Logger, serveErrMsg string) error { + lis, err := net.Listen("tcp", addr) + if err != nil { + return fmt.Errorf("could not create listener: %w, addr=%s", err, addr) + } + StartServerOnListener(srv, lis, healthCheck, logger, serveErrMsg) + return nil +} - return grpcServer +func StartServerOnListener(srv *grpc.Server, lis net.Listener, healthCheck bool, logger log.Logger, serveErrMsg string) { + var healthServer *health.Server + if healthCheck { + healthServer = health.NewServer() + grpc_health_v1.RegisterHealthServer(srv, healthServer) + } + go func() { + if healthServer != nil { + defer healthServer.Shutdown() + } + if err := srv.Serve(lis); err != nil { + logger.Error(serveErrMsg, "err", err) + } + }() } func Connect(creds credentials.TransportCredentials, dialAddress string) (*grpc.ClientConn, error) { diff --git a/node/privateapi/all.go b/node/privateapi/all.go index 31155c5a114..78bc4dca626 100644 --- a/node/privateapi/all.go +++ b/node/privateapi/all.go @@ -17,13 +17,8 @@ package privateapi import ( - "fmt" - "net" - "google.golang.org/grpc" "google.golang.org/grpc/credentials" - "google.golang.org/grpc/health" - "google.golang.org/grpc/health/grpc_health_v1" "github.com/erigontech/erigon/common/log/v3" "github.com/erigontech/erigon/db/kv/remotedbserver" @@ -38,10 +33,6 @@ func StartGrpc(kv *remotedbserver.KvServer, ethBackendSrv *EthBackendServer, txP miningServer txpoolproto.MiningServer, bridgeServer *bridge.BackendServer, heimdallServer *heimdall.BackendServer, addr string, rateLimit uint32, creds credentials.TransportCredentials, healthCheck bool, logger log.Logger) (*grpc.Server, error) { logger.Info("Starting private RPC server", "on", addr) - lis, err := net.Listen("tcp", addr) - if err != nil { - return nil, fmt.Errorf("could not create listener: %w, addr=%s", err, addr) - } grpcServer := grpcutil.NewServer(rateLimit, creds) remoteproto.RegisterETHBACKENDServer(grpcServer, ethBackendSrv) @@ -57,21 +48,10 @@ func StartGrpc(kv *remotedbserver.KvServer, ethBackendSrv *EthBackendServer, txP if heimdallServer != nil { remoteproto.RegisterHeimdallBackendServer(grpcServer, heimdallServer) } - remoteproto.RegisterKVServer(grpcServer, kv) - var healthServer *health.Server - if healthCheck { - healthServer = health.NewServer() - grpc_health_v1.RegisterHealthServer(grpcServer, healthServer) - } - go func() { - if healthCheck { - defer healthServer.Shutdown() - } - if err := grpcServer.Serve(lis); err != nil { - logger.Error("private RPC server fail", "err", err) - } - }() + if err := grpcutil.StartServer(grpcServer, addr, healthCheck, logger, "private RPC server fail"); err != nil { + return nil, err + } return grpcServer, nil } diff --git a/p2p/sentry/sentry_grpc_server.go b/p2p/sentry/sentry_grpc_server.go index cd14da58d0c..a02bf2586be 100644 --- a/p2p/sentry/sentry_grpc_server.go +++ b/p2p/sentry/sentry_grpc_server.go @@ -40,8 +40,6 @@ import ( "golang.org/x/sync/errgroup" "golang.org/x/time/rate" "google.golang.org/grpc" - "google.golang.org/grpc/health" - "google.golang.org/grpc/health/grpc_health_v1" "google.golang.org/protobuf/types/known/emptypb" "github.com/erigontech/erigon/common" @@ -887,20 +885,7 @@ func grpcSentryServer(ctx context.Context, sentryAddr string, ss *GrpcServer, he } grpcServer := grpcutil.NewServer(100, nil) sentryproto.RegisterSentryServer(grpcServer, ss) - var healthServer *health.Server - if healthCheck { - healthServer = health.NewServer() - grpc_health_v1.RegisterHealthServer(grpcServer, healthServer) - } - - go func() { - if healthCheck { - defer healthServer.Shutdown() - } - if err1 := grpcServer.Serve(lis); err1 != nil { - ss.logger.Error("Sentry gRPC server fail", "err", err1) - } - }() + grpcutil.StartServerOnListener(grpcServer, lis, healthCheck, ss.logger, "Sentry gRPC server fail") return grpcServer, nil } diff --git a/txnprovider/txpool/txpool_grpc_server.go b/txnprovider/txpool/txpool_grpc_server.go index 8b25bec2159..325b28adf06 100644 --- a/txnprovider/txpool/txpool_grpc_server.go +++ b/txnprovider/txpool/txpool_grpc_server.go @@ -19,19 +19,11 @@ package txpool import ( "context" "errors" - "fmt" "math" - "net" - "time" - "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/recovery" "github.com/holiman/uint256" "google.golang.org/grpc" "google.golang.org/grpc/credentials" - "google.golang.org/grpc/health" - "google.golang.org/grpc/health/grpc_health_v1" - "google.golang.org/grpc/keepalive" - "google.golang.org/grpc/reflection" "google.golang.org/protobuf/types/known/emptypb" "github.com/erigontech/erigon/common" @@ -325,45 +317,11 @@ func (s *GrpcServer) Nonce(ctx context.Context, in *txpoolproto.NonceRequest) (* // NewSlotsStreams - it's safe to use this class as non-pointer type NewSlotsStreams = grpcutil.StreamBroadcaster[txpoolproto.OnAddReply] -func StartGrpc(txPoolServer txpoolproto.TxpoolServer, miningServer txpoolproto.MiningServer, addr string, creds *credentials.TransportCredentials, logger log.Logger) (*grpc.Server, error) { - lis, err := net.Listen("tcp", addr) - if err != nil { - return nil, fmt.Errorf("could not create listener: %w, addr=%s", err, addr) - } - - var ( - streamInterceptors = make([]grpc.StreamServerInterceptor, 0, 2) - unaryInterceptors = make([]grpc.UnaryServerInterceptor, 0, 2) - ) - streamInterceptors = append(streamInterceptors, recovery.StreamServerInterceptor()) - unaryInterceptors = append(unaryInterceptors, recovery.UnaryServerInterceptor()) - - //if metrics.Enabled { - // streamInterceptors = append(streamInterceptors, grpc_prometheus.StreamServerInterceptor) - // unaryInterceptors = append(unaryInterceptors, grpc_prometheus.UnaryServerInterceptor) - //} - - //cpus := uint32(runtime.GOMAXPROCS(-1)) - opts := []grpc.ServerOption{ - //grpc.NumStreamWorkers(cpus), // reduce amount of goroutines +func StartGrpc(txPoolServer txpoolproto.TxpoolServer, miningServer txpoolproto.MiningServer, addr string, creds credentials.TransportCredentials, logger log.Logger) (*grpc.Server, error) { + grpcServer := grpcutil.NewServerWithOpts(creds, grpc.ReadBufferSize(0), // reduce buffers to save mem grpc.WriteBufferSize(0), // reduce buffers to save mem - // Don't drop the connection, settings accordign to this comment on GitHub - // https://github.com/grpc/grpc-go/issues/3171#issuecomment-552796779 - grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ - MinTime: 10 * time.Second, - PermitWithoutStream: true, - }), - grpc.ChainStreamInterceptor(streamInterceptors...), - grpc.ChainUnaryInterceptor(unaryInterceptors...), - } - if creds == nil { - // no specific opts - } else { - opts = append(opts, grpc.Creds(*creds)) - } - grpcServer := grpc.NewServer(opts...) - reflection.Register(grpcServer) // Register reflection service on gRPC server. + ) if txPoolServer != nil { txpoolproto.RegisterTxpoolServer(grpcServer, txPoolServer) } @@ -371,19 +329,9 @@ func StartGrpc(txPoolServer txpoolproto.TxpoolServer, miningServer txpoolproto.M txpoolproto.RegisterMiningServer(grpcServer, miningServer) } - //if metrics.Enabled { - // grpc_prometheus.Register(grpcServer) - //} - - healthServer := health.NewServer() - grpc_health_v1.RegisterHealthServer(grpcServer, healthServer) - - go func() { - defer healthServer.Shutdown() - if err := grpcServer.Serve(lis); err != nil { - logger.Error("private RPC server fail", "err", err) - } - }() + if err := grpcutil.StartServer(grpcServer, addr, true, logger, "txpool gRPC server fail"); err != nil { + return nil, err + } logger.Info("Started gRPC server", "on", addr) return grpcServer, nil }