Reduce hostd mem usage via event type filter

lib/portlayer/event/collector/vsphere/collector.go

@@ -17,6 +17,7 @@ package vsphere
 import (
 	"context"
 	"fmt"
+	"reflect"
 	"sync"
 
 	"github.com/vmware/vic/lib/portlayer/event/events"
@@ -107,6 +108,27 @@ func (ec *EventCollector) Stop() {
 	}
 }
 
+// eventTypes is used to filter the event collector so we only receive these event types.
+var eventTypes []string
+
+func init() {
+	events := []types.BaseEvent{
+		(*types.VmGuestShutdownEvent)(nil),
+		(*types.VmPoweredOnEvent)(nil),
+		(*types.DrsVmPoweredOnEvent)(nil),
+		(*types.VmPoweredOffEvent)(nil),
+		(*types.VmRemovedEvent)(nil),
+		(*types.VmSuspendedEvent)(nil),
+		(*types.VmMigratedEvent)(nil),
+		(*types.DrsVmMigratedEvent)(nil),
+		(*types.VmRelocatedEvent)(nil),
+	}
+
+	for _, event := range events {
+		eventTypes = append(eventTypes, reflect.TypeOf(event).Elem().Name())
+	}
+}
+
 // Start the event collector
 func (ec *EventCollector) Start() error {
 	// array of managed objects
@@ -125,8 +147,17 @@ func (ec *EventCollector) Start() error {
 	// pageSize is the number of events on the last page of the eventCollector
 	// as new events are added the oldest are removed.  Originally this value
 	// was 1 and we encountered missed events due to them being evicted
-	// before being processed.  A setting of 25 should provide ample buffer.
-	pageSize := int32(25)
+	// before being processed. We bumped to 25 but we still miss events during
+	// storms such as a host HA event.
+	// Setting pageSize to 1000 overwhelmed hostd via the task history and caused
+	// memory exhaustion. Setting pagesize to 200 while filtering for the specific
+	// types we require showed directly comparable memory overhead vs the 25 page
+	// size setting when running full ci. We may still have significantly higher
+	// memory usage in the scenario where we legitimately have events of interest
+	// at a rate of greater than 25 per page.
+	// This should eventually be replaced with a smaller maximum page size, a page
+	// cursor, and maybe a sliding window for the actual page size.
+	pageSize := int32(200)
 	// bool to follow the stream
 	followStream := true
 	// don't exceed the govmomi object limit
@@ -143,7 +174,7 @@ func (ec *EventCollector) Start() error {
 			err := ec.vmwManager.Events(ctx, refs, pageSize, followStream, force, func(_ types.ManagedObjectReference, page []types.BaseEvent) error {
 				evented(ec, page)
 				return nil
-			})
+			}, eventTypes...)
 			// TODO: this will disappear in the ether
 			if err != nil {
 				log.Debugf("Error configuring %s: %s", name, err.Error())
@@ -179,28 +210,7 @@ func evented(ec *EventCollector, page []types.BaseEvent) {
 	// events appear in page with most recent first - need to reverse for sane ordering
 	// we start from the first new event after the last one processed
 	for i := oldIndex - 1; i >= 0; i-- {
-		// what type of event do we have
-		switch page[i].(type) {
-		case *types.VmGuestShutdownEvent,
-			*types.VmPoweredOnEvent,
-			*types.DrsVmPoweredOnEvent,
-			*types.VmPoweredOffEvent,
-			*types.VmRemovedEvent,
-			*types.VmSuspendedEvent,
-			*types.VmMigratedEvent,
-			*types.DrsVmMigratedEvent,
-			*types.VmRelocatedEvent:
-
-			// we have an event we need to process
-			ec.callback(NewVMEvent(page[i]))
-		case *types.VmReconfiguredEvent:
-			// reconfigures happen often, so completely ignore for now
-			continue
-		default:
-			// log the skipped event
-			e := page[i].GetEvent()
-			log.Debugf("vSphere Event %s for eventID(%d) ignored by the event collector", e.FullFormattedMessage, int(e.Key))
-		}
+		ec.callback(NewVMEvent(page[i]))
 
 		ec.lastProcessedID = page[i].GetEvent().Key
 	}

lib/portlayer/event/collector/vsphere/collector_test.go

@@ -70,19 +70,6 @@ func TestEvented(t *testing.T) {
 	page := eventPage(3, LifeCycle)
 	evented(mgr, page)
 	assert.Equal(t, 3, callcount)
-
-	// non-lifecycle events, should be ignored
-	callcount = 0
-	page = eventPage(2, Reconfigure)
-	evented(mgr, page)
-	assert.Equal(t, 0, callcount)
-
-	// mixed events, ignore half
-	callcount = 0
-	page = eventPage(6, Mixed)
-	evented(mgr, page)
-	assert.Equal(t, 3, callcount)
-
 }
 
 func TestName(t *testing.T) {
@@ -97,6 +84,21 @@ func TestStart(t *testing.T) {
 	assert.Error(t, mgr.Start())
 }
 
+func TestEventTypes(t *testing.T) {
+	if len(eventTypes) != 9 {
+		t.Fatalf("eventTypes=%d", len(eventTypes))
+	}
+
+	f := types.TypeFunc()
+
+	for _, name := range eventTypes {
+		_, ok := f(name)
+		if !ok {
+			t.Errorf("unknown event type: %q", name)
+		}
+	}
+}
+
 func newCollector() *EventCollector {
 	return &EventCollector{mos: monitoredCache{mos: make(map[string]types.ManagedObjectReference)}, lastProcessedID: -1}
 }