Merge pull request #912 from weaveworks/812-rate-limit-proc

Rate-limit reading proc files

probe/endpoint/procspy/background_reader_linux.go

@@ -0,0 +1,141 @@
+package procspy
+
+import (
+	"bytes"
+	"io"
+	"sync"
+	"time"
+
+	log "github.com/Sirupsen/logrus"
+
+	"github.com/weaveworks/scope/probe/process"
+)
+
+const (
+	initialRateLimitPeriod = 50 * time.Millisecond  // Read 20 * fdBlockSize file descriptors (/proc/PID/fd/*) per namespace per second
+	maxRateLimitPeriod     = 500 * time.Millisecond // Read at least 2 * fdBlockSize file descriptors per namespace per second
+	minRateLimitPeriod     = initialRateLimitPeriod
+	fdBlockSize            = uint64(300) // Maximum number of /proc/PID/fd/* files to stat per rate-limit period
+	// (as a rule of thumb going through each block should be more expensive than reading /proc/PID/tcp{,6})
+	targetWalkTime = 10 * time.Second // Aim at walking all files in 10 seconds
+)
+
+type backgroundReader struct {
+	stopc         chan struct{}
+	mtx           sync.Mutex
+	latestBuf     *bytes.Buffer
+	latestSockets map[uint64]*Proc
+}
+
+// starts a rate-limited background goroutine to read the expensive files from
+// proc.
+func newBackgroundReader(walker process.Walker) *backgroundReader {
+	br := &backgroundReader{
+		stopc:         make(chan struct{}),
+		latestSockets: map[uint64]*Proc{},
+	}
+	go br.loop(walker)
+	return br
+}
+
+func (br *backgroundReader) stop() {
+	close(br.stopc)
+}
+
+func (br *backgroundReader) getWalkedProcPid(buf *bytes.Buffer) (map[uint64]*Proc, error) {
+	br.mtx.Lock()
+	defer br.mtx.Unlock()
+
+	_, err := io.Copy(buf, br.latestBuf)
+
+	return br.latestSockets, err
+}
+
+type walkResult struct {
+	buf     *bytes.Buffer
+	sockets map[uint64]*Proc
+}
+
+func performWalk(w pidWalker, c chan<- walkResult) {
+	var (
+		err    error
+		result = walkResult{
+			buf: bytes.NewBuffer(make([]byte, 0, 5000)),
+		}
+	)
+
+	result.sockets, err = w.walk(result.buf)
+	if err != nil {
+		log.Errorf("background /proc reader: error walking /proc: %s", err)
+		result.buf.Reset()
+		result.sockets = nil
+	}
+	c <- result
+}
+
+func (br *backgroundReader) loop(walker process.Walker) {
+	var (
+		begin           time.Time                      // when we started the last performWalk
+		tickc           = time.After(time.Millisecond) // fire immediately
+		walkc           chan walkResult                // initially nil, i.e. off
+		rateLimitPeriod = initialRateLimitPeriod
+		restInterval    time.Duration
+		ticker          = time.NewTicker(rateLimitPeriod)
+		pWalker         = newPidWalker(walker, ticker.C, fdBlockSize)
+	)
+
+	for {
+		select {
+		case <-tickc:
+			tickc = nil                      // turn off until the next loop
+			walkc = make(chan walkResult, 1) // turn on (need buffered so we don't leak performWalk)
+			begin = time.Now()               // reset counter
+			go performWalk(pWalker, walkc)   // do work
+
+		case result := <-walkc:
+			// Expose results
+			br.mtx.Lock()
+			br.latestBuf = result.buf
+			br.latestSockets = result.sockets
+			br.mtx.Unlock()
+
+			// Schedule next walk and adjust its rate limit
+			walkTime := time.Since(begin)
+			rateLimitPeriod, restInterval = scheduleNextWalk(rateLimitPeriod, walkTime)
+			ticker.Stop()
+			ticker = time.NewTicker(rateLimitPeriod)
+			pWalker.tickc = ticker.C
+
+			walkc = nil                      // turn off until the next loop
+			tickc = time.After(restInterval) // turn on
+
+		case <-br.stopc:
+			pWalker.stop()
+			ticker.Stop()
+			return // abort
+		}
+	}
+}
+
+// Adjust rate limit for next walk and calculate when it should be started
+func scheduleNextWalk(rateLimitPeriod time.Duration, took time.Duration) (newRateLimitPeriod time.Duration, restInterval time.Duration) {
+	log.Debugf("background /proc reader: full pass took %s", took)
+	if float64(took)/float64(targetWalkTime) > 1.5 {
+		log.Warnf(
+			"background /proc reader: full pass took %s: 50%% more than expected (%s)",
+			took,
+			targetWalkTime,
+		)
+	}
+
+	// Adjust rate limit to more-accurately meet the target walk time in next iteration
+	newRateLimitPeriod = time.Duration(float64(targetWalkTime) / float64(took) * float64(rateLimitPeriod))
+	if newRateLimitPeriod > maxRateLimitPeriod {
+		newRateLimitPeriod = maxRateLimitPeriod
+	} else if newRateLimitPeriod < minRateLimitPeriod {
+		newRateLimitPeriod = minRateLimitPeriod
+	}
+	log.Debugf("background /proc reader: new rate limit period %s", newRateLimitPeriod)
+
+	return newRateLimitPeriod, targetWalkTime - took
+}

probe/endpoint/procspy/benchmark_internal_test.go

@@ -1,36 +0,0 @@
-package procspy
-
-import (
-	"bytes"
-	"testing"
-)
-
-func BenchmarkParseConnectionsBaseline(b *testing.B) {
-	readFile = func(string, *bytes.Buffer) (int64, error) { return 0, nil }
-	benchmarkConnections(b)
-	// 333 ns/op, 0 allocs/op
-}
-
-func BenchmarkParseConnectionsFixture(b *testing.B) {
-	readFile = func(_ string, buf *bytes.Buffer) (int64, error) {
-		n, err := buf.Write(fixture)
-		return int64(n), err
-	}
-	benchmarkConnections(b)
-	// 15553 ns/op, 12 allocs/op
-}
-
-func benchmarkConnections(b *testing.B) {
-	b.ReportAllocs()
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		cbConnections(false, nil)
-	}
-}
-
-var fixture = []byte(`  sl  local_address rem_address   st tx_queue rx_queue tr tm->when retrnsmt   uid  timeout inode
-   0: 00000000:A6C0 00000000:0000 01 00000000:00000000 00:00000000 00000000   105        0 5107 1 ffff8800a6aaf040 100 0 0 10 0
-   1: 00000000:006F 00000000:0000 01 00000000:00000000 00:00000000 00000000     0        0 5084 1 ffff8800a6aaf740 100 0 0 10 0
-   2: 0100007F:0019 00000000:0000 01 00000000:00000000 00:00000000 00000000     0        0 10550 1 ffff8800a729b780 100 0 0 10 0
-   3: A12CF62E:E4D7 57FC1EC0:01BB 01 00000000:00000000 02:000006FA 00000000  1000        0 639474 2 ffff88007e75a740 48 4 26 10 -1
-`)

probe/endpoint/procspy/fixture.go

@@ -1,13 +1,5 @@
 package procspy
 
-import (
-	"github.com/weaveworks/scope/probe/process"
-)
-
-// SetFixtures declares constant Connection and ConnectionProcs which will
-// always be returned by the package-level Connections and Processes
-// functions. It's designed to be used in tests.
-
 type fixedConnIter []Connection
 
 func (f *fixedConnIter) Next() *Connection {
@@ -21,10 +13,15 @@ func (f *fixedConnIter) Next() *Connection {
 	return &car
 }
 
-// SetFixtures is used in test scenarios to have known output.
-func SetFixtures(c []Connection) {
-	cbConnections = func(bool, process.Walker) (ConnIter, error) {
-		f := fixedConnIter(c)
-		return &f, nil
-	}
+// FixedScanner implements ConnectionScanner and uses constant Connection and
+// ConnectionProcs.
+type FixedScanner []Connection
+
+// Connections implements ConnectionsScanner.Connections
+func (s FixedScanner) Connections(_ bool) (ConnIter, error) {
+	iter := fixedConnIter(s)
+	return &iter, nil
 }
+
+// Stop implements ConnectionsScanner.Stop (dummy since there is no background work)
+func (s FixedScanner) Stop() {}

probe/endpoint/procspy/proc_internal_test.go

@@ -5,6 +5,7 @@ import (
 	"reflect"
 	"syscall"
 	"testing"
+	"time"
 
 	fs_hook "github.com/weaveworks/scope/common/fs"
 	"github.com/weaveworks/scope/probe/process"
@@ -57,7 +58,11 @@ func TestWalkProcPid(t *testing.T) {
 	defer fs_hook.Restore()
 
 	buf := bytes.Buffer{}
-	have, err := walkProcPid(&buf, process.NewWalker(procRoot))
+	walker := process.NewWalker(procRoot)
+	ticker := time.NewTicker(time.Millisecond)
+	defer ticker.Stop()
+	pWalker := newPidWalker(walker, ticker.C, 1)
+	have, err := pWalker.walk(&buf)
 	if err != nil {
 		t.Fatal(err)
 	}

probe/endpoint/procspy/proc_linux.go

@@ -7,6 +7,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"syscall"
+	"time"
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/armon/go-metrics"
@@ -23,6 +24,23 @@ var (
 	netNamespacePathSuffix = ""
 )
 
+type pidWalker struct {
+	walker      process.Walker
+	tickc       <-chan time.Time // Rate-limit clock. Sets the pace when traversing namespaces and /proc/PID/fd/* files.
+	stopc       chan struct{}    // Abort walk
+	fdBlockSize uint64           // Maximum number of /proc/PID/fd/* files to stat() per tick
+}
+
+func newPidWalker(walker process.Walker, tickc <-chan time.Time, fdBlockSize uint64) pidWalker {
+	w := pidWalker{
+		walker:      walker,
+		tickc:       tickc,
+		fdBlockSize: fdBlockSize,
+		stopc:       make(chan struct{}),
+	}
+	return w
+}
+
 // SetProcRoot sets the location of the proc filesystem.
 func SetProcRoot(root string) {
 	procRoot = root
@@ -54,7 +72,7 @@ func getNetNamespacePathSuffix() string {
 
 	v, err := getKernelVersion()
 	if err != nil {
-		log.Errorf("getNeNameSpacePath: cannot get kernel version: %s\n", err)
+		log.Errorf("getNamespacePathSuffix: cannot get kernel version: %s\n", err)
 		netNamespacePathSuffix = post38Path
 		return netNamespacePathSuffix
 	}
@@ -68,51 +86,83 @@ func getNetNamespacePathSuffix() string {
 	return netNamespacePathSuffix
 }
 
-// walkNamespacePid does the work of walkProcPid for a single namespace
-func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProcs []*process.Process) {
+// Read the connections for a group of processes living in the same namespace,
+// which are found (identically) in /proc/PID/net/tcp{,6} for any of the
+// processes.
+func readProcessConnections(buf *bytes.Buffer, namespaceProcs []*process.Process) (bool, error) {
+	var (
+		errRead  error
+		errRead6 error
+		read     int64
+		read6    int64
+	)
 
-	// Read the connections for the namespace, which are found (identically) in
-	// /proc/PID/net/tcp{,6} for any of the processes in the namespace.
-	var tcpSuccess bool
 	for _, p := range namespaceProcs {
 		dirName := strconv.Itoa(p.PID)
 
-		read, errRead := readFile(filepath.Join(procRoot, dirName, "/net/tcp"), buf)
-		read6, errRead6 := readFile(filepath.Join(procRoot, dirName, "/net/tcp6"), buf)
+		read, errRead = readFile(filepath.Join(procRoot, dirName, "/net/tcp"), buf)
+		read6, errRead6 = readFile(filepath.Join(procRoot, dirName, "/net/tcp6"), buf)
 
 		if errRead != nil || errRead6 != nil {
 			// try next process
 			continue
 		}
-
-		if read+read6 == 0 {
-			// No connections, don't bother reading /fd/*
-			return
-		}
-
-		tcpSuccess = true
-		break
+		return read+read6 > 0, nil
 	}
 
-	if !tcpSuccess {
-		// There's no point in reading /fd/*
-		return
+	// would be cool to have an or operation between errors
+	if errRead != nil {
+		return false, errRead
+	}
+	if errRead6 != nil {
+		return false, errRead6
+	}
+
+	return false, nil
+
+}
+
+// walkNamespace does the work of walk for a single namespace
+func (w pidWalker) walkNamespace(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProcs []*process.Process) error {
+
+	if found, err := readProcessConnections(buf, namespaceProcs); err != nil || !found {
+		return err
 	}
 
-	// Get the sockets for all the processes in the namespace
 	var statT syscall.Stat_t
-	for _, p := range namespaceProcs {
+	var fdBlockCount uint64
+	for i, p := range namespaceProcs {
+
+		// Get the sockets for all the processes in the namespace
 		dirName := strconv.Itoa(p.PID)
 		fdBase := filepath.Join(procRoot, dirName, "fd")
 
+		if fdBlockCount > w.fdBlockSize {
+			// we surpassed the filedescriptor rate limit
+			select {
+			case <-w.tickc:
+			case <-w.stopc:
+				return nil // abort
+			}
+
+			fdBlockCount = 0
+			// read the connections again to
+			// avoid the race between between /net/tcp{,6} and /proc/PID/fd/*
+			if found, err := readProcessConnections(buf, namespaceProcs[i:]); err != nil || !found {
+				return err
+			}
+		}
+
 		fds, err := fs.ReadDirNames(fdBase)
 		if err != nil {
-			// Process is be gone by now, or we don't have access.
+			// Process is gone by now, or we don't have access.
 			continue
 		}
 
 		var proc *Proc
 		for _, fd := range fds {
+			fdBlockCount++
+
 			// Direct use of syscall.Stat() to save garbage.
 			err = fs.Stat(filepath.Join(fdBase, fd), &statT)
 			if err != nil {
@@ -137,13 +187,15 @@ func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProc
 		}
 
 	}
+
+	return nil
 }
 
-// walkProcPid walks over all numerical (PID) /proc entries. It reads
+// walk walks over all numerical (PID) /proc entries. It reads
 // /proc/PID/net/tcp{,6} for each namespace and sees if the ./fd/* files of each
 // process in that namespace are symlinks to sockets. Returns a map from socket
 // ID (inode) to PID.
-func walkProcPid(buf *bytes.Buffer, walker process.Walker) (map[uint64]*Proc, error) {
+func (w pidWalker) walk(buf *bytes.Buffer) (map[uint64]*Proc, error) {
 	var (
 		sockets    = map[uint64]*Proc{}              // map socket inode -> process
 		namespaces = map[uint64][]*process.Process{} // map network namespace id -> processes
@@ -158,7 +210,7 @@ func walkProcPid(buf *bytes.Buffer, walker process.Walker) (map[uint64]*Proc, er
 	// between reading /net/tcp{,6} of each namespace and /proc/PID/fd/* for
 	// the processes living in that namespace.
 
-	walker.Walk(func(p, _ process.Process) {
+	w.walker.Walk(func(p, _ process.Process) {
 		dirName := strconv.Itoa(p.PID)
 
 		netNamespacePath := filepath.Join(procRoot, dirName, getNetNamespacePathSuffix())
@@ -171,17 +223,24 @@ func walkProcPid(buf *bytes.Buffer, walker process.Walker) (map[uint64]*Proc, er
 	})
 
 	for _, procs := range namespaces {
-		walkNamespacePid(buf, sockets, procs)
+		select {
+		case <-w.tickc:
+			w.walkNamespace(buf, sockets, procs)
+		case <-w.stopc:
+			break // abort
+		}
 	}
 
 	metrics.SetGauge(namespaceKey, float32(len(namespaces)))
 	return sockets, nil
 }
 
-// readFile reads an arbitrary file into a buffer. It's a variable so it can
-// be overwritten for benchmarks. That's bad practice and we should change it
-// to be a dependency.
-var readFile = func(filename string, buf *bytes.Buffer) (int64, error) {
+func (w pidWalker) stop() {
+	close(w.stopc)
+}
+
+// readFile reads an arbitrary file into a buffer.
+func readFile(filename string, buf *bytes.Buffer) (int64, error) {
 	f, err := fs.Open(filename)
 	if err != nil {
 		return -1, err

probe/endpoint/procspy/spy.go

@@ -5,8 +5,6 @@ package procspy
 
 import (
 	"net"
-
-	"github.com/weaveworks/scope/probe/process"
 )
 
 const (
@@ -35,11 +33,14 @@ type ConnIter interface {
 	Next() *Connection
 }
 
-// Connections returns all established (TCP) connections.  If processes is
-// false we'll just list all TCP connections, and there is no need to be root.
-// If processes is true it'll additionally try to lookup the process owning the
-// connection, filling in the Proc field. You will need to run this as root to
-// find all processes.
-func Connections(processes bool, walker process.Walker) (ConnIter, error) {
-	return cbConnections(processes, walker)
+// ConnectionScanner scans the system for established (TCP) connections
+type ConnectionScanner interface {
+	// Connections returns all established (TCP) connections.  If processes is
+	// false we'll just list all TCP connections, and there is no need to be root.
+	// If processes is true it'll additionally try to lookup the process owning the
+	// connection, filling in the Proc field. You will need to run this as root to
+	// find all processes.
+	Connections(processes bool) (ConnIter, error)
+	// Stops the scanning
+	Stop()
 }

probe/endpoint/procspy/spy_darwin.go

@@ -13,10 +13,17 @@ const (
 	lsofBinary    = "lsof"
 )
 
+// NewConnectionScanner creates a new Darwin ConnectionScanner
+func NewConnectionScanner(_ process.Walker) ConnectionScanner {
+	return &darwinScanner{}
+}
+
+type darwinScanner struct{}
+
 // Connections returns all established (TCP) connections. No need to be root
 // to run this. If processes is true it also tries to fill in the process
 // fields of the connection. You need to be root to find all processes.
-var cbConnections = func(processes bool, walker process.Walker) (ConnIter, error) {
+func (s *darwinScanner) Connections(processes bool) (ConnIter, error) {
 	out, err := exec.Command(
 		netstatBinary,
 		"-n", // no number resolving
@@ -62,3 +69,6 @@ var cbConnections = func(processes bool, walker process.Walker) (ConnIter, error
 	f := fixedConnIter(connections)
 	return &f, nil
 }
+
+// Nothing to stop since there's nothing running in the background
+func (s *darwinScanner) Stop() {}

probe/endpoint/procspy/spy_linux.go

@@ -32,8 +32,17 @@ func (c *pnConnIter) Next() *Connection {
 	return n
 }
 
-// cbConnections sets Connections()
-var cbConnections = func(processes bool, walker process.Walker) (ConnIter, error) {
+// NewConnectionScanner creates a new Linux ConnectionScanner
+func NewConnectionScanner(walker process.Walker) ConnectionScanner {
+	br := newBackgroundReader(walker)
+	return &linuxScanner{br}
+}
+
+type linuxScanner struct {
+	br *backgroundReader
+}
+
+func (s *linuxScanner) Connections(processes bool) (ConnIter, error) {
 	// buffer for contents of /proc/<pid>/net/tcp
 	buf := bufPool.Get().(*bytes.Buffer)
 	buf.Reset()
@@ -41,7 +50,7 @@ var cbConnections = func(processes bool, walker process.Walker) (ConnIter, error
 	var procs map[uint64]*Proc
 	if processes {
 		var err error
-		if procs, err = walkProcPid(buf, walker); err != nil {
+		if procs, err = s.br.getWalkedProcPid(buf); err != nil {
 			return nil, err
 		}
 	}
@@ -57,3 +66,7 @@ var cbConnections = func(processes bool, walker process.Walker) (ConnIter, error
 		procs: procs,
 	}, nil
 }
+
+func (s *linuxScanner) Stop() {
+	s.br.stop()
+}

probe/endpoint/procspy/spy_linux_internal_test.go

@@ -4,6 +4,7 @@ import (
 	"net"
 	"reflect"
 	"testing"
+	"time"
 
 	fs_hook "github.com/weaveworks/scope/common/fs"
 	"github.com/weaveworks/scope/probe/process"
@@ -13,8 +14,13 @@ import (
 func TestLinuxConnections(t *testing.T) {
 	fs_hook.Mock(mockFS)
 	defer fs_hook.Restore()
+	scanner := NewConnectionScanner(process.NewWalker("/proc"))
+	defer scanner.Stop()
 
-	iter, err := cbConnections(true, process.NewWalker("/proc"))
+	// let the background scanner finish its first pass
+	time.Sleep(1 * time.Second)
+
+	iter, err := scanner.Connections(true)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -37,4 +43,5 @@ func TestLinuxConnections(t *testing.T) {
 	if have := iter.Next(); have != nil {
 		t.Fatal(have)
 	}
+
 }

probe/endpoint/reporter.go

@@ -26,7 +26,7 @@ type Reporter struct {
 	includeProcesses bool
 	includeNAT       bool
 	flowWalker       flowWalker // interface
-	procWalker       process.Walker
+	scanner          procspy.ConnectionScanner
 	natMapper        natMapper
 	reverseResolver  *reverseResolver
 }
@@ -48,7 +48,7 @@ var SpyDuration = prometheus.NewSummaryVec(
 // on the host machine, at the granularity of host and port. That information
 // is stored in the Endpoint topology. It optionally enriches that topology
 // with process (PID) information.
-func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bool, procWalker process.Walker) *Reporter {
+func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bool, scanner procspy.ConnectionScanner) *Reporter {
 	return &Reporter{
 		hostID:           hostID,
 		hostName:         hostName,
@@ -56,7 +56,7 @@ func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bo
 		flowWalker:       newConntrackFlowWalker(useConntrack),
 		natMapper:        makeNATMapper(newConntrackFlowWalker(useConntrack, "--any-nat")),
 		reverseResolver:  newReverseResolver(),
-		procWalker:       procWalker,
+		scanner:          scanner,
 	}
 }
 
@@ -68,6 +68,7 @@ func (r *Reporter) Stop() {
 	r.flowWalker.stop()
 	r.natMapper.stop()
 	r.reverseResolver.stop()
+	r.scanner.Stop()
 }
 
 // Report implements Reporter.
@@ -80,7 +81,7 @@ func (r *Reporter) Report() (report.Report, error) {
 	rpt := report.MakeReport()
 
 	{
-		conns, err := procspy.Connections(r.includeProcesses, r.procWalker)
+		conns, err := r.scanner.Connections(r.includeProcesses)
 		if err != nil {
 			return rpt, err
 		}

probe/endpoint/reporter_test.go

@@ -64,14 +64,13 @@ var (
 )
 
 func TestSpyNoProcesses(t *testing.T) {
-	procspy.SetFixtures(fixConnections)
-
 	const (
 		nodeID   = "heinz-tomato-ketchup" // TODO rename to hostID
 		nodeName = "frenchs-since-1904"   // TODO rename to hostNmae
 	)
 
-	reporter := endpoint.NewReporter(nodeID, nodeName, false, false, nil)
+	scanner := procspy.FixedScanner(fixConnections)
+	reporter := endpoint.NewReporter(nodeID, nodeName, false, false, scanner)
 	r, _ := reporter.Report()
 	//buf, _ := json.MarshalIndent(r, "", "    ")
 	//t.Logf("\n%s\n", buf)
@@ -101,14 +100,13 @@ func TestSpyNoProcesses(t *testing.T) {
 }
 
 func TestSpyWithProcesses(t *testing.T) {
-	procspy.SetFixtures(fixConnectionsWithProcesses)
-
 	const (
 		nodeID   = "nikon"             // TODO rename to hostID
 		nodeName = "fishermans-friend" // TODO rename to hostNmae
 	)
 
-	reporter := endpoint.NewReporter(nodeID, nodeName, true, false, nil)
+	scanner := procspy.FixedScanner(fixConnectionsWithProcesses)
+	reporter := endpoint.NewReporter(nodeID, nodeName, true, false, scanner)
 	r, _ := reporter.Report()
 	// buf, _ := json.MarshalIndent(r, "", "    ") ; t.Logf("\n%s\n", buf)
 

prog/probe.go

@@ -24,6 +24,7 @@ import (
 	"github.com/weaveworks/scope/probe/controls"
 	"github.com/weaveworks/scope/probe/docker"
 	"github.com/weaveworks/scope/probe/endpoint"
+	"github.com/weaveworks/scope/probe/endpoint/procspy"
 	"github.com/weaveworks/scope/probe/host"
 	"github.com/weaveworks/scope/probe/kubernetes"
 	"github.com/weaveworks/scope/probe/overlay"
@@ -139,8 +140,9 @@ func probeMain() {
 	defer resolver.Stop()
 
 	processCache := process.NewCachingWalker(process.NewWalker(*procRoot))
+	scanner := procspy.NewConnectionScanner(processCache)
 
-	endpointReporter := endpoint.NewReporter(hostID, hostName, *spyProcs, *useConntrack, processCache)
+	endpointReporter := endpoint.NewReporter(hostID, hostName, *spyProcs, *useConntrack, scanner)
 	defer endpointReporter.Stop()
 
 	p := probe.New(*spyInterval, *publishInterval, clients)