integration: speed up dockerized end2end test (#3719)

The overhead of running a binary in a docker container with
docker-compose exec is around 1 second.

Since our integration tests are serialized and a cartesian product between
all ASes is taken, this overhead easily amounts to multiple minutes.

With this PR, all servers are started in parallel for the end2end test.

All pairs with the same source are executed in on single docker exec
command. And all executions are done in parallel.

To get a progress update, the end2end binary talks through a unix socket
to the integration binary.

We can eventually move the server wait to this RPC framework as well.

With this style of invoking the end2end test, we can reduce the execution
time on CI from >10 minutes (timeout) to a bit more than 1 minute.
Most of the execution time is actually spent in starting the topology,
not in execution the tests.

Co-authored-by: Oncilla <roos@anapaya.net>

.buildkite/pipeline.yml

@@ -122,3 +122,20 @@ steps:
       automatic:
         - exit_status: -1  # Agent was lost
         - exit_status: 255 # Forced agent shutdown
+  - label: ":docker: Integration: end2end_integration full topology"
+    if: build.message !~ /\[doc\]/
+    command:
+    - bazel run //docker/perapp:prod >/dev/null 2>&1
+    - ./scion.sh topology -t -d
+    - ./scion.sh run
+    - docker-compose -f gen/scion-dc.yml -p scion up -d $(docker-compose -f gen/scion-dc.yml config --services | grep tester)
+    - sleep 10
+    - ./bin/end2end_integration -d -log.console warn
+    artifact_paths:
+      - "artifacts.out/**/*"
+    timeout_in_minutes: 10
+    key: docker_integration_e2e_default
+    retry:
+      automatic:
+        - exit_status: -1  # Agent was lost
+        - exit_status: 255 # Forced agent shutdown

go/examples/pingpong/pp_integration/main.go

@@ -72,7 +72,7 @@ func runTests(in integration.Integration, pairs []integration.IAPair) error {
 		for i, conn := range pairs {
 			testInfo := fmt.Sprintf("%v -> %v (%v/%v)", conn.Src.IA, conn.Dst.IA, i+1, len(pairs))
 			log.Info(fmt.Sprintf("Test %v: %s", in.Name(), testInfo))
-			if err := integration.RunClient(in, conn, 5*time.Second); err != nil {
+			if err := integration.RunClient(in, conn, 10*time.Second); err != nil {
 				log.Error(fmt.Sprintf("Error in client: %s", testInfo), "err", err)
 				return err
 			}

go/integration/BUILD.bazel

@@ -6,8 +6,10 @@ go_library(
     importpath = "github.com/scionproto/scion/go/integration",
     visibility = ["//visibility:public"],
     deps = [
+        "//go/lib/addr:go_default_library",
         "//go/lib/env:go_default_library",
         "//go/lib/integration:go_default_library",
+        "//go/lib/integration/progress:go_default_library",
         "//go/lib/log:go_default_library",
         "//go/lib/sciond:go_default_library",
         "//go/lib/snet:go_default_library",

go/integration/cert_req_integration/main.go

@@ -28,7 +28,7 @@ import (
 var (
 	name     = "cert_req_integration"
 	cmd      = "./bin/cert_req"
-	attempts = flag.Int("attempts", 2, "Number of attempts before giving up.")
+	attempts = flag.Int("attempts", 4, "Number of attempts before giving up.")
 )
 
 func main() {

go/integration/common.go

@@ -24,8 +24,10 @@ import (
 
 	"github.com/opentracing/opentracing-go"
 
+	"github.com/scionproto/scion/go/lib/addr"
 	"github.com/scionproto/scion/go/lib/env"
 	"github.com/scionproto/scion/go/lib/integration"
+	"github.com/scionproto/scion/go/lib/integration/progress"
 	"github.com/scionproto/scion/go/lib/log"
 	"github.com/scionproto/scion/go/lib/sciond"
 	"github.com/scionproto/scion/go/lib/snet"
@@ -41,6 +43,7 @@ const (
 var (
 	Local        snet.UDPAddr
 	Mode         string
+	Progress     string
 	sciondAddr   string
 	networksFile string
 	Attempts     int
@@ -55,6 +58,7 @@ func Setup() {
 func addFlags() {
 	flag.Var((*snet.UDPAddr)(&Local), "local", "(Mandatory) address to listen on")
 	flag.StringVar(&Mode, "mode", ModeClient, "Run in "+ModeClient+" or "+ModeServer+" mode")
+	flag.StringVar(&Progress, "progress", "", "Socket to write progress to")
 	flag.StringVar(&sciondAddr, "sciond", sciond.DefaultSCIONDAddress, "SCIOND address")
 	flag.StringVar(&networksFile, "networks", integration.SCIONDAddressesFile,
 		"File containing network definitions")
@@ -146,6 +150,16 @@ func AttemptRepeatedly(name string, attempt AttemptFunc) int {
 	return 1
 }
 
+// Done informs the integration test that a test binary has finished.
+func Done(src, dst addr.IA) {
+	if Progress == "" {
+		return
+	}
+	if doneErr := (progress.Client{Socket: Progress}).Done(src, dst); doneErr != nil {
+		log.Error("Unable to send done", "err", doneErr)
+	}
+}
+
 // LogFatal logs a critical error and exits with 1
 func LogFatal(msg string, a ...interface{}) {
 	log.Crit(msg, a...)

go/integration/end2end/main.go

@@ -49,7 +49,7 @@ const (
 
 var (
 	remote  snet.UDPAddr
-	timeout = &util.DurWrap{Duration: 2 * time.Second}
+	timeout = &util.DurWrap{Duration: 10 * time.Second}
 )
 
 func main() {
@@ -99,6 +99,9 @@ type server struct {
 }
 
 func (s server) run() {
+	log.Info("Starting server", "ia", integration.Local.IA)
+	defer log.Info("Finished server", "ia", integration.Local.IA)
+
 	connFactory := &snet.DefaultPacketDispatcherService{
 		Dispatcher: reliable.NewDispatcher(""),
 		SCMPHandler: snet.NewSCMPHandler(
@@ -113,7 +116,7 @@ func (s server) run() {
 	if len(os.Getenv(libint.GoIntegrationEnv)) > 0 {
 		// Needed for integration test ready signal.
 		fmt.Printf("Port=%d\n", port)
-		fmt.Printf("%s%s\n", libint.ReadySignal, integration.Local.IA)
+		fmt.Printf("%s%s\n\n", libint.ReadySignal, integration.Local.IA)
 	}
 	log.Debug("Listening", "local", fmt.Sprintf("%v:%d", integration.Local.Host, port))
 	// Receive ping message
@@ -127,8 +130,7 @@ func (s server) run() {
 		if string(p.Payload.(common.RawBytes)) != ping+integration.Local.IA.String() {
 			integration.LogFatal("Received unexpected data", "data", p.Payload.(common.RawBytes))
 		}
-		log.Debug(fmt.Sprintf("Ping received from %s, sending pong.",
-			p.Source))
+		log.Debug(fmt.Sprintf("Ping received from %s, sending pong.", p.Source))
 		// Send pong
 
 		if p.Path != nil {
@@ -143,7 +145,7 @@ func (s server) run() {
 		if err := conn.WriteTo(&p, &ov); err != nil {
 			integration.LogFatal("Unable to send reply", "err", err)
 		}
-		log.Debug(fmt.Sprintf("Sent pong to %s", p.Destination))
+		log.Info("Sent pong to", "client", p.Destination)
 	}
 }
 
@@ -154,6 +156,10 @@ type client struct {
 }
 
 func (c client) run() int {
+	pair := fmt.Sprintf("%s -> %s", integration.Local.IA, remote.IA)
+	log.Info("Starting", "pair", pair)
+	defer log.Info("Finished", "pair", pair)
+	defer integration.Done(integration.Local.IA, remote.IA)
 	connFactory := &snet.DefaultPacketDispatcherService{
 		Dispatcher: reliable.NewDispatcher(""),
 		SCMPHandler: snet.NewSCMPHandler(
@@ -195,7 +201,6 @@ func (c client) attemptRequest(n int) bool {
 		ext.Error.Set(span, true)
 		return false
 	}
-	logger.Info("Received pong")
 	return true
 }
 
@@ -275,7 +280,7 @@ func (c client) pong(ctx context.Context) error {
 		return common.NewBasicError("Received unexpected data", nil, "data",
 			string(p.Payload.(common.RawBytes)), "expected", expected)
 	}
-	log.Debug(fmt.Sprintf("Received pong from %s", remote.IA))
+	log.Info("Received pong", "server", p.Source)
 	return nil
 }
 

go/integration/end2end_integration/BUILD.bazel

@@ -11,6 +11,8 @@ go_library(
         "//go/lib/common:go_default_library",
         "//go/lib/integration:go_default_library",
         "//go/lib/log:go_default_library",
+        "//go/lib/serrors:go_default_library",
+        "//go/lib/snet:go_default_library",
         "//go/lib/util:go_default_library",
     ],
 )