Finish sequencer control (#8684)

op-conductor/client/sequencer.go

+package client
+
+import (
+	"context"
+
+	"github.com/ethereum/go-ethereum/common"
+
+	"github.com/ethereum-optimism/optimism/op-service/eth"
+	"github.com/ethereum-optimism/optimism/op-service/sources"
+)
+
+// SequencerControl defines the interface for controlling the sequencer.
+type SequencerControl interface {
+	StartSequencer(ctx context.Context, hash common.Hash) error
+	StopSequencer(ctx context.Context) (common.Hash, error)
+	LatestUnsafeBlock(ctx context.Context) (eth.BlockInfo, error)
+}
+
+// NewSequencerControl creates a new SequencerControl instance.
+func NewSequencerControl(exec *sources.EthClient, node *sources.RollupClient) SequencerControl {
+	return &sequencerController{
+		exec: exec,
+		node: node,
+	}
+}
+
+type sequencerController struct {
+	exec *sources.EthClient
+	node *sources.RollupClient
+}
+
+var _ SequencerControl = (*sequencerController)(nil)
+
+// LatestUnsafeBlock implements SequencerControl.
+func (s *sequencerController) LatestUnsafeBlock(ctx context.Context) (eth.BlockInfo, error) {
+	return s.exec.InfoByLabel(ctx, eth.Unsafe)
+}
+
+// StartSequencer implements SequencerControl.
+func (s *sequencerController) StartSequencer(ctx context.Context, hash common.Hash) error {
+	return s.node.StartSequencer(ctx, hash)
+}
+
+// StopSequencer implements SequencerControl.
+func (s *sequencerController) StopSequencer(ctx context.Context) (common.Hash, error) {
+	return s.node.StopSequencer(ctx)
+}

op-conductor/conductor/config.go

@@ -35,6 +35,9 @@ type Config struct {
 	// NodeRPC is the HTTP provider URL for op-node.
 	NodeRPC string
 
+	// ExecutionRPC is the HTTP provider URL for execution layer.
+	ExecutionRPC string
+
 	RollupCfg rollup.Config
 
 	LogConfig     oplog.CLIConfig
@@ -60,6 +63,9 @@ func (c *Config) Check() error {
 	if c.NodeRPC == "" {
 		return fmt.Errorf("missing node RPC")
 	}
+	if c.ExecutionRPC == "" {
+		return fmt.Errorf("missing geth RPC")
+	}
 	if err := c.RollupCfg.Check(); err != nil {
 		return errors.Wrap(err, "invalid rollup config")
 	}
@@ -92,6 +98,7 @@ func NewConfig(ctx *cli.Context, log log.Logger) (*Config, error) {
 		RaftServerID:   ctx.String(flags.RaftServerID.Name),
 		RaftStorageDir: ctx.String(flags.RaftStorageDir.Name),
 		NodeRPC:        ctx.String(flags.NodeRPC.Name),
+		ExecutionRPC:   ctx.String(flags.ExecutionRPC.Name),
 		RollupCfg:      *rollupCfg,
 		LogConfig:      oplog.ReadCLIConfig(ctx),
 		MetricsConfig:  opmetrics.ReadCLIConfig(ctx),

op-conductor/conductor/service.go

@@ -4,13 +4,66 @@ import (
 	"context"
 
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/hashicorp/go-multierror"
+	"github.com/pkg/errors"
 
+	"github.com/ethereum-optimism/optimism/op-conductor/client"
 	"github.com/ethereum-optimism/optimism/op-service/cliapp"
+	opclient "github.com/ethereum-optimism/optimism/op-service/client"
+	"github.com/ethereum-optimism/optimism/op-service/sources"
 )
 
 // New creates a new OpConductor instance.
 func New(ctx context.Context, cfg *Config, log log.Logger, version string) (*OpConductor, error) {
-	panic("unimplemented")
+	if err := cfg.Check(); err != nil {
+		return nil, errors.Wrap(err, "invalid config")
+	}
+
+	oc := &OpConductor{
+		log:     log,
+		version: version,
+		cfg:     cfg,
+	}
+
+	err := oc.init(ctx)
+	if err != nil {
+		log.Error("failed to initialize OpConductor", "err", err)
+		// ensure we always close the resources if we fail to initialize the conductor.
+		if closeErr := oc.Stop(ctx); closeErr != nil {
+			return nil, multierror.Append(err, closeErr)
+		}
+	}
+
+	return oc, nil
+}
+
+func (c *OpConductor) init(ctx context.Context) error {
+	c.log.Info("initializing OpConductor", "version", c.version)
+	if err := c.initSequencerControl(ctx); err != nil {
+		return errors.Wrap(err, "failed to initialize sequencer control")
+	}
+	return nil
+}
+
+func (c *OpConductor) initSequencerControl(ctx context.Context) error {
+	ec, err := opclient.NewRPC(ctx, c.log, c.cfg.ExecutionRPC)
+	if err != nil {
+		return errors.Wrap(err, "failed to create geth rpc client")
+	}
+	gethCfg := sources.L2ClientDefaultConfig(&c.cfg.RollupCfg, true)
+	// TODO: Add metrics tracer here. tracked by https://github.com/ethereum-optimism/protocol-quest/issues/45
+	geth, err := sources.NewEthClient(ec, c.log, nil, &gethCfg.EthClientConfig)
+	if err != nil {
+		return errors.Wrap(err, "failed to create geth client")
+	}
+
+	nc, err := opclient.NewRPC(ctx, c.log, c.cfg.NodeRPC)
+	if err != nil {
+		return errors.Wrap(err, "failed to create node rpc client")
+	}
+	node := sources.NewRollupClient(nc)
+	c.ctrl = client.NewSequencerControl(geth, node)
+	return nil
 }
 
 // OpConductor represents a full conductor instance and its resources, it does:
@@ -23,7 +76,12 @@ func New(ctx context.Context, cfg *Config, log log.Logger, version string) (*OpC
 //  2. paused: control loop (sequencer start/stop) is paused, but it still participates in leader election.
 //     it is paused for disaster recovery situation
 //  3. stopped: it is stopped, which means it is not participating in leader election and control loop. OpConductor cannot be started again from stopped mode.
-type OpConductor struct{}
+type OpConductor struct {
+	log     log.Logger
+	version string
+	cfg     *Config
+	ctrl    client.SequencerControl
+}
 
 var _ cliapp.Lifecycle = (*OpConductor)(nil)
 

op-conductor/flags/flags.go

@@ -43,6 +43,11 @@ var (
 		Usage:   "HTTP provider URL for op-node",
 		EnvVars: opservice.PrefixEnvVar(EnvVarPrefix, "NODE_RPC"),
 	}
+	ExecutionRPC = &cli.StringFlag{
+		Name:    "execution.rpc",
+		Usage:   "HTTP provider URL for execution layer",
+		EnvVars: opservice.PrefixEnvVar(EnvVarPrefix, "EXECUTION_RPC"),
+	}
 )
 
 var requiredFlags = []cli.Flag{
@@ -51,6 +56,7 @@ var requiredFlags = []cli.Flag{
 	RaftServerID,
 	RaftStorageDir,
 	NodeRPC,
+	ExecutionRPC,
 }
 
 var optionalFlags = []cli.Flag{}