Fix connection listener (#40)

Standardize flags, parse IPs in a more cautious manner, don't run
local-local traceroutes, add a dryRun flag.

Use icache.Trace and not scamper.Trace.

Actually, the pollers and listeners should not know about scamper at all. Only the ipcache.

Deleted old code.

Updated tests to preserve test coverage.

Don't risk a null pointer error in an error message

Author: pboothe

caller.go

@@ -4,6 +4,7 @@ package main
 import (
 	"context"
 	"flag"
+	"log"
 	"sync"
 	"time"
 
@@ -37,6 +38,8 @@ var (
 // go build
 // ./traceroute-caller --outputPath scamper_output
 func main() {
+	log.SetFlags(log.LstdFlags | log.Lshortfile)
+
 	flag.Parse()
 	rtx.Must(flagx.ArgsFromEnv(flag.CommandLine), "Could not get args from environment")
 
@@ -58,13 +61,13 @@ func main() {
 	}()
 
 	wg := sync.WaitGroup{}
-	cache := ipcache.New(ctx)
+	cache := ipcache.New(ctx, &daemon, *ipcache.IPCacheTimeout, *ipcache.IPCacheUpdatePeriod)
 	if *poll {
 		wg.Add(1)
 		go func() {
 			connPoller := connectionpoller.New(cache)
 			for ctx.Err() == nil {
-				connPoller.TraceClosedConnections(&daemon)
+				connPoller.TraceClosedConnections()
 
 				select {
 				case <-time.After(*waitTime):
@@ -82,9 +85,9 @@ func main() {
 			esdaemon := daemon
 			esdaemon.DryRun = *eventsocketDryRun
 			if *eventsocketDryRun {
-				cache = ipcache.New(ctx)
+				cache = ipcache.New(ctx, &esdaemon, *ipcache.IPCacheTimeout, *ipcache.IPCacheUpdatePeriod)
 			}
-			connListener := connectionlistener.New(&esdaemon, connCreator, cache)
+			connListener := connectionlistener.New(connCreator, cache)
 			eventsocket.MustRun(ctx, *eventsocket.Filename, connListener)
 			wg.Done()
 		}()

caller_test.go

@@ -19,9 +19,13 @@ func TestMetrics(t *testing.T) {
 }
 
 func TestMain(t *testing.T) {
+	dir, err := ioutil.TempDir("", "TestMain")
+	rtx.Must(err, "Could not create temp 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.
 	ctx, cancel = context.WithCancel(context.Background())
 	go func() {
@@ -32,12 +36,13 @@ func TestMain(t *testing.T) {
 }
 
 func TestMainWithConnectionListener(t *testing.T) {
-	dir, err := ioutil.TempDir("", "TestTracerouteCaller")
+	dir, err := ioutil.TempDir("", "TestMainWithConnectionListener")
 	rtx.Must(err, "Could not create temp dir")
 	srv := eventsocket.New(dir + "/events.sock")
 	rtx.Must(srv.Listen(), "Could not start the empty server")
 
 	*prometheusx.ListenAddress = ":0"
+	*scamperCtrlSocket = dir + "/scamper.sock"
 	*eventsocket.Filename = dir + "/events.sock"
 	*eventsocketDryRun = true
 

connectionlistener/connectionlistener.go

@@ -10,7 +10,6 @@ import (
 	"github.com/m-lab/tcp-info/inetdiag"
 	"github.com/m-lab/traceroute-caller/connection"
 	"github.com/m-lab/traceroute-caller/ipcache"
-	"github.com/m-lab/traceroute-caller/scamper"
 )
 
 // connectionListener implements the eventsocket.Handler interface, allowing us
@@ -19,7 +18,6 @@ import (
 type connectionListener struct {
 	mutex   sync.Mutex
 	conns   map[string]connection.Connection
-	tracer  scamper.Tracer
 	cache   *ipcache.RecentIPCache
 	creator connection.Creator
 }
@@ -46,19 +44,15 @@ func (cl *connectionListener) Close(ctx context.Context, timestamp time.Time, uu
 	cl.mutex.Unlock()
 
 	if ok {
-		if !cl.cache.Has(conn.RemoteIP) {
-			cl.cache.Add(conn.RemoteIP)
-			go cl.tracer.Trace(conn, timestamp)
-		}
+		go cl.cache.Trace(conn)
 	}
 }
 
 // New returns an eventsocket.Handler that will call the passed-in scamper
 // daemon on every closed connection.
-func New(tracer scamper.Tracer, creator connection.Creator, cache *ipcache.RecentIPCache) eventsocket.Handler {
+func New(creator connection.Creator, cache *ipcache.RecentIPCache) eventsocket.Handler {
 	return &connectionListener{
 		conns:   make(map[string]connection.Connection),
-		tracer:  tracer,
 		cache:   cache,
 		creator: creator,
 	}

connectionlistener/connectionlistener_test.go

@@ -55,13 +55,14 @@ func TestListener(t *testing.T) {
 	// Create a new connectionlistener with a fake tracer.
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	cache := ipcache.New(ctx)
 	ft := &fakeTracer{}
+	cache := ipcache.New(ctx, ft, 100*time.Second, time.Second)
+
 	ft.wg.Add(2)
 
 	localIP := net.ParseIP("10.0.0.1")
 	creator := connection.NewFakeCreator([]*net.IP{&localIP})
-	cl := connectionlistener.New(ft, creator, cache)
+	cl := connectionlistener.New(creator, cache)
 
 	// Connect the connectionlistener to the server
 	go eventsocket.MustRun(ctx, dir+"/tcpevents.sock", cl)

connectionpoller/connectionpoller.go

@@ -15,7 +15,6 @@ import (
 
 	"github.com/m-lab/traceroute-caller/connection"
 	"github.com/m-lab/traceroute-caller/ipcache"
-	"github.com/m-lab/traceroute-caller/scamper"
 )
 
 var (
@@ -137,18 +136,18 @@ type connectionPoller struct {
 // connections which it previously measured to be open, but it can no longer
 // measure to be open.
 type ConnectionPoller interface {
-	TraceClosedConnections(tracer scamper.Tracer)
+	TraceClosedConnections()
 }
 
 // TraceClosedConnections send trace for all closed connections.
-func (c *connectionPoller) TraceClosedConnections(tracer scamper.Tracer) {
+func (c *connectionPoller) TraceClosedConnections() {
 	oldConn := c.connectionPool
 	fmt.Printf("old connection size %d\n", len(oldConn))
 	c.connectionPool = c.GetConnections()
 	fmt.Printf("new connection size %d\n", len(c.connectionPool))
 	for conn := range oldConn {
 		if _, hasConn := c.connectionPool[conn]; !hasConn {
-			go c.recentIPCache.Trace(conn, tracer)
+			go c.recentIPCache.Trace(conn)
 		}
 	}
 }

connectionpoller/connectionpoller_test.go

@@ -147,7 +147,8 @@ func TestConnectionPollerConstruction(t *testing.T) {
 	// Which is not nothing, but it's not a lot.
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	cache := ipcache.New(ctx)
+	var tt testTracer
+	cache := ipcache.New(ctx, &tt, time.Second, time.Second)
 	connPoller := New(cache).(*connectionPoller)
 	connPoller.finder = &testFinder{}
 	connPoller.connectionPool = make(map[connection.Connection]struct{})
@@ -158,8 +159,7 @@ func TestConnectionPollerConstruction(t *testing.T) {
 		LocalPort:  58790,
 		Cookie:     "10f3d"}
 	connPoller.connectionPool[conn1] = struct{}{}
-	var tt testTracer
-	connPoller.TraceClosedConnections(&tt)
+	connPoller.TraceClosedConnections()
 
 	time.Sleep(200 * time.Millisecond)
 

ipcache/ipcache.go

@@ -32,10 +32,12 @@ type CacheTest struct {
 type RecentIPCache struct {
 	cache map[string]*CacheTest
 	mu    sync.RWMutex
+
+	tracer scamper.Tracer
 }
 
-// GetTrace returns test content in []byte given a connection and tracer
-func (rc *RecentIPCache) Trace(conn connection.Connection, sc scamper.Tracer) {
+// Trace performs a trace and adds it to the cache.
+func (rc *RecentIPCache) Trace(conn connection.Connection) {
 	ip := conn.RemoteIP
 	rc.mu.Lock()
 	_, ok := rc.cache[ip]
@@ -47,12 +49,12 @@ func (rc *RecentIPCache) Trace(conn connection.Connection, sc scamper.Tracer) {
 		rc.cache[ip] = nc
 		rc.mu.Unlock()
 
-		nc.data = sc.Trace(conn, nc.timeStamp)
+		nc.data = rc.tracer.Trace(conn, nc.timeStamp)
 		close(nc.done)
 		return
 	}
 	rc.mu.Unlock()
-	sc.CreateCacheTest(conn, time.Now(), rc.GetData(ip))
+	rc.tracer.CreateCacheTest(conn, time.Now(), rc.GetData(ip))
 }
 
 func (rc *RecentIPCache) GetCacheLength() int {
@@ -74,18 +76,20 @@ func (rc *RecentIPCache) GetData(ip string) string {
 
 // New creates and returns a RecentIPCache. It also starts up a background
 // goroutine that scrubs the cache.
-func New(ctx context.Context) *RecentIPCache {
-	m := &RecentIPCache{}
-	m.cache = make(map[string]*CacheTest)
+func New(ctx context.Context, tracer scamper.Tracer, ipCacheTimeout, ipCacheUpdatePeriod time.Duration) *RecentIPCache {
+	m := &RecentIPCache{
+		cache:  make(map[string]*CacheTest),
+		tracer: tracer,
+	}
 	go func() {
-		ticker := time.NewTicker(*IPCacheUpdatePeriod)
+		ticker := time.NewTicker(ipCacheUpdatePeriod)
 		defer ticker.Stop()
 		for now := range ticker.C {
 			if ctx.Err() != nil {
 				return
 			}
 			for k, v := range m.cache {
-				if now.Sub(v.timeStamp) > *IPCacheTimeout {
+				if now.Sub(v.timeStamp) > ipCacheTimeout {
 					fmt.Println("try to delete " + k)
 					m.mu.Lock()
 					delete(m.cache, k)
@@ -98,36 +102,3 @@ func New(ctx context.Context) *RecentIPCache {
 	}()
 	return m
 }
-
-func (m *RecentIPCache) len() int {
-	m.mu.RLock()
-	defer m.mu.RUnlock()
-	return len(m.cache)
-}
-
-// Add an IP to the cache.
-func (m *RecentIPCache) Add(ip string) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	fmt.Printf("func Add: Now is %d\n", time.Now().Unix())
-	_, present := m.cache[ip]
-	if !present {
-		m.cache[ip] = &CacheTest{
-			timeStamp: time.Now(),
-			done:      make(chan struct{}),
-		}
-		fmt.Printf("just add %s %d\n", ip, m.cache[ip].timeStamp.Unix())
-	}
-}
-
-// Has tests whether an IP is in the cache.
-func (m *RecentIPCache) Has(ip string) bool {
-	m.mu.RLock()
-	defer m.mu.RUnlock()
-	//fmt.Printf("func Has: Now is %d, length of cache: %d \n", time.Now().Unix(), m.Len())
-	if m.len() == 0 {
-		return false
-	}
-	_, ok := m.cache[ip]
-	return ok
-}

ipcache/ipcache_test.go

@@ -28,17 +28,17 @@ func (tf *testTracer) CreateCacheTest(conn connection.Connection, t time.Time, c
 func TestGetData(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	testCache := ipcache.New(ctx)
+	var tt testTracer
+	testCache := ipcache.New(ctx, &tt, 100*time.Second, time.Second)
 
 	conn1 := connection.Connection{
 		RemoteIP:   "1.1.1.2",
 		RemotePort: 5034,
 		LocalIP:    "1.1.1.3",
 		LocalPort:  58790,
 		Cookie:     "10f3d"}
-	var tt testTracer
 
-	testCache.Trace(conn1, &tt)
+	testCache.Trace(conn1)
 	time.Sleep(200 * time.Millisecond)
 	tmp := testCache.GetData("1.1.1.1")
 	if tmp != "" {
@@ -54,17 +54,17 @@ func TestGetData(t *testing.T) {
 func TestTrace(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	testCache := ipcache.New(ctx)
+	var tt testTracer
+	testCache := ipcache.New(ctx, &tt, 100*time.Second, time.Second)
 
 	conn1 := connection.Connection{
 		RemoteIP:   "1.1.1.2",
 		RemotePort: 5034,
 		LocalIP:    "1.1.1.3",
 		LocalPort:  58790,
 		Cookie:     "10f3d"}
-	var tt testTracer
 
-	testCache.Trace(conn1, &tt)
+	testCache.Trace(conn1)
 	time.Sleep(200 * time.Millisecond)
 	tmp := testCache.GetData("1.1.1.2")
 	if tmp != "Fake Trace test 1.1.1.2" {
@@ -78,8 +78,8 @@ func TestTrace(t *testing.T) {
 		LocalPort:  58790,
 		Cookie:     "aaaa"}
 
-	testCache.Trace(conn2, &tt)
-	testCache.Trace(conn1, &tt)
+	testCache.Trace(conn2)
+	testCache.Trace(conn1)
 
 	time.Sleep(200 * time.Millisecond)
 
@@ -89,20 +89,24 @@ func TestTrace(t *testing.T) {
 }
 
 func TestRecentIPCache(t *testing.T) {
-	*ipcache.IPCacheTimeout = 100 * time.Millisecond
-	*ipcache.IPCacheUpdatePeriod = 10 * time.Millisecond
-
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	tmp := ipcache.New(ctx)
-	tmp.Add("1.2.3.4")
-	if !tmp.Has("1.2.3.4") {
+	var tt testTracer
+	tmp := ipcache.New(ctx, &tt, 100*time.Millisecond, 10*time.Millisecond)
+	tmp.Trace(connection.Connection{
+		RemoteIP:   "1.2.3.4",
+		RemotePort: 5,
+		LocalIP:    "6.7.8.9",
+		LocalPort:  10,
+		Cookie:     "11",
+	})
+	if tmp.GetCacheLength() != 1 {
 		t.Error("cache not working correctly")
 	}
 
 	time.Sleep(300 * time.Millisecond)
-	if tmp.Has("1.2.3.4") {
-		t.Error("cache not expire correctly")
+	if tmp.GetCacheLength() != 0 {
+		t.Error("cache not working correctly")
 	}
 	cancel()
 	time.Sleep(200 * time.Millisecond)

scamper/scamper.go

@@ -70,7 +70,7 @@ func (d *Daemon) MustStart(ctx context.Context) {
 	derivedCtx, derivedCancel := context.WithCancel(ctx)
 	defer derivedCancel()
 	if _, err := os.Stat(d.ControlSocket); !os.IsNotExist(err) {
-		logFatal("The control socket file must not already exist")
+		logFatal("The control socket file must not already exist: ", err)
 	}
 	defer os.Remove(d.ControlSocket)
 	command := exec.Command(d.Binary, "-U", d.ControlSocket)