Move StatsBuffer to be time-based.

It now keeps stats for a certain amount of time before expiring them. It
used to keep a certain number of stats instead.

storage/memory/memory.go

@@ -29,7 +29,7 @@ import (
 type containerStorage struct {
 	ref         info.ContainerReference
 	recentStats *StatsBuffer
-	maxNumStats int
+	maxAge      time.Duration
 	lock        sync.RWMutex
 }
 
@@ -48,18 +48,18 @@ func (self *containerStorage) RecentStats(start, end time.Time, maxStats int) ([
 	return self.recentStats.InTimeRange(start, end, maxStats), nil
 }
 
-func newContainerStore(ref info.ContainerReference, maxNumStats int) *containerStorage {
+func newContainerStore(ref info.ContainerReference, maxAge time.Duration) *containerStorage {
 	return &containerStorage{
 		ref:         ref,
-		recentStats: NewStatsBuffer(maxNumStats),
-		maxNumStats: maxNumStats,
+		recentStats: NewStatsBuffer(maxAge),
+		maxAge:      maxAge,
 	}
 }
 
 type InMemoryStorage struct {
 	lock                sync.RWMutex
 	containerStorageMap map[string]*containerStorage
-	maxNumStats         int
+	maxAge              time.Duration
 	backend             storage.StorageDriver
 }
 
@@ -71,7 +71,7 @@ func (self *InMemoryStorage) AddStats(ref info.ContainerReference, stats *info.C
 		self.lock.Lock()
 		defer self.lock.Unlock()
 		if cstore, ok = self.containerStorageMap[ref.Name]; !ok {
-			cstore = newContainerStore(ref, self.maxNumStats)
+			cstore = newContainerStore(ref, self.maxAge)
 			self.containerStorageMap[ref.Name] = cstore
 		}
 	}()
@@ -113,12 +113,12 @@ func (self *InMemoryStorage) Close() error {
 }
 
 func New(
-	maxNumStats int,
+	maxAge time.Duration,
 	backend storage.StorageDriver,
 ) *InMemoryStorage {
 	ret := &InMemoryStorage{
 		containerStorageMap: make(map[string]*containerStorage, 32),
-		maxNumStats:         maxNumStats,
+		maxAge:              maxAge,
 		backend:             backend,
 	}
 	return ret

storage/memory/memory_test.go

@@ -47,7 +47,7 @@ func getRecentStats(t *testing.T, memoryStorage *InMemoryStorage, numStats int)
 }
 
 func TestAddStats(t *testing.T) {
-	memoryStorage := New(60, nil)
+	memoryStorage := New(60*time.Second, nil)
 
 	assert := assert.New(t)
 	assert.Nil(memoryStorage.AddStats(containerRef, makeStat(0)))
@@ -70,7 +70,7 @@ func TestRecentStatsNoRecentStats(t *testing.T) {
 
 // Make an instance of InMemoryStorage with n stats.
 func makeWithStats(n int) *InMemoryStorage {
-	memoryStorage := New(60, nil)
+	memoryStorage := New(60*time.Second, nil)
 
 	for i := 0; i < n; i++ {
 		memoryStorage.AddStats(containerRef, makeStat(i))

storage/memory/stats_buffer.go

@@ -21,38 +21,41 @@ import (
 	info "github.com/google/cadvisor/info/v1"
 )
 
-// A circular buffer for ContainerStats.
+// A time-based buffer for ContainerStats. Holds information for a specific time period.
 type StatsBuffer struct {
 	buffer []*info.ContainerStats
-	size   int
-	index  int
+	age    time.Duration
 }
 
 // Returns a new thread-compatible StatsBuffer.
-func NewStatsBuffer(size int) *StatsBuffer {
+func NewStatsBuffer(age time.Duration) *StatsBuffer {
 	return &StatsBuffer{
-		buffer: make([]*info.ContainerStats, size),
-		size:   0,
-		index:  size - 1,
+		buffer: make([]*info.ContainerStats, 0),
+		age:    age,
 	}
 }
 
 // Adds an element to the start of the buffer (removing one from the end if necessary).
 func (self *StatsBuffer) Add(item *info.ContainerStats) {
-	if self.size < len(self.buffer) {
-		self.size++
+	// Remove any elements before the eviction time.
+	evictTime := item.Timestamp.Add(-self.age)
+	index := sort.Search(len(self.buffer), func(index int) bool {
+		return self.buffer[index].Timestamp.After(evictTime)
+	})
+	if index < len(self.buffer) {
+		self.buffer = self.buffer[index:]
 	}
-	self.index = (self.index + 1) % len(self.buffer)
+
 	copied := *item
-	self.buffer[self.index] = &copied
+	self.buffer = append(self.buffer, &copied)
 }
 
 // Returns up to maxResult elements in the specified time period (inclusive).
 // Results are from first to last. maxResults of -1 means no limit. When first
 // and last are specified, maxResults is ignored.
 func (self *StatsBuffer) InTimeRange(start, end time.Time, maxResults int) []*info.ContainerStats {
 	// No stats, return empty.
-	if self.size == 0 {
+	if len(self.buffer) == 0 {
 		return []*info.ContainerStats{}
 	}
 
@@ -67,12 +70,12 @@ func (self *StatsBuffer) InTimeRange(start, end time.Time, maxResults int) []*in
 	var startIndex int
 	if start.IsZero() {
 		// None specified, start at the beginning.
-		startIndex = self.size - 1
+		startIndex = len(self.buffer) - 1
 	} else {
 		// Start is the index before the elements smaller than it. We do this by
 		// finding the first element smaller than start and taking the index
 		// before that element
-		startIndex = sort.Search(self.size, func(index int) bool {
+		startIndex = sort.Search(len(self.buffer), func(index int) bool {
 			// buffer[index] < start
 			return self.Get(index).Timestamp.Before(start)
 		}) - 1
@@ -88,12 +91,12 @@ func (self *StatsBuffer) InTimeRange(start, end time.Time, maxResults int) []*in
 		endIndex = 0
 	} else {
 		// End is the first index smaller than or equal to it (so, not larger).
-		endIndex = sort.Search(self.size, func(index int) bool {
+		endIndex = sort.Search(len(self.buffer), func(index int) bool {
 			// buffer[index] <= t -> !(buffer[index] > t)
 			return !self.Get(index).Timestamp.After(end)
 		})
 		// Check if end is before all the data we have.
-		if endIndex == self.size {
+		if endIndex == len(self.buffer) {
 			return []*info.ContainerStats{}
 		}
 	}
@@ -113,15 +116,11 @@ func (self *StatsBuffer) InTimeRange(start, end time.Time, maxResults int) []*in
 	return result
 }
 
-// Gets the element at the specified index. Note that elements are stored in LIFO order.
+// Gets the element at the specified index. Note that elements are output in LIFO order.
 func (self *StatsBuffer) Get(index int) *info.ContainerStats {
-	calculatedIndex := self.index - index
-	if calculatedIndex < 0 {
-		calculatedIndex += len(self.buffer)
-	}
-	return self.buffer[calculatedIndex]
+	return self.buffer[len(self.buffer)-index-1]
 }
 
 func (self *StatsBuffer) Size() int {
-	return self.size
+	return len(self.buffer)
 }

storage/memory/stats_buffer_test.go

@@ -53,18 +53,22 @@ func expectAllElements(t *testing.T, sb *StatsBuffer, expected []int32) {
 	expectElements(t, els, expected)
 }
 
+func getActualElements(actual []*info.ContainerStats) string {
+	actualElements := make([]string, len(actual))
+	for i, element := range actual {
+		actualElements[i] = strconv.Itoa(int(element.Cpu.LoadAverage))
+	}
+	return strings.Join(actualElements, ",")
+}
+
 func expectElements(t *testing.T, actual []*info.ContainerStats, expected []int32) {
 	if len(actual) != len(expected) {
-		t.Errorf("Expected elements %v, got %v", expected, actual)
+		t.Errorf("Expected elements %v, got %v", expected, getActualElements(actual))
 		return
 	}
 	for i, el := range actual {
 		if el.Cpu.LoadAverage != expected[i] {
-			actualElements := make([]string, len(actual))
-			for i, element := range actual {
-				actualElements[i] = strconv.Itoa(int(element.Cpu.LoadAverage))
-			}
-			t.Errorf("Expected elements %v, got %v", expected, strings.Join(actualElements, ","))
+			t.Errorf("Expected elements %v, got %v", expected, getActualElements(actual))
 			return
 		}
 	}
@@ -77,12 +81,12 @@ func expectElement(t *testing.T, stat *info.ContainerStats, expected int32) {
 }
 
 func TestAdd(t *testing.T) {
-	sb := NewStatsBuffer(5)
+	sb := NewStatsBuffer(5 * time.Second)
 
 	// Add 1.
-	sb.Add(createStats(1))
+	sb.Add(createStats(0))
 	expectSize(t, sb, 1)
-	expectAllElements(t, sb, []int32{1})
+	expectAllElements(t, sb, []int32{0})
 
 	// Fill the buffer.
 	for i := 1; i <= 5; i++ {
@@ -106,7 +110,7 @@ func TestAdd(t *testing.T) {
 }
 
 func TestGet(t *testing.T) {
-	sb := NewStatsBuffer(5)
+	sb := NewStatsBuffer(5 * time.Second)
 	sb.Add(createStats(1))
 	sb.Add(createStats(2))
 	sb.Add(createStats(3))
@@ -118,7 +122,7 @@ func TestGet(t *testing.T) {
 }
 
 func TestInTimeRange(t *testing.T) {
-	sb := NewStatsBuffer(5)
+	sb := NewStatsBuffer(5 * time.Second)
 	assert := assert.New(t)
 
 	var empty time.Time
@@ -195,7 +199,7 @@ func TestInTimeRange(t *testing.T) {
 }
 
 func TestInTimeRangeWithLimit(t *testing.T) {
-	sb := NewStatsBuffer(5)
+	sb := NewStatsBuffer(5 * time.Second)
 	sb.Add(createStats(1))
 	sb.Add(createStats(2))
 	sb.Add(createStats(3))

storagedriver.go

@@ -92,6 +92,6 @@ func NewMemoryStorage(backendStorageName string) (*memory.InMemoryStorage, error
 		glog.Infof("No backend storage selected")
 	}
 	glog.Infof("Caching %d stats in memory", statsToCache)
-	storageDriver = memory.New(statsToCache, backendStorage)
+	storageDriver = memory.New(time.Duration(statsToCache)*time.Second, backendStorage)
 	return storageDriver, nil
 }