Add component outputs (#64)

Author: ferranbt

internal/artifacts.go

@@ -37,6 +37,7 @@ import (
 
 var (
 	defaultDiscoveryPrivKey = "a11ac89899cd86e36b6fb881ec1255b8a92a688790b7d950f8b7d8dd626671fb"
+	defaultDiscoveryEnodeID = "3479db4d9217fb5d7a8ed4d61ac36e120b05d36c2eefb795dc42ff2e971f251a2315f5649ea1833271e020b9adc98d5db9973c7ed92d6b2f1f2223088c3d852f"
 )
 
 // minimumGenesisDelay is the minimum delay for the genesis time. This is required

internal/components.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"strings"
 	"time"
 )
 
@@ -91,6 +92,9 @@ func (o *OpNode) Run(service *service, ctx *ExContext) {
 
 type OpGeth struct {
 	UseDeterministicP2PKey bool
+
+	// outputs
+	Enode string
 }
 
 func logLevelToGethVerbosity(logLevel LogLevel) string {
@@ -155,6 +159,28 @@ func (o *OpGeth) Run(service *service, ctx *ExContext) {
 		)
 }
 
+var _ ServiceReady = &OpGeth{}
+
+func (o *OpGeth) Ready(out io.Writer, service *service, ctx context.Context) error {
+	logs := service.logs
+
+	if err := logs.WaitForLog("HTTP server started", 5*time.Second); err != nil {
+		return err
+	}
+
+	enodeLine, err := logs.FindLog("enode://")
+	if err != nil {
+		return err
+	}
+
+	parts := strings.Split(enodeLine, "enode://")[1]
+	enodeID := strings.Split(parts, "@")[0]
+
+	enode := fmt.Sprintf("enode://%s@127.0.0.1:%d?discport=0", enodeID, service.MustGetPort("rpc").HostPort)
+	o.Enode = enode
+	return nil
+}
+
 var _ ServiceWatchdog = &OpGeth{}
 
 func (o *OpGeth) Watchdog(out io.Writer, service *service, ctx context.Context) error {
@@ -293,7 +319,7 @@ var _ ServiceReady = &LighthouseBeaconNode{}
 func (l *LighthouseBeaconNode) Ready(logOutput io.Writer, service *service, ctx context.Context) error {
 	beaconNodeURL := fmt.Sprintf("http://localhost:%d", service.MustGetPort("http").HostPort)
 
-	if err := waitForChainAlive(logOutput, beaconNodeURL, 30*time.Second); err != nil {
+	if err := waitForChainAlive(ctx, logOutput, beaconNodeURL, 30*time.Second); err != nil {
 		return err
 	}
 	return nil

internal/local_runner.go

@@ -56,10 +56,15 @@ type LocalRunner struct {
 
 	// tasks tracks the status of each service
 	tasksMtx     sync.Mutex
-	tasks        map[string]string
+	tasks        map[string]*task
 	taskUpdateCh chan struct{}
 }
 
+type task struct {
+	status string
+	logs   *os.File
+}
+
 var (
 	taskStatusPending = "pending"
 	taskStatusStarted = "started"
@@ -86,9 +91,12 @@ func NewLocalRunner(out *output, manifest *Manifest, overrides map[string]string
 		overrides[k] = v
 	}
 
-	tasks := map[string]string{}
+	tasks := map[string]*task{}
 	for _, svc := range manifest.services {
-		tasks[svc.Name] = taskStatusPending
+		tasks[svc.Name] = &task{
+			status: taskStatusPending,
+			logs:   nil,
+		}
 	}
 
 	d := &LocalRunner{
@@ -153,7 +161,7 @@ func (d *LocalRunner) printStatus() {
 			lineOffset = 0
 			// Use ordered services instead of ranging over map
 			for _, name := range orderedServices {
-				status := d.tasks[name]
+				status := d.tasks[name].status
 				var statusLine string
 
 				switch status {
@@ -183,7 +191,7 @@ func (d *LocalRunner) printStatus() {
 func (d *LocalRunner) updateTaskStatus(name string, status string) {
 	d.tasksMtx.Lock()
 	defer d.tasksMtx.Unlock()
-	d.tasks[name] = status
+	d.tasks[name].status = status
 
 	if status == taskStatusDie {
 		d.exitErr <- fmt.Errorf("container %s failed", name)
@@ -482,9 +490,12 @@ func (d *LocalRunner) runOnHost(ss *service) error {
 
 // trackLogs tracks the logs of a container and writes them to the log output
 func (d *LocalRunner) trackLogs(serviceName string, containerID string) error {
-	log_output, err := d.out.LogOutput(serviceName)
-	if err != nil {
-		return fmt.Errorf("error getting log output: %w", err)
+	d.tasksMtx.Lock()
+	log_output := d.tasks[serviceName].logs
+	d.tasksMtx.Unlock()
+
+	if log_output == nil {
+		panic("BUG: log output not found for service " + serviceName)
 	}
 
 	logs, err := d.client.ContainerLogs(context.Background(), containerID, container.LogsOptions{
@@ -546,6 +557,18 @@ func (d *LocalRunner) Run() error {
 		return fmt.Errorf("failed to write docker-compose.yaml: %w", err)
 	}
 
+	// generate the output log file for each service so that it is available after Run is done
+	for _, svc := range d.manifest.services {
+		log_output, err := d.out.LogOutput(svc.Name)
+		if err != nil {
+			return fmt.Errorf("error getting log output: %w", err)
+		}
+		svc.logs = &serviceLogs{
+			path: log_output.Name(),
+		}
+		d.tasks[svc.Name].logs = log_output
+	}
+
 	// First start the services that are running in docker-compose
 	cmd := exec.Command("docker", "compose", "-f", d.out.dst+"/docker-compose.yaml", "up", "-d")
 

internal/manifest.go

@@ -4,9 +4,11 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"os"
 	"strings"
 	"sync"
 	"text/template"
+	"time"
 
 	flag "github.com/spf13/pflag"
 )
@@ -19,6 +21,7 @@ type Recipe interface {
 	Flags() *flag.FlagSet
 	Artifacts() *ArtifactsBuilder
 	Apply(ctx *ExContext, artifacts *Artifacts) *Manifest
+	Output(manifest *Manifest) map[string]interface{}
 }
 
 // Manifest describes a list of services and their dependencies
@@ -80,7 +83,7 @@ type ServiceReady interface {
 	Ready(out io.Writer, service *service, ctx context.Context) error
 }
 
-func WaitForReady(manifest *Manifest) error {
+func WaitForReady(ctx context.Context, manifest *Manifest) error {
 	var wg sync.WaitGroup
 	readyErr := make(chan error, len(manifest.Services()))
 
@@ -96,7 +99,7 @@ func WaitForReady(manifest *Manifest) error {
 			go func() {
 				defer wg.Done()
 
-				if err := readyFn.Ready(output, s, context.Background()); err != nil {
+				if err := readyFn.Ready(output, s, ctx); err != nil {
 					readyErr <- fmt.Errorf("service %s failed to start: %w", s.Name, err)
 				}
 			}()
@@ -240,6 +243,52 @@ type NodeRef struct {
 	PortLabel string
 }
 
+// serviceLogs is a service to access the logs of the running service
+type serviceLogs struct {
+	path string
+}
+
+func (s *serviceLogs) readLogs() (string, error) {
+	content, err := os.ReadFile(s.path)
+	if err != nil {
+		return "", fmt.Errorf("failed to read logs: %w", err)
+	}
+	return string(content), nil
+}
+
+func (s *serviceLogs) WaitForLog(pattern string, timeout time.Duration) error {
+	timer := time.After(timeout)
+	for {
+		select {
+		case <-timer:
+			return fmt.Errorf("timeout waiting for log pattern %s", pattern)
+		case <-time.After(500 * time.Millisecond):
+			logs, err := s.readLogs()
+			if err != nil {
+				return fmt.Errorf("failed to read logs: %w", err)
+			}
+			if strings.Contains(logs, pattern) {
+				return nil
+			}
+		}
+	}
+}
+
+func (s *serviceLogs) FindLog(pattern string) (string, error) {
+	logs, err := s.readLogs()
+	if err != nil {
+		return "", fmt.Errorf("failed to read logs: %w", err)
+	}
+
+	lines := strings.Split(logs, "\n")
+	for _, line := range lines {
+		if strings.Contains(line, pattern) {
+			return line, nil
+		}
+	}
+	return "", fmt.Errorf("log pattern %s not found", pattern)
+}
+
 type service struct {
 	Name string
 	args []string
@@ -253,6 +302,7 @@ type service struct {
 	image      string
 	entrypoint string
 
+	logs      *serviceLogs
 	component Service
 }
 

internal/recipe_l1.go

@@ -88,3 +88,7 @@ func (l *L1Recipe) Apply(ctx *ExContext, artifacts *Artifacts) *Manifest {
 	})
 	return svcManager
 }
+
+func (l *L1Recipe) Output(manifest *Manifest) map[string]interface{} {
+	return map[string]interface{}{}
+}

internal/recipe_opstack.go

@@ -66,3 +66,14 @@ func (o *OpRecipe) Apply(ctx *ExContext, artifacts *Artifacts) *Manifest {
 	})
 	return svcManager
 }
+
+func (o *OpRecipe) Output(manifest *Manifest) map[string]interface{} {
+	opGeth := manifest.MustGetService("op-geth").component.(*OpGeth)
+	if opGeth.Enode != "" {
+		// Only output if enode was set
+		return map[string]interface{}{
+			"op-geth-enode": opGeth.Enode,
+		}
+	}
+	return map[string]interface{}{}
+}

internal/watchers.go

@@ -14,7 +14,7 @@ import (
 	mevRCommon "github.com/flashbots/mev-boost-relay/common"
 )
 
-func waitForChainAlive(logOutput io.Writer, beaconNodeURL string, timeout time.Duration) error {
+func waitForChainAlive(ctx context.Context, logOutput io.Writer, beaconNodeURL string, timeout time.Duration) error {
 	// Test that blocks are being produced
 	log := mevRCommon.LogSetup(false, "info").WithField("context", "waitForChainAlive")
 	log.Logger.Out = logOutput
@@ -51,6 +51,8 @@ func waitForChainAlive(logOutput io.Writer, beaconNodeURL string, timeout time.D
 				select {
 				case <-syncTimeoutCh:
 					return fmt.Errorf("beacon client failed to start")
+				case <-ctx.Done():
+					return fmt.Errorf("timeout waiting for chain to start")
 				default:
 					time.Sleep(1 * time.Second)
 				}

main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	_ "embed"
 	"fmt"
 	"log"
@@ -120,14 +121,20 @@ func runIt(recipe internal.Recipe) error {
 	sig := make(chan os.Signal, 1)
 	signal.Notify(sig, os.Interrupt)
 
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		<-sig
+		cancel()
+	}()
+
 	if err := dockerRunner.Run(); err != nil {
 		dockerRunner.Stop()
 		return fmt.Errorf("failed to run docker: %w", err)
 	}
 
 	if !interactive {
 		// print services info
-		fmt.Printf("Services started:\n==================\n")
+		fmt.Printf("\n========= Services started =========\n")
 		for _, ss := range svcManager.Services() {
 			ports := ss.Ports()
 			sort.Slice(ports, func(i, j int) bool {
@@ -142,33 +149,39 @@ func runIt(recipe internal.Recipe) error {
 		}
 	}
 
-	watchdogErr := make(chan error, 1)
-	readyErr := make(chan error, 1)
+	if err := internal.WaitForReady(ctx, svcManager); err != nil {
+		dockerRunner.Stop()
+		return fmt.Errorf("failed to wait for service readiness: %w", err)
+	}
 
-	go func() {
-		if err := internal.WaitForReady(svcManager); err != nil {
-			readyErr <- fmt.Errorf("failed to wait for service readiness: %w", err)
+	// get the output from the recipe
+	output := recipe.Output(svcManager)
+	if len(output) > 0 {
+		fmt.Printf("\n========= Output =========\n")
+		for k, v := range output {
+			fmt.Printf("- %s: %v\n", k, v)
 		}
+	}
 
-		if watchdog {
+	watchdogErr := make(chan error, 1)
+	if watchdog {
+		go func() {
 			if err := internal.RunWatchdog(svcManager); err != nil {
 				watchdogErr <- fmt.Errorf("watchdog failed: %w", err)
 			}
-		}
-	}()
+		}()
+	}
 
 	var timerCh <-chan time.Time
 	if timeout > 0 {
 		timerCh = time.After(timeout)
 	}
 
 	select {
-	case <-sig:
+	case <-ctx.Done():
 		fmt.Println("Stopping...")
 	case err := <-dockerRunner.ExitErr():
 		fmt.Println("Service failed:", err)
-	case err := <-readyErr:
-		fmt.Println("Service failed to start:", err)
 	case err := <-watchdogErr:
 		fmt.Println("Watchdog failed:", err)
 	case <-timerCh: