Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
N
nebula
Project
Project
Details
Activity
Releases
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
exchain
nebula
Commits
7e4821e7
Unverified
Commit
7e4821e7
authored
Oct 09, 2023
by
protolambda
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
op-service: Stop(ctx) calls, shutdown testing, op-node rpc server update
parent
246d7dbe
Changes
16
Show whitespace changes
Inline
Side-by-side
Showing
16 changed files
with
246 additions
and
85 deletions
+246
-85
endpoint_monitor.go
endpoint-monitor/endpoint_monitor.go
+5
-1
api.go
indexer/api/api.go
+22
-18
indexer.go
indexer/indexer.go
+2
-2
batch_submitter.go
op-batcher/batcher/batch_submitter.go
+10
-2
metrics.go
op-batcher/metrics/metrics.go
+1
-2
entrypoint.go
op-bootnode/bootnode/entrypoint.go
+5
-1
service.go
op-challenger/game/service.go
+15
-10
service.go
op-heartbeat/service.go
+11
-10
service_test.go
op-heartbeat/service_test.go
+2
-2
node.go
op-node/node/node.go
+9
-8
server.go
op-node/node/server.go
+8
-16
server_test.go
op-node/node/server_test.go
+9
-3
l2_output_submitter.go
op-proposer/proposer/l2_output_submitter.go
+10
-2
server.go
op-service/httputil/server.go
+32
-7
server_test.go
op-service/httputil/server_test.go
+100
-0
commands.go
op-wheel/commands.go
+5
-1
No files found.
endpoint-monitor/endpoint_monitor.go
View file @
7e4821e7
...
...
@@ -52,7 +52,11 @@ func Main(version string) func(cliCtx *cli.Context) error {
l
.
Error
(
"error starting metrics server"
,
err
)
return
err
}
defer
srv
.
Close
()
defer
func
()
{
if
err
:=
srv
.
Stop
(
cliCtx
.
Context
);
err
!=
nil
{
l
.
Error
(
"failed to stop metrics server"
,
"err"
,
err
)
}
}()
opio
.
BlockOnInterrupts
()
return
nil
...
...
indexer/api/api.go
View file @
7e4821e7
...
...
@@ -2,15 +2,18 @@ package api
import
(
"context"
"errors"
"fmt"
"net"
"net/http"
"runtime/debug"
"strconv"
"sync"
"github.com/ethereum-optimism/optimism/indexer/api/routes"
"github.com/ethereum-optimism/optimism/indexer/config"
"github.com/ethereum-optimism/optimism/indexer/database"
"github.com/ethereum-optimism/optimism/op-service/httputil"
"github.com/ethereum-optimism/optimism/op-service/metrics"
"github.com/ethereum/go-ethereum/log"
"github.com/go-chi/chi/v5"
...
...
@@ -120,41 +123,42 @@ func (a *API) Port() int {
// startServer ... Starts the API server
func
(
a
*
API
)
startServer
(
ctx
context
.
Context
)
error
{
a
.
log
.
Info
(
"api server listening..."
,
"port"
,
a
.
serverConfig
.
Port
)
server
:=
http
.
Server
{
Addr
:
fmt
.
Sprintf
(
":%d"
,
a
.
serverConfig
.
Port
),
Handler
:
a
.
router
}
a
.
log
.
Debug
(
"api server listening..."
,
"port"
,
a
.
serverConfig
.
Port
)
addr
:=
net
.
JoinHostPort
(
a
.
serverConfig
.
Host
,
strconv
.
Itoa
(
a
.
serverConfig
.
Port
))
srv
,
err
:=
httputil
.
StartHTTPServer
(
addr
,
a
.
router
)
if
err
!=
nil
{
return
fmt
.
Errorf
(
"failed to start API server: %w"
,
err
)
}
addr
:=
fmt
.
Sprintf
(
":%d"
,
a
.
serverConfig
.
Port
)
listener
,
err
:=
net
.
Listen
(
"tcp"
,
addr
)
host
,
portStr
,
err
:=
net
.
SplitHostPort
(
srv
.
Addr
()
.
String
())
if
err
!=
nil
{
a
.
log
.
Error
(
"Listen:"
,
err
)
return
err
return
errors
.
Join
(
err
,
srv
.
Close
())
}
tcpAddr
,
ok
:=
listener
.
Addr
()
.
(
*
net
.
TCPAdd
r
)
if
!
ok
{
return
fmt
.
Errorf
(
"failed to get TCP address from network listener"
)
port
,
err
:=
strconv
.
Atoi
(
portSt
r
)
if
err
!=
nil
{
return
errors
.
Join
(
err
,
srv
.
Close
()
)
}
// Update the port in the config in case the OS chose a different port
// than the one we requested (e.g. using port 0 to fetch a random open port)
a
.
serverConfig
.
Port
=
tcpAddr
.
Port
a
.
serverConfig
.
Host
=
host
a
.
serverConfig
.
Port
=
port
err
=
http
.
Serve
(
listener
,
server
.
Handler
)
if
err
!=
nil
{
a
.
log
.
Error
(
"api server stopped with error"
,
"err"
,
err
)
}
else
{
a
.
log
.
Info
(
"api server stopped"
)
<-
ctx
.
Done
()
if
err
:=
srv
.
Stop
(
context
.
Background
());
err
!=
nil
{
return
fmt
.
Errorf
(
"failed to shutdown api server: %w"
,
err
)
}
return
err
return
nil
}
// startMetricsServer ... Starts the metrics server
func
(
a
*
API
)
startMetricsServer
(
ctx
context
.
Context
)
error
{
a
.
log
.
Info
(
"starting metrics server..."
,
"port"
,
a
.
metricsConfig
.
Port
)
a
.
log
.
Debug
(
"starting metrics server..."
,
"port"
,
a
.
metricsConfig
.
Port
)
srv
,
err
:=
metrics
.
StartServer
(
a
.
metricsRegistry
,
a
.
metricsConfig
.
Host
,
a
.
metricsConfig
.
Port
)
if
err
!=
nil
{
return
fmt
.
Errorf
(
"failed to start metrics server: %w"
,
err
)
}
<-
ctx
.
Done
()
defer
a
.
log
.
Info
(
"metrics server stopped"
)
return
srv
.
Close
(
)
return
srv
.
Stop
(
context
.
Background
()
)
}
indexer/indexer.go
View file @
7e4821e7
...
...
@@ -119,7 +119,7 @@ func (i *Indexer) startHttpServer(ctx context.Context) error {
i
.
log
.
Info
(
"http server started"
,
"addr"
,
srv
.
Addr
())
<-
ctx
.
Done
()
defer
i
.
log
.
Info
(
"http server stopped"
)
return
srv
.
Close
(
)
return
srv
.
Stop
(
context
.
Background
()
)
}
func
(
i
*
Indexer
)
startMetricsServer
(
ctx
context
.
Context
)
error
{
...
...
@@ -131,7 +131,7 @@ func (i *Indexer) startMetricsServer(ctx context.Context) error {
i
.
log
.
Info
(
"metrics server started"
,
"addr"
,
srv
.
Addr
())
<-
ctx
.
Done
()
defer
i
.
log
.
Info
(
"metrics server stopped"
)
return
srv
.
Close
(
)
return
srv
.
Stop
(
context
.
Background
()
)
}
// Start starts the indexing service on L1 and L2 chains
...
...
op-batcher/batcher/batch_submitter.go
View file @
7e4821e7
...
...
@@ -61,7 +61,11 @@ func Main(version string, cliCtx *cli.Context) error {
return
err
}
l
.
Info
(
"started pprof server"
,
"addr"
,
pprofSrv
.
Addr
())
defer
pprofSrv
.
Close
()
defer
func
()
{
if
err
:=
pprofSrv
.
Stop
(
context
.
Background
());
err
!=
nil
{
l
.
Error
(
"failed to stop pprof server"
,
"err"
,
err
)
}
}()
}
metricsCfg
:=
cfg
.
MetricsConfig
...
...
@@ -72,7 +76,11 @@ func Main(version string, cliCtx *cli.Context) error {
return
fmt
.
Errorf
(
"failed to start metrics server: %w"
,
err
)
}
l
.
Info
(
"started metrics server"
,
"addr"
,
metricsSrv
.
Addr
())
defer
metricsSrv
.
Close
()
defer
func
()
{
if
err
:=
metricsSrv
.
Stop
(
context
.
Background
());
err
!=
nil
{
l
.
Error
(
"failed to stop pprof server"
,
"err"
,
err
)
}
}()
ctx
,
cancel
:=
context
.
WithCancel
(
context
.
Background
())
defer
cancel
()
m
.
StartBalanceMetrics
(
ctx
,
l
,
batchSubmitter
.
L1Client
,
batchSubmitter
.
TxManager
.
From
())
...
...
op-batcher/metrics/metrics.go
View file @
7e4821e7
...
...
@@ -3,8 +3,6 @@ package metrics
import
(
"context"
"github.com/ethereum-optimism/optimism/op-service/httputil"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/ethclient"
...
...
@@ -13,6 +11,7 @@ import (
"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
"github.com/ethereum-optimism/optimism/op-service/eth"
"github.com/ethereum-optimism/optimism/op-service/httputil"
opmetrics
"github.com/ethereum-optimism/optimism/op-service/metrics"
txmetrics
"github.com/ethereum-optimism/optimism/op-service/txmgr/metrics"
)
...
...
op-bootnode/bootnode/entrypoint.go
View file @
7e4821e7
...
...
@@ -78,7 +78,11 @@ func Main(cliCtx *cli.Context) error {
if
err
!=
nil
{
return
fmt
.
Errorf
(
"failed to start metrics server: %w"
,
err
)
}
defer
metricsSrv
.
Close
()
defer
func
()
{
if
err
:=
metricsSrv
.
Stop
(
context
.
Background
());
err
!=
nil
{
log
.
Error
(
"failed to stop metrics server"
,
"err"
,
err
)
}
}()
log
.
Info
(
"started metrics server"
,
"addr"
,
metricsSrv
.
Addr
())
m
.
RecordUp
()
}
...
...
op-challenger/game/service.go
View file @
7e4821e7
...
...
@@ -31,13 +31,16 @@ type Service struct {
metricsSrv
*
httputil
.
HTTPServer
}
func
(
s
*
Service
)
Close
(
)
error
{
func
(
s
*
Service
)
Stop
(
ctx
context
.
Context
)
error
{
var
result
error
if
s
.
sched
!=
nil
{
result
=
errors
.
Join
(
result
,
s
.
sched
.
Close
())
}
if
s
.
pprofSrv
!=
nil
{
result
=
errors
.
Join
(
result
,
s
.
pprofSrv
.
Close
(
))
result
=
errors
.
Join
(
result
,
s
.
pprofSrv
.
Stop
(
ctx
))
}
if
s
.
metricsSrv
!=
nil
{
result
=
errors
.
Join
(
result
,
s
.
metricsSrv
.
Close
(
))
result
=
errors
.
Join
(
result
,
s
.
metricsSrv
.
Stop
(
ctx
))
}
return
result
}
...
...
@@ -66,7 +69,7 @@ func NewService(ctx context.Context, logger log.Logger, cfg *config.Config) (*Se
logger
.
Debug
(
"starting pprof"
,
"addr"
,
pprofConfig
.
ListenAddr
,
"port"
,
pprofConfig
.
ListenPort
)
pprofSrv
,
err
:=
oppprof
.
StartServer
(
pprofConfig
.
ListenAddr
,
pprofConfig
.
ListenPort
)
if
err
!=
nil
{
return
nil
,
errors
.
Join
(
fmt
.
Errorf
(
"failed to start pprof server: %w"
,
err
),
s
.
Close
(
))
return
nil
,
errors
.
Join
(
fmt
.
Errorf
(
"failed to start pprof server: %w"
,
err
),
s
.
Stop
(
ctx
))
}
s
.
pprofSrv
=
pprofSrv
logger
.
Info
(
"started pprof server"
,
"addr"
,
pprofSrv
.
Addr
())
...
...
@@ -77,7 +80,7 @@ func NewService(ctx context.Context, logger log.Logger, cfg *config.Config) (*Se
logger
.
Debug
(
"starting metrics server"
,
"addr"
,
metricsCfg
.
ListenAddr
,
"port"
,
metricsCfg
.
ListenPort
)
metricsSrv
,
err
:=
m
.
Start
(
metricsCfg
.
ListenAddr
,
metricsCfg
.
ListenPort
)
if
err
!=
nil
{
return
nil
,
errors
.
Join
(
fmt
.
Errorf
(
"failed to start metrics server: %w"
,
err
),
s
.
Close
(
))
return
nil
,
errors
.
Join
(
fmt
.
Errorf
(
"failed to start metrics server: %w"
,
err
),
s
.
Stop
(
ctx
))
}
logger
.
Info
(
"started metrics server"
,
"addr"
,
metricsSrv
.
Addr
())
s
.
metricsSrv
=
metricsSrv
...
...
@@ -86,7 +89,7 @@ func NewService(ctx context.Context, logger log.Logger, cfg *config.Config) (*Se
factory
,
err
:=
bindings
.
NewDisputeGameFactory
(
cfg
.
GameFactoryAddress
,
l1Client
)
if
err
!=
nil
{
return
nil
,
errors
.
Join
(
fmt
.
Errorf
(
"failed to bind the fault dispute game factory contract: %w"
,
err
),
s
.
Close
(
))
return
nil
,
errors
.
Join
(
fmt
.
Errorf
(
"failed to bind the fault dispute game factory contract: %w"
,
err
),
s
.
Stop
(
ctx
))
}
loader
:=
NewGameLoader
(
factory
)
...
...
@@ -102,7 +105,7 @@ func NewService(ctx context.Context, logger log.Logger, cfg *config.Config) (*Se
pollClient
,
err
:=
opClient
.
NewRPCWithClient
(
ctx
,
logger
,
cfg
.
L1EthRpc
,
opClient
.
NewBaseRPCClient
(
l1Client
.
Client
()),
cfg
.
PollInterval
)
if
err
!=
nil
{
return
nil
,
errors
.
Join
(
fmt
.
Errorf
(
"failed to create RPC client: %w"
,
err
),
s
.
Close
(
))
return
nil
,
errors
.
Join
(
fmt
.
Errorf
(
"failed to create RPC client: %w"
,
err
),
s
.
Stop
(
ctx
))
}
s
.
monitor
=
newGameMonitor
(
logger
,
cl
,
loader
,
s
.
sched
,
cfg
.
GameWindow
,
l1Client
.
BlockNumber
,
cfg
.
GameAllowlist
,
pollClient
)
...
...
@@ -115,7 +118,9 @@ func NewService(ctx context.Context, logger log.Logger, cfg *config.Config) (*Se
// MonitorGame monitors the fault dispute game and attempts to progress it.
func
(
s
*
Service
)
MonitorGame
(
ctx
context
.
Context
)
error
{
s
.
sched
.
Start
(
ctx
)
defer
s
.
sched
.
Close
()
defer
s
.
Close
()
return
s
.
monitor
.
MonitorGames
(
ctx
)
err
:=
s
.
monitor
.
MonitorGames
(
ctx
)
// The other ctx is the close-trigger.
// We need to refactor Service more to allow for graceful/force-shutdown granularity.
err
=
errors
.
Join
(
err
,
s
.
Stop
(
context
.
Background
()))
return
err
}
op-heartbeat/service.go
View file @
7e4821e7
package
op_heartbeat
import
(
"context"
"encoding/json"
"errors"
"fmt"
...
...
@@ -41,7 +42,7 @@ func Main(version string) func(ctx *cli.Context) error {
oplog
.
SetGlobalLogHandler
(
l
.
GetHandler
())
l
.
Info
(
"starting heartbeat monitor"
,
"version"
,
version
)
srv
,
err
:=
Start
(
l
,
cfg
,
version
)
srv
,
err
:=
Start
(
cliCtx
.
Context
,
l
,
cfg
,
version
)
if
err
!=
nil
{
l
.
Crit
(
"error starting application"
,
"err"
,
err
)
}
...
...
@@ -54,7 +55,7 @@ func Main(version string) func(ctx *cli.Context) error {
syscall
.
SIGQUIT
,
}
...
)
<-
doneCh
return
srv
.
Close
(
)
return
srv
.
Stop
(
context
.
Background
()
)
}
}
...
...
@@ -62,21 +63,21 @@ type HeartbeatService struct {
pprof
,
metrics
,
http
*
httputil
.
HTTPServer
}
func
(
hs
*
HeartbeatService
)
Close
(
)
error
{
func
(
hs
*
HeartbeatService
)
Stop
(
ctx
context
.
Context
)
error
{
var
result
error
if
hs
.
pprof
!=
nil
{
result
=
errors
.
Join
(
result
,
hs
.
pprof
.
Close
(
))
result
=
errors
.
Join
(
result
,
hs
.
pprof
.
Stop
(
ctx
))
}
if
hs
.
metrics
!=
nil
{
result
=
errors
.
Join
(
result
,
hs
.
metrics
.
Close
(
))
result
=
errors
.
Join
(
result
,
hs
.
metrics
.
Stop
(
ctx
))
}
if
hs
.
http
!=
nil
{
result
=
errors
.
Join
(
result
,
hs
.
http
.
Close
(
))
result
=
errors
.
Join
(
result
,
hs
.
http
.
Stop
(
ctx
))
}
return
result
}
func
Start
(
l
log
.
Logger
,
cfg
Config
,
version
string
)
(
*
HeartbeatService
,
error
)
{
func
Start
(
ctx
context
.
Context
,
l
log
.
Logger
,
cfg
Config
,
version
string
)
(
*
HeartbeatService
,
error
)
{
hs
:=
&
HeartbeatService
{}
registry
:=
opmetrics
.
NewRegistry
()
...
...
@@ -85,7 +86,7 @@ func Start(l log.Logger, cfg Config, version string) (*HeartbeatService, error)
l
.
Debug
(
"starting metrics server"
,
"addr"
,
metricsCfg
.
ListenAddr
,
"port"
,
metricsCfg
.
ListenPort
)
metricsSrv
,
err
:=
opmetrics
.
StartServer
(
registry
,
metricsCfg
.
ListenAddr
,
metricsCfg
.
ListenPort
)
if
err
!=
nil
{
return
nil
,
errors
.
Join
(
fmt
.
Errorf
(
"failed to start metrics server: %w"
,
err
),
hs
.
Close
(
))
return
nil
,
errors
.
Join
(
fmt
.
Errorf
(
"failed to start metrics server: %w"
,
err
),
hs
.
Stop
(
ctx
))
}
hs
.
metrics
=
metricsSrv
l
.
Info
(
"started metrics server"
,
"addr"
,
metricsSrv
.
Addr
())
...
...
@@ -96,7 +97,7 @@ func Start(l log.Logger, cfg Config, version string) (*HeartbeatService, error)
l
.
Debug
(
"starting pprof"
,
"addr"
,
pprofCfg
.
ListenAddr
,
"port"
,
pprofCfg
.
ListenPort
)
pprofSrv
,
err
:=
oppprof
.
StartServer
(
pprofCfg
.
ListenAddr
,
pprofCfg
.
ListenPort
)
if
err
!=
nil
{
return
nil
,
errors
.
Join
(
fmt
.
Errorf
(
"failed to start pprof server: %w"
,
err
),
hs
.
Close
(
))
return
nil
,
errors
.
Join
(
fmt
.
Errorf
(
"failed to start pprof server: %w"
,
err
),
hs
.
Stop
(
ctx
))
}
l
.
Info
(
"started pprof server"
,
"addr"
,
pprofSrv
.
Addr
())
hs
.
pprof
=
pprofSrv
...
...
@@ -121,7 +122,7 @@ func Start(l log.Logger, cfg Config, version string) (*HeartbeatService, error)
}),
httputil
.
WithMaxHeaderBytes
(
HTTPMaxHeaderSize
))
if
err
!=
nil
{
return
nil
,
errors
.
Join
(
fmt
.
Errorf
(
"failed to start HTTP server: %w"
,
err
),
hs
.
Close
(
))
return
nil
,
errors
.
Join
(
fmt
.
Errorf
(
"failed to start HTTP server: %w"
,
err
),
hs
.
Stop
(
ctx
))
}
hs
.
http
=
srv
...
...
op-heartbeat/service_test.go
View file @
7e4821e7
...
...
@@ -31,13 +31,13 @@ func TestService(t *testing.T) {
}
ctx
,
cancel
:=
context
.
WithCancel
(
context
.
Background
())
srv
,
err
:=
Start
(
log
.
New
(),
cfg
,
"foobar"
)
srv
,
err
:=
Start
(
ctx
,
log
.
New
(),
cfg
,
"foobar"
)
// Make sure that the service properly starts
require
.
NoError
(
t
,
err
)
defer
cancel
()
defer
func
()
{
require
.
NoError
(
t
,
srv
.
Close
(
),
"close heartbeat server"
)
require
.
NoError
(
t
,
srv
.
Stop
(
ctx
),
"close heartbeat server"
)
}()
tests
:=
[]
struct
{
...
...
op-node/node/node.go
View file @
7e4821e7
...
...
@@ -555,7 +555,9 @@ func (n *OpNode) Stop(ctx context.Context) error {
var
result
*
multierror
.
Error
if
n
.
server
!=
nil
{
n
.
server
.
Stop
()
if
err
:=
n
.
server
.
Stop
(
ctx
);
err
!=
nil
{
result
=
multierror
.
Append
(
result
,
fmt
.
Errorf
(
"failed to close RPC server: %w"
,
err
))
}
}
if
n
.
p2pNode
!=
nil
{
if
err
:=
n
.
p2pNode
.
Close
();
err
!=
nil
{
...
...
@@ -623,12 +625,12 @@ func (n *OpNode) Stop(ctx context.Context) error {
// Close metrics and pprof only after we are done idling
if
n
.
pprofSrv
!=
nil
{
if
err
:=
n
.
pprofSrv
.
Close
(
);
err
!=
nil
{
if
err
:=
n
.
pprofSrv
.
Stop
(
ctx
);
err
!=
nil
{
result
=
multierror
.
Append
(
result
,
fmt
.
Errorf
(
"failed to close pprof server: %w"
,
err
))
}
}
if
n
.
metricsSrv
!=
nil
{
if
err
:=
n
.
metricsSrv
.
Close
(
);
err
!=
nil
{
if
err
:=
n
.
metricsSrv
.
Stop
(
ctx
);
err
!=
nil
{
result
=
multierror
.
Append
(
result
,
fmt
.
Errorf
(
"failed to close metrics server: %w"
,
err
))
}
}
...
...
@@ -640,10 +642,9 @@ func (n *OpNode) Stopped() bool {
return
n
.
closed
.
Load
()
}
func
(
n
*
OpNode
)
ListenAddr
()
string
{
return
n
.
server
.
listenAddr
.
String
()
}
func
(
n
*
OpNode
)
HTTPEndpoint
()
string
{
return
fmt
.
Sprintf
(
"http://%s"
,
n
.
ListenAddr
())
if
n
.
server
==
nil
{
return
""
}
return
fmt
.
Sprintf
(
"http://%s"
,
n
.
server
.
Addr
()
.
String
())
}
op-node/node/server.go
View file @
7e4821e7
...
...
@@ -2,7 +2,7 @@ package node
import
(
"context"
"
errors
"
"
fmt
"
"net"
"net/http"
"strconv"
...
...
@@ -21,9 +21,8 @@ import (
type
rpcServer
struct
{
endpoint
string
apis
[]
rpc
.
API
httpServer
*
http
.
Server
httpServer
*
ophttp
.
HTTP
Server
appVersion
string
listenAddr
net
.
Addr
log
log
.
Logger
sources
.
L2Client
}
...
...
@@ -79,27 +78,20 @@ func (s *rpcServer) Start() error {
mux
.
Handle
(
"/"
,
nodeHandler
)
mux
.
HandleFunc
(
"/healthz"
,
healthzHandler
(
s
.
appVersion
))
listener
,
err
:=
net
.
Listen
(
"tcp"
,
s
.
endpoint
)
hs
,
err
:=
ophttp
.
StartHTTPServer
(
s
.
endpoint
,
mux
)
if
err
!=
nil
{
return
err
}
s
.
listenAddr
=
listener
.
Addr
()
s
.
httpServer
=
ophttp
.
NewHttpServer
(
mux
)
go
func
()
{
if
err
:=
s
.
httpServer
.
Serve
(
listener
);
err
!=
nil
&&
!
errors
.
Is
(
err
,
http
.
ErrServerClosed
)
{
// todo improve error handling
s
.
log
.
Error
(
"http server failed"
,
"err"
,
err
)
return
fmt
.
Errorf
(
"failed to start HTTP RPC server: %w"
,
err
)
}
}()
s
.
httpServer
=
hs
return
nil
}
func
(
r
*
rpcServer
)
Stop
(
)
{
_
=
r
.
httpServer
.
Shutdown
(
context
.
Background
()
)
func
(
r
*
rpcServer
)
Stop
(
ctx
context
.
Context
)
error
{
return
r
.
httpServer
.
Stop
(
ctx
)
}
func
(
r
*
rpcServer
)
Addr
()
net
.
Addr
{
return
r
.
listenAddr
return
r
.
httpServer
.
Addr
()
}
func
healthzHandler
(
appVersion
string
)
http
.
HandlerFunc
{
...
...
op-node/node/server_test.go
View file @
7e4821e7
...
...
@@ -104,7 +104,9 @@ func TestOutputAtBlock(t *testing.T) {
server
,
err
:=
newRPCServer
(
context
.
Background
(),
rpcCfg
,
rollupCfg
,
l2Client
,
drClient
,
log
,
"0.0"
,
metrics
.
NoopMetrics
)
require
.
NoError
(
t
,
err
)
require
.
NoError
(
t
,
server
.
Start
())
defer
server
.
Stop
()
defer
func
()
{
require
.
NoError
(
t
,
server
.
Stop
(
context
.
Background
()))
}()
client
,
err
:=
rpcclient
.
NewRPC
(
context
.
Background
(),
log
,
"http://"
+
server
.
Addr
()
.
String
(),
rpcclient
.
WithDialBackoff
(
3
))
require
.
NoError
(
t
,
err
)
...
...
@@ -136,7 +138,9 @@ func TestVersion(t *testing.T) {
server
,
err
:=
newRPCServer
(
context
.
Background
(),
rpcCfg
,
rollupCfg
,
l2Client
,
drClient
,
log
,
"0.0"
,
metrics
.
NoopMetrics
)
assert
.
NoError
(
t
,
err
)
assert
.
NoError
(
t
,
server
.
Start
())
defer
server
.
Stop
()
defer
func
()
{
require
.
NoError
(
t
,
server
.
Stop
(
context
.
Background
()))
}()
client
,
err
:=
rpcclient
.
NewRPC
(
context
.
Background
(),
log
,
"http://"
+
server
.
Addr
()
.
String
(),
rpcclient
.
WithDialBackoff
(
3
))
assert
.
NoError
(
t
,
err
)
...
...
@@ -180,7 +184,9 @@ func TestSyncStatus(t *testing.T) {
server
,
err
:=
newRPCServer
(
context
.
Background
(),
rpcCfg
,
rollupCfg
,
l2Client
,
drClient
,
log
,
"0.0"
,
metrics
.
NoopMetrics
)
assert
.
NoError
(
t
,
err
)
assert
.
NoError
(
t
,
server
.
Start
())
defer
server
.
Stop
()
defer
func
()
{
require
.
NoError
(
t
,
server
.
Stop
(
context
.
Background
()))
}()
client
,
err
:=
rpcclient
.
NewRPC
(
context
.
Background
(),
log
,
"http://"
+
server
.
Addr
()
.
String
(),
rpcclient
.
WithDialBackoff
(
3
))
assert
.
NoError
(
t
,
err
)
...
...
op-proposer/proposer/l2_output_submitter.go
View file @
7e4821e7
...
...
@@ -79,7 +79,11 @@ func Main(version string, cliCtx *cli.Context) error {
return
err
}
l
.
Info
(
"started pprof server"
,
"addr"
,
pprofSrv
.
Addr
())
defer
pprofSrv
.
Close
()
defer
func
()
{
if
err
:=
pprofSrv
.
Stop
(
context
.
Background
());
err
!=
nil
{
l
.
Error
(
"failed to stop pprof server"
,
"err"
,
err
)
}
}()
}
metricsCfg
:=
cfg
.
MetricsConfig
...
...
@@ -90,7 +94,11 @@ func Main(version string, cliCtx *cli.Context) error {
return
fmt
.
Errorf
(
"failed to start metrics server: %w"
,
err
)
}
l
.
Info
(
"started metrics server"
,
"addr"
,
metricsSrv
.
Addr
())
defer
metricsSrv
.
Close
()
defer
func
()
{
if
err
:=
metricsSrv
.
Stop
(
context
.
Background
());
err
!=
nil
{
l
.
Error
(
"failed to stop metrics server"
,
"err"
,
err
)
}
}()
ctx
,
cancel
:=
context
.
WithCancel
(
context
.
Background
())
defer
cancel
()
m
.
StartBalanceMetrics
(
ctx
,
l
,
proposerConfig
.
L1Client
,
proposerConfig
.
TxManager
.
From
())
...
...
op-service/httputil/server.go
View file @
7e4821e7
package
httputil
import
(
"context"
"errors"
"fmt"
"net"
...
...
@@ -25,21 +26,27 @@ func StartHTTPServer(addr string, handler http.Handler, opts ...HTTPOption) (*HT
return
nil
,
fmt
.
Errorf
(
"failed to bind to address %q: %w"
,
addr
,
err
)
}
srvCtx
,
srvCancel
:=
context
.
WithCancel
(
context
.
Background
())
srv
:=
&
http
.
Server
{
Handler
:
handler
,
ReadTimeout
:
DefaultTimeouts
.
ReadTimeout
,
ReadHeaderTimeout
:
DefaultTimeouts
.
ReadHeaderTimeout
,
WriteTimeout
:
DefaultTimeouts
.
WriteTimeout
,
IdleTimeout
:
DefaultTimeouts
.
IdleTimeout
,
BaseContext
:
func
(
listener
net
.
Listener
)
context
.
Context
{
return
srvCtx
},
}
out
:=
&
HTTPServer
{
listener
:
listener
,
srv
:
srv
}
for
_
,
opt
:=
range
opts
{
if
err
:=
opt
(
out
);
err
!=
nil
{
srvCancel
()
return
nil
,
errors
.
Join
(
fmt
.
Errorf
(
"failed to apply HTTP option: %w"
,
err
),
listener
.
Close
())
}
}
go
func
()
{
err
:=
out
.
srv
.
Serve
(
listener
)
srvCancel
()
// no error, unless ErrServerClosed (or unused base context closes, or unused http2 config error)
if
errors
.
Is
(
err
,
http
.
ErrServerClosed
)
{
out
.
closed
.
Store
(
true
)
...
...
@@ -54,17 +61,35 @@ func (s *HTTPServer) Closed() bool {
return
s
.
closed
.
Load
()
}
func
(
s
*
HTTPServer
)
Close
()
error
{
// closes the underlying listener too
err
:=
s
.
srv
.
Close
()
if
errors
.
Is
(
err
,
http
.
ErrServerClosed
)
{
return
nil
// Stop is a convenience method to gracefully shut down the server, but force-close if the ctx is cancelled.
// The ctx error is not returned when the force-close is successful.
func
(
s
*
HTTPServer
)
Stop
(
ctx
context
.
Context
)
error
{
if
err
:=
s
.
Shutdown
(
ctx
);
err
!=
nil
{
if
errors
.
Is
(
err
,
ctx
.
Err
())
{
// force-close connections if we cancelled the stopping
return
s
.
Close
()
}
return
err
}
return
nil
}
// Shutdown shuts down the HTTP server and its listener,
// but allows active connections to close gracefully.
// If the function exits due to a ctx cancellation the listener is closed but active connections may remain,
// a call to Close() can force-close any remaining active connections.
func
(
s
*
HTTPServer
)
Shutdown
(
ctx
context
.
Context
)
error
{
// closes the underlying listener too.
return
s
.
srv
.
Shutdown
(
ctx
)
}
// Close force-closes the HTTPServer, its listener, and all its active connections.
func
(
s
*
HTTPServer
)
Close
()
error
{
// closes the underlying listener too
return
s
.
srv
.
Close
()
}
func
(
s
*
HTTPServer
)
Addr
()
string
{
return
s
.
listener
.
Addr
()
.
String
()
func
(
s
*
HTTPServer
)
Addr
()
net
.
Addr
{
return
s
.
listener
.
Addr
()
}
func
WithMaxHeaderBytes
(
max
int
)
HTTPOption
{
...
...
op-service/httputil/server_test.go
0 → 100644
View file @
7e4821e7
package
httputil
import
(
"context"
"net/http"
"sync"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func
TestStartHTTPServer
(
t
*
testing
.
T
)
{
testSetup
:=
func
(
t
*
testing
.
T
)
(
srv
*
HTTPServer
,
reqRespBlock
chan
chan
chan
struct
{})
{
reqRespBlock
=
make
(
chan
chan
chan
struct
{},
10
)
h
:=
http
.
HandlerFunc
(
func
(
w
http
.
ResponseWriter
,
r
*
http
.
Request
)
{
require
.
NoError
(
t
,
r
.
Context
()
.
Err
())
respBlock
:=
make
(
chan
chan
struct
{})
reqRespBlock
<-
respBlock
select
{
case
block
:=
<-
respBlock
:
block
<-
struct
{}{}
w
.
WriteHeader
(
http
.
StatusTeapot
)
case
<-
r
.
Context
()
.
Done
()
:
w
.
WriteHeader
(
http
.
StatusServiceUnavailable
)
}
})
srv
,
err
:=
StartHTTPServer
(
"localhost:0"
,
h
,
WithTimeouts
(
HTTPTimeouts
{
ReadTimeout
:
time
.
Minute
,
ReadHeaderTimeout
:
time
.
Minute
,
WriteTimeout
:
time
.
Minute
,
IdleTimeout
:
time
.
Minute
,
}))
require
.
NoError
(
t
,
err
)
require
.
False
(
t
,
srv
.
Closed
())
return
srv
,
reqRespBlock
}
t
.
Run
(
"basics"
,
func
(
t
*
testing
.
T
)
{
srv
,
reqRespBlock
:=
testSetup
(
t
)
// test basics
go
func
()
{
req
:=
<-
reqRespBlock
// take request
block
:=
make
(
chan
struct
{})
req
<-
block
// start response
<-
block
// unblock response
}()
resp
,
err
:=
http
.
Get
(
"http://"
+
srv
.
Addr
()
.
String
()
+
"/"
)
require
.
NoError
(
t
,
err
)
assert
.
NoError
(
t
,
resp
.
Body
.
Close
())
assert
.
Equal
(
t
,
http
.
StatusTeapot
,
resp
.
StatusCode
,
"I am a teapot"
)
assert
.
NoError
(
t
,
srv
.
Close
())
assert
.
True
(
t
,
srv
.
Closed
())
})
t
.
Run
(
"force-shutdown"
,
func
(
t
*
testing
.
T
)
{
srv
,
reqRespBlock
:=
testSetup
(
t
)
var
wg
sync
.
WaitGroup
wg
.
Add
(
1
)
go
func
()
{
resp
,
err
:=
http
.
Get
(
"http://"
+
srv
.
Addr
()
.
String
()
+
"/"
)
assert
.
ErrorContains
(
t
,
err
,
"EOF"
)
// error must indicate connection is force-closed
if
resp
!=
nil
{
assert
.
NoError
(
t
,
resp
.
Body
.
Close
())
// makes linter happy
}
wg
.
Done
()
}()
req
:=
<-
reqRespBlock
// take the request
block
:=
make
(
chan
struct
{})
req
<-
block
// start response
// just force-shutdown the server
assert
.
NoError
(
t
,
srv
.
Close
())
wg
.
Wait
()
// only now unblock the response
<-
block
require
.
True
(
t
,
srv
.
Closed
())
})
t
.
Run
(
"graceful"
,
func
(
t
*
testing
.
T
)
{
srv
,
reqRespBlock
:=
testSetup
(
t
)
var
wg
sync
.
WaitGroup
wg
.
Add
(
1
)
go
func
()
{
resp
,
err
:=
http
.
Get
(
"http://"
+
srv
.
Addr
()
.
String
()
+
"/"
)
assert
.
NoError
(
t
,
err
)
assert
.
NoError
(
t
,
resp
.
Body
.
Close
())
assert
.
Equal
(
t
,
http
.
StatusServiceUnavailable
,
resp
.
StatusCode
,
"service unavailable when shutting down"
)
wg
.
Done
()
}()
// Wait for a request, but don't start a response to it, just try to shut down the server
// The base-context will be shut down, allowing the server to stop waiting for the user,
// and gracefully tell the user it's not able to continue.
<-
reqRespBlock
assert
.
NoError
(
t
,
srv
.
Shutdown
(
context
.
Background
()))
wg
.
Wait
()
require
.
True
(
t
,
srv
.
Closed
())
})
}
op-wheel/commands.go
View file @
7e4821e7
...
...
@@ -423,7 +423,11 @@ var (
if
err
!=
nil
{
return
fmt
.
Errorf
(
"failed to start metrics server: %w"
,
err
)
}
defer
metricsSrv
.
Close
()
defer
func
()
{
if
err
:=
metricsSrv
.
Stop
(
context
.
Background
());
err
!=
nil
{
l
.
Error
(
"failed to stop metrics server: %w"
,
err
)
}
}()
}
return
engine
.
Auto
(
ctx
,
metrics
,
client
,
l
,
shutdown
,
settings
)
})
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment