Skip to content

Commit

Permalink
Merge pull request #4411 from UlyanaAndrukhiv/UlyanaAndrukhiv/4217-si…
Browse files Browse the repository at this point in the history 
…ngle-gRPC-port
  • Loading branch information
@durkmurder
durkmurder authored Jul 21, 2023
2 parents ca27c87 + d6eb702 commit 3f3c15f
Show file tree
Hide file tree
Showing 11 changed files with 757 additions and 215 deletions.
59 changes: 55 additions & 4 deletions cmd/access/node_builder/access_node_builder.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@ import (
"github.com/onflow/flow-go/module/executiondatasync/execution_data"
execdatacache "github.com/onflow/flow-go/module/executiondatasync/execution_data/cache"
finalizer "github.com/onflow/flow-go/module/finalizer/consensus"
"github.com/onflow/flow-go/module/grpcserver"
"github.com/onflow/flow-go/module/id"
"github.com/onflow/flow-go/module/mempool/herocache"
"github.com/onflow/flow-go/module/mempool/stdmap"
Expand Down Expand Up @@ -241,6 +242,11 @@ type FlowAccessNodeBuilder struct {
FollowerEng *followereng.ComplianceEngine
SyncEng *synceng.Engine
StateStreamEng *state_stream.Engine

// grpc servers
secureGrpcServer *grpcserver.GrpcServer
unsecureGrpcServer *grpcserver.GrpcServer
stateStreamGrpcServer *grpcserver.GrpcServer
}

func (builder *FlowAccessNodeBuilder) buildFollowerState() *FlowAccessNodeBuilder {
Expand Down Expand Up @@ -613,8 +619,7 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionDataRequester() *FlowAccessN
node.RootChainID,
builder.executionDataConfig.InitialBlockHeight,
highestAvailableHeight,
builder.apiRatelimits,
builder.apiBurstlimits,
builder.stateStreamGrpcServer,
)
if err != nil {
return nil, fmt.Errorf("could not create state stream engine: %w", err)
Expand Down Expand Up @@ -996,6 +1001,38 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
builder.rpcConf.TransportCredentials = credentials.NewTLS(tlsConfig)
return nil
}).
Module("creating grpc servers", func(node *cmd.NodeConfig) error {
builder.secureGrpcServer = grpcserver.NewGrpcServerBuilder(
node.Logger,
builder.rpcConf.SecureGRPCListenAddr,
builder.rpcConf.MaxMsgSize,
builder.rpcMetricsEnabled,
builder.apiRatelimits,
builder.apiBurstlimits,
grpcserver.WithTransportCredentials(builder.rpcConf.TransportCredentials)).Build()

builder.stateStreamGrpcServer = grpcserver.NewGrpcServerBuilder(
node.Logger,
builder.stateStreamConf.ListenAddr,
builder.stateStreamConf.MaxExecutionDataMsgSize,
builder.rpcMetricsEnabled,
builder.apiRatelimits,
builder.apiBurstlimits,
grpcserver.WithStreamInterceptor()).Build()

if builder.rpcConf.UnsecureGRPCListenAddr != builder.stateStreamConf.ListenAddr {
builder.unsecureGrpcServer = grpcserver.NewGrpcServerBuilder(node.Logger,
builder.rpcConf.UnsecureGRPCListenAddr,
builder.rpcConf.MaxMsgSize,
builder.rpcMetricsEnabled,
builder.apiRatelimits,
builder.apiBurstlimits).Build()
} else {
builder.unsecureGrpcServer = builder.stateStreamGrpcServer
}

return nil
}).
Component("RPC engine", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
engineBuilder, err := rpc.NewBuilder(
node.Logger,
Expand All @@ -1015,9 +1052,9 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
builder.executionGRPCPort,
builder.retryEnabled,
builder.rpcMetricsEnabled,
builder.apiRatelimits,
builder.apiBurstlimits,
builder.Me,
builder.secureGrpcServer,
builder.unsecureGrpcServer,
)
if err != nil {
return nil, err
Expand Down Expand Up @@ -1107,6 +1144,20 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
builder.BuildExecutionDataRequester()
}

builder.Component("secure grpc server", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
return builder.secureGrpcServer, nil
})

builder.Component("state stream unsecure grpc server", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
return builder.stateStreamGrpcServer, nil
})

if builder.rpcConf.UnsecureGRPCListenAddr != builder.stateStreamConf.ListenAddr {
builder.Component("unsecure grpc server", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
return builder.unsecureGrpcServer, nil
})
}

builder.Component("ping engine", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
ping, err := pingeng.New(
node.Logger,
Expand Down
37 changes: 35 additions & 2 deletions cmd/observer/node_builder/observer_builder.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ import (
"github.com/onflow/flow-go/module"
"github.com/onflow/flow-go/module/chainsync"
finalizer "github.com/onflow/flow-go/module/finalizer/consensus"
"github.com/onflow/flow-go/module/grpcserver"
"github.com/onflow/flow-go/module/id"
"github.com/onflow/flow-go/module/local"
"github.com/onflow/flow-go/module/metrics"
Expand Down Expand Up @@ -162,6 +163,10 @@ type ObserverServiceBuilder struct {

// Public network
peerID peer.ID

// grpc servers
secureGrpcServer *grpcserver.GrpcServer
unsecureGrpcServer *grpcserver.GrpcServer
}

// deriveBootstrapPeerIdentities derives the Flow Identity of the bootstrap peers from the parameters.
Expand Down Expand Up @@ -844,6 +849,24 @@ func (builder *ObserverServiceBuilder) enqueueConnectWithStakedAN() {
}

func (builder *ObserverServiceBuilder) enqueueRPCServer() {
builder.Module("creating grpc servers", func(node *cmd.NodeConfig) error {
builder.secureGrpcServer = grpcserver.NewGrpcServerBuilder(node.Logger,
builder.rpcConf.SecureGRPCListenAddr,
builder.rpcConf.MaxMsgSize,
builder.rpcMetricsEnabled,
builder.apiRatelimits,
builder.apiBurstlimits,
grpcserver.WithTransportCredentials(builder.rpcConf.TransportCredentials)).Build()

builder.unsecureGrpcServer = grpcserver.NewGrpcServerBuilder(node.Logger,
builder.rpcConf.UnsecureGRPCListenAddr,
builder.rpcConf.MaxMsgSize,
builder.rpcMetricsEnabled,
builder.apiRatelimits,
builder.apiBurstlimits).Build()

return nil
})
builder.Component("RPC engine", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
engineBuilder, err := rpc.NewBuilder(
node.Logger,
Expand All @@ -863,9 +886,9 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
0,
false,
builder.rpcMetricsEnabled,
builder.apiRatelimits,
builder.apiBurstlimits,
builder.Me,
builder.secureGrpcServer,
builder.unsecureGrpcServer,
)
if err != nil {
return nil, err
Expand Down Expand Up @@ -900,6 +923,16 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
builder.FollowerDistributor.AddOnBlockFinalizedConsumer(builder.RpcEng.OnFinalizedBlock)
return builder.RpcEng, nil
})

// build secure grpc server
builder.Component("secure grpc server", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
return builder.secureGrpcServer, nil
})

// build unsecure grpc server
builder.Component("unsecure grpc server", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
return builder.unsecureGrpcServer, nil
})
}

// initMiddleware creates the network.Middleware implementation with the libp2p factory function, metrics, peer update
Expand Down
Loading

0 comments on commit 3f3c15f

Please sign in to comment.