Repository: vmware/govmomi (go)

Issue:
Incorrect behavior of event history collector in vcsim simulator
Compared to a regular vCenter installation (tested with vCenter Server v6.7) the event history collector and some of its APIs, e.g. `ReadNextEvent` in particular, show some different and likely undesired behavior:

- The embedded `EventHistoryCollector` in `vcsim` and its ring buffer purely relies on (and serves from) `latestPage`
- To comply with `latestPage` API behavior, events are not sorted (thus `govc` leverages a sort function to print the events in order) - the internal ring buffer reads backwards via `page.Prev()`
- `ReadNextPage` in `vcsim` also relies on `latestPage` which is not the case in a real vCenter
- `PostEvent` uses `Map.Update(c, []types.PropertyChange{{Name: "latestPage", Val: c.GetLatestPage()}})` which means custom `EventHistoryCollector` created without this option/with specific `EventFilter`s won't be triggered

For example, given the following code:

```go
func newHistoryCollector(ctx context.Context, vcClient *vim25.Client, begin *time.Time) (*event.HistoryCollector, error) {
	mgr := event.NewManager(vcClient)
	root := vcClient.ServiceContent.RootFolder
	// configure the event stream filter (begin of stream)
	filter := types.EventFilterSpec{
		// EventTypeId: []string{...}, // only stream specific types, e.g. VmEvent
		Entity: &types.EventFilterSpecByEntity{
			Entity:    root,
			Recursion: types.EventFilterSpecRecursionOptionAll,
		},
		Time: &types.EventFilterSpecByTime{
			BeginTime: types.NewTime(*begin),
		},
	}
	return mgr.CreateCollectorForEvents(ctx, filter)
}
```

`vcsim` will only serve the first call to `ReadNextEvents` but no later events, such as `VmPoweredOffEvent` triggered via `govc` against the simulator.

Since more users are relying on vCenter events/programmatically using its APIs, it would be nice to see `vcsim` events API behave like a real vCenter, especially for integration testing. `latestPage` might not be the standard use case for retrieving events from vCenter, especially when a user wants control over the polling interval or start of the event stream (via filter/`ReadNext`). `latestPage` is also used in the event manager in govmomi and hardcoded via `props := []string{"latestPage"}` in `run()`.

Pull request description:
vcsim: fix EventHistoryCollector fixes
Fix Next/Previous methods:

The original implementation only applied to the LatestPage ring buffer.
Switch EventManager's history to use a linked list that goes beyond the
LatestPage size.

Fix SetCollectorPageSize: had caused duplicate events in LatestPage

Fixes #2134

Patch:
diff --git a/simulator/event_manager.go b/simulator/event_manager.go
index b7d5d6b2..49eda136 100644
--- a/simulator/event_manager.go
+++ b/simulator/event_manager.go
@@ -18,6 +18,7 @@ package simulator
 
 import (
 	"bytes"
+	"container/list"
 	"container/ring"
 	"log"
 	"reflect"
@@ -40,7 +41,7 @@ type EventManager struct {
 	mo.EventManager
 
 	root       types.ManagedObjectReference
-	page       *ring.Ring
+	history    *list.List
 	key        int32
 	collectors map[types.ManagedObjectReference]*EventHistoryCollector
 	templates  map[string]*template.Template
@@ -54,7 +55,7 @@ func (m *EventManager) init(r *Registry) {
 		m.MaxCollector = 1000
 	}
 	m.root = r.content().RootFolder
-	m.page = ring.New(maxPageSize)
+	m.history = list.New()
 	m.collectors = make(map[types.ManagedObjectReference]*EventHistoryCollector)
 	m.templates = make(map[string]*template.Template)
 }
@@ -155,10 +156,15 @@ func (m *EventManager) PostEvent(ctx *Context, req *types.PostEvent) soap.HasFau
 	event.CreatedTime = time.Now()
 	event.UserName = ctx.Session.UserName
 
-	m.page = m.page.Prev()
-	m.page.Value = req.EventToPost
 	m.formatMessage(req.EventToPost)
 
+	if m.history.Len() > maxPageSize*5 {
+		// Prune history
+		m.history.Remove(m.history.Front())
+	}
+
+	m.history.PushBack(req.EventToPost)
+
 	for _, c := range m.collectors {
 		ctx.WithLock(c, func() {
 			if c.eventMatches(req.EventToPost) {
@@ -179,7 +185,7 @@ type EventHistoryCollector struct {
 
 	m    *EventManager
 	page *ring.Ring
-	pos  int
+	pos  *list.Element
 }
 
 // doEntityEventArgument calls f for each entity argument in the event.
@@ -327,31 +333,11 @@ func (c *EventHistoryCollector) eventMatches(event types.BaseEvent) bool {
 
 // filePage copies the manager's latest events into the collector's page with Filter applied.
 func (c *EventHistoryCollector) fillPage(size int) {
-	c.pos = 0
-	l := c.page.Len()
-	delta := size - l
-
-	if delta < 0 {
-		// Shrink ring size
-		c.page = c.page.Unlink(-delta)
-		return
-	}
-
 	matches := 0
-	mpage := c.m.page
 	page := c.page
 
-	if delta != 0 {
-		// Grow ring size
-		c.page = c.page.Link(ring.New(delta))
-	}
-
-	for i := 0; i < maxPageSize; i++ {
-		event, ok := mpage.Value.(types.BaseEvent)
-		mpage = mpage.Prev()
-		if !ok {
-			continue
-		}
+	for e := c.m.history.Back(); e != nil; e = e.Prev() {
+		event := e.Value.(types.BaseEvent)
 
 		if c.eventMatches(event) {
 			page.Value = event
@@ -385,6 +371,7 @@ func (c *EventHistoryCollector) SetCollectorPageSize(ctx *Context, req *types.Se
 	}
 
 	ctx.WithLock(c.m, func() {
+		c.page = ring.New(size)
 		c.fillPage(size)
 	})
 
@@ -393,7 +380,7 @@ func (c *EventHistoryCollector) SetCollectorPageSize(ctx *Context, req *types.Se
 }
 
 func (c *EventHistoryCollector) ResetCollector(ctx *Context, req *types.ResetCollector) soap.HasFault {
-	c.pos = len(c.GetLatestPage())
+	c.pos = c.m.history.Back()
 
 	return &methods.ResetCollectorBody{
 		Res: new(types.ResetCollectorResponse),
@@ -401,12 +388,30 @@ func (c *EventHistoryCollector) ResetCollector(ctx *Context, req *types.ResetCol
 }
 
 func (c *EventHistoryCollector) RewindCollector(ctx *Context, req *types.RewindCollector) soap.HasFault {
-	c.pos = 0
+	c.pos = c.m.history.Front()
+
 	return &methods.RewindCollectorBody{
 		Res: new(types.RewindCollectorResponse),
 	}
 }
 
+// readEvents returns the next max Events from the EventManager's history
+func (c *EventHistoryCollector) readEvents(ctx *Context, max int32, next func() *list.Element) []types.BaseEvent {
+	var events []types.BaseEvent
+
+	for i := 0; i < int(max); i++ {
+		e := next()
+		if e == nil {
+			break
+		}
+
+		events = append(events, e.Value.(types.BaseEvent))
+		c.pos = e
+	}
+
+	return events
+}
+
 func (c *EventHistoryCollector) ReadNextEvents(ctx *Context, req *types.ReadNextEvents) soap.HasFault {
 	body := &methods.ReadNextEventsBody{}
 	if req.MaxCount <= 0 {
@@ -415,21 +420,14 @@ func (c *EventHistoryCollector) ReadNextEvents(ctx *Context, req *types.ReadNext
 	}
 	body.Res = new(types.ReadNextEventsResponse)
 
-	events := c.GetLatestPage()
-	nevents := len(events)
-	if c.pos == nevents {
-		return body // already read to EOF
-	}
-
-	start := c.pos
-	end := start + int(req.MaxCount)
-	c.pos += int(req.MaxCount)
-	if end > nevents {
-		end = nevents
-		c.pos = nevents
+	next := func() *list.Element {
+		if c.pos != nil {
+			return c.pos.Next()
+		}
+		return c.m.history.Front()
 	}
 
-	body.Res.Returnval = events[start:end]
+	body.Res.Returnval = c.readEvents(ctx, req.MaxCount, next)
 
 	return body
 }
@@ -442,20 +440,14 @@ func (c *EventHistoryCollector) ReadPreviousEvents(ctx *Context, req *types.Read
 	}
 	body.Res = new(types.ReadPreviousEventsResponse)
 
-	events := c.GetLatestPage()
-	if c.pos == 0 {
-		return body // already read to EOF
-	}
-
-	start := c.pos - int(req.MaxCount)
-	end := c.pos
-	c.pos -= int(req.MaxCount)
-	if start < 0 {
-		start = 0
-		c.pos = 0
+	next := func() *list.Element {
+		if c.pos != nil {
+			return c.pos.Prev()
+		}
+		return c.m.history.Back()
 	}
 
-	body.Res.Returnval = events[start:end]
+	body.Res.Returnval = c.readEvents(ctx, req.MaxCount, next)
 
 	return body
 }


Test patch:
diff --git a/simulator/event_manager_test.go b/simulator/event_manager_test.go
index 3699e40d..b95d38c5 100644
--- a/simulator/event_manager_test.go
+++ b/simulator/event_manager_test.go
@@ -157,7 +157,8 @@ func TestEventManagerRead(t *testing.T) {
 	}()
 
 	spec := types.EventFilterSpec{}
-	c, err := event.NewManager(vc.Client).CreateCollectorForEvents(ctx, spec)
+	em := event.NewManager(vc.Client)
+	c, err := em.CreateCollectorForEvents(ctx, spec)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -224,6 +225,19 @@ func TestEventManagerRead(t *testing.T) {
 		t.Errorf("expected 0 events, got %d", len(events))
 	}
 
+	err = em.PostEvent(ctx, &types.GeneralEvent{Message: "vcsim"})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	events, err = c.ReadNextEvents(ctx, int32(nevents))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(events) != 1 {
+		t.Errorf("expected 1 events, got %d", len(events))
+	}
+
 	count := 0
 	for {
 		events, err = c.ReadPreviousEvents(ctx, 3)
@@ -235,7 +249,7 @@ func TestEventManagerRead(t *testing.T) {
 		}
 		count += len(events)
 	}
-	if nevents != count {
+	if count < nevents {
 		t.Errorf("expected %d events, got %d", nevents, count)
 	}
 }
diff --git a/simulator/race_test.go b/simulator/race_test.go
index 1596cd58..0daabca1 100644
--- a/simulator/race_test.go
+++ b/simulator/race_test.go
@@ -80,6 +80,40 @@ func TestRace(t *testing.T) {
 		}
 	}()
 
+	collectors.Add(1)
+	go func() {
+		defer collectors.Done()
+
+		ec, werr := em.CreateCollectorForEvents(ctx, types.EventFilterSpec{})
+		if werr != nil {
+			t.Error(werr)
+		}
+
+		n := 0
+		for {
+			events, werr := ec.ReadNextEvents(ctx, 10)
+			if werr != nil {
+				t.Error(werr)
+			}
+
+			n += len(events)
+			if len(events) != 0 {
+				continue
+			}
+
+			select {
+			case <-wctx.Done():
+				logf := t.Logf
+				if n == 0 {
+					logf = t.Errorf
+				}
+				logf("ReadNextEvents=%d", n)
+				return
+			case <-time.After(time.Millisecond * 100):
+			}
+		}
+	}()
+
 	ntasks := -1
 	tv, err := view.NewManager(c.Client).CreateTaskView(ctx, content.TaskManager)
 	if err != nil {