app: add teku simnet integration test

.github/workflows/test.yml

@@ -20,9 +20,28 @@ jobs:
         key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
         restore-keys: |
           ${{ runner.os }}-go-
-    - run: go test -coverprofile=coverage.out -covermode=atomic ./...
+    - run: go test -coverprofile=coverage.out -covermode=atomic -timeout=5m ./...
     - name: Upload coverage to Codecov
       uses: codecov/codecov-action@v2.1.0
       with:
         token: ${{ secrets.CODECOV_TOKEN }}
         files: coverage.out
+
+  integration_tests:
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-go@v2
+        with:
+          go-version: '^1.17.1'
+      - uses: actions/cache@v2
+        with:
+          path: |
+            ~/go/pkg/mod
+            ~/.cache/go-build
+          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+          restore-keys: |
+            ${{ runner.os }}-go-
+      - run: docker pull consensys/teku:latest
+      - run: go test -timeout=5m -v github.com/obolnetwork/charon/app -integration -slow

app/app.go

@@ -252,7 +252,7 @@ func wireSimNetCoreWorkflow(life *lifecycle.Manager, conf Config, manifest Manif
 	if err != nil {
 		return err
 	}
-	conf.BeaconNodeAddr = bmock.HTTPServerAddr
+	conf.BeaconNodeAddr = bmock.HTTPAddr()
 
 	sched, err := scheduler.New(corePubkeys, bmock)
 	if err != nil {
@@ -316,6 +316,7 @@ func wireSimNetCoreWorkflow(life *lifecycle.Manager, conf Config, manifest Manif
 	life.RegisterStart(lifecycle.AsyncAppCtx, lifecycle.StartLeaderCast, lifecycle.HookFunc(consensus.Run))
 	life.RegisterStart(lifecycle.AsyncBackground, lifecycle.StartScheduler, lifecycle.HookFuncErr(sched.Run))
 	life.RegisterStop(lifecycle.StopScheduler, lifecycle.HookFuncMin(sched.Stop))
+	life.RegisterStop(lifecycle.StopBeaconMock, lifecycle.HookFuncErr(bmock.Close))
 
 	return nil
 }
@@ -441,14 +442,15 @@ func wireValidatorMock(conf Config, pubshares []eth2p0.BLSPubKey, sched core.Sch
 			addr := "http://" + conf.ValidatorAPIAddr
 			cl, err := eth2http.New(ctx, eth2http.WithLogLevel(1), eth2http.WithAddress(addr))
 			if err != nil {
-				log.Warn(ctx, "validatorapi client", z.Err(err))
+				log.Warn(ctx, "Cannot connect to validatorapi", z.Err(err))
+				return
 			}
 
 			err = validatormock.Attest(ctx, cl.(*eth2http.Service), signer, eth2p0.Slot(duty.Slot), pubshares...)
 			if err != nil {
-				log.Warn(ctx, "attestation failed", z.Err(err))
+				log.Warn(ctx, "Attestation failed", z.Err(err))
 			} else {
-				log.Info(ctx, "attestation success", z.I64("slot", duty.Slot))
+				log.Info(ctx, "Attestation success", z.I64("slot", duty.Slot))
 			}
 		}()
 

app/lifecycle/manager.go

@@ -225,7 +225,7 @@ func run(appCtx context.Context, startHooks, stopHooks []hook) error {
 	stop := func(hook hook) {
 		err := hook.Func.Call(stopCtx)
 		if errors.Is(stopCtx.Err(), context.DeadlineExceeded) {
-			cacheErr(errors.New("shutdown timeout"))
+			cacheErr(errors.New("shutdown timeout", z.Str("hook", hook.Label)))
 		} else if err != nil && !errors.Is(err, context.Canceled) {
 			cacheErr(errors.Wrap(err, "stop hook", z.Str("hook", hook.Label)))
 			cancel() // Cancel the graceful stop context.

app/lifecycle/order.go

@@ -38,5 +38,6 @@ const (
 	StopP2PTCPNode
 	StopP2PUDPNode
 	StopMonitoringAPI
+	StopBeaconMock // Need to close this before validator API, since it can hold long lived connections.
 	StopValidatorAPI
 )

app/simnet_test.go

@@ -16,6 +16,14 @@ package app_test
 
 import (
 	"context"
+	"crypto/ecdsa"
+	"flag"
+	"fmt"
+	"net"
+	"os"
+	"os/exec"
+	"path"
+	"strings"
 	"testing"
 	"time"
 
@@ -24,18 +32,47 @@ import (
 	"golang.org/x/sync/errgroup"
 
 	"github.com/obolnetwork/charon/app"
+	"github.com/obolnetwork/charon/app/errors"
 	"github.com/obolnetwork/charon/core"
 	"github.com/obolnetwork/charon/core/leadercast"
 	"github.com/obolnetwork/charon/core/parsigex"
 	"github.com/obolnetwork/charon/p2p"
 	"github.com/obolnetwork/charon/tbls/tblsconv"
 	"github.com/obolnetwork/charon/testutil"
 	"github.com/obolnetwork/charon/testutil/beaconmock"
+	"github.com/obolnetwork/charon/testutil/keystore"
 )
 
-func TestSimnetNoNetwork(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+//go:generate go test . -run=TestSimnetNoNetwork_TekuVC -integration -v
+var integration = flag.Bool("integration", false, "Enable docker based integration test")
+
+func TestSimnetNoNetwork_TekuVC(t *testing.T) {
+	if !*integration {
+		t.Skip("Skipping Teku integration test")
+	}
+
+	args := newSimnetArgs(t)
+	args = startTeku(t, args, 0)
+	testSimnet(t, args)
+}
+
+func TestSimnetNoNetwork_MockVCs(t *testing.T) {
+	testSimnet(t, newSimnetArgs(t))
+}
+
+type simnetArgs struct {
+	N          int
+	VMocks     []bool
+	VAPIAddrs  []string
+	P2PKeys    []*ecdsa.PrivateKey
+	SimnetKeys []*bls_sig.SecretKey
+	Manifest   app.Manifest
+	ErrChan    chan error
+}
+
+// newSimnetArgs defines the default simnet test args.
+func newSimnetArgs(t *testing.T) simnetArgs {
+	t.Helper()
 
 	const n = 3
 	manifest, p2pKeys, secretShares := app.NewClusterForT(t, 1, n, n, 99)
@@ -47,23 +84,55 @@ func TestSimnetNoNetwork(t *testing.T) {
 		secrets = append(secrets, secret)
 	}
 
+	var (
+		vmocks    []bool
+		vapiAddrs []string
+	)
+	for i := 0; i < n; i++ {
+		vmocks = append(vmocks, true)
+		vapiAddrs = append(vapiAddrs, testutil.AvailableAddr(t).String())
+	}
+
+	return simnetArgs{
+		N:          n,
+		VMocks:     vmocks,
+		VAPIAddrs:  vapiAddrs,
+		P2PKeys:    p2pKeys,
+		SimnetKeys: secrets,
+		Manifest:   manifest,
+		ErrChan:    make(chan error, 1),
+	}
+}
+
+// testSimnet spins of a simnet cluster or N charon nodes connected via in-memory transports.
+// It asserts successful end-2-end attestation broadcast from all nodes for 2 slots.
+func testSimnet(t *testing.T, args simnetArgs) {
+	t.Helper()
+	ctx, cancel := context.WithCancel(context.Background())
+
 	parSigExFunc := parsigex.NewMemExFunc()
 	lcastTransportFunc := leadercast.NewMemTransportFunc(ctx)
 
+	type simResult struct {
+		Duty   core.Duty
+		Pubkey core.PubKey
+		Data   core.AggSignedData
+	}
+
 	var (
 		eg      errgroup.Group
 		results = make(chan simResult)
 	)
-	for i := 0; i < n; i++ {
+	for i := 0; i < args.N; i++ {
 		conf := app.Config{
-			SimnetVMock:      true,
+			SimnetVMock:      args.VMocks[i],
 			MonitoringAddr:   testutil.AvailableAddr(t).String(), // Random monitoring address
-			ValidatorAPIAddr: testutil.AvailableAddr(t).String(), // Random validatorapi address
+			ValidatorAPIAddr: args.VAPIAddrs[i],
 			TestConfig: app.TestConfig{
-				Manifest:           &manifest,
-				P2PKey:             p2pKeys[i],
+				Manifest:           &args.Manifest,
+				P2PKey:             args.P2PKeys[i],
 				DisablePing:        true,
-				SimnetKeys:         []*bls_sig.SecretKey{secrets[i]},
+				SimnetKeys:         []*bls_sig.SecretKey{args.SimnetKeys[i]},
 				ParSigExFunc:       parSigExFunc,
 				LcastTransportFunc: lcastTransportFunc,
 				BroadcastCallback: func(ctx context.Context, duty core.Duty, key core.PubKey, data core.AggSignedData) error {
@@ -83,7 +152,7 @@ func TestSimnetNoNetwork(t *testing.T) {
 		})
 	}
 
-	pubkey, err := tblsconv.KeyToCore(manifest.PublicKeys()[0])
+	pubkey, err := tblsconv.KeyToCore(args.Manifest.PublicKeys()[0])
 	require.NoError(t, err)
 
 	// Assert results
@@ -107,7 +176,7 @@ func TestSimnetNoNetwork(t *testing.T) {
 
 			// Assert we get results from all peers.
 			counts[res.Duty]++
-			if counts[res.Duty] == n {
+			if counts[res.Duty] == args.N {
 				remaining--
 			}
 			if remaining != 0 {
@@ -120,11 +189,120 @@ func TestSimnetNoNetwork(t *testing.T) {
 		}
 	}()
 
+	// Wire err channel (for docker errors)
+	eg.Go(func() error {
+		select {
+		case <-ctx.Done():
+			return nil
+		case err := <-args.ErrChan:
+			cancel()
+			return err
+		}
+	})
+
 	require.NoError(t, eg.Wait())
 }
 
-type simResult struct {
-	Duty   core.Duty
-	Pubkey core.PubKey
-	Data   core.AggSignedData
+// startTeku starts a teku validator client for the provided node and returns updated args.
+func startTeku(t *testing.T, args simnetArgs, node int) simnetArgs {
+	t.Helper()
+
+	// Configure teku as VC for node0
+	args.VMocks[node] = false
+
+	// Write private share keystore and password
+	tempDir, err := os.MkdirTemp("", "")
+	require.NoError(t, err)
+	err = keystore.StoreSimnetKeys([]*bls_sig.SecretKey{args.SimnetKeys[node]}, tempDir)
+	require.NoError(t, err)
+	err = os.WriteFile(path.Join(tempDir, "keystore-simnet-0.txt"), []byte("simnet"), 0o644)
+	require.NoError(t, err)
+
+	// Change VAPI bind address to host external IP
+	args.VAPIAddrs[node] = strings.Replace(args.VAPIAddrs[node], "127.0.0.1", externalIP(t), 1)
+
+	// Teku arguments
+	tekuArgs := []string{
+		"validator-client",
+		"--network=auto",
+		"--validator-keys=/keys:/keys",
+		fmt.Sprintf("--beacon-node-api-endpoint=http://%s", args.VAPIAddrs[node]),
+	}
+
+	// Configure docker
+	name := fmt.Sprint(time.Now().UnixNano())
+	dockerArgs := []string{
+		"run",
+		"--rm",
+		fmt.Sprintf("--name=%s", name),
+		fmt.Sprintf("--volume=%s:/keys", tempDir),
+		"--user=root", // Root required to read volume files in GitHub actions.
+		"consensys/teku:latest",
+	}
+	dockerArgs = append(dockerArgs, tekuArgs...)
+	t.Logf("docker args: %v", dockerArgs)
+
+	// Start teku
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		c := exec.CommandContext(ctx, "docker", dockerArgs...)
+		c.Stdout = os.Stdout
+		c.Stderr = os.Stderr
+		err = c.Run()
+		if ctx.Err() != nil {
+			// Expected shutdown
+			return
+		}
+		args.ErrChan <- errors.Wrap(err, "docker command failed (see logging)")
+	}()
+
+	// Kill the container when done (context cancel is not enough for some reason).
+	t.Cleanup(func() {
+		cancel()
+		_ = exec.Command("docker", "kill", name).Run()
+	})
+
+	return args
+}
+
+// externalIP returns the hosts external IP.
+// Copied from https://stackoverflow.com/questions/23558425/how-do-i-get-the-local-ip-address-in-go.
+func externalIP(t *testing.T) string {
+	t.Helper()
+
+	ifaces, err := net.Interfaces()
+	require.NoError(t, err)
+
+	for _, iface := range ifaces {
+		if iface.Flags&net.FlagUp == 0 {
+			continue // interface down
+		}
+		if iface.Flags&net.FlagLoopback != 0 {
+			continue // loopback interface
+		}
+		addrs, err := iface.Addrs()
+		require.NoError(t, err)
+		for _, addr := range addrs {
+			var ip net.IP
+			switch v := addr.(type) {
+			case *net.IPNet:
+				ip = v.IP
+			case *net.IPAddr:
+				ip = v.IP
+			}
+			if ip == nil || ip.IsLoopback() {
+				continue
+			}
+			ip = ip.To4()
+			if ip == nil {
+				continue // not an ipv4 address
+			}
+
+			return ip.String()
+		}
+	}
+
+	t.Fatal("no network?")
+
+	return ""
 }