Add events StoragePolicy.
This commit is contained in:
parent
872ab737fb
commit
28dfea761c
@ -100,10 +100,8 @@ type events struct {
|
|||||||
watcherLock sync.RWMutex
|
watcherLock sync.RWMutex
|
||||||
// last allocated watch id.
|
// last allocated watch id.
|
||||||
lastId int
|
lastId int
|
||||||
// Max duration for which to keep events (per event type).
|
// Event storage policy.
|
||||||
maxAge time.Duration
|
storagePolicy StoragePolicy
|
||||||
// Max number of events to keep (per event type).
|
|
||||||
maxNumEvents int
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// initialized by a call to WatchEvents(), a watch struct will then be added
|
// initialized by a call to WatchEvents(), a watch struct will then be added
|
||||||
@ -129,15 +127,34 @@ func NewEventChannel(watchId int) *EventChannel {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Policy specifying how many events to store.
|
||||||
|
// MaxAge is the max duration for which to keep events.
|
||||||
|
// MaxNumEvents is the max number of events to keep (-1 for no limit).
|
||||||
|
type StoragePolicy struct {
|
||||||
|
// Defaults limites, used if a per-event limit is not set.
|
||||||
|
DefaultMaxAge time.Duration
|
||||||
|
DefaultMaxNumEvents int
|
||||||
|
|
||||||
|
// Per-event type limits.
|
||||||
|
PerTypeMaxAge map[info.EventType]time.Duration
|
||||||
|
PerTypeMaxNumEvents map[info.EventType]int
|
||||||
|
}
|
||||||
|
|
||||||
|
func DefaultStoragePolicy() StoragePolicy {
|
||||||
|
return StoragePolicy{
|
||||||
|
DefaultMaxAge: 24 * time.Hour,
|
||||||
|
DefaultMaxNumEvents: 100000,
|
||||||
|
PerTypeMaxAge: make(map[info.EventType]time.Duration),
|
||||||
|
PerTypeMaxNumEvents: make(map[info.EventType]int),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// returns a pointer to an initialized Events object.
|
// returns a pointer to an initialized Events object.
|
||||||
// eventMaxAge is the max duration for which to keep events per type.
|
func NewEventManager(storagePolicy StoragePolicy) *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{
|
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,
|
storagePolicy: storagePolicy,
|
||||||
maxNumEvents: maxNumEvents,
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -266,7 +283,16 @@ func (self *events) updateEventStore(e *info.Event) {
|
|||||||
self.eventsLock.Lock()
|
self.eventsLock.Lock()
|
||||||
defer self.eventsLock.Unlock()
|
defer self.eventsLock.Unlock()
|
||||||
if _, ok := self.eventStore[e.EventType]; !ok {
|
if _, ok := self.eventStore[e.EventType]; !ok {
|
||||||
self.eventStore[e.EventType] = utils.NewTimedStore(self.maxAge, self.maxNumEvents)
|
maxAge := self.storagePolicy.DefaultMaxAge
|
||||||
|
maxNumEvents := self.storagePolicy.DefaultMaxNumEvents
|
||||||
|
if age, ok := self.storagePolicy.PerTypeMaxAge[e.EventType]; ok {
|
||||||
|
maxAge = age
|
||||||
|
}
|
||||||
|
if numEvents, ok := self.storagePolicy.PerTypeMaxNumEvents[e.EventType]; ok {
|
||||||
|
maxNumEvents = numEvents
|
||||||
|
}
|
||||||
|
|
||||||
|
self.eventStore[e.EventType] = utils.NewTimedStore(maxAge, maxNumEvents)
|
||||||
}
|
}
|
||||||
self.eventStore[e.EventType].Add(e.Timestamp, e)
|
self.eventStore[e.EventType].Add(e.Timestamp, e)
|
||||||
}
|
}
|
||||||
|
@ -47,7 +47,7 @@ func initializeScenario(t *testing.T) (*events, *Request, *info.Event, *info.Eve
|
|||||||
fakeEvent := makeEvent(createOldTime(t), "/")
|
fakeEvent := makeEvent(createOldTime(t), "/")
|
||||||
fakeEvent2 := makeEvent(time.Now(), "/")
|
fakeEvent2 := makeEvent(time.Now(), "/")
|
||||||
|
|
||||||
return NewEventManager(time.Hour, -1), NewRequest(), fakeEvent, fakeEvent2
|
return NewEventManager(DefaultStoragePolicy()), NewRequest(), fakeEvent, fakeEvent2
|
||||||
}
|
}
|
||||||
|
|
||||||
func checkNumberOfEvents(t *testing.T, numEventsExpected int, numEventsReceived int) {
|
func checkNumberOfEvents(t *testing.T, numEventsExpected int, numEventsReceived int) {
|
||||||
|
@ -136,8 +136,8 @@ func New(memoryStorage *memory.InMemoryStorage, sysfs sysfs.SysFs) (Manager, err
|
|||||||
newManager.versionInfo = *versionInfo
|
newManager.versionInfo = *versionInfo
|
||||||
glog.Infof("Version: %+v", newManager.versionInfo)
|
glog.Infof("Version: %+v", newManager.versionInfo)
|
||||||
|
|
||||||
// TODO(vmarmol): Make configurable.
|
storagePolicy := events.DefaultStoragePolicy()
|
||||||
newManager.eventHandler = events.NewEventManager(24*time.Hour, 100000)
|
newManager.eventHandler = events.NewEventManager(storagePolicy)
|
||||||
|
|
||||||
// Register Docker container factory.
|
// Register Docker container factory.
|
||||||
err = docker.Register(newManager, fsInfo)
|
err = docker.Register(newManager, fsInfo)
|
||||||
|
Loading…
Reference in New Issue
Block a user