Merge pull request #671 from vmarmol/limit

Add an item limit to TimedStore

events/handler.go

@@ -100,8 +100,10 @@ type events struct {
 	watcherLock sync.RWMutex
 	// last allocated watch id.
 	lastId int
-	// Max duration for which to keep events.
+	// Max duration for which to keep events (per event type).
 	maxAge time.Duration
+	// Max number of events to keep (per event type).
+	maxNumEvents int
 }
 
 // initialized by a call to WatchEvents(), a watch struct will then be added
@@ -128,12 +130,14 @@ func NewEventChannel(watchId int) *EventChannel {
 }
 
 // returns a pointer to an initialized Events object.
-// eventMaxAge is the max duration for which to keep events.
+// eventMaxAge is the max duration for which to keep events per type.
-func NewEventManager(eventMaxAge time.Duration) *events {
+// maxNumEvents is the max number of events to keep per type (-1 for no limit).
+func NewEventManager(eventMaxAge time.Duration, maxNumEvents int) *events {
 	return &events{
-		eventStore: make(map[info.EventType]*utils.TimedStore, 0),
+		eventStore:   make(map[info.EventType]*utils.TimedStore, 0),
-		watchers:   make(map[int]*watch),
+		watchers:     make(map[int]*watch),
-		maxAge:     eventMaxAge,
+		maxAge:       eventMaxAge,
+		maxNumEvents: maxNumEvents,
 	}
 }
 
@@ -262,7 +266,7 @@ func (self *events) updateEventStore(e *info.Event) {
 	self.eventsLock.Lock()
 	defer self.eventsLock.Unlock()
 	if _, ok := self.eventStore[e.EventType]; !ok {
-		self.eventStore[e.EventType] = utils.NewTimedStore(self.maxAge)
+		self.eventStore[e.EventType] = utils.NewTimedStore(self.maxAge, self.maxNumEvents)
 	}
 	self.eventStore[e.EventType].Add(e.Timestamp, e)
 }

events/handler_test.go

@@ -47,7 +47,7 @@ func initializeScenario(t *testing.T) (*events, *Request, *info.Event, *info.Eve
 	fakeEvent := makeEvent(createOldTime(t), "/")
 	fakeEvent2 := makeEvent(time.Now(), "/")
 
-	return NewEventManager(time.Hour), NewRequest(), fakeEvent, fakeEvent2
+	return NewEventManager(time.Hour, -1), NewRequest(), fakeEvent, fakeEvent2
 }
 
 func checkNumberOfEvents(t *testing.T, numEventsExpected int, numEventsReceived int) {

manager/manager.go

@@ -137,7 +137,7 @@ func New(memoryStorage *memory.InMemoryStorage, sysfs sysfs.SysFs) (Manager, err
 	glog.Infof("Version: %+v", newManager.versionInfo)
 
 	// TODO(vmarmol): Make configurable.
-	newManager.eventHandler = events.NewEventManager(24 * time.Hour)
+	newManager.eventHandler = events.NewEventManager(24*time.Hour, 100000)
 
 	// Register Docker container factory.
 	err = docker.Register(newManager, fsInfo)

storage/memory/memory.go

@@ -57,7 +57,7 @@ func (self *containerStorage) RecentStats(start, end time.Time, maxStats int) ([
 func newContainerStore(ref info.ContainerReference, maxAge time.Duration) *containerStorage {
 	return &containerStorage{
 		ref:         ref,
-		recentStats: utils.NewTimedStore(maxAge),
+		recentStats: utils.NewTimedStore(maxAge, -1),
 		maxAge:      maxAge,
 	}
 }

utils/timed_store.go

@@ -19,10 +19,12 @@ import (
 	"time"
 )
 
-// A time-based buffer for ContainerStats. Holds information for a specific time period.
+// A time-based buffer for ContainerStats.
+// Holds information for a specific time period and/or a max number of items.
 type TimedStore struct {
-	buffer []timedStoreData
+	buffer   []timedStoreData
-	age    time.Duration
+	age      time.Duration
+	maxItems int
 }
 
 type timedStoreData struct {
@@ -31,10 +33,12 @@ type timedStoreData struct {
 }
 
 // Returns a new thread-compatible TimedStore.
-func NewTimedStore(age time.Duration) *TimedStore {
+// A maxItems value of -1 means no limit.
+func NewTimedStore(age time.Duration, maxItems int) *TimedStore {
 	return &TimedStore{
-		buffer: make([]timedStoreData, 0),
+		buffer:   make([]timedStoreData, 0),
-		age:    age,
+		age:      age,
+		maxItems: maxItems,
 	}
 }
 
@@ -49,6 +53,12 @@ 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)+1) > self.maxItems {
+		startIndex := len(self.buffer) + 1 - self.maxItems
+		self.buffer = self.buffer[startIndex:]
+	}
+
 	copied := item
 	self.buffer = append(self.buffer, timedStoreData{
 		timestamp: timestamp,

utils/timed_store_test.go

@@ -55,7 +55,7 @@ func expectElements(t *testing.T, actual []interface{}, expected []int) {
 }
 
 func TestAdd(t *testing.T) {
-	sb := NewTimedStore(5 * time.Second)
+	sb := NewTimedStore(5*time.Second, 100)
 
 	// Add 1.
 	sb.Add(createTime(0), 0)
@@ -84,7 +84,7 @@ func TestAdd(t *testing.T) {
 }
 
 func TestGet(t *testing.T) {
-	sb := NewTimedStore(5 * time.Second)
+	sb := NewTimedStore(5*time.Second, -1)
 	sb.Add(createTime(1), 1)
 	sb.Add(createTime(2), 2)
 	sb.Add(createTime(3), 3)
@@ -97,7 +97,7 @@ func TestGet(t *testing.T) {
 }
 
 func TestInTimeRange(t *testing.T) {
-	sb := NewTimedStore(5 * time.Second)
+	sb := NewTimedStore(5*time.Second, -1)
 	assert := assert.New(t)
 
 	var empty time.Time
@@ -174,7 +174,7 @@ func TestInTimeRange(t *testing.T) {
 }
 
 func TestInTimeRangeWithLimit(t *testing.T) {
-	sb := NewTimedStore(5 * time.Second)
+	sb := NewTimedStore(5*time.Second, -1)
 	sb.Add(createTime(1), 1)
 	sb.Add(createTime(2), 2)
 	sb.Add(createTime(3), 3)
@@ -190,3 +190,32 @@ func TestInTimeRangeWithLimit(t *testing.T) {
 	expectElements(t, sb.InTimeRange(empty, empty, 1), []int{4})
 	assert.Empty(t, sb.InTimeRange(empty, empty, 0))
 }
+
+func TestLimitedSize(t *testing.T) {
+	sb := NewTimedStore(time.Hour, 5)
+
+	// Add 1.
+	sb.Add(createTime(0), 0)
+	expectSize(t, sb, 1)
+	expectAllElements(t, sb, []int{0})
+
+	// Fill the buffer.
+	for i := 1; i <= 5; i++ {
+		expectSize(t, sb, i)
+		sb.Add(createTime(i), i)
+	}
+	expectSize(t, sb, 5)
+	expectAllElements(t, sb, []int{1, 2, 3, 4, 5})
+
+	// Add more than is available in the buffer
+	sb.Add(createTime(6), 6)
+	expectSize(t, sb, 5)
+	expectAllElements(t, sb, []int{2, 3, 4, 5, 6})
+
+	// Replace all elements.
+	for i := 7; i <= 10; i++ {
+		sb.Add(createTime(i), i)
+	}
+	expectSize(t, sb, 5)
+	expectAllElements(t, sb, []int{6, 7, 8, 9, 10})
+}