mirror/prysm-pulse
1
0
Fork 0
mirror of https://gitlab.com/pulsechaincom/prysm-pulse.git synced 2024-12-22 11:32:09 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Unify gRPC Gateways into Single, Shared lib (#8799)

Browse Source 
* moved validator/rpc/gateway under shared,  validator/rpc/web is not visiable to the new shared gateway

* decided to have one lib with two methods if needed, status and stop are the same, however New and Start are not

* moved beacon's gateway to shared and moved the main func to its own /server folder

* goftm

* gofmt

* removed the extra loggin

* reduce visibility

* tighter visibilty of the shared lib for beacon and validator only for now

* fix patterns , ctx needs to be first param

* fix comments

* gofmt

* added enum for the caller id

* added unit test for gateway

* deprecated .WithDialer to .WithContextDialer

* changed the string callerId to uint8, and rearranged the Gateway struct based on the compiler

* fix 1 based on comment

* iota type

Co-authored-by: Raul Jordan <raul@prysmaticlabs.com>
This commit is contained in:
ahadda5 2021-05-04 19:20:58 +04:00 committed by GitHub
parent 3c9b9e7bf5
commit 8dd2915d3c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 308 additions and 336 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
beacon-chain/gateway/cors.go
View File

@ -1,20 +0,0 @@
package gateway
import (
"net/http"
"github.com/rs/cors"
)
func newCorsHandler(srv http.Handler, allowedOrigins []string) http.Handler {
if len(allowedOrigins) == 0 {
return srv
}
c := cors.New(cors.Options{
AllowedOrigins: allowedOrigins,
AllowedMethods: []string{http.MethodPost, http.MethodGet},
MaxAge: 600,
AllowedHeaders: []string{"*"},
})
return c.Handler(srv)
}

26
beacon-chain/gateway/handlers.go
View File

@ -1,26 +0,0 @@
package gateway
import (
"net/http"
"path"
"strings"
)
// Swagger directory for the runtime files provided by bazel data.
const swaggerDir = "proto/beacon/rpc/v1/"
// SwaggerServer returns swagger specification files located under "/swagger/"
func SwaggerServer() http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
if !strings.HasSuffix(r.URL.Path, ".swagger.json") {
log.Debugf("Not found: %s", r.URL.Path)
http.NotFound(w, r)
return
}
log.Debugf("Serving %s\n", r.URL.Path)
p := strings.TrimPrefix(r.URL.Path, "/swagger/")
p = path.Join(swaggerDir, p)
http.ServeFile(w, r, p)
}
}

50
beacon-chain/gateway/server/BUILD.bazel
View File

@ -1,50 +0,0 @@
load("@prysm//tools/go:def.bzl", "go_library")
load("@io_bazel_rules_go//go:def.bzl", "go_binary")
load("@io_bazel_rules_docker//go:image.bzl", "go_image")
load("@io_bazel_rules_docker//container:container.bzl", "container_bundle")
load("@io_bazel_rules_docker//contrib:push-all.bzl", "docker_push")
go_library(
name = "go_default_library",
srcs = [
"log.go",
"main.go",
],
importpath = "github.com/prysmaticlabs/prysm/beacon-chain/gateway/server",
visibility = ["//visibility:private"],
deps = [
"//beacon-chain/gateway:go_default_library",
"//shared/maxprocs:go_default_library",
"@com_github_joonix_log//:go_default_library",
"@com_github_sirupsen_logrus//:go_default_library",
],
)
go_binary(
name = "server",
embed = [":go_default_library"],
visibility = ["//visibility:private"],
)
go_image(
name = "image",
base = "//tools:go_image",
binary = ":server",
tags = ["manual"],
visibility = ["//visibility:private"],
)
container_bundle(
name = "image_bundle",
images = {
"gcr.io/prysmaticlabs/prysm/beacon-chain/gateway:latest": ":image",
"gcr.io/prysmaticlabs/prysm/beacon-chain/gateway:{DOCKER_TAG}": ":image",
},
tags = ["manual"],
)
docker_push(
name = "push_images",
bundle = ":image_bundle",
tags = ["manual"],
)

2
beacon-chain/node/BUILD.bazel
View File

@ -22,7 +22,6 @@ go_library(
"//beacon-chain/db/kv:go_default_library",
"//beacon-chain/forkchoice:go_default_library",
"//beacon-chain/forkchoice/protoarray:go_default_library",
"//beacon-chain/gateway:go_default_library",
"//beacon-chain/interop-cold-start:go_default_library",
"//beacon-chain/node/registration:go_default_library",
"//beacon-chain/operations/attestations:go_default_library",
@ -41,6 +40,7 @@ go_library(
"//shared/debug:go_default_library",
"//shared/event:go_default_library",
"//shared/featureconfig:go_default_library",
"//shared/gateway:go_default_library",
"//shared/params:go_default_library",
"//shared/prereq:go_default_library",
"//shared/prometheus:go_default_library",

4
beacon-chain/node/node.go
View File

@ -23,7 +23,6 @@ import (
"github.com/prysmaticlabs/prysm/beacon-chain/db/kv"
"github.com/prysmaticlabs/prysm/beacon-chain/forkchoice"
"github.com/prysmaticlabs/prysm/beacon-chain/forkchoice/protoarray"
"github.com/prysmaticlabs/prysm/beacon-chain/gateway"
interopcoldstart "github.com/prysmaticlabs/prysm/beacon-chain/interop-cold-start"
"github.com/prysmaticlabs/prysm/beacon-chain/node/registration"
"github.com/prysmaticlabs/prysm/beacon-chain/operations/attestations"
@ -42,6 +41,7 @@ import (
"github.com/prysmaticlabs/prysm/shared/debug"
"github.com/prysmaticlabs/prysm/shared/event"
"github.com/prysmaticlabs/prysm/shared/featureconfig"
"github.com/prysmaticlabs/prysm/shared/gateway"
"github.com/prysmaticlabs/prysm/shared/params"
"github.com/prysmaticlabs/prysm/shared/prereq"
"github.com/prysmaticlabs/prysm/shared/prometheus"
@ -650,7 +650,7 @@ func (b *BeaconNode) registerGRPCGateway() error {
enableDebugRPCEndpoints := b.cliCtx.Bool(flags.EnableDebugRPCEndpoints.Name)
selfCert := b.cliCtx.String(flags.CertFlag.Name)
return b.services.RegisterService(
gateway.New(
gateway.NewBeacon(
b.ctx,
selfAddress,
selfCert,

24
beacon-chain/server/BUILD.bazel Normal file
View File

@ -0,0 +1,24 @@
load("@io_bazel_rules_go//go:def.bzl", "go_binary")
load("@prysm//tools/go:def.bzl", "go_library")
go_library(
name = "go_default_library",
srcs = [
"log.go",
"main.go",
],
importpath = "github.com/prysmaticlabs/prysm/beacon-chain/server",
visibility = ["//visibility:private"],
deps = [
"//shared/gateway:go_default_library",
"//shared/maxprocs:go_default_library",
"@com_github_joonix_log//:go_default_library",
"@com_github_sirupsen_logrus//:go_default_library",
],
)
go_binary(
name = "server",
embed = [":go_default_library"],
visibility = ["//visibility:private"],
)

0
beacon-chain/gateway/server/log.go → beacon-chain/server/log.go
View File

4
beacon-chain/gateway/server/main.go → beacon-chain/server/main.go
View File

@ -10,7 +10,7 @@ import (
"strings"
joonix "github.com/joonix/log"
"github.com/prysmaticlabs/prysm/beacon-chain/gateway"
"github.com/prysmaticlabs/prysm/shared/gateway"
_ "github.com/prysmaticlabs/prysm/shared/maxprocs"
"github.com/sirupsen/logrus"
)
@ -36,7 +36,7 @@ func main() {
}
mux := http.NewServeMux()
gw := gateway.New(
gw := gateway.NewBeacon(
context.Background(),
*beaconRPC,
"", // remoteCert

1
cmd/beacon-chain/flags/BUILD.bazel
View File

@ -12,6 +12,7 @@ go_library(
visibility = [
"//beacon-chain:__subpackages__",
"//cmd/beacon-chain:__subpackages__",
"//shared/gateway:__pkg__",
],
deps = [
"//shared/cmd:go_default_library",

25
beacon-chain/gateway/BUILD.bazel → shared/gateway/BUILD.bazel
View File

@ -1,22 +1,23 @@
# gazelle:ignore
load("@io_bazel_rules_go//go:def.bzl", "go_library")
load("@io_bazel_rules_go//go:def.bzl", "go_test")
load("@prysm//tools/go:def.bzl", "go_library")
go_library(
name = "go_default_library",
srcs = [
"cors.go",
"gateway.go",
"handlers.go",
"log.go",
],
importpath = "github.com/prysmaticlabs/prysm/beacon-chain/gateway",
importpath = "github.com/prysmaticlabs/prysm/shared/gateway",
visibility = [
"//beacon-chain/gateway/server:__pkg__",
"//beacon-chain/node:__pkg__",
"//beacon-chain:__subpackages__",
"//validator:__subpackages__",
],
deps = [
"//proto/beacon/rpc/v1:go_grpc_gateway_library",
"//proto/validator/accounts/v2:ethereum_validator_account_gateway_proto",
"//shared:go_default_library",
"//validator/web:go_default_library",
"@com_github_grpc_ecosystem_grpc_gateway//runtime:go_default_library",
"@com_github_pkg_errors//:go_default_library",
"@com_github_prysmaticlabs_ethereumapis//eth/v1alpha1:go_grpc_gateway_library",
@ -27,3 +28,15 @@ go_library(
"@org_golang_google_grpc//credentials:go_default_library",
],
)
go_test(
name = "go_default_test",
srcs = ["gateway_test.go"],
embed = [":go_default_library"],
deps = [
"//cmd/beacon-chain/flags:go_default_library",
"//shared/testutil/require:go_default_library",
"@com_github_sirupsen_logrus//hooks/test:go_default_library",
"@com_github_urfave_cli_v2//:go_default_library",
],
)

236
beacon-chain/gateway/gateway.go → shared/gateway/gateway.go
View File

@ -1,5 +1,5 @@
// Package gateway defines a gRPC gateway to serve HTTP-JSON
// traffic as a proxy and forward it to a beacon node's gRPC service.
// traffic as a proxy and forward it to a beacon or validator node's gRPC service.
package gateway
import (
@ -7,13 +7,18 @@ import (
"fmt"
"net"
"net/http"
"path"
"strings"
"time"
gwruntime "github.com/grpc-ecosystem/grpc-gateway/runtime"
"github.com/pkg/errors"
ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1_gateway"
pbrpc "github.com/prysmaticlabs/prysm/proto/beacon/rpc/v1_gateway"
pb "github.com/prysmaticlabs/prysm/proto/validator/accounts/v2_gateway"
"github.com/prysmaticlabs/prysm/shared"
"github.com/prysmaticlabs/prysm/validator/web"
"github.com/rs/cors"
"google.golang.org/grpc"
"google.golang.org/grpc/connectivity"
"google.golang.org/grpc/credentials"
@ -21,70 +26,158 @@ import (
var _ shared.Service = (*Gateway)(nil)
// Gateway is the gRPC gateway to serve HTTP JSON traffic as a proxy and forward
// it to the beacon-chain gRPC server.
// CallerId defines whether the caller node is a beacon
// or a validator node. This helps register the handlers accordingly.
type CallerId uint8
const (
Beacon CallerId = iota
Validator
)
// Gateway is the gRPC gateway to serve HTTP JSON traffic as a
// proxy and forward it to the gRPC server.
type Gateway struct {
conn *grpc.ClientConn
ctx context.Context
cancel context.CancelFunc
gatewayAddr string
remoteAddr string
remoteCert string
server *http.Server
mux *http.ServeMux
allowedOrigins []string
startFailure error
enableDebugRPCEndpoints bool
callerId CallerId
maxCallRecvMsgSize uint64
mux *http.ServeMux
server *http.Server
cancel context.CancelFunc
remoteCert string
gatewayAddr string
ctx context.Context
startFailure error
remoteAddr string
allowedOrigins []string
}
// Start the gateway service. This serves the HTTP JSON traffic on the specified
// port.
// New returns a new gateway server which translates HTTP into gRPC.
// Accepts a context.
func NewValidator(
ctx context.Context,
remoteAddress,
gatewayAddress string,
allowedOrigins []string,
) *Gateway {
return &Gateway{
callerId: Validator,
remoteAddr: remoteAddress,
gatewayAddr: gatewayAddress,
ctx: ctx,
mux: http.NewServeMux(),
allowedOrigins: allowedOrigins,
}
}
// New returns a new gateway server which translates HTTP into gRPC.
// Accepts a context and optional http.ServeMux.
func NewBeacon(
ctx context.Context,
remoteAddress,
remoteCert,
gatewayAddress string,
mux *http.ServeMux,
allowedOrigins []string,
enableDebugRPCEndpoints bool,
maxCallRecvMsgSize uint64,
) *Gateway {
if mux == nil {
mux = http.NewServeMux()
}
return &Gateway{
callerId: Beacon,
remoteAddr: remoteAddress,
remoteCert: remoteCert,
gatewayAddr: gatewayAddress,
ctx: ctx,
mux: mux,
allowedOrigins: allowedOrigins,
enableDebugRPCEndpoints: enableDebugRPCEndpoints,
maxCallRecvMsgSize: maxCallRecvMsgSize,
}
}
// Start the gateway service. This serves the HTTP JSON traffic.
// The beacon node supports TCP and Unix domain socket communications.
// Beacon and validator have different handlers.
func (g *Gateway) Start() {
ctx, cancel := context.WithCancel(g.ctx)
g.cancel = cancel
log.WithField("address", g.gatewayAddr).Info("Starting JSON-HTTP API")
if g.callerId == Beacon {
conn, err := g.dial(ctx, "tcp", g.remoteAddr)
if err != nil {
log.WithError(err).Error("Failed to connect to gRPC server")
g.startFailure = err
return
}
conn, err := g.dial(ctx, "tcp", g.remoteAddr)
if err != nil {
log.WithError(err).Error("Failed to connect to gRPC server")
g.startFailure = err
return
g.conn = conn
}
g.conn = conn
gwmux := gwruntime.NewServeMux(
gwruntime.WithMarshalerOption(
gwruntime.MIMEWildcard,
&gwruntime.JSONPb{OrigName: false, EmitDefaults: true},
),
)
handlers := []func(context.Context, *gwruntime.ServeMux, *grpc.ClientConn) error{
ethpb.RegisterNodeHandler,
ethpb.RegisterBeaconChainHandler,
ethpb.RegisterBeaconNodeValidatorHandler,
pbrpc.RegisterHealthHandler,
}
if g.enableDebugRPCEndpoints {
handlers = append(handlers, pbrpc.RegisterDebugHandler)
}
for _, f := range handlers {
if err := f(ctx, gwmux, conn); err != nil {
log.WithError(err).Error("Failed to start gateway")
g.startFailure = err
return
if g.callerId == Beacon {
handlers := []func(context.Context, *gwruntime.ServeMux, *grpc.ClientConn) error{
ethpb.RegisterNodeHandler,
ethpb.RegisterBeaconChainHandler,
ethpb.RegisterBeaconNodeValidatorHandler,
pbrpc.RegisterHealthHandler,
}
if g.enableDebugRPCEndpoints {
handlers = append(handlers, pbrpc.RegisterDebugHandler)
}
for _, f := range handlers {
if err := f(ctx, gwmux, g.conn); err != nil {
log.WithError(err).Error("Failed to start gateway")
g.startFailure = err
return
}
}
g.mux.Handle("/", gwmux)
g.server = &http.Server{
Addr: g.gatewayAddr,
Handler: g.corsMiddleware(g.mux),
}
} else {
opts := []grpc.DialOption{grpc.WithInsecure()}
handlers := []func(context.Context, *gwruntime.ServeMux, string, []grpc.DialOption) error{
pb.RegisterAuthHandlerFromEndpoint,
pb.RegisterWalletHandlerFromEndpoint,
pb.RegisterHealthHandlerFromEndpoint,
pb.RegisterAccountsHandlerFromEndpoint,
pb.RegisterBeaconHandlerFromEndpoint,
pb.RegisterSlashingProtectionHandlerFromEndpoint,
}
for _, h := range handlers {
if err := h(ctx, gwmux, g.remoteAddr, opts); err != nil {
log.Fatalf("Could not register API handler with grpc endpoint: %v", err)
}
}
apiHandler := g.corsMiddleware(gwmux)
g.mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
if strings.HasPrefix(r.URL.Path, "/api") {
http.StripPrefix("/api", apiHandler).ServeHTTP(w, r)
} else {
web.Handler(w, r)
}
})
g.server = &http.Server{
Addr: g.gatewayAddr,
Handler: g.mux,
}
}
g.mux.Handle("/", gwmux)
g.server = &http.Server{
Addr: g.gatewayAddr,
Handler: newCorsHandler(g.mux, g.allowedOrigins),
}
go func() {
log.WithField("address", g.gatewayAddr).Info("Starting gRPC gateway")
if err := g.server.ListenAndServe(); err != http.ErrServerClosed {
log.WithError(err).Error("Failed to listen and serve")
g.startFailure = err
@ -127,31 +220,32 @@ func (g *Gateway) Stop() error {
return nil
}
// New returns a new gateway server which translates HTTP into gRPC.
// Accepts a context and optional http.ServeMux.
func New(
ctx context.Context,
remoteAddress,
remoteCert,
gatewayAddress string,
mux *http.ServeMux,
allowedOrigins []string,
enableDebugRPCEndpoints bool,
maxCallRecvMsgSize uint64,
) *Gateway {
if mux == nil {
mux = http.NewServeMux()
}
func (g *Gateway) corsMiddleware(h http.Handler) http.Handler {
c := cors.New(cors.Options{
AllowedOrigins: g.allowedOrigins,
AllowedMethods: []string{http.MethodPost, http.MethodGet, http.MethodOptions},
AllowCredentials: true,
MaxAge: 600,
AllowedHeaders: []string{"*"},
})
return c.Handler(h)
}
return &Gateway{
remoteAddr: remoteAddress,
remoteCert: remoteCert,
gatewayAddr: gatewayAddress,
ctx: ctx,
mux: mux,
allowedOrigins: allowedOrigins,
enableDebugRPCEndpoints: enableDebugRPCEndpoints,
maxCallRecvMsgSize: maxCallRecvMsgSize,
const swaggerDir = "proto/beacon/rpc/v1/"
// SwaggerServer returns swagger specification files located under "/swagger/"
func SwaggerServer() http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
if !strings.HasSuffix(r.URL.Path, ".swagger.json") {
log.Debugf("Not found: %s", r.URL.Path)
http.NotFound(w, r)
return
}
log.Debugf("Serving %s\n", r.URL.Path)
p := strings.TrimPrefix(r.URL.Path, "/swagger/")
p = path.Join(swaggerDir, p)
http.ServeFile(w, r, p)
}
}
@ -196,9 +290,15 @@ func (g *Gateway) dialUnix(ctx context.Context, addr string) (*grpc.ClientConn,
d := func(addr string, timeout time.Duration) (net.Conn, error) {
return net.DialTimeout("unix", addr, timeout)
}
f := func(ctx context.Context, addr string) (net.Conn, error) {
if deadline, ok := ctx.Deadline(); ok {
return d(addr, time.Until(deadline))
}
return d(addr, 0)
}
opts := []grpc.DialOption{
grpc.WithInsecure(),
grpc.WithDialer(d),
grpc.WithContextDialer(f),
grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(int(g.maxCallRecvMsgSize))),
}
return grpc.DialContext(ctx, addr, opts...)

84
shared/gateway/gateway_test.go Normal file
View File

@ -0,0 +1,84 @@
package gateway
import (
"flag"
"fmt"
"strings"
"testing"
"github.com/prysmaticlabs/prysm/cmd/beacon-chain/flags"
"github.com/prysmaticlabs/prysm/shared/testutil/require"
logTest "github.com/sirupsen/logrus/hooks/test"
"github.com/urfave/cli/v2"
)
// Test that beacon gateway Start, Stop.
func TestBeaconGateway_StartStop(t *testing.T) {
hook := logTest.NewGlobal()
app := cli.App{}
set := flag.NewFlagSet("test", 0)
ctx := cli.NewContext(&app, set, nil)
gatewayPort := ctx.Int(flags.GRPCGatewayPort.Name)
gatewayHost := ctx.String(flags.GRPCGatewayHost.Name)
rpcHost := ctx.String(flags.RPCHost.Name)
selfAddress := fmt.Sprintf("%s:%d", rpcHost, ctx.Int(flags.RPCPort.Name))
gatewayAddress := fmt.Sprintf("%s:%d", gatewayHost, gatewayPort)
allowedOrigins := strings.Split(ctx.String(flags.GPRCGatewayCorsDomain.Name), ",")
enableDebugRPCEndpoints := ctx.Bool(flags.EnableDebugRPCEndpoints.Name)
selfCert := ctx.String(flags.CertFlag.Name)
beaconGateway := NewBeacon(
ctx.Context,
selfAddress,
selfCert,
gatewayAddress,
nil, /*optional mux*/
allowedOrigins,
enableDebugRPCEndpoints,
ctx.Uint64("grpc-max-msg-size"),
)
beaconGateway.Start()
go func() {
require.LogsContain(t, hook, "Starting gRPC gateway")
}()
err := beaconGateway.Stop()
require.NoError(t, err)
}
// Test that validator gateway Start, Stop.
func TestValidatorGateway_StartStop(t *testing.T) {
hook := logTest.NewGlobal()
app := cli.App{}
set := flag.NewFlagSet("test", 0)
ctx := cli.NewContext(&app, set, nil)
gatewayHost := ctx.String(flags.GRPCGatewayHost.Name)
gatewayPort := ctx.Int(flags.GRPCGatewayPort.Name)
rpcHost := ctx.String(flags.RPCHost.Name)
rpcPort := ctx.Int(flags.RPCPort.Name)
rpcAddr := fmt.Sprintf("%s:%d", rpcHost, rpcPort)
gatewayAddress := fmt.Sprintf("%s:%d", gatewayHost, gatewayPort)
allowedOrigins := strings.Split(ctx.String(flags.GPRCGatewayCorsDomain.Name), ",")
validatorGateway := NewValidator(
ctx.Context,
rpcAddr,
gatewayAddress,
allowedOrigins,
)
validatorGateway.Start()
go func() {
require.LogsContain(t, hook, "Starting gRPC gateway")
}()
err := validatorGateway.Stop()
require.NoError(t, err)
}

0
beacon-chain/gateway/log.go → shared/gateway/log.go
View File

2
validator/node/BUILD.bazel
View File

@ -37,6 +37,7 @@ go_library(
"//shared/event:go_default_library",
"//shared/featureconfig:go_default_library",
"//shared/fileutil:go_default_library",
"//shared/gateway:go_default_library",
"//shared/params:go_default_library",
"//shared/prereq:go_default_library",
"//shared/prometheus:go_default_library",
@ -50,7 +51,6 @@ go_library(
"//validator/keymanager:go_default_library",
"//validator/keymanager/imported:go_default_library",
"//validator/rpc:go_default_library",
"//validator/rpc/gateway:go_default_library",
"//validator/slashing-protection:go_default_library",
"//validator/slashing-protection/iface:go_default_library",
"@com_github_pkg_errors//:go_default_library",

4
validator/node/node.go
View File

@ -22,6 +22,7 @@ import (
"github.com/prysmaticlabs/prysm/shared/event"
"github.com/prysmaticlabs/prysm/shared/featureconfig"
"github.com/prysmaticlabs/prysm/shared/fileutil"
"github.com/prysmaticlabs/prysm/shared/gateway"
"github.com/prysmaticlabs/prysm/shared/params"
"github.com/prysmaticlabs/prysm/shared/prereq"
"github.com/prysmaticlabs/prysm/shared/prometheus"
@ -35,7 +36,6 @@ import (
"github.com/prysmaticlabs/prysm/validator/keymanager"
"github.com/prysmaticlabs/prysm/validator/keymanager/imported"
"github.com/prysmaticlabs/prysm/validator/rpc"
"github.com/prysmaticlabs/prysm/validator/rpc/gateway"
slashingprotection "github.com/prysmaticlabs/prysm/validator/slashing-protection"
"github.com/prysmaticlabs/prysm/validator/slashing-protection/iface"
"github.com/sirupsen/logrus"
@ -510,7 +510,7 @@ func (c *ValidatorClient) registerRPCGatewayService(cliCtx *cli.Context) error {
rpcAddr := fmt.Sprintf("%s:%d", rpcHost, rpcPort)
gatewayAddress := fmt.Sprintf("%s:%d", gatewayHost, gatewayPort)
allowedOrigins := strings.Split(cliCtx.String(flags.GPRCGatewayCorsDomain.Name), ",")
gatewaySrv := gateway.New(
gatewaySrv := gateway.NewValidator(
cliCtx.Context,
rpcAddr,
gatewayAddress,

20
validator/rpc/gateway/BUILD.bazel
View File

@ -1,20 +0,0 @@
load("@prysm//tools/go:def.bzl", "go_library")
go_library(
name = "go_default_library",
srcs = [
"gateway.go",
"log.go",
],
importpath = "github.com/prysmaticlabs/prysm/validator/rpc/gateway",
visibility = ["//validator:__subpackages__"],
deps = [
"//proto/validator/accounts/v2:ethereum_validator_account_gateway_proto",
"//validator/web:go_default_library",
"@com_github_grpc_ecosystem_grpc_gateway//runtime:go_default_library",
"@com_github_pkg_errors//:go_default_library",
"@com_github_rs_cors//:go_default_library",
"@com_github_sirupsen_logrus//:go_default_library",
"@org_golang_google_grpc//:go_default_library",
],
)

132
validator/rpc/gateway/gateway.go
View File

@ -1,132 +0,0 @@
package gateway
import (
"context"
"net/http"
"strings"
"time"
gwruntime "github.com/grpc-ecosystem/grpc-gateway/runtime"
"github.com/pkg/errors"
pb "github.com/prysmaticlabs/prysm/proto/validator/accounts/v2_gateway"
"github.com/prysmaticlabs/prysm/validator/web"
"github.com/rs/cors"
"google.golang.org/grpc"
)
// Gateway is the gRPC gateway to serve HTTP JSON traffic as a
// proxy and forward it to the gRPC server.
type Gateway struct {
ctx context.Context
cancel context.CancelFunc
gatewayAddr string
remoteAddr string
server *http.Server
mux *http.ServeMux
allowedOrigins []string
startFailure error
}
// New returns a new gateway server which translates HTTP into gRPC.
// Accepts a context and optional http.ServeMux.
func New(
ctx context.Context,
remoteAddress,
gatewayAddress string,
allowedOrigins []string,
) *Gateway {
return &Gateway{
remoteAddr: remoteAddress,
gatewayAddr: gatewayAddress,
ctx: ctx,
mux: http.NewServeMux(),
allowedOrigins: allowedOrigins,
}
}
// Start the gateway service. This serves the HTTP JSON traffic.
func (g *Gateway) Start() {
ctx, cancel := context.WithCancel(g.ctx)
g.cancel = cancel
gwmux := gwruntime.NewServeMux(
gwruntime.WithMarshalerOption(
gwruntime.MIMEWildcard,
&gwruntime.JSONPb{OrigName: false},
),
)
opts := []grpc.DialOption{grpc.WithInsecure()}
handlers := []func(context.Context, *gwruntime.ServeMux, string, []grpc.DialOption) error{
pb.RegisterAuthHandlerFromEndpoint,
pb.RegisterWalletHandlerFromEndpoint,
pb.RegisterHealthHandlerFromEndpoint,
pb.RegisterAccountsHandlerFromEndpoint,
pb.RegisterBeaconHandlerFromEndpoint,
pb.RegisterSlashingProtectionHandlerFromEndpoint,
}
for _, h := range handlers {
if err := h(ctx, gwmux, g.remoteAddr, opts); err != nil {
log.Fatalf("Could not register API handler with grpc endpoint: %v", err)
}
}
apiHandler := g.corsMiddleware(gwmux)
g.mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
if strings.HasPrefix(r.URL.Path, "/api") {
http.StripPrefix("/api", apiHandler).ServeHTTP(w, r)
} else {
web.Handler(w, r)
}
})
g.server = &http.Server{
Addr: g.gatewayAddr,
Handler: g.mux,
}
go func() {
log.WithField("address", g.gatewayAddr).Info("Starting gRPC gateway")
if err := g.server.ListenAndServe(); err != http.ErrServerClosed {
log.WithError(err).Error("Failed to listen and serve")
g.startFailure = err
return
}
}()
}
// Status of grpc gateway. Returns an error if this service is unhealthy.
func (g *Gateway) Status() error {
if g.startFailure != nil {
return g.startFailure
}
return nil
}
// Stop the gateway with a graceful shutdown.
func (g *Gateway) Stop() error {
shutdownCtx, shutdownCancel := context.WithTimeout(g.ctx, 2*time.Second)
defer shutdownCancel()
if err := g.server.Shutdown(shutdownCtx); err != nil {
if errors.Is(err, context.DeadlineExceeded) {
log.Warn("Existing connections terminated")
} else {
log.WithError(err).Error("Failed to gracefully shut down server")
}
}
if g.cancel != nil {
g.cancel()
}
return nil
}
func (g *Gateway) corsMiddleware(h http.Handler) http.Handler {
c := cors.New(cors.Options{
AllowedOrigins: g.allowedOrigins,
AllowedMethods: []string{http.MethodPost, http.MethodGet, http.MethodOptions},
AllowCredentials: true,
MaxAge: 600,
AllowedHeaders: []string{"*"},
})
return c.Handler(h)
}

5
validator/rpc/gateway/log.go
View File

@ -1,5 +0,0 @@
package gateway
import "github.com/sirupsen/logrus"
var log = logrus.WithField("prefix", "gateway")

5
validator/web/BUILD.bazel
View File

@ -12,7 +12,10 @@ go_library(
":site_data", # keep
],
importpath = "github.com/prysmaticlabs/prysm/validator/web",
visibility = ["//validator:__subpackages__"],
visibility = [
"//shared/gateway:__pkg__",
"//validator:__subpackages__",
],
deps = [
"//shared:go_default_library",
"@com_github_sirupsen_logrus//:go_default_library",