From dfc136904cebd84ae221bbbb7413e6eb2e711e6c Mon Sep 17 00:00:00 2001 From: Alfonso Acosta Date: Tue, 9 Feb 2016 10:00:04 +0000 Subject: [PATCH] Review feedback --- .../procspy/background_reader_linux.go | 88 +++++++++++-------- probe/endpoint/procspy/proc_linux.go | 25 +++--- 2 files changed, 65 insertions(+), 48 deletions(-) diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go index 6199ac3d3..6b1a06858 100644 --- a/probe/endpoint/procspy/background_reader_linux.go +++ b/probe/endpoint/procspy/background_reader_linux.go @@ -3,7 +3,6 @@ package procspy import ( "bytes" "io" - "math" "sync" "time" @@ -15,7 +14,8 @@ import ( 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 - fdBlockSize = uint64(300) // Maximum number of /proc/PID/fd/* files to stat per rate-limit period + 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 ) @@ -51,14 +51,35 @@ func (br *backgroundReader) getWalkedProcPid(buf *bytes.Buffer) (map[uint64]*Pro 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 map[uint64]*Proc // initially nil, i.e. off - walkBuf = bytes.NewBuffer(make([]byte, 0, 5000)) + walkc chan walkResult // initially nil, i.e. off rateLimitPeriod = initialRateLimitPeriod - nextInterval time.Duration + restInterval time.Duration ticker = time.NewTicker(rateLimitPeriod) pWalker = newPidWalker(walker, ticker.C, fdBlockSize) ) @@ -66,28 +87,27 @@ func (br *backgroundReader) loop(walker process.Walker) { for { select { case <-tickc: - tickc = nil // turn off until the next loop - walkc = make(chan map[uint64]*Proc, 1) // turn on (need buffered so we don't leak performWalk) - begin = time.Now() // reset counter - go performWalk(pWalker, walkBuf, walkc) // do work + 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 sockets := <-walkc: - // Swap buffers + case result := <-walkc: + // Expose results br.mtx.Lock() - br.latestBuf, walkBuf = walkBuf, br.latestBuf - br.latestSockets = sockets + br.latestBuf = result.buf + br.latestSockets = result.sockets br.mtx.Unlock() - walkBuf.Reset() - // Schedule next walk and adjust rate limit + // Schedule next walk and adjust its rate limit walkTime := time.Since(begin) - rateLimitPeriod, nextInterval = scheduleNextWalk(rateLimitPeriod, walkTime) + rateLimitPeriod, restInterval = scheduleNextWalk(rateLimitPeriod, walkTime) ticker.Stop() ticker = time.NewTicker(rateLimitPeriod) - pWalker.ticker = ticker.C + pWalker.tickc = ticker.C walkc = nil // turn off until the next loop - tickc = time.After(nextInterval) // turn on + tickc = time.After(restInterval) // turn on case <-br.stopc: pWalker.stop() @@ -97,9 +117,8 @@ func (br *backgroundReader) loop(walker process.Walker) { } } -// Adjust rate limit for next walk and calculate how long to wait until it should be started -func scheduleNextWalk(rateLimitPeriod time.Duration, took time.Duration) (time.Duration, time.Duration) { - +// 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( @@ -110,21 +129,16 @@ func scheduleNextWalk(rateLimitPeriod time.Duration, took time.Duration) (time.D } // Adjust rate limit to more-accurately meet the target walk time in next iteration - scaledRateLimitPeriod := float64(targetWalkTime) / float64(took) * float64(rateLimitPeriod) - rateLimitPeriod = time.Duration(math.Min(scaledRateLimitPeriod, float64(maxRateLimitPeriod))) - - log.Debugf("background /proc reader: new rate limit %s", rateLimitPeriod) - - return rateLimitPeriod, targetWalkTime - took -} - -func performWalk(w pidWalker, buf *bytes.Buffer, c chan<- map[uint64]*Proc) { - sockets, err := w.walk(buf) - if err != nil { - log.Errorf("background /proc reader: error walking /proc: %s", err) - buf.Reset() - c <- nil - return + newRateLimitPeriod = time.Duration(float64(targetWalkTime) / float64(took) * float64(rateLimitPeriod)) + if newRateLimitPeriod > maxRateLimitPeriod { + newRateLimitPeriod = maxRateLimitPeriod + } else if newRateLimitPeriod < minRateLimitPeriod { + newRateLimitPeriod = minRateLimitPeriod } - c <- sockets + log.Debugf("background /proc reader: new rate limit period %s", newRateLimitPeriod) + + // when to start next walk + restInterval = targetWalkTime - took + + return } diff --git a/probe/endpoint/procspy/proc_linux.go b/probe/endpoint/procspy/proc_linux.go index e475aee18..f0cd8c69b 100644 --- a/probe/endpoint/procspy/proc_linux.go +++ b/probe/endpoint/procspy/proc_linux.go @@ -26,15 +26,15 @@ var ( type pidWalker struct { walker process.Walker - ticker <-chan time.Time - stopc chan struct{} - fdBlockSize uint64 + 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, ticker <-chan time.Time, fdBlockSize uint64) pidWalker { +func newPidWalker(walker process.Walker, tickc <-chan time.Time, fdBlockSize uint64) pidWalker { w := pidWalker{ walker: walker, - ticker: ticker, + tickc: tickc, fdBlockSize: fdBlockSize, stopc: make(chan struct{}), } @@ -72,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 } @@ -139,10 +139,13 @@ func (w pidWalker) walkNamespace(buf *bytes.Buffer, sockets map[uint64]*Proc, na if fdBlockCount > w.fdBlockSize { // we surpassed the filedescriptor rate limit - // TODO: worth selecting on w.stopc? - <-w.ticker - fdBlockCount = 0 + 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 { @@ -221,10 +224,10 @@ func (w pidWalker) walk(buf *bytes.Buffer) (map[uint64]*Proc, error) { for _, procs := range namespaces { select { - case <-w.ticker: + case <-w.tickc: w.walkNamespace(buf, sockets, procs) case <-w.stopc: - break + break // abort } }