agent: Adding event ingestion

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{}),
 	}
 

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)