diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go new file mode 100644 index 000000000..ce4ee2a4f --- /dev/null +++ b/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 +} diff --git a/probe/endpoint/procspy/benchmark_internal_test.go b/probe/endpoint/procspy/benchmark_internal_test.go deleted file mode 100644 index 1e48aab06..000000000 --- a/probe/endpoint/procspy/benchmark_internal_test.go +++ /dev/null @@ -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 -`) diff --git a/probe/endpoint/procspy/fixture.go b/probe/endpoint/procspy/fixture.go index 2e53c8190..0731f624a 100644 --- a/probe/endpoint/procspy/fixture.go +++ b/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() {} diff --git a/probe/endpoint/procspy/proc_internal_test.go b/probe/endpoint/procspy/proc_internal_test.go index 1035e485a..54cfa20f4 100644 --- a/probe/endpoint/procspy/proc_internal_test.go +++ b/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) } diff --git a/probe/endpoint/procspy/proc_linux.go b/probe/endpoint/procspy/proc_linux.go index 87ac4bd43..f0cd8c69b 100644 --- a/probe/endpoint/procspy/proc_linux.go +++ b/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 diff --git a/probe/endpoint/procspy/spy.go b/probe/endpoint/procspy/spy.go index 27a748256..73d4a560d 100644 --- a/probe/endpoint/procspy/spy.go +++ b/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() } diff --git a/probe/endpoint/procspy/spy_darwin.go b/probe/endpoint/procspy/spy_darwin.go index 56edb52a7..bec6f2a13 100644 --- a/probe/endpoint/procspy/spy_darwin.go +++ b/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() {} diff --git a/probe/endpoint/procspy/spy_linux.go b/probe/endpoint/procspy/spy_linux.go index 68b0d7c0b..61f189881 100644 --- a/probe/endpoint/procspy/spy_linux.go +++ b/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//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() +} diff --git a/probe/endpoint/procspy/spy_linux_internal_test.go b/probe/endpoint/procspy/spy_linux_internal_test.go index fc5ed3890..bdbf11c8e 100644 --- a/probe/endpoint/procspy/spy_linux_internal_test.go +++ b/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) } + } diff --git a/probe/endpoint/reporter.go b/probe/endpoint/reporter.go index c2f9c83f0..5999312c6 100644 --- a/probe/endpoint/reporter.go +++ b/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 } diff --git a/probe/endpoint/reporter_test.go b/probe/endpoint/reporter_test.go index 4a03636e7..ba3c2e738 100644 --- a/probe/endpoint/reporter_test.go +++ b/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) diff --git a/prog/probe.go b/prog/probe.go index 51081a6aa..8a2054d26 100644 --- a/prog/probe.go +++ b/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)