Prepare traceroute-caller for supporting hop annotations (#119)

* Prepare traceroute-caller for supporting hop annotations

Although this commit seems big, it does not change the behavior
of traceroute-caller (hence this work was done in feature branch
sandbox-saied-noop). Here's a summary of the changes:

  - Remove poll mode because it's not used and its package
    testing was intermittently failing due to race conditions.
  - Remove legacy debug code for scamper-daemon-with-scamper-backup.
  - Make changes to comply with Go style guidelines with respect
    to comments, error messages, etc.
  - Improve package testing code and coverage.
  - Introduce a no-op hopannotation code to be followed with
    an actual implementation in the next commit.

The changes were tested with "go test ./... -race" and
"docker-compose up" on my local workstation.

* Make changes based on code review comments

Author: Saied Kazemi

.dockerignore

@@ -0,0 +1 @@
+**/local

caller.go

@@ -1,64 +1,52 @@
-// traceroute-caller is a wrapper around the
-// `scamper` commands and can be invoked in two different poll and
-// listen modes:
+// traceroute-caller is a wrapper around scamper, a tool that actively
+// probes the Internet in order to analyze topology and performance.
+// For details, visit https://www.caida.org/catalog/software/scamper.
 //
-//   - Poll mode uses the `connectionpoller` package to get a complete list
-//     of all connections by executing `/bin/ss -e -n` every 5 seconds
-//     and running a traceroute on all closed connections.  This mode is
-//     mostly for local test and debugging purposes as it doesn't require
-//     any services such as `tcp-info` or `uuid-annotator`.
-//
-//   - Listen mode uses the `tcp-info/eventsocket` package to listen for
-//     open and close connection events, and runs a traceroute measurement
-//     on closed connections.
-//
-// traceroute-caller on M-Lab servers always runs in the listen mode.
-// To see all available flags:
-//
-//   $ go build
-//   $ ./traceroute-caller --help
+// traceroute-caller uses the tcp-info/eventsocket package to listen for
+// open and close connection events, and runs a traceroute measurement
+// on closed connections.
 package main
 
 import (
 	"context"
+	"errors"
 	"flag"
 	"log"
 	"os"
 	"sync"
 	"time"
 
 	"github.com/m-lab/traceroute-caller/connection"
+	"github.com/m-lab/traceroute-caller/connectionlistener"
+	"github.com/m-lab/traceroute-caller/hopannotation"
+	"github.com/m-lab/traceroute-caller/ipcache"
 	"github.com/m-lab/traceroute-caller/tracer"
 
 	"github.com/m-lab/go/flagx"
+	"github.com/m-lab/go/prometheusx"
 	"github.com/m-lab/go/rtx"
 	"github.com/m-lab/tcp-info/eventsocket"
-
-	"github.com/m-lab/go/prometheusx"
-	"github.com/m-lab/traceroute-caller/connectionlistener"
-	"github.com/m-lab/traceroute-caller/connectionpoller"
-	"github.com/m-lab/traceroute-caller/ipcache"
+	"github.com/m-lab/uuid-annotator/ipservice"
 )
 
 var (
-	// TODO: scamper and its commands (e.g., tracelb) support a
-	// relatively large number of flags.  Instead of adding these
-	// flags one by one to traceroute-caller flags, going forward
-	// it's much better to have traceroute-caller read a configuration
-	// file in textproto format that would support all scamper and
-	// its command flags.
-	scamperBin        = flag.String("scamper.bin", "scamper", "The path to the scamper binary.")
-	scattachBin       = flag.String("scamper.sc_attach", "sc_attach", "The path to the sc_attach binary.")
-	scwarts2jsonBin   = flag.String("scamper.sc_warts2json", "sc_warts2json", "The path to the sc_warts2json binary.")
-	scamperCtrlSocket = flag.String("scamper.unixsocket", "/tmp/scamperctrl", "The name of the UNIX-domain socket that the scamper daemon should listen on.")
-	scamperTimeout    = flag.Duration("scamper.timeout", 900*time.Second, "How long to wait to complete a scamper trace.")
-	scamperPTR        = flag.Bool("scamper.tracelb-ptr", true, "Look up DNS pointer records for IP addresses.")
-	scamperWaitProbe  = flag.Int("scamper.tracelb-W", 25, "How long to wait between probes in 1/100ths of seconds (min 15, max 200).")
-	outputPath        = flag.String("outputPath", "/var/spool/scamper", "The path of output.")
-	waitTime          = flag.Duration("waitTime", 5*time.Second, "How long to wait between subsequent listings of open connections.")
-	poll              = flag.Bool("poll", true, "Whether the polling method should be used to see new connections.")
-	tracerType        = flagx.Enum{
-		Options: []string{"scamper", "scamper-daemon", "scamper-daemon-with-scamper-backup"},
+	// TODO(SaiedKazemi): scamper and its commands (e.g., tracelb)
+	//     support a large number of flags.  Instead of adding
+	//     these flags one by one to traceroute-caller flags, it
+	//     will be much better to have traceroute-caller read a
+	//     configuration file in textproto format that would support
+	//     all scamper and its command flags.
+	scamperBin          = flag.String("scamper.bin", "scamper", "The path to the scamper binary.")
+	scattachBin         = flag.String("scamper.sc_attach", "sc_attach", "The path to the sc_attach binary.")
+	scwarts2jsonBin     = flag.String("scamper.sc_warts2json", "sc_warts2json", "The path to the sc_warts2json binary.")
+	scamperCtrlSocket   = flag.String("scamper.unixsocket", "/tmp/scamperctrl", "The name of the UNIX-domain socket that the scamper daemon should listen on.")
+	scamperTimeout      = flag.Duration("scamper.timeout", 900*time.Second, "How long to wait to complete a scamper trace.")
+	scamperPTR          = flag.Bool("scamper.tracelb-ptr", true, "Look up DNS pointer records for IP addresses.")
+	scamperWaitProbe    = flag.Int("scamper.tracelb-W", 25, "How long to wait between probes in 1/100ths of seconds (min 15, max 200).")
+	tracerouteOutput    = flag.String("traceroute-output", "/var/spool/scamper", "The path to store traceroute output.")
+	hopAnnotationOutput = flag.String("hopannotation-output", "/var/spool/hopannotation1", "The path to store hop annotations.")
+	tracerType          = flagx.Enum{
+		Options: []string{"scamper", "scamper-daemon"},
 		Value:   "scamper",
 	}
 
@@ -75,18 +63,26 @@ func main() {
 	log.SetFlags(log.LstdFlags | log.Lshortfile)
 
 	flag.Parse()
-	rtx.Must(flagx.ArgsFromEnv(flag.CommandLine), "Could not get args from environment")
-	rtx.Must(os.MkdirAll(*outputPath, 0777), "Could not create data directory")
+	rtx.Must(flagx.ArgsFromEnv(flag.CommandLine), "failed to get args from environment")
+	if *eventsocket.Filename == "" {
+		logFatal("tcpinfo.eventsocket was set to \"\"")
+	}
+	rtx.Must(os.MkdirAll(*tracerouteOutput, 0777), "failed to create directory for traceroute results")
+	rtx.Must(os.MkdirAll(*hopAnnotationOutput, 0777), "failed to create directory for hop annotation results")
 
 	defer cancel()
 	wg := sync.WaitGroup{}
 
 	promSrv := prometheusx.MustServeMetrics()
-	defer promSrv.Shutdown(ctx)
+	defer func() {
+		if err := promSrv.Shutdown(ctx); err != nil && !errors.Is(err, context.Canceled) {
+			log.Printf("failed to shut down Prometheus server (error: %v)", err)
+		}
+	}()
 
 	scamper := &tracer.Scamper{
 		Binary:           *scamperBin,
-		OutputPath:       *outputPath,
+		OutputPath:       *tracerouteOutput,
 		ScamperTimeout:   *scamperTimeout,
 		TracelbPTR:       *scamperPTR,
 		TracelbWaitProbe: *scamperWaitProbe,
@@ -98,58 +94,28 @@ func main() {
 		ControlSocket:    *scamperCtrlSocket,
 	}
 
-	var cache *ipcache.RecentIPCache
-
-	// Set up the cache three different ways, depending on the trace method requested.
+	// Set up the ipCache depending on the trace method requested.
+	var ipCache *ipcache.RecentIPCache
 	switch tracerType.Value {
 	case "scamper":
-		cache = ipcache.New(ctx, scamper, *ipcache.IPCacheTimeout, *ipcache.IPCacheUpdatePeriod)
+		ipCache = ipcache.New(ctx, scamper, *ipcache.IPCacheTimeout, *ipcache.IPCacheUpdatePeriod)
 	case "scamper-daemon":
-		cache = ipcache.New(ctx, scamperDaemon, *ipcache.IPCacheTimeout, *ipcache.IPCacheUpdatePeriod)
+		ipCache = ipcache.New(ctx, scamperDaemon, *ipcache.IPCacheTimeout, *ipcache.IPCacheUpdatePeriod)
 		wg.Add(1)
 		go func() {
 			scamperDaemon.MustStart(ctx)
 			// When the scamper daemon dies, cancel main() and exit.
 			cancel()
 			wg.Done()
 		}()
-	// These are hacks - the scamper daemon should not fail at all.
-	case "scamper-daemon-with-scamper-backup":
-		cache = ipcache.New(ctx, scamperDaemon, *ipcache.IPCacheTimeout, *ipcache.IPCacheUpdatePeriod)
-		wg.Add(1)
-		go func() {
-			scamperDaemon.MustStart(ctx)
-			// When the scamper daemon dies, switch to scamper
-			cache.UpdateTracer(scamper)
-			wg.Done()
-		}()
 	}
 
-	if *poll {
-		wg.Add(1)
-		go func(c *ipcache.RecentIPCache) {
-			connPoller := connectionpoller.New(c)
-			for ctx.Err() == nil {
-				connPoller.TraceClosedConnections()
-
-				select {
-				case <-time.After(*waitTime):
-				case <-ctx.Done():
-				}
-			}
-			wg.Done()
-		}(cache)
-	} else if *eventsocket.Filename != "" {
-		wg.Add(1)
-		go func() {
-			connCreator, err := connection.NewCreator()
-			rtx.Must(err, "Could not discover local IPs")
-			connListener := connectionlistener.New(connCreator, cache)
-			eventsocket.MustRun(ctx, *eventsocket.Filename, connListener)
-			wg.Done()
-		}()
-	} else {
-		logFatal("--poll was false but --tcpinfo.eventsocket was set to \"\". This is a nonsensical configuration.")
-	}
+	localIPs, err := connection.NewLocalRemoteIPs()
+	rtx.Must(err, "failed to discover local IPs")
+	ipserviceClient := ipservice.NewClient(*ipservice.SocketFilename)
+	hopAnnotator := hopannotation.New(ctx, ipserviceClient, *hopAnnotationOutput)
+	connListener := connectionlistener.New(localIPs, ipCache, hopAnnotator)
+	eventsocket.MustRun(ctx, *eventsocket.Filename, connListener)
+	cancel()
 	wg.Wait()
 }

caller_test.go

@@ -5,7 +5,6 @@ import (
 	"io/ioutil"
 	"log"
 	"os"
-	"runtime"
 	"testing"
 	"time"
 
@@ -23,95 +22,25 @@ func TestMetrics(t *testing.T) {
 	promtest.LintMetrics(t)
 }
 
-func TestMain(t *testing.T) {
-	if runtime.GOOS != "linux" {
-		t.Skip("Skipping for non-linux environment", runtime.GOOS)
-	}
-	dir, err := ioutil.TempDir("", "TestMain")
-	rtx.Must(err, "Could not create temp dir")
-	defer os.RemoveAll(dir)
-
-	// Verify that main doesn't crash, and that it does exit when the context is canceled.
-	// TODO: verify more in this test.
-	*prometheusx.ListenAddress = ":0"
-	*scamperCtrlSocket = dir + "/scamper.sock"
-	*waitTime = time.Nanosecond // Run through the loop a few times.
-	*outputPath = dir
-	*poll = true
-	*scamperBin = "scamper"
-	tracerType.Value = "scamper-daemon"
-	ctx, cancel = context.WithCancel(context.Background())
-	go func() {
-		time.Sleep(1 * time.Second)
-		cancel()
-	}()
-	main()
-}
-
-func TestScamper(t *testing.T) {
-	if runtime.GOOS != "linux" {
-		t.Skip("Skipping for non-linux environment", runtime.GOOS)
-	}
-	dir, err := ioutil.TempDir("", "TestMain")
-	rtx.Must(err, "Could not create temp dir")
-	defer os.RemoveAll(dir)
-
-	// Verify that main doesn't crash, and that it does exit when the context is canceled.
-	// TODO: verify more in this test.
-	*prometheusx.ListenAddress = ":0"
-	*scamperCtrlSocket = ""
-	*waitTime = time.Nanosecond // Run through the loop a few times.
-	*outputPath = dir
-	*poll = true
-	*scamperBin = "scamper"
-	tracerType.Value = "scamper"
-	ctx, cancel = context.WithCancel(context.Background())
-	go func() {
-		time.Sleep(1 * time.Second)
-		cancel()
-	}()
-	main()
-}
-
 func TestMainWithConnectionListener(t *testing.T) {
 	dir, err := ioutil.TempDir("", "TestMainWithConnectionListener")
-	rtx.Must(err, "Could not create temp dir")
+	rtx.Must(err, "failed to create temp dir")
 	defer os.RemoveAll(dir)
 	srv := eventsocket.New(dir + "/events.sock")
-	rtx.Must(srv.Listen(), "Could not start the empty server")
+	rtx.Must(srv.Listen(), "failed to start the empty server")
 
 	*prometheusx.ListenAddress = ":0"
 	*eventsocket.Filename = dir + "/events.sock"
-	*outputPath = dir
-	*poll = false
+	*tracerouteOutput = dir
+	*hopAnnotationOutput = dir
 	tracerType.Value = "scamper"
 
 	ctx, cancel = context.WithCancel(context.Background())
-	go srv.Serve(ctx)
-	go func() {
-		time.Sleep(1 * time.Second)
-		cancel()
-	}()
-	main()
-}
-
-func TestMainWithBackupScamper(t *testing.T) {
-	dir, err := ioutil.TempDir("", "TestMainWithBackupScamper")
-	rtx.Must(err, "Could not create temp dir")
-	defer os.RemoveAll(dir)
-	srv := eventsocket.New(dir + "/events.sock")
-	rtx.Must(srv.Listen(), "Could not start the empty server")
-
-	*prometheusx.ListenAddress = ":0"
-	*eventsocket.Filename = dir + "/events.sock"
-	*outputPath = dir
-	*poll = false
-	*scamperCtrlSocket = dir + "/scamper.sock"
-	*scamperBin = "false"
-	tracerType.Value = "scamper-daemon-with-scamper-backup"
-
-	ctx, cancel = context.WithCancel(context.Background())
-	go srv.Serve(ctx)
+	go func(t *testing.T) {
+		if err := srv.Serve(ctx); err != nil {
+			t.Logf("failed to start eventsocket server (error: %v)", err)
+		}
+	}(t)
 	go func() {
 		time.Sleep(1 * time.Second)
 		cancel()
@@ -122,8 +51,8 @@ func TestMainWithBackupScamper(t *testing.T) {
 func TestMainWithBadArgs(t *testing.T) {
 	tracerType.Value = "scamper"
 	*eventsocket.Filename = ""
-	*outputPath = "/tmp/"
-	*poll = false
+	*tracerouteOutput = "/tmp/"
+	*hopAnnotationOutput = "/tmp/"
 
 	logFatal = func(_ ...interface{}) {
 		panic("testpanic")

connection/connection.go

@@ -29,7 +29,7 @@ type Connection struct {
 
 // UUID returns uuid from cookie parsed from "ss -e" output.
 func (c *Connection) UUID() (string, error) {
-	// cookie is a hexdecimal string
+	// cookie is a hexadecimal string
 	result, err := strconv.ParseUint(c.Cookie, 16, 64)
 	return uuid.FromCookie(result), err
 }
@@ -79,9 +79,9 @@ type Creator interface {
 	FromSockID(sockid inetdiag.SockID) (Connection, error)
 }
 
-// NewCreator makes an object that can convert src and dst into local and remote
-// IPs.
-func NewCreator() (Creator, error) {
+// NewLocalRemoteIPs makes an object that can convert src and dst into local
+// and remote IPs.
+func NewLocalRemoteIPs() (Creator, error) {
 	c := &creator{
 		localIPs: make([]*net.IP, 0),
 	}
@@ -108,9 +108,9 @@ func NewCreator() (Creator, error) {
 	return c, err
 }
 
-// NewFakeCreator makes a fake creator with hardcoded local IPs to enable
-// testing in diverse network environments.
-func NewFakeCreator(localIPs []*net.IP) Creator {
+// NewFakeLocalIPs makes a fake creator with hardcoded local IPs to
+// enable testing in diverse network environments.
+func NewFakeLocalIPs(localIPs []*net.IP) Creator {
 	return &creator{
 		localIPs: localIPs,
 	}