Ready check inside the runtime (#82)

* X

* Improve

* Remove log

* Better graph

* Fix

* Fix

Author: ferranbt

internal/artifacts.go

@@ -57,6 +57,9 @@ var opState []byte
 //go:embed config.yaml.tmpl
 var clConfigContent []byte
 
+//go:embed utils/query.sh
+var queryReadyCheck []byte
+
 type ArtifactsBuilder struct {
 	outputDir         string
 	applyLatestL1Fork bool
@@ -230,6 +233,7 @@ func (b *ArtifactsBuilder) Build() (*Artifacts, error) {
 		"testnet/genesis_validators_root.txt": hex.EncodeToString(state.GenesisValidatorsRoot()),
 		"data_validator/":                     &lighthouseKeystore{privKeys: priv},
 		"deterministic_p2p_key.txt":           defaultDiscoveryPrivKey,
+		"scripts/query.sh":                    queryReadyCheck,
 	})
 	if err != nil {
 		return nil, err

internal/components.go

@@ -178,14 +178,8 @@ func (o *OpGeth) Name() string {
 
 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://")
+func (o *OpGeth) Ready(service *service) error {
+	enodeLine, err := service.logs.FindLog("enode://")
 	if err != nil {
 		return err
 	}
@@ -324,7 +318,14 @@ func (l *LighthouseBeaconNode) Run(svc *service, ctx *ExContext) {
 			"--always-prepare-payload",
 			"--prepare-payload-lookahead", "8000",
 			"--suggested-fee-recipient", "0x690B9A9E9aa1C9dB991C7721a92d351Db4FaC990",
-		)
+		).
+		WithReady(ReadyCheck{
+			QueryURL:    "http://localhost:3500/eth/v1/node/syncing",
+			Interval:    1 * time.Second,
+			Timeout:     30 * time.Second,
+			Retries:     3,
+			StartPeriod: 1 * time.Second,
+		})
 
 	if l.MevBoostNode != "" {
 		svc.WithArgs(
@@ -339,17 +340,6 @@ func (l *LighthouseBeaconNode) Name() string {
 	return "lighthouse-beacon-node"
 }
 
-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(ctx, logOutput, beaconNodeURL, 30*time.Second); err != nil {
-		return err
-	}
-	return nil
-}
-
 type LighthouseValidator struct {
 	BeaconNode string
 }
@@ -407,6 +397,7 @@ func (m *MevBoostRelay) Run(service *service, ctx *ExContext) {
 		WithImage("docker.io/flashbots/playground-utils").
 		WithTag("latest").
 		WithEntrypoint("mev-boost-relay").
+		DependsOnHealthy(m.BeaconClient).
 		WithArgs(
 			"--api-listen-addr", "0.0.0.0",
 			"--api-listen-port", `{{Port "http" 5555}}`,

internal/local_runner.go

@@ -11,6 +11,7 @@ import (
 	"strings"
 	"sync"
 	"text/template"
+	"time"
 
 	"github.com/charmbracelet/bubbles/spinner"
 	"github.com/charmbracelet/lipgloss"
@@ -62,13 +63,15 @@ type LocalRunner struct {
 
 type task struct {
 	status string
+	ready  bool
 	logs   *os.File
 }
 
 var (
 	taskStatusPending = "pending"
 	taskStatusStarted = "started"
 	taskStatusDie     = "die"
+	taskStatusHealthy = "healthy"
 )
 
 type taskUI struct {
@@ -188,10 +191,57 @@ func (d *LocalRunner) printStatus() {
 	}
 }
 
+func (d *LocalRunner) AreReady() bool {
+	d.tasksMtx.Lock()
+	defer d.tasksMtx.Unlock()
+
+	for name, task := range d.tasks {
+		// ensure the task is not a host service
+		if d.isHostService(name) {
+			continue
+		}
+
+		// first ensure the task has started
+		if task.status != taskStatusStarted {
+			return false
+		}
+
+		// then ensure it is ready if it has a ready function
+		svc := d.getService(name)
+		if svc.readyCheck != nil {
+			if !task.ready {
+				return false
+			}
+		}
+	}
+	return true
+}
+
+func (d *LocalRunner) WaitForReady(ctx context.Context, timeout time.Duration) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+
+		case <-time.After(1 * time.Second):
+			if d.AreReady() {
+				return nil
+			}
+
+		case err := <-d.exitErr:
+			return err
+		}
+	}
+}
+
 func (d *LocalRunner) updateTaskStatus(name string, status string) {
 	d.tasksMtx.Lock()
 	defer d.tasksMtx.Unlock()
-	d.tasks[name].status = status
+	if status == taskStatusHealthy {
+		d.tasks[name].ready = true
+	} else {
+		d.tasks[name].status = status
+	}
 
 	if status == taskStatusDie {
 		d.exitErr <- fmt.Errorf("container %s failed", name)
@@ -387,6 +437,39 @@ func (d *LocalRunner) toDockerComposeService(s *service) (map[string]interface{}
 		service["environment"] = s.env
 	}
 
+	if s.readyCheck != nil {
+		var test []string
+		if s.readyCheck.QueryURL != "" {
+			// This is pretty much hardcoded for now.
+			test = []string{"CMD-SHELL", "chmod +x /artifacts/scripts/query.sh && /artifacts/scripts/query.sh " + s.readyCheck.QueryURL}
+		} else {
+			test = s.readyCheck.Test
+		}
+
+		service["healthcheck"] = map[string]interface{}{
+			"test":         test,
+			"interval":     s.readyCheck.Interval.String(),
+			"timeout":      s.readyCheck.Timeout.String(),
+			"retries":      s.readyCheck.Retries,
+			"start_period": s.readyCheck.StartPeriod.String(),
+		}
+	}
+
+	if s.dependsOn != nil {
+		depends := map[string]interface{}{}
+
+		for _, d := range s.dependsOn {
+			if d.Condition == "" {
+				depends[d.Name] = struct{}{}
+			} else {
+				depends[d.Name] = map[string]interface{}{
+					"condition": d.Condition,
+				}
+			}
+		}
+		service["depends_on"] = depends
+	}
+
 	if runtime.GOOS == "linux" {
 		// We rely on host.docker.internal as the DNS address for the host inside
 		// the container. But, this is only available on Macos and Windows.
@@ -541,6 +624,10 @@ func (d *LocalRunner) trackContainerStatusAndLogs() {
 			case events.ActionDie:
 				d.updateTaskStatus(name, taskStatusDie)
 				log.Info("container died", "name", name)
+
+			case events.ActionHealthStatusHealthy:
+				d.updateTaskStatus(name, taskStatusHealthy)
+				log.Info("container is healthy", "name", name)
 			}
 
 		case err := <-errCh:

internal/manifest.go

@@ -81,37 +81,15 @@ type Service interface {
 }
 
 type ServiceReady interface {
-	Ready(out io.Writer, service *service, ctx context.Context) error
+	Ready(ouservice *service) error
 }
 
-func WaitForReady(ctx context.Context, manifest *Manifest) error {
-	var wg sync.WaitGroup
-	readyErr := make(chan error, len(manifest.Services()))
-
-	output, err := manifest.out.LogOutput("ready")
-	if err != nil {
-		return fmt.Errorf("failed to create log output: %w", err)
-	}
-
-	for _, s := range manifest.Services() {
+func (m *Manifest) CompleteReady() error {
+	for _, s := range m.services {
 		if readyFn, ok := s.component.(ServiceReady); ok {
-			wg.Add(1)
-
-			go func() {
-				defer wg.Done()
-
-				if err := readyFn.Ready(output, s, ctx); err != nil {
-					readyErr <- fmt.Errorf("service %s failed to start: %w", s.Name, err)
-				}
-			}()
-		}
-	}
-	wg.Wait()
-
-	close(readyErr)
-	for err := range readyErr {
-		if err != nil {
-			return err
+			if err := readyFn.Ready(s); err != nil {
+				return err
+			}
 		}
 	}
 	return nil
@@ -192,6 +170,7 @@ func (s *Manifest) GetService(name string) (*service, bool) {
 // - downloads any local release artifacts for the services that require host execution
 func (s *Manifest) Validate() error {
 	for _, ss := range s.services {
+		// validate node port references
 		for _, nodeRef := range ss.nodeRefs {
 			targetService, ok := s.GetService(nodeRef.Service)
 			if !ok {
@@ -202,6 +181,21 @@ func (s *Manifest) Validate() error {
 				return fmt.Errorf("service %s depends on service %s, but it does not expose port %s", ss.Name, nodeRef.Service, nodeRef.PortLabel)
 			}
 		}
+
+		// validate depends_on statements
+		for _, dep := range ss.dependsOn {
+			service, ok := s.GetService(dep.Name)
+			if !ok {
+				return fmt.Errorf("service %s depends on service %s, but it is not defined", ss.Name, dep.Name)
+			}
+
+			if dep.Condition == DependsOnConditionHealthy {
+				// if we depedn on the service to be healthy, it must have a ready check
+				if service.readyCheck == nil {
+					return fmt.Errorf("service %s depends on service %s, but it does not have a ready check", ss.Name, dep.Name)
+				}
+			}
+		}
 	}
 
 	// download any local release artifacts for the services that require them
@@ -257,24 +251,6 @@ func (s *serviceLogs) readLogs() (string, error) {
 	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 {
@@ -299,6 +275,10 @@ type service struct {
 	// list of environment variables to set for the service
 	env map[string]string
 
+	readyCheck *ReadyCheck
+
+	dependsOn []DependsOn
+
 	ports    []*Port
 	nodeRefs []*NodeRef
 
@@ -310,6 +290,18 @@ type service struct {
 	component Service
 }
 
+type DependsOnCondition string
+
+const (
+	DependsOnConditionRunning DependsOnCondition = "service_started"
+	DependsOnConditionHealthy DependsOnCondition = "service_healthy"
+)
+
+type DependsOn struct {
+	Name      string
+	Condition DependsOnCondition
+}
+
 func (s *service) Ports() []*Port {
 	return s.ports
 }
@@ -402,6 +394,30 @@ func (s *service) WithArgs(args ...string) *service {
 	return s
 }
 
+func (s *service) WithReady(check ReadyCheck) *service {
+	s.readyCheck = &check
+	return s
+}
+
+type ReadyCheck struct {
+	QueryURL    string
+	Test        []string
+	Interval    time.Duration
+	StartPeriod time.Duration
+	Timeout     time.Duration
+	Retries     int
+}
+
+func (s *service) DependsOnHealthy(name string) *service {
+	s.dependsOn = append(s.dependsOn, DependsOn{Name: name, Condition: DependsOnConditionHealthy})
+	return s
+}
+
+func (s *service) DependsOnRunning(name string) *service {
+	s.dependsOn = append(s.dependsOn, DependsOn{Name: name, Condition: DependsOnConditionRunning})
+	return s
+}
+
 func applyTemplate(templateStr string) (string, []Port, []NodeRef) {
 	// use template substitution to load constants
 	// pass-through the Dir template because it has to be resolved at the runtime
@@ -494,6 +510,18 @@ func (s *Manifest) GenerateDotGraph() string {
 		}
 	}
 
+	// Add edges for dependws_on
+	for _, ss := range s.services {
+		for _, dep := range ss.dependsOn {
+			sourceNode := strings.ReplaceAll(ss.Name, "-", "_")
+			targetNode := strings.ReplaceAll(dep.Name, "-", "_")
+			b.WriteString(fmt.Sprintf("  %s -> %s [style=dashed, color=gray, constraint=true, label=\"depends_on\"];\n",
+				sourceNode,
+				targetNode,
+			))
+		}
+	}
+
 	b.WriteString("}\n")
 	return b.String()
 }