Add watchdog command in service (#61)

* Add watchdog command in service

* Go

Author: ferranbt

internal/components.go

@@ -153,6 +153,13 @@ func (o *OpGeth) Run(service *service, ctx *ExContext) {
 		)
 }
 
+var _ ServiceWatchdog = &OpGeth{}
+
+func (o *OpGeth) Watchdog(out io.Writer, service *service, ctx context.Context) error {
+	rethURL := fmt.Sprintf("http://localhost:%d", service.MustGetPort("http").HostPort)
+	return watchChainHead(out, rethURL, 2*time.Second)
+}
+
 type RethEL struct {
 	UseRethForValidation bool
 	UseNativeReth        bool
@@ -229,6 +236,13 @@ func (r *RethEL) Run(svc *service, ctx *ExContext) {
 	}
 }
 
+var _ ServiceWatchdog = &RethEL{}
+
+func (r *RethEL) Watchdog(out io.Writer, service *service, ctx context.Context) error {
+	rethURL := fmt.Sprintf("http://localhost:%d", service.MustGetPort("http").HostPort)
+	return watchChainHead(out, rethURL, 12*time.Second)
+}
+
 type LighthouseBeaconNode struct {
 	ExecutionNode string
 	MevBoostNode  string
@@ -344,3 +358,19 @@ func (m *MevBoostRelay) Run(service *service, ctx *ExContext) {
 		service.WithArgs("--validation-server-addr", Connect(m.ValidationServer, "http"))
 	}
 }
+
+var _ ServiceWatchdog = &MevBoostRelay{}
+
+func (m *MevBoostRelay) Watchdog(out io.Writer, service *service, ctx context.Context) error {
+	beaconNodeURL := fmt.Sprintf("http://localhost:%d", service.MustGetPort("http").HostPort)
+
+	watchGroup := newWatchGroup()
+	watchGroup.watch(func() error {
+		return watchProposerPayloads(beaconNodeURL)
+	})
+	watchGroup.watch(func() error {
+		return validateProposerPayloads(out, beaconNodeURL)
+	})
+
+	return watchGroup.wait()
+}

internal/manifest.go

@@ -19,7 +19,6 @@ type Recipe interface {
 	Flags() *flag.FlagSet
 	Artifacts() *ArtifactsBuilder
 	Apply(ctx *ExContext, artifacts *Artifacts) *Manifest
-	Watchdog(manifest *Manifest, out *output) error
 }
 
 // Manifest describes a list of services and their dependencies
@@ -114,6 +113,42 @@ func WaitForReady(manifest *Manifest) error {
 	return nil
 }
 
+type ServiceWatchdog interface {
+	Watchdog(out io.Writer, service *service, ctx context.Context) error
+}
+
+func RunWatchdog(manifest *Manifest) error {
+	var wg sync.WaitGroup
+	watchdogErr := make(chan error, len(manifest.Services()))
+
+	output, err := manifest.out.LogOutput("watchdog")
+	if err != nil {
+		return fmt.Errorf("failed to create log output: %w", err)
+	}
+
+	for _, s := range manifest.Services() {
+		if watchdogFn, ok := s.component.(ServiceWatchdog); ok {
+			wg.Add(1)
+
+			go func() {
+				defer wg.Done()
+				if err := watchdogFn.Watchdog(output, s, context.Background()); err != nil {
+					watchdogErr <- fmt.Errorf("service %s watchdog failed: %w", s.Name, err)
+				}
+			}()
+		}
+	}
+	wg.Wait()
+
+	close(watchdogErr)
+	for err := range watchdogErr {
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (s *Manifest) Services() []*service {
 	return s.services
 }

internal/recipe_l1.go

@@ -2,7 +2,6 @@ package internal
 
 import (
 	"fmt"
-	"time"
 
 	flag "github.com/spf13/pflag"
 )
@@ -89,32 +88,3 @@ func (l *L1Recipe) Apply(ctx *ExContext, artifacts *Artifacts) *Manifest {
 	})
 	return svcManager
 }
-
-func (l *L1Recipe) Watchdog(manifest *Manifest, out *output) error {
-	beaconNode := manifest.MustGetService("beacon")
-	beaconNodeEL := manifest.MustGetService("el")
-
-	watchDogOut, err := out.LogOutput("watchdog")
-	if err != nil {
-		return err
-	}
-
-	beaconNodeURL := fmt.Sprintf("http://localhost:%d", beaconNode.MustGetPort("http").HostPort)
-	beaconNodeELURL := fmt.Sprintf("http://localhost:%d", beaconNodeEL.MustGetPort("http").HostPort)
-
-	watchGroup := newWatchGroup()
-	watchGroup.watch(func() error {
-		return watchProposerPayloads(beaconNodeURL)
-	})
-	watchGroup.watch(func() error {
-		return validateProposerPayloads(watchDogOut, beaconNodeURL)
-	})
-	watchGroup.watch(func() error {
-		return watchChainHead(watchDogOut, beaconNodeELURL, 12*time.Second)
-	})
-
-	if err := watchGroup.wait(); err != nil {
-		return err
-	}
-	return nil
-}

internal/recipe_opstack.go

@@ -1,9 +1,6 @@
 package internal
 
 import (
-	"fmt"
-	"time"
-
 	flag "github.com/spf13/pflag"
 )
 
@@ -67,36 +64,3 @@ func (o *OpRecipe) Apply(ctx *ExContext, artifacts *Artifacts) *Manifest {
 	})
 	return svcManager
 }
-
-func (o *OpRecipe) Watchdog(manifest *Manifest, out *output) error {
-	beaconNode := manifest.MustGetService("beacon")
-	beaconNodeEL := manifest.MustGetService("el")
-	opNodeEL := manifest.MustGetService("op-geth")
-
-	watchDogOut, err := out.LogOutput("watchdog")
-	if err != nil {
-		return err
-	}
-
-	beaconNodeURL := fmt.Sprintf("http://localhost:%d", beaconNode.MustGetPort("http").HostPort)
-	if err := waitForChainAlive(watchDogOut, beaconNodeURL, 50*time.Second); err != nil {
-		return err
-	}
-
-	watchGroup := newWatchGroup()
-
-	beaconNodeELURL := fmt.Sprintf("http://localhost:%d", beaconNodeEL.MustGetPort("http").HostPort)
-	watchGroup.watch(func() error {
-		return watchChainHead(watchDogOut, beaconNodeELURL, 12*time.Second)
-	})
-
-	opNodeELURL := fmt.Sprintf("http://localhost:%d", opNodeEL.MustGetPort("http").HostPort)
-	watchGroup.watch(func() error {
-		return watchChainHead(watchDogOut, opNodeELURL, 2*time.Second)
-	})
-
-	if err := watchGroup.wait(); err != nil {
-		return err
-	}
-	return nil
-}

main.go

@@ -147,7 +147,7 @@ func runIt(recipe internal.Recipe) error {
 	watchdogErr := make(chan error, 1)
 	if watchdog {
 		go func() {
-			if err := recipe.Watchdog(svcManager, artifacts.Out); err != nil {
+			if err := internal.RunWatchdog(svcManager); err != nil {
 				watchdogErr <- fmt.Errorf("watchdog failed: %w", err)
 			}
 		}()