Bump cAdvisor (and dependencies) godeps version

vendor/github.com/google/cadvisor/utils/cloudinfo/aws.go

@@ -15,24 +15,26 @@
 package cloudinfo
 
 import (
-	"time"
-
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/ec2metadata"
 	"github.com/aws/aws-sdk-go/aws/session"
+	"io/ioutil"
+	"strings"
 
 	info "github.com/google/cadvisor/info/v1"
 )
 
+const (
+	ProductVerFileName = "/sys/class/dmi/id/product_version"
+	Amazon             = "amazon"
+)
+
 func onAWS() bool {
-	// the default client behavior retried the operation multiple times with a 5s timeout per attempt.
-	// if you were not on aws, you would block for 20s when invoking this operation.
-	// we reduce retries to 0 and set the timeout to 2s to reduce the time this blocks when not on aws.
-	client := ec2metadata.New(session.New(&aws.Config{MaxRetries: aws.Int(0)}))
-	if client.Config.HTTPClient != nil {
-		client.Config.HTTPClient.Timeout = time.Duration(2 * time.Second)
+	data, err := ioutil.ReadFile(ProductVerFileName)
+	if err != nil {
+		return false
 	}
-	return client.Available()
+	return strings.Contains(string(data), Amazon)
 }
 
 func getAwsMetadata(name string) string {

vendor/github.com/google/cadvisor/utils/cpuload/cpuload.go

@@ -41,6 +41,6 @@ func New() (CpuLoadReader, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to create a netlink based cpuload reader: %v", err)
 	}
-	glog.Info("Using a netlink-based load reader")
+	glog.V(3).Info("Using a netlink-based load reader")
 	return reader, nil
 }

vendor/github.com/google/cadvisor/utils/docker/docker.go

@@ -0,0 +1,58 @@
+// Copyright 2016 Google Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package docker
+
+import (
+	"fmt"
+	"strings"
+
+	dockertypes "github.com/docker/engine-api/types"
+)
+
+const (
+	DockerInfoDriver         = "Driver"
+	DockerInfoDriverStatus   = "DriverStatus"
+	DriverStatusPoolName     = "Pool Name"
+	DriverStatusDataLoopFile = "Data loop file"
+	DriverStatusMetadataFile = "Metadata file"
+)
+
+func DriverStatusValue(status [][2]string, target string) string {
+	for _, v := range status {
+		if strings.EqualFold(v[0], target) {
+			return v[1]
+		}
+	}
+
+	return ""
+}
+
+func DockerThinPoolName(info dockertypes.Info) (string, error) {
+	poolName := DriverStatusValue(info.DriverStatus, DriverStatusPoolName)
+	if len(poolName) == 0 {
+		return "", fmt.Errorf("Could not get devicemapper pool name")
+	}
+
+	return poolName, nil
+}
+
+func DockerMetadataDevice(info dockertypes.Info) (string, error) {
+	metadataDevice := DriverStatusValue(info.DriverStatus, DriverStatusMetadataFile)
+	if len(metadataDevice) == 0 {
+		return "", fmt.Errorf("Could not get the devicemapper metadata device")
+	}
+
+	return metadataDevice, nil
+}

vendor/github.com/google/cadvisor/utils/machine/machine.go

@@ -1,315 +0,0 @@
-// Copyright 2015 Google Inc. All Rights Reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package machine
-
-import (
-	"fmt"
-	"io/ioutil"
-	"regexp"
-	"strconv"
-	"strings"
-
-	// s390/s390x changes
-	"runtime"
-	"syscall"
-
-	info "github.com/google/cadvisor/info/v1"
-	"github.com/google/cadvisor/utils"
-	"github.com/google/cadvisor/utils/sysfs"
-	"github.com/google/cadvisor/utils/sysinfo"
-
-	"github.com/golang/glog"
-)
-
-// The utils/machine package contains functions that extract machine-level specs.
-
-var (
-	cpuRegExp  = regexp.MustCompile(`^processor\s*:\s*([0-9]+)$`)
-	coreRegExp = regexp.MustCompile(`^core id\s*:\s*([0-9]+)$`)
-	nodeRegExp = regexp.MustCompile(`^physical id\s*:\s*([0-9]+)$`)
-	// Power systems have a different format so cater for both
-	cpuClockSpeedMHz     = regexp.MustCompile(`(?:cpu MHz|clock)\s*:\s*([0-9]+\.[0-9]+)(?:MHz)?`)
-	memoryCapacityRegexp = regexp.MustCompile(`MemTotal:\s*([0-9]+) kB`)
-	swapCapacityRegexp   = regexp.MustCompile(`SwapTotal:\s*([0-9]+) kB`)
-)
-
-const maxFreqFile = "/sys/devices/system/cpu/cpu0/cpufreq/cpuinfo_max_freq"
-
-// GetClockSpeed returns the CPU clock speed, given a []byte formatted as the /proc/cpuinfo file.
-func GetClockSpeed(procInfo []byte) (uint64, error) {
-	// s390/s390x and aarch64 changes
-	if true == isSystemZ() || true == isAArch64() {
-		return 0, nil
-	}
-
-	// First look through sys to find a max supported cpu frequency.
-	if utils.FileExists(maxFreqFile) {
-		val, err := ioutil.ReadFile(maxFreqFile)
-		if err != nil {
-			return 0, err
-		}
-		var maxFreq uint64
-		n, err := fmt.Sscanf(string(val), "%d", &maxFreq)
-		if err != nil || n != 1 {
-			return 0, fmt.Errorf("could not parse frequency %q", val)
-		}
-		return maxFreq, nil
-	}
-	// Fall back to /proc/cpuinfo
-	matches := cpuClockSpeedMHz.FindSubmatch(procInfo)
-	if len(matches) != 2 {
-		return 0, fmt.Errorf("could not detect clock speed from output: %q", string(procInfo))
-	}
-
-	speed, err := strconv.ParseFloat(string(matches[1]), 64)
-	if err != nil {
-		return 0, err
-	}
-	// Convert to kHz
-	return uint64(speed * 1000), nil
-}
-
-// GetMachineMemoryCapacity returns the machine's total memory from /proc/meminfo.
-// Returns the total memory capacity as an uint64 (number of bytes).
-func GetMachineMemoryCapacity() (uint64, error) {
-	out, err := ioutil.ReadFile("/proc/meminfo")
-	if err != nil {
-		return 0, err
-	}
-
-	memoryCapacity, err := parseCapacity(out, memoryCapacityRegexp)
-	if err != nil {
-		return 0, err
-	}
-	return memoryCapacity, err
-}
-
-// GetMachineSwapCapacity returns the machine's total swap from /proc/meminfo.
-// Returns the total swap capacity as an uint64 (number of bytes).
-func GetMachineSwapCapacity() (uint64, error) {
-	out, err := ioutil.ReadFile("/proc/meminfo")
-	if err != nil {
-		return 0, err
-	}
-
-	swapCapacity, err := parseCapacity(out, swapCapacityRegexp)
-	if err != nil {
-		return 0, err
-	}
-	return swapCapacity, err
-}
-
-// parseCapacity matches a Regexp in a []byte, returning the resulting value in bytes.
-// Assumes that the value matched by the Regexp is in KB.
-func parseCapacity(b []byte, r *regexp.Regexp) (uint64, error) {
-	matches := r.FindSubmatch(b)
-	if len(matches) != 2 {
-		return 0, fmt.Errorf("failed to match regexp in output: %q", string(b))
-	}
-	m, err := strconv.ParseUint(string(matches[1]), 10, 64)
-	if err != nil {
-		return 0, err
-	}
-
-	// Convert to bytes.
-	return m * 1024, err
-}
-
-func GetTopology(sysFs sysfs.SysFs, cpuinfo string) ([]info.Node, int, error) {
-	nodes := []info.Node{}
-
-	// s390/s390x changes
-	if true == isSystemZ() {
-		return nodes, getNumCores(), nil
-	}
-
-	numCores := 0
-	lastThread := -1
-	lastCore := -1
-	lastNode := -1
-	for _, line := range strings.Split(cpuinfo, "\n") {
-		if line == "" {
-			continue
-		}
-		ok, val, err := extractValue(line, cpuRegExp)
-		if err != nil {
-			return nil, -1, fmt.Errorf("could not parse cpu info from %q: %v", line, err)
-		}
-		if ok {
-			thread := val
-			numCores++
-			if lastThread != -1 {
-				// New cpu section. Save last one.
-				nodeIdx, err := addNode(&nodes, lastNode)
-				if err != nil {
-					return nil, -1, fmt.Errorf("failed to add node %d: %v", lastNode, err)
-				}
-				nodes[nodeIdx].AddThread(lastThread, lastCore)
-				lastCore = -1
-				lastNode = -1
-			}
-			lastThread = thread
-			continue
-		}
-		ok, val, err = extractValue(line, coreRegExp)
-		if err != nil {
-			return nil, -1, fmt.Errorf("could not parse core info from %q: %v", line, err)
-		}
-		if ok {
-			lastCore = val
-			continue
-		}
-		ok, val, err = extractValue(line, nodeRegExp)
-		if err != nil {
-			return nil, -1, fmt.Errorf("could not parse node info from %q: %v", line, err)
-		}
-		if ok {
-			lastNode = val
-			continue
-		}
-	}
-	nodeIdx, err := addNode(&nodes, lastNode)
-	if err != nil {
-		return nil, -1, fmt.Errorf("failed to add node %d: %v", lastNode, err)
-	}
-	nodes[nodeIdx].AddThread(lastThread, lastCore)
-	if numCores < 1 {
-		return nil, numCores, fmt.Errorf("could not detect any cores")
-	}
-	for idx, node := range nodes {
-		caches, err := sysinfo.GetCacheInfo(sysFs, node.Cores[0].Threads[0])
-		if err != nil {
-			glog.Errorf("failed to get cache information for node %d: %v", node.Id, err)
-			continue
-		}
-		numThreadsPerCore := len(node.Cores[0].Threads)
-		numThreadsPerNode := len(node.Cores) * numThreadsPerCore
-		for _, cache := range caches {
-			c := info.Cache{
-				Size:  cache.Size,
-				Level: cache.Level,
-				Type:  cache.Type,
-			}
-			if cache.Cpus == numThreadsPerNode && cache.Level > 2 {
-				// Add a node-level cache.
-				nodes[idx].AddNodeCache(c)
-			} else if cache.Cpus == numThreadsPerCore {
-				// Add to each core.
-				nodes[idx].AddPerCoreCache(c)
-			}
-			// Ignore unknown caches.
-		}
-	}
-	return nodes, numCores, nil
-}
-
-func extractValue(s string, r *regexp.Regexp) (bool, int, error) {
-	matches := r.FindSubmatch([]byte(s))
-	if len(matches) == 2 {
-		val, err := strconv.ParseInt(string(matches[1]), 10, 32)
-		if err != nil {
-			return false, -1, err
-		}
-		return true, int(val), nil
-	}
-	return false, -1, nil
-}
-
-func findNode(nodes []info.Node, id int) (bool, int) {
-	for i, n := range nodes {
-		if n.Id == id {
-			return true, i
-		}
-	}
-	return false, -1
-}
-
-func addNode(nodes *[]info.Node, id int) (int, error) {
-	var idx int
-	if id == -1 {
-		// Some VMs don't fill topology data. Export single package.
-		id = 0
-	}
-
-	ok, idx := findNode(*nodes, id)
-	if !ok {
-		// New node
-		node := info.Node{Id: id}
-		// Add per-node memory information.
-		meminfo := fmt.Sprintf("/sys/devices/system/node/node%d/meminfo", id)
-		out, err := ioutil.ReadFile(meminfo)
-		// Ignore if per-node info is not available.
-		if err == nil {
-			m, err := parseCapacity(out, memoryCapacityRegexp)
-			if err != nil {
-				return -1, err
-			}
-			node.Memory = uint64(m)
-		}
-		*nodes = append(*nodes, node)
-		idx = len(*nodes) - 1
-	}
-	return idx, nil
-}
-
-// s390/s390x changes
-func getMachineArch() (string, error) {
-	uname := syscall.Utsname{}
-	err := syscall.Uname(&uname)
-	if err != nil {
-		return "", err
-	}
-
-	var arch string
-	for _, val := range uname.Machine {
-		arch += string(int(val))
-	}
-
-	return arch, nil
-}
-
-// aarch64 changes
-func isAArch64() bool {
-	arch, err := getMachineArch()
-	if err == nil {
-		if true == strings.Contains(arch, "aarch64") {
-			return true
-		}
-	}
-	return false
-}
-
-// s390/s390x changes
-func isSystemZ() bool {
-	arch, err := getMachineArch()
-	if err == nil {
-		if true == strings.Contains(arch, "390") {
-			return true
-		}
-	}
-	return false
-}
-
-// s390/s390x changes
-func getNumCores() int {
-	maxProcs := runtime.GOMAXPROCS(0)
-	numCPU := runtime.NumCPU()
-
-	if maxProcs < numCPU {
-		return maxProcs
-	}
-
-	return numCPU
-}

vendor/github.com/google/cadvisor/utils/oomparser/oomparser.go

@@ -18,7 +18,6 @@ import (
 	"bufio"
 	"fmt"
 	"io"
-	"os"
 	"os/exec"
 	"path"
 	"regexp"
@@ -26,6 +25,7 @@ import (
 	"time"
 
 	"github.com/google/cadvisor/utils"
+	"github.com/google/cadvisor/utils/tail"
 
 	"github.com/golang/glog"
 )
@@ -105,27 +105,29 @@ func checkIfStartOfOomMessages(line string) bool {
 // Should prevent EOF errors that occur when lines are read before being fully
 // written to the log. It reads line by line splitting on
 // the "\n" character.
-func readLinesFromFile(lineChannel chan string, ioreader *bufio.Reader) {
+func readLinesFromFile(lineChannel chan string, ioreader *bufio.Reader) error {
 	linefragment := ""
 	var line string
 	var err error
 	for true {
 		line, err = ioreader.ReadString('\n')
-		if err == io.EOF {
-			if line != "" {
-				linefragment += line
-			}
-			time.Sleep(100 * time.Millisecond)
-		} else if err == nil {
-			if linefragment != "" {
-				line = linefragment + line
-				linefragment = ""
-			}
-			lineChannel <- line
-		} else if err != nil && err != io.EOF {
+		if err != nil && err != io.EOF {
 			glog.Errorf("exiting analyzeLinesHelper with error %v", err)
+			close(lineChannel)
+			break
+		}
+		if line == "" {
+			time.Sleep(100 * time.Millisecond)
+			continue
+		}
+		if err == nil {
+			lineChannel <- linefragment + line
+			linefragment = ""
+		} else { // err == io.EOF
+			linefragment += line
 		}
 	}
+	return err
 }
 
 // Calls goroutine for readLinesFromFile, which feeds it complete lines.
@@ -144,22 +146,23 @@ func (self *OomParser) StreamOoms(outStream chan *OomInstance) {
 			oomCurrentInstance := &OomInstance{
 				ContainerName: "/",
 			}
-			finished := false
-			for !finished {
+			for line := range lineChannel {
 				err := getContainerName(line, oomCurrentInstance)
 				if err != nil {
 					glog.Errorf("%v", err)
 				}
-				finished, err = getProcessNamePid(line, oomCurrentInstance)
+				finished, err := getProcessNamePid(line, oomCurrentInstance)
 				if err != nil {
 					glog.Errorf("%v", err)
 				}
-				line = <-lineChannel
+				if finished {
+					break
+				}
 			}
 			outStream <- oomCurrentInstance
 		}
 	}
-	glog.Infof("exiting analyzeLines")
+	glog.Infof("exiting analyzeLines. OOM events will not be reported.")
 }
 
 func callJournalctl() (io.ReadCloser, error) {
@@ -183,7 +186,6 @@ func trySystemd() (*OomParser, error) {
 	return &OomParser{
 		ioreader: bufio.NewReader(readcloser),
 	}, nil
-
 }
 
 // List of possible kernel log files. These are prioritized in order so that
@@ -192,7 +194,7 @@ var kernelLogFiles = []string{"/var/log/kern.log", "/var/log/messages", "/var/lo
 
 // looks for system files that contain kernel messages and if one is found, sets
 // the systemFile attribute of the OomParser object
-func getSystemFile() (string, error) {
+func getLogFile() (string, error) {
 	for _, logFile := range kernelLogFiles {
 		if utils.FileExists(logFile) {
 			glog.Infof("OOM parser using kernel log file: %q", logFile)
@@ -202,18 +204,29 @@ func getSystemFile() (string, error) {
 	return "", fmt.Errorf("unable to find any kernel log file available from our set: %v", kernelLogFiles)
 }
 
-// initializes an OomParser object and calls getSystemFile to set the systemFile
-// attribute.  Returns and OomParser object and an error
-func New() (*OomParser, error) {
-	systemFile, err := getSystemFile()
+func tryLogFile() (*OomParser, error) {
+	logFile, err := getLogFile()
 	if err != nil {
-		return trySystemd()
+		return nil, err
 	}
-	file, err := os.Open(systemFile)
+	tail, err := tail.NewTail(logFile)
 	if err != nil {
-		return trySystemd()
+		return nil, err
 	}
 	return &OomParser{
-		ioreader: bufio.NewReader(file),
+		ioreader: bufio.NewReader(tail),
 	}, nil
 }
+
+// initializes an OomParser object. Returns an OomParser object and an error.
+func New() (*OomParser, error) {
+	parser, err := trySystemd()
+	if err == nil {
+		return parser, nil
+	}
+	parser, err = tryLogFile()
+	if err == nil {
+		return parser, nil
+	}
+	return nil, err
+}

vendor/github.com/google/cadvisor/utils/tail/tail.go

@@ -0,0 +1,146 @@
+// Copyright 2016 Google Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package tail implements "tail -F" functionality following rotated logs
+package tail
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/golang/glog"
+	"golang.org/x/exp/inotify"
+)
+
+type Tail struct {
+	reader     *bufio.Reader
+	readerErr  error
+	readerLock sync.RWMutex
+	filename   string
+	file       *os.File
+	stop       chan bool
+	watcher    *inotify.Watcher
+}
+
+const (
+	defaultRetryInterval = 100 * time.Millisecond
+	maxRetryInterval     = 30 * time.Second
+)
+
+// NewTail starts opens the given file and watches it for deletion/rotation
+func NewTail(filename string) (*Tail, error) {
+	t := &Tail{
+		filename: filename,
+	}
+	var err error
+	t.stop = make(chan bool)
+	t.watcher, err = inotify.NewWatcher()
+	if err != nil {
+		return nil, fmt.Errorf("inotify init failed on %s: %v", t.filename, err)
+	}
+	go t.watchLoop()
+	return t, nil
+}
+
+// Read implements the io.Reader interface for Tail
+func (t *Tail) Read(p []byte) (int, error) {
+	t.readerLock.RLock()
+	defer t.readerLock.RUnlock()
+	if t.reader == nil || t.readerErr != nil {
+		return 0, t.readerErr
+	}
+	return t.reader.Read(p)
+}
+
+var _ io.Reader = &Tail{}
+
+// Close stops watching and closes the file
+func (t *Tail) Close() {
+	close(t.stop)
+}
+
+func (t *Tail) attemptOpen() error {
+	t.readerLock.Lock()
+	defer t.readerLock.Unlock()
+	t.reader = nil
+	t.readerErr = nil
+	attempt := 0
+	for interval := defaultRetryInterval; ; interval *= 2 {
+		attempt++
+		glog.V(4).Infof("Opening %s (attempt %d)", t.filename, attempt)
+		var err error
+		t.file, err = os.Open(t.filename)
+		if err == nil {
+			// TODO: not interested in old events?
+			//t.file.Seek(0, os.SEEK_END)
+			t.reader = bufio.NewReader(t.file)
+			return nil
+		}
+		if interval >= maxRetryInterval {
+			break
+		}
+		select {
+		case <-time.After(interval):
+		case <-t.stop:
+			t.readerErr = io.EOF
+			return fmt.Errorf("watch was cancelled")
+		}
+	}
+	err := fmt.Errorf("can't open log file %s", t.filename)
+	t.readerErr = err
+	return err
+}
+
+func (t *Tail) watchLoop() {
+	for {
+		err := t.watchFile()
+		if err != nil {
+			glog.Errorf("Tail failed on %s: %v", t.filename, err)
+			break
+		}
+	}
+}
+
+func (t *Tail) watchFile() error {
+	err := t.attemptOpen()
+	if err != nil {
+		return err
+	}
+	defer t.file.Close()
+
+	watchDir := filepath.Dir(t.filename)
+	err = t.watcher.AddWatch(watchDir, inotify.IN_MOVED_FROM|inotify.IN_DELETE)
+	if err != nil {
+		return fmt.Errorf("Failed to add watch to directory %s: %v", watchDir, err)
+	}
+	defer t.watcher.RemoveWatch(watchDir)
+
+	for {
+		select {
+		case event := <-t.watcher.Event:
+			eventPath := filepath.Clean(event.Name) // Directory events have an extra '/'
+			if eventPath == t.filename {
+				glog.V(4).Infof("Log file %s moved/deleted", t.filename)
+				return nil
+			}
+		case <-t.stop:
+			return fmt.Errorf("watch was cancelled")
+		}
+	}
+}

vendor/github.com/google/cadvisor/utils/timed_store.go

@@ -58,19 +58,23 @@ func NewTimedStore(age time.Duration, maxItems int) *TimedStore {
 
 // Adds an element to the start of the buffer (removing one from the end if necessary).
 func (self *TimedStore) Add(timestamp time.Time, item interface{}) {
-	// Remove any elements if over our max size.
-	if self.maxItems >= 0 && (len(self.buffer)+1) > self.maxItems {
-		startIndex := len(self.buffer) + 1 - self.maxItems
-		self.buffer = self.buffer[startIndex:]
-	}
-	// Add the new element first and sort. We can then remove an expired element, if required.
-	copied := item
-	self.buffer = append(self.buffer, timedStoreData{
+	data := timedStoreData{
 		timestamp: timestamp,
-		data:      copied,
-	})
+		data:      item,
+	}
+	// Common case: data is added in order.
+	if len(self.buffer) == 0 || !timestamp.Before(self.buffer[len(self.buffer)-1].timestamp) {
+		self.buffer = append(self.buffer, data)
+	} else {
+		// Data is out of order; insert it in the correct position.
+		index := sort.Search(len(self.buffer), func(index int) bool {
+			return self.buffer[index].timestamp.After(timestamp)
+		})
+		self.buffer = append(self.buffer, timedStoreData{}) // Make room to shift the elements
+		copy(self.buffer[index+1:], self.buffer[index:])    // Shift the elements over
+		self.buffer[index] = data
+	}
 
-	sort.Sort(self.buffer)
 	// Remove any elements before eviction time.
 	// TODO(rjnagal): This is assuming that the added entry has timestamp close to now.
 	evictTime := timestamp.Add(-self.age)
@@ -81,6 +85,11 @@ func (self *TimedStore) Add(timestamp time.Time, item interface{}) {
 		self.buffer = self.buffer[index:]
 	}
 
+	// Remove any elements if over our max size.
+	if self.maxItems >= 0 && len(self.buffer) > self.maxItems {
+		startIndex := len(self.buffer) - self.maxItems
+		self.buffer = self.buffer[startIndex:]
+	}
 }
 
 // Returns up to maxResult elements in the specified time period (inclusive).