testutil/compose: add definition version matrix test

.github/workflows/test.yml

@@ -62,4 +62,4 @@ jobs:
           restore-keys: |
             ${{ runner.os }}-go-
       - run: GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -o testutil/compose/compose # Pre-build current SHA charon binary
-      - run: go test github.com/obolnetwork/charon/testutil/compose/compose -integration -sudo-perms -prebuilt-binary=charon
+      - run: go test github.com/obolnetwork/charon/testutil/compose/compose -v -integration -sudo-perms -prebuilt-binary=charon -timeout=20m

dkg/frostp2p.go

@@ -19,6 +19,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"sync"
 
 	"github.com/coinbase/kryptology/pkg/core/curves"
 	"github.com/coinbase/kryptology/pkg/dkg/frost"
@@ -42,6 +43,7 @@ func newFrostP2P(ctx context.Context, tcpNode host.Host, peers map[uint32]peer.I
 	var (
 		round1Recv  = make(chan *pb.FrostRound1Msg, len(peers))
 		round2Recv  = make(chan *pb.FrostRound2Msg, len(peers))
+		mu          sync.Mutex
 		dedupRound1 = make(map[peer.ID]bool)
 		dedupRound2 = make(map[peer.ID]bool)
 		knownPeers  = make(map[peer.ID]bool)
@@ -66,6 +68,9 @@ func newFrostP2P(ctx context.Context, tcpNode host.Host, peers map[uint32]peer.I
 			return
 		}
 
+		mu.Lock()
+		defer mu.Unlock()
+
 		pID := s.Conn().RemotePeer()
 		if !knownPeers[pID] {
 			log.Warn(ctx, "Ignoring unknown round 1 peer", nil, z.Any("peer", p2p.PeerName(pID)))
@@ -94,6 +99,9 @@ func newFrostP2P(ctx context.Context, tcpNode host.Host, peers map[uint32]peer.I
 			return
 		}
 
+		mu.Lock()
+		defer mu.Unlock()
+
 		pID := s.Conn().RemotePeer()
 		if !knownPeers[pID] {
 			log.Warn(ctx, "Ignoring unknown round 2 peer", nil, z.Any("peer", p2p.PeerName(pID)))

p2p/sender.go

@@ -118,7 +118,7 @@ func Send(ctx context.Context, tcpNode host.Host, protoID protocol.ID, peerID pe
 	// Circuit relay connections are transient
 	s, err := tcpNode.NewStream(network.WithUseTransient(ctx, ""), peerID, protoID)
 	if err != nil {
-		return errors.Wrap(err, "tcpNode stream")
+		return errors.Wrap(err, "tcpNode stream", z.Str("protocol", string(protoID)))
 	}
 
 	_, err = s.Write(b)

testutil/compose/compose/alert.go

@@ -27,13 +27,18 @@ import (
 	"github.com/obolnetwork/charon/app/z"
 )
 
+const alertsPolled = "alerts_polled"
+
 // startAlertCollector starts a goroutine that polls prometheus alerts until the context is closed and returns
 // a channel on which the received alert descriptions will be sent.
 func startAlertCollector(ctx context.Context, dir string) (chan string, error) {
-	dedup := make(map[string]bool)
 	resp := make(chan string, 100)
 
 	go func() {
+		var (
+			success bool
+			dedup   = make(map[string]bool)
+		)
 		defer close(resp)
 		for ctx.Err() == nil {
 			time.Sleep(time.Second * 5)
@@ -59,6 +64,11 @@ func startAlertCollector(ctx context.Context, dir string) (chan string, error) {
 				continue
 			}
 
+			if !success {
+				resp <- alertsPolled // Push initial "fake alert" so logic can fail is not alerts polled.
+				success = true
+			}
+
 			for _, active := range getActiveAlerts(alerts) {
 				if dedup[active] {
 					continue

testutil/compose/compose/main.go

@@ -124,7 +124,7 @@ func newDockerCmd(use string, short string, runFunc runFunc) *cobra.Command {
 }
 
 //nolint:gocognit // TODO(corver): Move this to compose package and improve API.
-func newAutoCmd(tmplCallback func(data *compose.TmplData)) *cobra.Command {
+func newAutoCmd(tmplCallbacks map[string]func(data *compose.TmplData)) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "auto",
 		Short: "Convenience function that runs `compose define && compose lock && compose run`",
@@ -137,17 +137,17 @@ func newAutoCmd(tmplCallback func(data *compose.TmplData)) *cobra.Command {
 	printYML := cmd.Flags().Bool("print-yml", false, "Print generated docker-compose.yml files.")
 
 	cmd.RunE = func(cmd *cobra.Command, _ []string) (err error) {
-		runFuncs := []func(context.Context) (compose.TmplData, error){
-			newRunnerFunc("define", *dir, false, compose.Define),
-			newRunnerFunc("lock", *dir, false, compose.Lock),
-			newRunnerFunc("run", *dir, false, compose.Run),
+		runFuncs := map[string]func(context.Context) (compose.TmplData, error){
+			"define": newRunnerFunc("define", *dir, false, compose.Define),
+			"lock":   newRunnerFunc("lock", *dir, false, compose.Lock),
+			"run":    newRunnerFunc("run", *dir, false, compose.Run),
 		}
 
 		rootCtx := log.WithTopic(cmd.Context(), "auto")
 
 		var lastTmpl compose.TmplData
-		for i, runFunc := range runFuncs {
-			lastTmpl, err = runFunc(rootCtx)
+		for i, step := range []string{"define", "lock", "run"} {
+			lastTmpl, err = runFuncs[step](rootCtx)
 			if err != nil {
 				return err
 			}
@@ -158,6 +158,14 @@ func newAutoCmd(tmplCallback func(data *compose.TmplData)) *cobra.Command {
 				}
 			}
 
+			if tmplCallbacks[step] != nil {
+				tmplCallbacks[step](&lastTmpl)
+				err := compose.WriteDockerCompose(*dir, lastTmpl)
+				if err != nil {
+					return err
+				}
+			}
+
 			if *printYML {
 				if err := printDockerCompose(rootCtx, *dir); err != nil {
 					return err
@@ -171,20 +179,6 @@ func newAutoCmd(tmplCallback func(data *compose.TmplData)) *cobra.Command {
 			}
 		}
 
-		if tmplCallback != nil {
-			tmplCallback(&lastTmpl)
-			err := compose.WriteDockerCompose(*dir, lastTmpl)
-			if err != nil {
-				return err
-			}
-		}
-
-		if *printYML {
-			if err := printDockerCompose(rootCtx, *dir); err != nil {
-				return err
-			}
-		}
-
 		ctx := rootCtx
 		if *alertTimeout != 0 {
 			// Ensure everything is clean before we start with alert test.
@@ -208,14 +202,23 @@ func newAutoCmd(tmplCallback func(data *compose.TmplData)) *cobra.Command {
 			return err
 		}
 
-		var fail bool
+		var (
+			alertMsgs    []string
+			alertSuccess bool
+		)
 		for alert := range alerts {
-			log.Error(rootCtx, "Received alert", nil, z.Str("alert", alert))
-			fail = true
+			if alert == alertsPolled {
+				alertSuccess = true
+			} else {
+				alertMsgs = append(alertMsgs, alert)
+			}
 		}
-		if fail {
-			return errors.New("alerts detected")
+		if !alertSuccess {
+			return errors.New("alerts couldn't be polled")
+		} else if len(alertMsgs) > 0 {
+			return errors.New("alerts detected", z.Any("alerts", alertMsgs))
 		}
+
 		log.Info(ctx, "No alerts detected")
 
 		return nil
@@ -317,10 +320,17 @@ func addUpFlag(flags *pflag.FlagSet) *bool {
 
 // execUp executes `docker-compose up`.
 func execUp(ctx context.Context, dir string) error {
+	// Build first so containers start at the same time below.
+	log.Info(ctx, "Executing docker-compose build")
+	cmd := exec.CommandContext(ctx, "docker-compose", "build", "--parallel")
+	cmd.Dir = dir
+	if out, err := cmd.CombinedOutput(); err != nil {
+		return errors.Wrap(err, "exec docker-compose build", z.Str("output", string(out)))
+	}
+
 	log.Info(ctx, "Executing docker-compose up")
-	cmd := exec.CommandContext(ctx, "docker-compose", "up",
+	cmd = exec.CommandContext(ctx, "docker-compose", "up",
 		"--remove-orphans",
-		"--build",
 		"--abort-on-container-exit",
 		"--quiet-pull",
 	)
@@ -333,7 +343,7 @@ func execUp(ctx context.Context, dir string) error {
 			err = ctx.Err()
 		}
 
-		return errors.Wrap(err, "run up")
+		return errors.Wrap(err, "exec docker-compose up")
 	}
 
 	return nil

testutil/compose/compose/smoke_internal_test.go

@@ -43,12 +43,15 @@ func TestSmoke(t *testing.T) {
 	}
 
 	tests := []struct {
-		Name       string
-		ConfigFunc func(*compose.Config)
-		TmplFunc   func(*compose.TmplData)
+		Name           string
+		ConfigFunc     func(*compose.Config)
+		RunTmplFunc    func(*compose.TmplData)
+		DefineTmplFunc func(*compose.TmplData)
+		PrintYML       bool
 	}{
 		{
-			Name: "default alpha",
+			Name:     "default alpha",
+			PrintYML: true,
 			ConfigFunc: func(conf *compose.Config) {
 				conf.KeyGen = compose.KeyGenCreate
 				conf.FeatureSet = "alpha"
@@ -84,33 +87,28 @@ func TestSmoke(t *testing.T) {
 			},
 		},
 		{
-			Name: "version matrix with dkg",
+			Name:     "run version matrix with dkg",
+			PrintYML: true,
 			ConfigFunc: func(conf *compose.Config) {
 				conf.KeyGen = compose.KeyGenDKG
 			},
-			TmplFunc: func(data *compose.TmplData) {
-				data.Nodes[0].ImageTag = "latest"
-				// Use default entrypoint
-
-				data.Nodes[1].ImageTag = "latest"
-				// TODO(dhruv): Using default entrypoint for now need to update with data.Nodes[1].Entrypoint = containerBinary
-				// Use default entrypoint
-
-				data.Nodes[2].ImageTag = "v0.5.0" // TODO(corver): Update this with new releases.
-				// TODO(dhruv): Using default entrypoint for now need to update with data.Nodes[1].Entrypoint = containerBinary
-				// Use default entrypoint
-
-				data.Nodes[3].ImageTag = "v0.5.0"
-				// TODO(dhruv): Using default entrypoint for now need to update with data.Nodes[1].Entrypoint = containerBinary
-				// Use default entrypoint
+			DefineTmplFunc: func(data *compose.TmplData) {
+				// v8.0.0 of charon generates v1.0.0 definition files required by older versions.
+				pegImageTag(data.Nodes, 0, "v0.8.0")
+			},
+			RunTmplFunc: func(data *compose.TmplData) {
+				// Node 0 is latest
+				pegImageTag(data.Nodes, 1, "v0.8.0")
+				pegImageTag(data.Nodes, 2, "v0.8.0")
+				pegImageTag(data.Nodes, 3, "v0.7.0")
 			},
 		},
 		{
 			Name: "teku versions", // TODO(corver): Do the same for lighthouse.
 			ConfigFunc: func(conf *compose.Config) {
 				conf.VCs = []compose.VCType{compose.VCTeku}
 			},
-			TmplFunc: func(data *compose.TmplData) {
+			RunTmplFunc: func(data *compose.TmplData) {
 				data.VCs[0].Image = "consensys/teku:latest"
 				data.VCs[1].Image = "consensys/teku:22.5"
 				data.VCs[2].Image = "consensys/teku:22.4"
@@ -119,7 +117,7 @@ func TestSmoke(t *testing.T) {
 		},
 		{
 			Name: "1 of 4 down",
-			TmplFunc: func(data *compose.TmplData) {
+			RunTmplFunc: func(data *compose.TmplData) {
 				node0 := data.Nodes[0]
 				for i := 0; i < len(node0.EnvVars); i++ {
 					if strings.HasPrefix(node0.EnvVars[i].Key, "p2p") {
@@ -131,13 +129,13 @@ func TestSmoke(t *testing.T) {
 	}
 
 	for _, test := range tests {
+		test := test // Copy iterator for async usage
 		t.Run(test.Name, func(t *testing.T) {
-			t.Parallel()
-
 			dir, err := os.MkdirTemp("", "")
 			require.NoError(t, err)
 
 			conf := compose.NewDefaultConfig()
+			conf.DisableMonitoringPorts = true
 			if *prebuiltBinary != "" {
 				copyPrebuiltBinary(t, dir, *prebuiltBinary)
 				conf.PrebuiltBinary = true
@@ -147,16 +145,14 @@ func TestSmoke(t *testing.T) {
 			}
 			require.NoError(t, compose.WriteConfig(dir, conf))
 
-			cmd := newAutoCmd(func(data *compose.TmplData) {
-				data.MonitoringPorts = false
-				if test.TmplFunc != nil {
-					test.TmplFunc(data)
-				}
+			cmd := newAutoCmd(map[string]func(data *compose.TmplData){
+				"define": test.DefineTmplFunc,
+				"run":    test.RunTmplFunc,
 			})
 			require.NoError(t, cmd.Flags().Set("compose-dir", dir))
-			require.NoError(t, cmd.Flags().Set("alert-timeout", "30s"))
+			require.NoError(t, cmd.Flags().Set("alert-timeout", "45s"))
 			require.NoError(t, cmd.Flags().Set("sudo-perms", fmt.Sprint(*sudoPerms)))
-			require.NoError(t, cmd.Flags().Set("print-yml", "true"))
+			require.NoError(t, cmd.Flags().Set("print-yml", fmt.Sprint(test.PrintYML)))
 
 			os.Args = []string{"cobra.test"}
 
@@ -174,3 +170,10 @@ func copyPrebuiltBinary(t *testing.T, dir string, binary string) {
 
 	require.NoError(t, os.WriteFile(path.Join(dir, "charon"), b, 0o555))
 }
+
+// pegImageTag pegs the charon docker image tag for one of the nodes.
+// It overrides the default that uses locally built latest version.
+func pegImageTag(nodes []compose.TmplNode, index int, imageTag string) {
+	nodes[index].ImageTag = imageTag
+	nodes[index].Entrypoint = "/usr/local/bin/charon" // Use contains binary, not locally built latest version.
+}

testutil/compose/config.go

@@ -104,6 +104,9 @@ type Config struct {
 
 	// FeatureSet defines the minimum feature set to enable.
 	FeatureSet string `json:"feature_set"`
+
+	// DisableMonitoringPorts defines whether to disable prometheus and jaeger monitoring  port binding.
+	DisableMonitoringPorts bool `json:"disable_monitoring_ports"`
 }
 
 // entrypoint returns the path to the charon binary based on the BuildLocal field.

testutil/compose/define.go

@@ -139,7 +139,7 @@ func Define(ctx context.Context, dir string, conf Config) (TmplData, error) {
 			enrs = append(enrs, enrStr)
 		}
 
-		n := node{EnvVars: []kv{
+		n := TmplNode{EnvVars: []kv{
 			{"name", "compose"},
 			{"num_validators", fmt.Sprint(conf.NumValidators)},
 			{"operator_enrs", strings.Join(enrs, ",")},
@@ -154,7 +154,7 @@ func Define(ctx context.Context, dir string, conf Config) (TmplData, error) {
 			CharonImageTag:   conf.ImageTag,
 			CharonEntrypoint: conf.entrypoint(),
 			CharonCommand:    cmdCreateDKG,
-			Nodes:            []node{n},
+			Nodes:            []TmplNode{n},
 		}
 	} else {
 		// Other keygens only need a noop docker-compose, since charon-compose.yml
@@ -165,7 +165,7 @@ func Define(ctx context.Context, dir string, conf Config) (TmplData, error) {
 			CharonImageTag:   conf.ImageTag,
 			CharonEntrypoint: "echo",
 			CharonCommand:    fmt.Sprintf("No charon commands needed for keygen=%s define step", conf.KeyGen),
-			Nodes:            []node{{}},
+			Nodes:            []TmplNode{{}},
 		}
 	}
 

testutil/compose/lock.go

@@ -44,7 +44,7 @@ func Lock(ctx context.Context, dir string, conf Config) (TmplData, error) {
 		}
 
 		// Only single node to call charon create cluster generate keys
-		n := node{EnvVars: []kv{
+		n := TmplNode{EnvVars: []kv{
 			{"threshold", fmt.Sprint(conf.Threshold)},
 			{"nodes", fmt.Sprint(conf.NumNodes)},
 			{"cluster-dir", "/compose"},
@@ -57,13 +57,13 @@ func Lock(ctx context.Context, dir string, conf Config) (TmplData, error) {
 			CharonImageTag:   conf.ImageTag,
 			CharonEntrypoint: conf.entrypoint(),
 			CharonCommand:    cmdCreateCluster,
-			Nodes:            []node{n},
+			Nodes:            []TmplNode{n},
 		}
 	case KeyGenDKG:
 
-		var nodes []node
+		var nodes []TmplNode
 		for i := 0; i < conf.NumNodes; i++ {
-			n := node{EnvVars: newNodeEnvs(i, true, conf)}
+			n := TmplNode{EnvVars: newNodeEnvs(i, true, conf)}
 			nodes = append(nodes, n)
 		}