Feedback #8

* Wait for servers to be ready
* Rename pingpong.go -> main.go
* Move logLogEntry to log.Log
* Other small fixes

go/examples/pingpong/pingpong.go

@@ -346,6 +346,7 @@ func (s server) run() {
 	if err != nil {
 		LogFatal("Unable to listen", "err", err)
 	}
+	fmt.Printf("Listening ia=%s\n", local.IA) // Needed for integration test ready signal.
 	log.Info("Listening", "local", qsock.Addr())
 	for {
 		qsess, err := qsock.Accept()

go/examples/pingpong/pp_integration/pingpong.go → go/examples/pingpong/pp_integration/main.go

@@ -64,13 +64,10 @@ func runTests(in integration.Integration, pairs []integration.IAPair) error {
 			}
 			defer c.Close()
 		}
-		// give the servers some time to start.
-		time.Sleep(100 * time.Millisecond)
 		// Now start the clients for srcDest pair
 		for i, conn := range pairs {
 			log.Info(fmt.Sprintf("Test %v: %v -> %v (%v/%v)",
 				in.Name(), conn.Src, conn.Dst, i+1, len(pairs)))
-
 			if err := integration.RunClient(in, conn, 1*time.Second); err != nil {
 				fmt.Fprintf(os.Stderr, "Error during client execution: %s\n", err)
 				return err

go/lib/integration/binary.go

@@ -15,7 +15,9 @@
 package integration
 
 import (
+	"bufio"
 	"context"
+	"fmt"
 	"io"
 	"os/exec"
 	"strings"
@@ -30,6 +32,10 @@ const (
 	SrcIAReplace = "<SRCIA>"
 	// DstIAReplace is a placeholder for the destination IA in the arguments.
 	DstIAReplace = "<DSTIA>"
+	// ReadySignal should be written to Stdout by the server once it is read to accept clients.
+	// The message should always be `Listening ia=<IA>`
+	// where <IA> is the IA the server is listening on.
+	ReadySignal = "Listening ia="
 )
 
 var _ Integration = (*binaryIntegration)(nil)
@@ -44,6 +50,7 @@ type binaryIntegration struct {
 // Start* will run the binary programm with name and use the given arguments for the client/server.
 // Use SrcIAReplace and DstIAReplace in arguments as placeholder for the source and destination IAs.
 // When starting a client/server the placeholders will be replaced with the actual values.
+// The server should output the ReadySignal to Stdout once it is ready to accept clients.
 func NewBinaryIntegration(name string, clientArgs, serverArgs []string) Integration {
 	return &binaryIntegration{
 		name:       name,
@@ -56,6 +63,7 @@ func (bi *binaryIntegration) Name() string {
 	return bi.name
 }
 
+// StartServer starts a server and blocks until the ReadySignal is received on Stdout.
 func (bi *binaryIntegration) StartServer(ctx context.Context, dst addr.IA) (Waiter, error) {
 	args := replacePattern(DstIAReplace, dst.String(), bi.serverArgs)
 	r := &binaryWaiter{
@@ -65,8 +73,35 @@ func (bi *binaryIntegration) StartServer(ctx context.Context, dst addr.IA) (Wait
 	if err != nil {
 		return nil, err
 	}
-	go redirectLog("Server", "ServerErr", ep)
-	return r, r.Start()
+	sp, err := r.StdoutPipe()
+	if err != nil {
+		return nil, err
+	}
+	ready := make(chan struct{})
+	// parse until we have the ready signal.
+	go func() {
+		defer log.LogPanicAndExit()
+		defer sp.Close()
+		scanner := bufio.NewScanner(sp)
+		for scanner.Scan() {
+			line := scanner.Text()
+			if fmt.Sprintf("%s%s", ReadySignal, dst) == line {
+				close(ready)
+				return
+			}
+		}
+	}()
+	go redirectLog("Server", "ServerErr", dst, ep)
+	err = r.Start()
+	if err != nil {
+		return nil, err
+	}
+	select {
+	case <-ready:
+		return r, err
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
 }
 
 func (bi *binaryIntegration) StartClient(ctx context.Context, src, dst addr.IA) (Waiter, error) {
@@ -79,7 +114,7 @@ func (bi *binaryIntegration) StartClient(ctx context.Context, src, dst addr.IA)
 	if err != nil {
 		return nil, err
 	}
-	go redirectLog("Client", "ClientErr", ep)
+	go redirectLog("Client", "ClientErr", src, ep)
 	return r, r.Start()
 }
 
@@ -94,30 +129,16 @@ func replacePattern(pattern string, replacement string, args []string) []string
 	return argsCopy
 }
 
-func redirectLog(name, pName string, ep io.ReadCloser) {
+func redirectLog(name, pName string, local addr.IA, ep io.ReadCloser) {
 	defer log.LogPanicAndExit()
 	defer ep.Close()
 	logparse.ParseFrom(ep, pName, pName, func(e logparse.LogEntry) {
-		logLogEntry(name, e)
+		log.Log(e.Level, fmt.Sprintf("%s@%s: %s", name, local, strings.Join(e.Lines, "\n")))
 	})
 }
 
-func logLogEntry(name string, e logparse.LogEntry) {
-	var logFun func(string, ...interface{})
-	switch e.Level {
-	case logparse.LvlDebug:
-		logFun = log.Debug
-	case logparse.LvlInfo:
-		logFun = log.Info
-	case logparse.LvlWarn:
-		logFun = log.Warn
-	case logparse.LvlError:
-		logFun = log.Error
-	case logparse.LvlCrit:
-		logFun = log.Crit
-	}
-	indent := strings.Repeat(" ", len(name)+2)
-	logFun(name + ": " + strings.Join(e.Lines, "\n"+indent))
+func readyConsumer(local addr.IA) {
+
 }
 
 var _ Waiter = (*binaryWaiter)(nil)

go/lib/integration/integration.go

@@ -32,14 +32,15 @@ import (
 
 type iaArgs []addr.IA
 
-func (a *iaArgs) String() string {
-	rawIAs := make([]string, len(*a))
-	for i, ia := range *a {
+func (a iaArgs) String() string {
+	rawIAs := make([]string, len(a))
+	for i, ia := range a {
 		rawIAs[i] = ia.String()
 	}
 	return strings.Join(rawIAs, ",")
 }
 
+// Set implements flag.Value.Set().
 func (a *iaArgs) Set(value string) error {
 	rawIAs := strings.Split(value, ",")
 	for _, rawIA := range rawIAs {
@@ -62,7 +63,7 @@ type Integration interface {
 	// Name returns the name of the test
 	Name() string
 	// StartServer should start the server listening on the address dst.
-	// StartServer should return immediately.
+	// StartServer should return after it is ready to accept clients.
 	// The context should be used to make the server cancellable.
 	StartServer(ctx context.Context, dst addr.IA) (Waiter, error)
 	// StartClient should start the client on the src address connecting to the dst address.
@@ -160,6 +161,7 @@ func (s *serverStop) Close() error {
 }
 
 // StartServer runs a server. The server can be stopped by calling Close() on the returned Closer.
+// To start a server with a custom context use in.StartServer directly.
 func StartServer(in Integration, dst addr.IA) (io.Closer, error) {
 	serverCtx, serverCancel := context.WithCancel(context.Background())
 	s, err := in.StartServer(serverCtx, dst)

go/lib/log/log.go

@@ -1,4 +1,5 @@
 // Copyright 2016 ETH Zurich
+// Copyright 2018 ETH Zurich, Anapaya Systems
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -32,6 +33,38 @@ import (
 	"github.com/scionproto/scion/go/lib/common"
 )
 
+type Lvl log15.Lvl
+
+const (
+	LvlCrit  = Lvl(log15.LvlCrit)
+	LvlError = Lvl(log15.LvlError)
+	LvlWarn  = Lvl(log15.LvlWarn)
+	LvlInfo  = Lvl(log15.LvlInfo)
+	LvlDebug = Lvl(log15.LvlDebug)
+)
+
+func LvlFromString(lvl string) (Lvl, error) {
+	// Since we also parse python log entries we also have to handle the levels of python.
+	switch strings.ToUpper(lvl) {
+	case "DEBUG", "DBUG":
+		return LvlDebug, nil
+	case "INFO":
+		return LvlInfo, nil
+	case "WARN", "WARNING":
+		return LvlWarn, nil
+	case "ERROR", "EROR":
+		return LvlError, nil
+	case "CRIT", "CRITICAL":
+		return LvlCrit, nil
+	default:
+		return LvlDebug, fmt.Errorf("Unknown level: %v", lvl)
+	}
+}
+
+func (l Lvl) String() string {
+	return strings.ToUpper(log15.Lvl(l).String())
+}
+
 type Logger log15.Logger
 type Handler log15.Handler
 
@@ -204,6 +237,23 @@ func Crit(msg string, ctx ...interface{}) {
 	log15.Crit(msg, ctx...)
 }
 
+func Log(lvl Lvl, msg string, ctx ...interface{}) {
+	var logFun func(string, ...interface{})
+	switch lvl {
+	case LvlDebug:
+		logFun = Debug
+	case LvlInfo:
+		logFun = Info
+	case LvlWarn:
+		logFun = Warn
+	case LvlError:
+		logFun = Error
+	case LvlCrit:
+		logFun = Crit
+	}
+	logFun(msg, ctx...)
+}
+
 func RandId(idlen int) string {
 	return logext.RandId(idlen)
 }

go/lib/log/logparse/logentry.go

@@ -22,54 +22,20 @@ import (
 	"strings"
 	"time"
 
-	"github.com/inconshreveable/log15"
-
 	"github.com/scionproto/scion/go/lib/common"
 	"github.com/scionproto/scion/go/lib/log"
 )
 
-type Lvl log15.Lvl
-
-const (
-	LvlCrit  = Lvl(log15.LvlCrit)
-	LvlError = Lvl(log15.LvlError)
-	LvlWarn  = Lvl(log15.LvlWarn)
-	LvlInfo  = Lvl(log15.LvlInfo)
-	LvlDebug = Lvl(log15.LvlDebug)
-)
-
 var (
 	lineRegex = regexp.MustCompile(` \[(\w+)\] (.+)`)
 )
 
-func LvlFromString(lvl string) (Lvl, error) {
-	// Since we also parse python log entries we also have to handle the levels of python.
-	switch strings.ToUpper(lvl) {
-	case "DEBUG", "DBUG":
-		return LvlDebug, nil
-	case "INFO":
-		return LvlInfo, nil
-	case "WARN", "WARNING":
-		return LvlWarn, nil
-	case "ERROR", "EROR":
-		return LvlError, nil
-	case "CRIT", "CRITICAL":
-		return LvlCrit, nil
-	default:
-		return LvlDebug, fmt.Errorf("Unknown level: %v", lvl)
-	}
-}
-
-func (l Lvl) String() string {
-	return strings.ToUpper(log15.Lvl(l).String())
-}
-
 // LogEntry is one entry in a log.
 type LogEntry struct {
 	Timestamp time.Time
 	// Element describes the source of this LogEntry, e.g. the file name.
 	Element string
-	Level   Lvl
+	Level   log.Lvl
 	Lines   []string
 }
 
@@ -134,7 +100,7 @@ func parseInitialEntry(line, fileName, element string, lineno int) *LogEntry {
 			fileName, lineno, lineRegex))
 		return nil
 	}
-	lvl, err := LvlFromString(matches[1])
+	lvl, err := log.LvlFromString(matches[1])
 	if err != nil {
 		log.Error(fmt.Sprintf("%s:%d: Unknown log level: %v", fileName, lineno, err))
 	}

go/lib/log/logparse/logentry_test.go

@@ -40,7 +40,7 @@ func TestParseFrom(t *testing.T) {
 			Entries: []LogEntry{
 				{
 					Timestamp: defaultTs,
-					Level:     LvlError,
+					Level:     log.LvlError,
 					Lines:     []string{"Txt"},
 				},
 			},
@@ -52,7 +52,7 @@ func TestParseFrom(t *testing.T) {
 			Entries: []LogEntry{
 				{
 					Timestamp: defaultTs,
-					Level:     LvlCrit,
+					Level:     log.LvlCrit,
 					Lines:     []string{"(CliSrvExt 2-ff00:0: > ...", "> SCIONDPathReplyEntry:"},
 				},
 			},
@@ -64,7 +64,7 @@ func TestParseFrom(t *testing.T) {
 			Entries: []LogEntry{
 				{
 					Timestamp: defaultTs,
-					Level:     LvlCrit,
+					Level:     log.LvlCrit,
 					Lines:     []string{"(CliSrvExt 2-ff00:0: > ...", " SCIONDPathReplyEntry:"},
 				},
 			},
@@ -80,12 +80,12 @@ func TestParseFrom(t *testing.T) {
 			Entries: []LogEntry{
 				{
 					Timestamp: defaultTs,
-					Level:     LvlError,
+					Level:     log.LvlError,
 					Lines:     []string{"Txt"},
 				},
 				{
 					Timestamp: mustParse("2018-07-19 14:39:30.489625+0000", t),
-					Level:     LvlInfo,
+					Level:     log.LvlInfo,
 					Lines:     []string{"Txt2"},
 				},
 			},