Add InTimeRange() to StatsBuffer.

This will allow us to accept queries over a narrower window of time.

storage/memory/stats_buffer.go

@@ -15,6 +15,9 @@
 package memory
 
 import (
+	"sort"
+	"time"
+
 	"github.com/google/cadvisor/info"
 )
 
@@ -43,7 +46,69 @@ func (self *StatsBuffer) Add(item *info.ContainerStats) {
 	self.buffer[self.index] = *item
 }
 
+// Returns up to maxResult elements in the specified time period (inclusive).
+// Results are from first to last. maxResults of -1 means no limit.
+func (self *StatsBuffer) InTimeRange(start, end time.Time, maxResults int) []*info.ContainerStats {
+	// No stats, return empty.
+	if self.size == 0 {
+		return []*info.ContainerStats{}
+	}
+
+	// NOTE: Since we store the elments in descending timestamp order "start" will
+	// be a higher index than "end".
+
+	var startIndex int
+	if start.IsZero() {
+		// None specified, start at the beginning.
+		startIndex = self.size - 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 {
+			// buffer[index] < start
+			return self.Get(index).Timestamp.Before(start)
+		}) - 1
+		// Check if start is after all the data we have.
+		if startIndex < 0 {
+			return []*info.ContainerStats{}
+		}
+	}
+
+	var endIndex int
+	if end.IsZero() {
+		// None specified, end with the latest stats.
+		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 {
+			// 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 {
+			return []*info.ContainerStats{}
+		}
+	}
+
+	// Trim to maxResults size.
+	numResults := startIndex - endIndex + 1
+	if maxResults != -1 && numResults > maxResults {
+		startIndex -= numResults - maxResults
+		numResults = maxResults
+	}
+
+	// Return in sorted timestamp order so from the "back" to "front".
+	result := make([]*info.ContainerStats, numResults)
+	for i := 0; i < numResults; i++ {
+		result[i] = self.Get(startIndex - i)
+	}
+	return result
+}
+
+// TODO(vmarmol): Remove this function as it will no longer be neededt.
 // Returns the first N elements in the buffer. If N > size of buffer, size of buffer elements are returned.
+// Returns the elements in ascending timestamp order.
 func (self *StatsBuffer) FirstN(n int) []*info.ContainerStats {
 	// Cap n at the number of elements we have.
 	if n > self.size {
@@ -65,6 +130,15 @@ func (self *StatsBuffer) FirstN(n int) []*info.ContainerStats {
 	return res
 }
 
+// Gets the element at the specified index. Note that elements are stored 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]
+}
+
 func (self *StatsBuffer) Size() int {
 	return self.size
 }

storage/memory/stats_buffer_test.go

@@ -15,13 +15,23 @@
 package memory
 
 import (
+	"strconv"
+	"strings"
 	"testing"
+	"time"
 
 	"github.com/google/cadvisor/info"
+	"github.com/stretchr/testify/assert"
 )
 
+func createTime(id int) time.Time {
+	var zero time.Time
+	return zero.Add(time.Duration(id+1) * time.Second)
+}
+
 func createStats(id int32) *info.ContainerStats {
 	return &info.ContainerStats{
+		Timestamp: createTime(int(id)),
 		Cpu: info.CpuStats{
 			LoadAverage: id,
 		},
@@ -34,26 +44,40 @@ func expectSize(t *testing.T, sb *StatsBuffer, expectedSize int) {
 	}
 }
 
-func expectElements(t *testing.T, sb *StatsBuffer, expected []int32) {
+func expectFirstN(t *testing.T, sb *StatsBuffer, expected []int32) {
-	res := sb.FirstN(sb.Size())
+	expectElements(t, sb.FirstN(sb.Size()), expected)
-	if len(res) != len(expected) {
+}
-		t.Errorf("Expected elements %v, got %v", expected, res)
+
+func expectElements(t *testing.T, actual []*info.ContainerStats, expected []int32) {
+	if len(actual) != len(expected) {
+		t.Errorf("Expected elements %v, got %v", expected, actual)
 		return
 	}
-	for i, el := range res {
+	for i, el := range actual {
 		if el.Cpu.LoadAverage != expected[i] {
-			t.Errorf("Expected elements %v, got %v", expected, res)
+			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, ","))
+			return
 		}
 	}
 }
 
+func expectElement(t *testing.T, stat *info.ContainerStats, expected int32) {
+	if stat.Cpu.LoadAverage != expected {
+		t.Errorf("Expected %d, but received %d", expected, stat.Cpu.LoadAverage)
+	}
+}
+
 func TestAddAndFirstN(t *testing.T) {
 	sb := NewStatsBuffer(5)
 
 	// Add 1.
 	sb.Add(createStats(1))
 	expectSize(t, sb, 1)
-	expectElements(t, sb, []int32{1})
+	expectFirstN(t, sb, []int32{1})
 
 	// Fill the buffer.
 	for i := 1; i <= 5; i++ {
@@ -61,17 +85,122 @@ func TestAddAndFirstN(t *testing.T) {
 		sb.Add(createStats(int32(i)))
 	}
 	expectSize(t, sb, 5)
-	expectElements(t, sb, []int32{1, 2, 3, 4, 5})
+	expectFirstN(t, sb, []int32{1, 2, 3, 4, 5})
 
 	// Add more than is available in the buffer
 	sb.Add(createStats(6))
 	expectSize(t, sb, 5)
-	expectElements(t, sb, []int32{2, 3, 4, 5, 6})
+	expectFirstN(t, sb, []int32{2, 3, 4, 5, 6})
 
 	// Replace all elements.
 	for i := 7; i <= 10; i++ {
 		sb.Add(createStats(int32(i)))
 	}
 	expectSize(t, sb, 5)
-	expectElements(t, sb, []int32{6, 7, 8, 9, 10})
+	expectFirstN(t, sb, []int32{6, 7, 8, 9, 10})
+}
+
+func TestGet(t *testing.T) {
+	sb := NewStatsBuffer(5)
+	sb.Add(createStats(1))
+	sb.Add(createStats(2))
+	sb.Add(createStats(3))
+	expectSize(t, sb, 3)
+	expectFirstN(t, sb, []int32{1, 2, 3})
+
+	expectElement(t, sb.Get(0), 3)
+	expectElement(t, sb.Get(1), 2)
+	expectElement(t, sb.Get(2), 1)
+}
+
+func TestInTimeRange(t *testing.T) {
+	sb := NewStatsBuffer(5)
+	assert := assert.New(t)
+
+	var empty time.Time
+
+	// No elements.
+	assert.Empty(sb.InTimeRange(createTime(0), createTime(5), 10))
+	assert.Empty(sb.InTimeRange(createTime(0), empty, 10))
+	assert.Empty(sb.InTimeRange(empty, createTime(5), 10))
+	assert.Empty(sb.InTimeRange(empty, empty, 10))
+
+	// One element.
+	sb.Add(createStats(1))
+	expectSize(t, sb, 1)
+	expectElements(t, sb.InTimeRange(createTime(0), createTime(5), 10), []int32{1})
+	expectElements(t, sb.InTimeRange(createTime(1), createTime(5), 10), []int32{1})
+	expectElements(t, sb.InTimeRange(createTime(0), createTime(1), 10), []int32{1})
+	expectElements(t, sb.InTimeRange(createTime(1), createTime(1), 10), []int32{1})
+	assert.Empty(sb.InTimeRange(createTime(2), createTime(5), 10))
+
+	// Two element.
+	sb.Add(createStats(2))
+	expectSize(t, sb, 2)
+	expectElements(t, sb.InTimeRange(createTime(0), createTime(5), 10), []int32{1, 2})
+	expectElements(t, sb.InTimeRange(createTime(1), createTime(5), 10), []int32{1, 2})
+	expectElements(t, sb.InTimeRange(createTime(0), createTime(2), 10), []int32{1, 2})
+	expectElements(t, sb.InTimeRange(createTime(1), createTime(2), 10), []int32{1, 2})
+	expectElements(t, sb.InTimeRange(createTime(1), createTime(1), 10), []int32{1})
+	expectElements(t, sb.InTimeRange(createTime(2), createTime(2), 10), []int32{2})
+	assert.Empty(sb.InTimeRange(createTime(3), createTime(5), 10))
+
+	// Many elements.
+	sb.Add(createStats(3))
+	sb.Add(createStats(4))
+	expectSize(t, sb, 4)
+	expectElements(t, sb.InTimeRange(createTime(0), createTime(5), 10), []int32{1, 2, 3, 4})
+	expectElements(t, sb.InTimeRange(createTime(0), createTime(5), 10), []int32{1, 2, 3, 4})
+	expectElements(t, sb.InTimeRange(createTime(1), createTime(5), 10), []int32{1, 2, 3, 4})
+	expectElements(t, sb.InTimeRange(createTime(0), createTime(4), 10), []int32{1, 2, 3, 4})
+	expectElements(t, sb.InTimeRange(createTime(1), createTime(4), 10), []int32{1, 2, 3, 4})
+	expectElements(t, sb.InTimeRange(createTime(0), createTime(2), 10), []int32{1, 2})
+	expectElements(t, sb.InTimeRange(createTime(1), createTime(2), 10), []int32{1, 2})
+	expectElements(t, sb.InTimeRange(createTime(2), createTime(3), 10), []int32{2, 3})
+	expectElements(t, sb.InTimeRange(createTime(3), createTime(4), 10), []int32{3, 4})
+	expectElements(t, sb.InTimeRange(createTime(3), createTime(5), 10), []int32{3, 4})
+	assert.Empty(sb.InTimeRange(createTime(5), createTime(5), 10))
+
+	// No start time.
+	expectElements(t, sb.InTimeRange(empty, createTime(5), 10), []int32{1, 2, 3, 4})
+	expectElements(t, sb.InTimeRange(empty, createTime(4), 10), []int32{1, 2, 3, 4})
+	expectElements(t, sb.InTimeRange(empty, createTime(3), 10), []int32{1, 2, 3})
+	expectElements(t, sb.InTimeRange(empty, createTime(2), 10), []int32{1, 2})
+	expectElements(t, sb.InTimeRange(empty, createTime(1), 10), []int32{1})
+
+	// No end time.
+	expectElements(t, sb.InTimeRange(createTime(0), empty, 10), []int32{1, 2, 3, 4})
+	expectElements(t, sb.InTimeRange(createTime(1), empty, 10), []int32{1, 2, 3, 4})
+	expectElements(t, sb.InTimeRange(createTime(2), empty, 10), []int32{2, 3, 4})
+	expectElements(t, sb.InTimeRange(createTime(3), empty, 10), []int32{3, 4})
+	expectElements(t, sb.InTimeRange(createTime(4), empty, 10), []int32{4})
+
+	// No start or end time.
+	expectElements(t, sb.InTimeRange(empty, empty, 10), []int32{1, 2, 3, 4})
+
+	// Start after data.
+	assert.Empty(sb.InTimeRange(createTime(5), createTime(5), 10))
+	assert.Empty(sb.InTimeRange(createTime(5), empty, 10))
+
+	// End before data.
+	assert.Empty(sb.InTimeRange(createTime(0), createTime(0), 10))
+	assert.Empty(sb.InTimeRange(empty, createTime(0), 10))
+}
+
+func TestInTimeRangeWithLimit(t *testing.T) {
+	sb := NewStatsBuffer(5)
+	sb.Add(createStats(1))
+	sb.Add(createStats(2))
+	sb.Add(createStats(3))
+	sb.Add(createStats(4))
+	expectSize(t, sb, 4)
+
+	var empty time.Time
+
+	// Limit cuts off from latest timestamp.
+	expectElements(t, sb.InTimeRange(empty, empty, 4), []int32{1, 2, 3, 4})
+	expectElements(t, sb.InTimeRange(empty, empty, 3), []int32{2, 3, 4})
+	expectElements(t, sb.InTimeRange(empty, empty, 2), []int32{3, 4})
+	expectElements(t, sb.InTimeRange(empty, empty, 1), []int32{4})
+	assert.Empty(t, sb.InTimeRange(empty, empty, 0))
 }