diff --git a/command/agent/agent.go b/command/agent/agent.go index 08964d3257..2170d0cee9 100644 --- a/command/agent/agent.go +++ b/command/agent/agent.go @@ -50,6 +50,15 @@ type Agent struct { // eventCh is used to receive user events eventCh chan serf.UserEvent + // eventBuf stores the most recent events in a ring buffer + // using eventIndex as the next index to insert into. This + // is guarded by eventLock. When an insert happens, the + // eventNotify group is notified. + eventBuf []*userEventEnc + eventIndex int + eventLock sync.RWMutex + eventNotify consul.NotifyGroup + shutdown bool shutdownCh chan struct{} shutdownLock sync.Mutex @@ -93,6 +102,7 @@ func Create(config *Config, logOutput io.Writer) (*Agent, error) { checkMonitors: make(map[string]*CheckMonitor), checkTTLs: make(map[string]*CheckTTL), eventCh: make(chan serf.UserEvent, 1024), + eventBuf: make([]*userEventEnc, 256), shutdownCh: make(chan struct{}), } diff --git a/command/agent/user_event.go b/command/agent/user_event.go index 501b58d4ed..58580af74c 100644 --- a/command/agent/user_event.go +++ b/command/agent/user_event.go @@ -115,7 +115,8 @@ func (a *Agent) handleEvents() { continue } - // TODO: Process event + // Ingest the event + a.ingestUserEvent(msg) case <-a.shutdownCh: return @@ -195,6 +196,19 @@ func (a *Agent) shouldProcessUserEvent(msg *userEventEnc) bool { return true } +// ingestUserEvent is used to process an event that passes filtering +func (a *Agent) ingestUserEvent(msg *userEventEnc) { + a.eventLock.Lock() + defer func() { + a.eventLock.Unlock() + a.eventNotify.Notify() + }() + + idx := a.eventIndex + a.eventBuf[idx] = msg + a.eventIndex = (idx + 1) % len(a.eventBuf) +} + // Decode is used to decode a MsgPack encoded object func decodeUserEvent(buf []byte, out interface{}) error { return codec.NewDecoder(bytes.NewReader(buf), msgpackHandle).Decode(out)