Merge pull request #7682 from ethereum-optimism/batcher-lifecycle

op-batcher: service lifecycle cleanup

op-batcher/batcher/batch_submitter.go

@@ -3,110 +3,32 @@ package batcher
 import (
 	"context"
 	"fmt"
-	_ "net/http/pprof"
 
 	"github.com/urfave/cli/v2"
 
 	"github.com/ethereum-optimism/optimism/op-batcher/flags"
-	"github.com/ethereum-optimism/optimism/op-batcher/metrics"
-	"github.com/ethereum-optimism/optimism/op-batcher/rpc"
 	opservice "github.com/ethereum-optimism/optimism/op-service"
+	"github.com/ethereum-optimism/optimism/op-service/cliapp"
 	oplog "github.com/ethereum-optimism/optimism/op-service/log"
-	"github.com/ethereum-optimism/optimism/op-service/opio"
-	oppprof "github.com/ethereum-optimism/optimism/op-service/pprof"
-	oprpc "github.com/ethereum-optimism/optimism/op-service/rpc"
 )
 
-// Main is the entrypoint into the Batch Submitter. This method returns a
-// closure that executes the service and blocks until the service exits. The use
-// of a closure allows the parameters bound to the top-level main package, e.g.
-// GitVersion, to be captured and used once the function is executed.
-func Main(version string, cliCtx *cli.Context) error {
-	if err := flags.CheckRequired(cliCtx); err != nil {
-		return err
-	}
-	cfg := NewConfig(cliCtx)
-	if err := cfg.Check(); err != nil {
-		return fmt.Errorf("invalid CLI flags: %w", err)
-	}
-
-	l := oplog.NewLogger(oplog.AppOut(cliCtx), cfg.LogConfig)
-	oplog.SetGlobalLogHandler(l.GetHandler())
-	opservice.ValidateEnvVars(flags.EnvVarPrefix, flags.Flags, l)
-	procName := "default"
-	m := metrics.NewMetrics(procName)
-	l.Info("Initializing Batch Submitter")
-
-	batchSubmitter, err := NewBatchSubmitterFromCLIConfig(cfg, l, m)
-	if err != nil {
-		l.Error("Unable to create Batch Submitter", "error", err)
-		return err
-	}
-
-	if !cfg.Stopped {
-		if err := batchSubmitter.Start(); err != nil {
-			l.Error("Unable to start Batch Submitter", "error", err)
-			return err
-		}
-	}
-
-	defer batchSubmitter.StopIfRunning(context.Background())
-
-	pprofConfig := cfg.PprofConfig
-	if pprofConfig.Enabled {
-		l.Debug("starting pprof", "addr", pprofConfig.ListenAddr, "port", pprofConfig.ListenPort)
-		pprofSrv, err := oppprof.StartServer(pprofConfig.ListenAddr, pprofConfig.ListenPort)
-		if err != nil {
-			l.Error("failed to start pprof server", "err", err)
-			return err
+// Main is the entrypoint into the Batch Submitter.
+// This method returns a cliapp.LifecycleAction, to create an op-service CLI-lifecycle-managed batch-submitter with.
+func Main(version string) cliapp.LifecycleAction {
+	return func(cliCtx *cli.Context, closeApp context.CancelCauseFunc) (cliapp.Lifecycle, error) {
+		if err := flags.CheckRequired(cliCtx); err != nil {
+			return nil, err
 		}
-		l.Info("started pprof server", "addr", pprofSrv.Addr())
-		defer func() {
-			if err := pprofSrv.Stop(context.Background()); err != nil {
-				l.Error("failed to stop pprof server", "err", err)
-			}
-		}()
-	}
-
-	metricsCfg := cfg.MetricsConfig
-	if metricsCfg.Enabled {
-		l.Debug("starting metrics server", "addr", metricsCfg.ListenAddr, "port", metricsCfg.ListenPort)
-		metricsSrv, err := m.Start(metricsCfg.ListenAddr, metricsCfg.ListenPort)
-		if err != nil {
-			return fmt.Errorf("failed to start metrics server: %w", err)
+		cfg := NewConfig(cliCtx)
+		if err := cfg.Check(); err != nil {
+			return nil, fmt.Errorf("invalid CLI flags: %w", err)
 		}
-		l.Info("started metrics server", "addr", metricsSrv.Addr())
-		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())
-	}
-
-	server := oprpc.NewServer(
-		cfg.RPCFlag.ListenAddr,
-		cfg.RPCFlag.ListenPort,
-		version,
-		oprpc.WithLogger(l),
-	)
-	if cfg.RPCFlag.EnableAdmin {
-		adminAPI := rpc.NewAdminAPI(batchSubmitter, &m.RPCMetrics, l)
-		server.AddAPI(rpc.GetAdminAPI(adminAPI))
-		l.Info("Admin RPC enabled")
-	}
-	if err := server.Start(); err != nil {
-		return fmt.Errorf("error starting RPC server: %w", err)
-	}
 
-	m.RecordInfo(version)
-	m.RecordUp()
+		l := oplog.NewLogger(oplog.AppOut(cliCtx), cfg.LogConfig)
+		oplog.SetGlobalLogHandler(l.GetHandler())
+		opservice.ValidateEnvVars(flags.EnvVarPrefix, flags.Flags, l)
 
-	opio.BlockOnInterrupts()
-	if err := server.Stop(); err != nil {
-		l.Error("Error shutting down http server: %w", err)
+		l.Info("Initializing Batch Submitter")
+		return BatcherServiceFromCLIConfig(cliCtx.Context, version, cfg, l)
 	}
-	return nil
 }

op-batcher/batcher/config.go

@@ -3,52 +3,17 @@ package batcher
 import (
 	"time"
 
-	"github.com/ethereum/go-ethereum/ethclient"
-	"github.com/ethereum/go-ethereum/log"
 	"github.com/urfave/cli/v2"
 
 	"github.com/ethereum-optimism/optimism/op-batcher/compressor"
 	"github.com/ethereum-optimism/optimism/op-batcher/flags"
-	"github.com/ethereum-optimism/optimism/op-batcher/metrics"
-	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	oplog "github.com/ethereum-optimism/optimism/op-service/log"
 	opmetrics "github.com/ethereum-optimism/optimism/op-service/metrics"
 	oppprof "github.com/ethereum-optimism/optimism/op-service/pprof"
 	oprpc "github.com/ethereum-optimism/optimism/op-service/rpc"
-	"github.com/ethereum-optimism/optimism/op-service/sources"
 	"github.com/ethereum-optimism/optimism/op-service/txmgr"
 )
 
-type Config struct {
-	log        log.Logger
-	metr       metrics.Metricer
-	L1Client   *ethclient.Client
-	L2Client   *ethclient.Client
-	RollupNode *sources.RollupClient
-	TxManager  txmgr.TxManager
-
-	NetworkTimeout         time.Duration
-	PollInterval           time.Duration
-	MaxPendingTransactions uint64
-
-	// RollupConfig is queried at startup
-	Rollup *rollup.Config
-
-	// Channel builder parameters
-	Channel ChannelConfig
-}
-
-// Check ensures that the [Config] is valid.
-func (c *Config) Check() error {
-	if err := c.Rollup.Check(); err != nil {
-		return err
-	}
-	if err := c.Channel.Check(); err != nil {
-		return err
-	}
-	return nil
-}
-
 type CLIConfig struct {
 	// L1EthRpc is the HTTP provider URL for L1.
 	L1EthRpc string
@@ -92,11 +57,11 @@ type CLIConfig struct {
 	MetricsConfig    opmetrics.CLIConfig
 	PprofConfig      oppprof.CLIConfig
 	CompressorConfig compressor.CLIConfig
-	RPCFlag          oprpc.CLIConfig
+	RPC              oprpc.CLIConfig
 }
 
-func (c CLIConfig) Check() error {
-	// TODO: check the sanity of flags loaded directly https://github.com/ethereum-optimism/optimism/issues/7512
+func (c *CLIConfig) Check() error {
+	// TODO(7512): check the sanity of flags loaded directly https://github.com/ethereum-optimism/optimism/issues/7512
 
 	if err := c.MetricsConfig.Check(); err != nil {
 		return err
@@ -107,15 +72,15 @@ func (c CLIConfig) Check() error {
 	if err := c.TxMgrConfig.Check(); err != nil {
 		return err
 	}
-	if err := c.RPCFlag.Check(); err != nil {
+	if err := c.RPC.Check(); err != nil {
 		return err
 	}
 	return nil
 }
 
 // NewConfig parses the Config from the provided flags or environment variables.
-func NewConfig(ctx *cli.Context) CLIConfig {
-	return CLIConfig{
+func NewConfig(ctx *cli.Context) *CLIConfig {
+	return &CLIConfig{
 		/* Required Flags */
 		L1EthRpc:        ctx.String(flags.L1EthRpcFlag.Name),
 		L2EthRpc:        ctx.String(flags.L2EthRpcFlag.Name),
@@ -133,6 +98,6 @@ func NewConfig(ctx *cli.Context) CLIConfig {
 		MetricsConfig:          opmetrics.ReadCLIConfig(ctx),
 		PprofConfig:            oppprof.ReadCLIConfig(ctx),
 		CompressorConfig:       compressor.ReadCLIConfig(ctx),
-		RPCFlag:                oprpc.ReadCLIConfig(ctx),
+		RPC:                    oprpc.ReadCLIConfig(ctx),
 	}
 }

op-batcher/cmd/main.go

@@ -30,7 +30,7 @@ func main() {
 	app.Name = "op-batcher"
 	app.Usage = "Batch Submitter Service"
 	app.Description = "Service for generating and submitting L2 tx batches to L1"
-	app.Action = curryMain(Version)
+	app.Action = cliapp.LifecycleCmd(batcher.Main(Version))
 	app.Commands = []*cli.Command{
 		{
 			Name:        "doc",
@@ -43,11 +43,3 @@ func main() {
 		log.Crit("Application failed", "message", err)
 	}
 }
-
-// curryMain transforms the batcher.Main function into an app.Action
-// This is done to capture the Version of the batcher.
-func curryMain(version string) func(ctx *cli.Context) error {
-	return func(ctx *cli.Context) error {
-		return batcher.Main(version, ctx)
-	}
-}

op-batcher/metrics/metrics.go

 package metrics
 
 import (
-	"context"
+	"io"
+
+	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/ethereum/go-ethereum/log"
-	"github.com/prometheus/client_golang/prometheus"
 
 	"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"
 )
@@ -28,6 +28,10 @@ type Metricer interface {
 	// Record Tx metrics
 	txmetrics.TxMetricer
 
+	opmetrics.RPCMetricer
+
+	StartBalanceMetrics(l log.Logger, client *ethclient.Client, account common.Address) io.Closer
+
 	RecordLatestL1Block(l1ref eth.L1BlockRef)
 	RecordL2BlocksLoaded(l2ref eth.L2BlockRef)
 	RecordChannelOpened(id derive.ChannelID, numPendingBlocks int)
@@ -79,6 +83,9 @@ type Metrics struct {
 
 var _ Metricer = (*Metrics)(nil)
 
+// implements the Registry getter, for metrics HTTP server to hook into
+var _ opmetrics.RegistryMetricer = (*Metrics)(nil)
+
 func NewMetrics(procName string) *Metrics {
 	if procName == "" {
 		procName = "default"
@@ -179,17 +186,16 @@ func NewMetrics(procName string) *Metrics {
 	}
 }
 
-func (m *Metrics) Start(host string, port int) (*httputil.HTTPServer, error) {
-	return opmetrics.StartServer(m.registry, host, port)
+func (m *Metrics) Registry() *prometheus.Registry {
+	return m.registry
 }
 
 func (m *Metrics) Document() []opmetrics.DocumentedMetric {
 	return m.factory.Document()
 }
 
-func (m *Metrics) StartBalanceMetrics(ctx context.Context,
-	l log.Logger, client *ethclient.Client, account common.Address) {
-	opmetrics.LaunchBalanceMetrics(ctx, l, m.registry, m.ns, client, account)
+func (m *Metrics) StartBalanceMetrics(l log.Logger, client *ethclient.Client, account common.Address) io.Closer {
+	return opmetrics.LaunchBalanceMetrics(l, m.registry, m.ns, client, account)
 }
 
 // RecordInfo sets a pseudo-metric that contains versioning and

op-batcher/metrics/noop.go

 package metrics
 
 import (
+	"io"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/ethereum/go-ethereum/log"
+
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 	opmetrics "github.com/ethereum-optimism/optimism/op-service/metrics"
 	txmetrics "github.com/ethereum-optimism/optimism/op-service/txmgr/metrics"
-	"github.com/ethereum/go-ethereum/core/types"
 )
 
 type noopMetrics struct {
 	opmetrics.NoopRefMetrics
 	txmetrics.NoopTxMetrics
+	opmetrics.NoopRPCMetrics
 }
 
 var NoopMetrics Metricer = new(noopMetrics)
@@ -35,3 +42,6 @@ func (*noopMetrics) RecordChannelTimedOut(derive.ChannelID)       {}
 func (*noopMetrics) RecordBatchTxSubmitted() {}
 func (*noopMetrics) RecordBatchTxSuccess()   {}
 func (*noopMetrics) RecordBatchTxFailed()    {}
+func (*noopMetrics) StartBalanceMetrics(log.Logger, *ethclient.Client, common.Address) io.Closer {
+	return nil
+}

op-batcher/rpc/api.go

@@ -10,17 +10,17 @@ import (
 	"github.com/ethereum-optimism/optimism/op-service/rpc"
 )
 
-type batcherClient interface {
-	Start() error
-	Stop(ctx context.Context) error
+type BatcherDriver interface {
+	StartBatchSubmitting() error
+	StopBatchSubmitting(ctx context.Context) error
 }
 
 type adminAPI struct {
 	*rpc.CommonAdminAPI
-	b batcherClient
+	b BatcherDriver
 }
 
-func NewAdminAPI(dr batcherClient, m metrics.RPCMetricer, log log.Logger) *adminAPI {
+func NewAdminAPI(dr BatcherDriver, m metrics.RPCMetricer, log log.Logger) *adminAPI {
 	return &adminAPI{
 		CommonAdminAPI: rpc.NewCommonAdminAPI(m, log),
 		b:              dr,
@@ -35,9 +35,9 @@ func GetAdminAPI(api *adminAPI) gethrpc.API {
 }
 
 func (a *adminAPI) StartBatcher(_ context.Context) error {
-	return a.b.Start()
+	return a.b.StartBatchSubmitting()
 }
 
 func (a *adminAPI) StopBatcher(ctx context.Context) error {
-	return a.b.Stop(ctx)
+	return a.b.StopBatchSubmitting(ctx)
 }

op-challenger/game/fault/trace/outputs/provider.go

@@ -34,7 +34,7 @@ type OutputTraceProvider struct {
 }
 
 func NewTraceProvider(ctx context.Context, logger log.Logger, rollupRpc string, gameDepth, prestateBlock, poststateBlock uint64) (*OutputTraceProvider, error) {
-	rollupClient, err := dial.DialRollupClientWithTimeout(dial.DefaultDialTimeout, logger, rollupRpc)
+	rollupClient, err := dial.DialRollupClientWithTimeout(ctx, dial.DefaultDialTimeout, logger, rollupRpc)
 	if err != nil {
 		return nil, err
 	}

op-challenger/game/service.go

@@ -55,7 +55,7 @@ func NewService(ctx context.Context, logger log.Logger, cfg *config.Config) (*Se
 		return nil, fmt.Errorf("failed to create the transaction manager: %w", err)
 	}
 
-	l1Client, err := dial.DialEthClientWithTimeout(dial.DefaultDialTimeout, logger, cfg.L1EthRpc)
+	l1Client, err := dial.DialEthClientWithTimeout(ctx, dial.DefaultDialTimeout, logger, cfg.L1EthRpc)
 	if err != nil {
 		return nil, fmt.Errorf("failed to dial L1: %w", err)
 	}

op-challenger/metrics/metrics.go

@@ -133,7 +133,12 @@ func (m *Metrics) StartBalanceMetrics(
 	client *ethclient.Client,
 	account common.Address,
 ) {
-	opmetrics.LaunchBalanceMetrics(ctx, l, m.registry, m.ns, client, account)
+	// TODO(7684): util was refactored to close, but ctx is still being used by caller for shutdown
+	balanceMetric := opmetrics.LaunchBalanceMetrics(l, m.registry, m.ns, client, account)
+	go func() {
+		<-ctx.Done()
+		_ = balanceMetric.Close()
+	}()
 }
 
 // RecordInfo sets a pseudo-metric that contains versioning and

op-e2e/setup.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/ecdsa"
 	"crypto/rand"
+	"errors"
 	"fmt"
 	"math/big"
 	"net"
@@ -37,7 +38,6 @@ import (
 
 	bss "github.com/ethereum-optimism/optimism/op-batcher/batcher"
 	"github.com/ethereum-optimism/optimism/op-batcher/compressor"
-	batchermetrics "github.com/ethereum-optimism/optimism/op-batcher/metrics"
 	"github.com/ethereum-optimism/optimism/op-bindings/predeploys"
 	"github.com/ethereum-optimism/optimism/op-chain-ops/genesis"
 	"github.com/ethereum-optimism/optimism/op-e2e/config"
@@ -254,7 +254,7 @@ type System struct {
 	RawClients        map[string]*rpc.Client
 	RollupNodes       map[string]*rollupNode.OpNode
 	L2OutputSubmitter *l2os.L2OutputSubmitter
-	BatchSubmitter    *bss.BatchSubmitter
+	BatchSubmitter    *bss.BatcherService
 	Mocknet           mocknet.Mocknet
 
 	// TimeTravelClock is nil unless SystemConfig.SupportL1TimeTravel was set to true
@@ -270,18 +270,16 @@ func (sys *System) NodeEndpoint(name string) string {
 }
 
 func (sys *System) Close() {
+	postCtx, postCancel := context.WithCancel(context.Background())
+	postCancel() // immediate shutdown, no allowance for idling
+
 	if sys.L2OutputSubmitter != nil {
 		sys.L2OutputSubmitter.Stop()
 	}
 	if sys.BatchSubmitter != nil {
-		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-		defer cancel()
-		sys.BatchSubmitter.StopIfRunning(ctx)
+		_ = sys.BatchSubmitter.Kill()
 	}
 
-	postCtx, postCancel := context.WithCancel(context.Background())
-	postCancel() // immediate shutdown, no allowance for idling
-
 	for _, node := range sys.RollupNodes {
 		_ = node.Stop(postCtx)
 	}
@@ -681,8 +679,7 @@ func (cfg SystemConfig) Start(t *testing.T, _opts ...SystemConfigOption) (*Syste
 		return nil, fmt.Errorf("unable to start l2 output submitter: %w", err)
 	}
 
-	// Batch Submitter
-	sys.BatchSubmitter, err = bss.NewBatchSubmitterFromCLIConfig(bss.CLIConfig{
+	batcherCLIConfig := &bss.CLIConfig{
 		L1EthRpc:               sys.EthInstances["l1"].WSEndpoint(),
 		L2EthRpc:               sys.EthInstances["sequencer"].WSEndpoint(),
 		RollupRpc:              sys.RollupNodes["sequencer"].HTTPEndpoint(),
@@ -701,17 +698,17 @@ func (cfg SystemConfig) Start(t *testing.T, _opts ...SystemConfigOption) (*Syste
 			Level:  log.LvlInfo,
 			Format: oplog.FormatText,
 		},
-	}, sys.cfg.Loggers["batcher"], batchermetrics.NoopMetrics)
+		Stopped: sys.cfg.DisableBatcher, // Batch submitter may be enabled later
+	}
+	// Batch Submitter
+	batcher, err := bss.BatcherServiceFromCLIConfig(context.Background(), "0.0.1", batcherCLIConfig, sys.cfg.Loggers["batcher"])
 	if err != nil {
 		return nil, fmt.Errorf("failed to setup batch submitter: %w", err)
 	}
-
-	// Batcher may be enabled later
-	if !sys.cfg.DisableBatcher {
-		if err := sys.BatchSubmitter.Start(); err != nil {
-			return nil, fmt.Errorf("unable to start batch submitter: %w", err)
-		}
+	if err := batcher.Start(context.Background()); err != nil {
+		return nil, errors.Join(fmt.Errorf("failed to start batch submitter: %w", err), batcher.Stop(context.Background()))
 	}
+	sys.BatchSubmitter = batcher
 
 	return sys, nil
 }

op-e2e/system_fpp_test.go

@@ -93,7 +93,7 @@ func testVerifyL2OutputRootEmptyBlock(t *testing.T, detached bool) {
 	l2OutputRoot := agreedL2Output.OutputRoot
 
 	t.Log("=====Stopping batch submitter=====")
-	err = sys.BatchSubmitter.Stop(ctx)
+	err = sys.BatchSubmitter.Driver().StopBatchSubmitting(ctx)
 	require.NoError(t, err, "could not stop batch submitter")
 
 	// Wait for the sequencer to catch up with the current L1 head so we know all submitted batches are processed
@@ -121,7 +121,7 @@ func testVerifyL2OutputRootEmptyBlock(t *testing.T, detached bool) {
 	l2Claim := l2Output.OutputRoot
 
 	t.Log("=====Restarting batch submitter=====")
-	err = sys.BatchSubmitter.Start()
+	err = sys.BatchSubmitter.Driver().StartBatchSubmitting()
 	require.NoError(t, err, "could not start batch submitter")
 
 	t.Log("Add a transaction to the next batch after sequence of empty blocks")
@@ -258,7 +258,7 @@ func testFaultProofProgramScenario(t *testing.T, ctx context.Context, sys *Syste
 
 	t.Log("Shutting down network")
 	// Shutdown the nodes from the actual chain. Should now be able to run using only the pre-fetched data.
-	sys.BatchSubmitter.StopIfRunning(context.Background())
+	require.NoError(t, sys.BatchSubmitter.Kill())
 	sys.L2OutputSubmitter.Stop()
 	sys.L2OutputSubmitter = nil
 	for _, node := range sys.EthInstances {

op-e2e/system_test.go

@@ -1266,7 +1266,7 @@ func TestStopStartBatcher(t *testing.T) {
 	require.Greater(t, newSeqStatus.SafeL2.Number, seqStatus.SafeL2.Number, "Safe chain did not advance")
 
 	// stop the batch submission
-	err = sys.BatchSubmitter.Stop(context.Background())
+	err = sys.BatchSubmitter.Driver().StopBatchSubmitting(context.Background())
 	require.Nil(t, err)
 
 	// wait for any old safe blocks being submitted / derived
@@ -1286,7 +1286,7 @@ func TestStopStartBatcher(t *testing.T) {
 	require.Equal(t, newSeqStatus.SafeL2.Number, seqStatus.SafeL2.Number, "Safe chain advanced while batcher was stopped")
 
 	// start the batch submission
-	err = sys.BatchSubmitter.Start()
+	err = sys.BatchSubmitter.Driver().StartBatchSubmitting()
 	require.Nil(t, err)
 	time.Sleep(safeBlockInclusionDuration)
 
@@ -1325,7 +1325,7 @@ func TestBatcherMultiTx(t *testing.T) {
 	require.Nil(t, err)
 
 	// start batch submission
-	err = sys.BatchSubmitter.Start()
+	err = sys.BatchSubmitter.Driver().StartBatchSubmitting()
 	require.Nil(t, err)
 
 	totalTxCount := 0

op-node/metrics/metrics.go

@@ -604,7 +604,9 @@ func (m *Metrics) ReportProtocolVersions(local, engine, recommended, required pa
 	m.ProtocolVersions.WithLabelValues(local.String(), engine.String(), recommended.String(), required.String()).Set(1)
 }
 
-type noopMetricer struct{}
+type noopMetricer struct {
+	metrics.NoopRPCMetrics
+}
 
 var NoopMetrics Metricer = new(noopMetricer)
 
@@ -614,17 +616,6 @@ func (n *noopMetricer) RecordInfo(version string) {
 func (n *noopMetricer) RecordUp() {
 }
 
-func (n *noopMetricer) RecordRPCServerRequest(method string) func() {
-	return func() {}
-}
-
-func (n *noopMetricer) RecordRPCClientRequest(method string) func(err error) {
-	return func(err error) {}
-}
-
-func (n *noopMetricer) RecordRPCClientResponse(method string, err error) {
-}
-
 func (n *noopMetricer) SetDerivationIdle(status bool) {
 }
 

op-proposer/metrics/metrics.go

@@ -84,7 +84,12 @@ func (m *Metrics) Start(host string, port int) (*httputil.HTTPServer, error) {
 
 func (m *Metrics) StartBalanceMetrics(ctx context.Context,
 	l log.Logger, client *ethclient.Client, account common.Address) {
-	opmetrics.LaunchBalanceMetrics(ctx, l, m.registry, m.ns, client, account)
+	// TODO(7684): util was refactored to close, but ctx is still being used by caller for shutdown
+	balanceMetric := opmetrics.LaunchBalanceMetrics(l, m.registry, m.ns, client, account)
+	go func() {
+		<-ctx.Done()
+		_ = balanceMetric.Close()
+	}()
 }
 
 // RecordInfo sets a pseudo-metric that contains versioning and

op-proposer/proposer/l2_output_submitter.go

@@ -172,12 +172,12 @@ func NewL2OutputSubmitterConfigFromCLIConfig(cfg CLIConfig, l log.Logger, m metr
 	}
 
 	// Connect to L1 and L2 providers. Perform these last since they are the most expensive.
-	l1Client, err := dial.DialEthClientWithTimeout(dial.DefaultDialTimeout, l, cfg.L1EthRpc)
+	l1Client, err := dial.DialEthClientWithTimeout(context.Background(), dial.DefaultDialTimeout, l, cfg.L1EthRpc)
 	if err != nil {
 		return nil, err
 	}
 
-	rollupClient, err := dial.DialRollupClientWithTimeout(dial.DefaultDialTimeout, l, cfg.RollupRpc)
+	rollupClient, err := dial.DialRollupClientWithTimeout(context.Background(), dial.DefaultDialTimeout, l, cfg.RollupRpc)
 	if err != nil {
 		return nil, err
 	}

op-service/clock/loop.go

+package clock
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+// LoopFn is a simple ticker-loop with io.Closer support.
+// Note that ticks adapt; slow function calls may result in lost ticks.
+type LoopFn struct {
+	ctx    context.Context
+	cancel context.CancelFunc
+
+	ticker  Ticker
+	fn      func(ctx context.Context)
+	onClose func() error
+
+	wg sync.WaitGroup
+}
+
+// Close cancels the context of the ongoing function call, waits for the call to complete, and cancels further calls.
+// Close is safe to call again or concurrently. The onClose callback will be called for each Close call.
+func (lf *LoopFn) Close() error {
+	lf.cancel()  // stop any ongoing function call, and close the main loop
+	lf.wg.Wait() // wait for completion
+	if lf.onClose != nil {
+		return lf.onClose() // optional: user can specify function to close resources with
+	}
+	return nil
+}
+
+func (lf *LoopFn) work() {
+	defer lf.wg.Done()
+	defer lf.ticker.Stop() // clean up the timer
+	for {
+		select {
+		case <-lf.ctx.Done():
+			return
+		case <-lf.ticker.Ch():
+			ctx, cancel := context.WithCancel(lf.ctx)
+			func() {
+				defer cancel()
+				lf.fn(ctx)
+			}()
+		}
+	}
+}
+
+// NewLoopFn creates a periodic function call, which can be closed,
+// with an optional onClose callback to clean up resources.
+func NewLoopFn(clock Clock, fn func(ctx context.Context), onClose func() error, interval time.Duration) *LoopFn {
+	ctx, cancel := context.WithCancel(context.Background())
+	lf := &LoopFn{
+		ctx:     ctx,
+		cancel:  cancel,
+		fn:      fn,
+		ticker:  clock.NewTicker(interval),
+		onClose: onClose,
+	}
+	lf.wg.Add(1)
+	go lf.work()
+	return lf
+}

op-service/clock/loop_test.go

+package clock
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestLoopFn(t *testing.T) {
+	cl := NewDeterministicClock(time.Now())
+	calls := make(chan struct{}, 10)
+	testErr := errors.New("test close error")
+	loopFn := NewLoopFn(cl, func(ctx context.Context) {
+		calls <- struct{}{}
+	}, func() error {
+		close(calls)
+		return testErr
+	}, time.Second*10)
+	cl.AdvanceTime(time.Second * 15)
+	<-calls
+	cl.AdvanceTime(time.Second * 10)
+	<-calls
+	select {
+	case <-calls:
+		t.Fatal("more calls than expected")
+	default:
+	}
+	require.ErrorIs(t, loopFn.Close(), testErr)
+}

op-service/dial/dial.go

@@ -21,8 +21,8 @@ const defaultRetryTime = 2 * time.Second
 // DialEthClientWithTimeout attempts to dial the L1 provider using the provided
 // URL. If the dial doesn't complete within defaultDialTimeout seconds, this
 // method will return an error.
-func DialEthClientWithTimeout(timeout time.Duration, log log.Logger, url string) (*ethclient.Client, error) {
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+func DialEthClientWithTimeout(ctx context.Context, timeout time.Duration, log log.Logger, url string) (*ethclient.Client, error) {
+	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 
 	c, err := dialRPCClientWithBackoff(ctx, log, url)
@@ -35,8 +35,8 @@ func DialEthClientWithTimeout(timeout time.Duration, log log.Logger, url string)
 
 // DialRollupClientWithTimeout attempts to dial the RPC provider using the provided URL.
 // If the dial doesn't complete within timeout seconds, this method will return an error.
-func DialRollupClientWithTimeout(timeout time.Duration, log log.Logger, url string) (*sources.RollupClient, error) {
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+func DialRollupClientWithTimeout(ctx context.Context, timeout time.Duration, log log.Logger, url string) (*sources.RollupClient, error) {
+	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 
 	rpcCl, err := dialRPCClientWithBackoff(ctx, log, url)

op-service/metrics/balance.go

@@ -5,12 +5,15 @@ import (
 	"math/big"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/params"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
+
+	"github.com/ethereum-optimism/optimism/op-service/clock"
 )
 
 // weiToEther divides the wei value by 10^18 to get a number in ether as a float64
@@ -22,38 +25,27 @@ func weiToEther(wei *big.Int) float64 {
 	return f
 }
 
-// LaunchBalanceMetrics fires off a go rountine that queries the balance of the supplied account & periodically records it
-// to the balance metric of the namespace. The balance of the account is recorded in Ether (not Wei).
+// LaunchBalanceMetrics starts a periodic query of the balance of the supplied account and records it
+// to the "balance" metric of the namespace. The balance of the account is recorded in Ether (not Wei).
 // Cancel the supplied context to shut down the go routine
-func LaunchBalanceMetrics(ctx context.Context, log log.Logger, r *prometheus.Registry, ns string, client *ethclient.Client, account common.Address) {
-	go func() {
-		balanceGuage := promauto.With(r).NewGauge(prometheus.GaugeOpts{
-			Namespace: ns,
-			Name:      "balance",
-			Help:      "balance (in ether) of account " + account.String(),
-		})
-
-		ticker := time.NewTicker(10 * time.Second)
-		defer ticker.Stop()
-
-		for {
-			select {
-			case <-ticker.C:
-				ctx, cancel := context.WithTimeout(ctx, 1*time.Minute)
-				bigBal, err := client.BalanceAt(ctx, account, nil)
-				if err != nil {
-					log.Warn("failed to get balance of account", "err", err, "address", account)
-					cancel()
-					continue
-				}
-				bal := weiToEther(bigBal)
-				balanceGuage.Set(bal)
-				cancel()
-			case <-ctx.Done():
-				log.Info("balance metrics shutting down")
-				return
-			}
+func LaunchBalanceMetrics(log log.Logger, r *prometheus.Registry, ns string, client *ethclient.Client, account common.Address) *clock.LoopFn {
+	balanceGuage := promauto.With(r).NewGauge(prometheus.GaugeOpts{
+		Namespace: ns,
+		Name:      "balance",
+		Help:      "balance (in ether) of account " + account.String(),
+	})
+	return clock.NewLoopFn(clock.SystemClock, func(ctx context.Context) {
+		ctx, cancel := context.WithTimeout(ctx, 1*time.Minute)
+		defer cancel()
+		bigBal, err := client.BalanceAt(ctx, account, nil)
+		if err != nil {
+			log.Warn("failed to get balance of account", "err", err, "address", account)
+			return
 		}
-
-	}()
+		bal := weiToEther(bigBal)
+		balanceGuage.Set(bal)
+	}, func() error {
+		log.Info("balance metrics shutting down")
+		return nil
+	}, 10*time.Second)
 }

op-service/metrics/registry.go

@@ -11,3 +11,7 @@ func NewRegistry() *prometheus.Registry {
 	registry.MustRegister(collectors.NewGoCollector())
 	return registry
 }
+
+type RegistryMetricer interface {
+	Registry() *prometheus.Registry
+}

op-service/metrics/rpc_metrics.go

@@ -125,3 +125,17 @@ func (m *RPCMetrics) RecordRPCClientResponse(method string, err error) {
 	}
 	m.RPCClientResponsesTotal.WithLabelValues(method, errStr).Inc()
 }
+
+type NoopRPCMetrics struct{}
+
+func (n *NoopRPCMetrics) RecordRPCServerRequest(method string) func() {
+	return func() {}
+}
+
+func (n *NoopRPCMetrics) RecordRPCClientRequest(method string) func(err error) {
+	return func(err error) {}
+}
+func (n *NoopRPCMetrics) RecordRPCClientResponse(method string, err error) {
+}
+
+var _ RPCMetricer = (*NoopRPCMetrics)(nil)

op-service/sources/rollupclient.go

@@ -63,3 +63,7 @@ func (r *RollupClient) SequencerActive(ctx context.Context) (bool, error) {
 func (r *RollupClient) SetLogLevel(ctx context.Context, lvl log.Lvl) error {
 	return r.rpc.CallContext(ctx, nil, "admin_setLogLevel", lvl.String())
 }
+
+func (r *RollupClient) Close() {
+	r.rpc.Close()
+}