Repository: vmware/govmomi
Pull request #3517 (2024-08-10 03:38:38)

fix: govc object.collect truncation
In the PR #3331 refactor WaitOptions was changed to pass by value, rather than reference.
The Truncated value needs to be propagated from WaitForUpdatesEx response to the caller.

Fixes #3501

vcsim: implement WaitForUpdatesEx pagination support

Default [WaitOptions.MaxObjectUpdates](https://dp-downloads.broadcom.com/api-content/apis/API_VWSA_001/8.0U3/html/ReferenceGuides/vmodl.query.PropertyCollector.WaitOptions.html) default to 100 as real vCenter does.


Patch:
diff --git a/property/collector.go b/property/collector.go
index 263621a0..9ee78d0d 100644
--- a/property/collector.go
+++ b/property/collector.go
@@ -276,7 +276,7 @@ func (p *Collector) RetrieveOne(ctx context.Context, obj types.ManagedObjectRefe
 // propagation.
 func (p *Collector) WaitForUpdatesEx(
 	ctx context.Context,
-	opts WaitOptions,
+	opts *WaitOptions,
 	onUpdatesFn func([]types.ObjectUpdate) bool) error {
 
 	if !p.mu.TryLock() {
diff --git a/property/wait.go b/property/wait.go
index 07ea3cb5..dae7bf38 100644
--- a/property/wait.go
+++ b/property/wait.go
@@ -123,7 +123,7 @@ func WaitForUpdates(
 		return err
 	}
 
-	return pc.WaitForUpdatesEx(ctx, filter.WaitOptions, onUpdatesFn)
+	return pc.WaitForUpdatesEx(ctx, &filter.WaitOptions, onUpdatesFn)
 }
 
 // WaitForUpdates waits for any of the specified properties of the specified
@@ -166,5 +166,5 @@ func WaitForUpdatesEx(
 
 	}()
 
-	return pc.WaitForUpdatesEx(ctx, filter.WaitOptions, onUpdatesFn)
+	return pc.WaitForUpdatesEx(ctx, &filter.WaitOptions, onUpdatesFn)
 }
diff --git a/simulator/property_collector.go b/simulator/property_collector.go
index 9d37f2e9..3fdb21cf 100644
--- a/simulator/property_collector.go
+++ b/simulator/property_collector.go
@@ -41,6 +41,7 @@ type PropertyCollector struct {
 	mo.PropertyCollector
 
 	nopLocker
+	pending *types.UpdateSet
 	updates []types.ObjectUpdate
 	mu      sync.Mutex
 	cancel  context.CancelFunc
@@ -818,9 +819,47 @@ func (pc *PropertyCollector) apply(ctx *Context, update *types.UpdateSet) types.
 	return nil
 }
 
+// pageUpdateSet limits the given UpdateSet to max number of object updates.
+// nil is returned when not truncated, otherwise the remaining UpdateSet.
+func pageUpdateSet(update *types.UpdateSet, max int) *types.UpdateSet {
+	for i := range update.FilterSet {
+		set := update.FilterSet[i].ObjectSet
+		n := len(set)
+		if n+1 > max {
+			update.Truncated = types.NewBool(true)
+			f := types.PropertyFilterUpdate{
+				Filter:    update.FilterSet[i].Filter,
+				ObjectSet: update.FilterSet[i].ObjectSet[max:],
+			}
+			update.FilterSet[i].ObjectSet = update.FilterSet[i].ObjectSet[:max]
+
+			pending := &types.UpdateSet{
+				Version:   "P",
+				FilterSet: []types.PropertyFilterUpdate{f},
+			}
+
+			if len(update.FilterSet) > i {
+				pending.FilterSet = append(pending.FilterSet, update.FilterSet[i+1:]...)
+				update.FilterSet = update.FilterSet[:i+1]
+			}
+
+			return pending
+		}
+		max -= n
+	}
+	return nil
+}
+
+// WaitOptions.maxObjectUpdates says:
+// > PropertyCollector policy may still limit the total count
+// > to something less than maxObjectUpdates.
+// Seems to be "may" == "will" and the default max is 100.
+const defaultMaxObjectUpdates = 100 // vCenter's default
+
 func (pc *PropertyCollector) WaitForUpdatesEx(ctx *Context, r *types.WaitForUpdatesEx) soap.HasFault {
 	wait, cancel := context.WithCancel(context.Background())
 	oneUpdate := false
+	maxObject := defaultMaxObjectUpdates
 	if r.Options != nil {
 		if max := r.Options.MaxWaitSeconds; max != nil {
 			// A value of 0 causes WaitForUpdatesEx to do one update calculation and return any results.
@@ -829,6 +868,9 @@ func (pc *PropertyCollector) WaitForUpdatesEx(ctx *Context, r *types.WaitForUpda
 				wait, cancel = context.WithTimeout(context.Background(), time.Second*time.Duration(*max))
 			}
 		}
+		if max := r.Options.MaxObjectUpdates; max > 0 && max < defaultMaxObjectUpdates {
+			maxObject = int(max)
+		}
 	}
 	pc.mu.Lock()
 	pc.cancel = cancel
@@ -844,6 +886,12 @@ func (pc *PropertyCollector) WaitForUpdatesEx(ctx *Context, r *types.WaitForUpda
 		Returnval: set,
 	}
 
+	if pc.pending != nil {
+		body.Res.Returnval = pc.pending
+		pc.pending = pageUpdateSet(body.Res.Returnval, maxObject)
+		return body
+	}
+
 	apply := func() bool {
 		if fault := pc.apply(ctx, set); fault != nil {
 			body.Fault_ = Fault("", fault)
@@ -857,6 +905,7 @@ func (pc *PropertyCollector) WaitForUpdatesEx(ctx *Context, r *types.WaitForUpda
 		ctx.Map.AddHandler(pc) // Listen for create, update, delete of managed objects
 		apply()                // Collect current state
 		set.Version = "-"      // Next request with Version set will wait via loop below
+		pc.pending = pageUpdateSet(body.Res.Returnval, maxObject)
 		return body
 	}
 
@@ -932,6 +981,7 @@ func (pc *PropertyCollector) WaitForUpdatesEx(ctx *Context, r *types.WaitForUpda
 				}
 			}
 			if len(set.FilterSet) != 0 {
+				pc.pending = pageUpdateSet(body.Res.Returnval, maxObject)
 				return body
 			}
 			if oneUpdate {


Test patch:
diff --git a/govc/test/vcsim.bats b/govc/test/vcsim.bats
index 3828fda5..7b5bcbe0 100755
--- a/govc/test/vcsim.bats
+++ b/govc/test/vcsim.bats
@@ -70,6 +70,21 @@ EOF
   assert_matches "govmomi simulator"
 }
 
+@test "truncated WaitForUpdatesEx" {
+  total=142 # 100 is the default MaxObjectUpdates
+  vcsim_env -standalone-host 0 -vm $total
+
+  url="https://$(govc env GOVC_URL)"
+
+  model=$(curl -sfk "$url/debug/vars" | jq .vcsim.model)
+  vms=$(jq .machine <<<"$model")
+  assert_equal $total "$vms" # sanity check
+
+  run govc object.collect -type m / name
+  assert_success
+  [ ${#lines[@]} -eq "$total" ]
+}
+
 @test "vcsim host placement" {
   vcsim_start -dc 0
 
diff --git a/property/example_test.go b/property/example_test.go
index 5621e400..9a15a8c0 100644
--- a/property/example_test.go
+++ b/property/example_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/vmware/govmomi/find"
 	"github.com/vmware/govmomi/property"
 	"github.com/vmware/govmomi/simulator"
+	"github.com/vmware/govmomi/view"
 	"github.com/vmware/govmomi/vim25"
 	"github.com/vmware/govmomi/vim25/mo"
 	"github.com/vmware/govmomi/vim25/types"
@@ -153,7 +154,7 @@ func ExampleCollector_WaitForUpdatesEx_addingRemovingPropertyFilters() {
 		go func() {
 			if err := pc.WaitForUpdatesEx(
 				cancelCtx,
-				property.WaitOptions{},
+				&property.WaitOptions{},
 				func(updates []types.ObjectUpdate) bool {
 					return waitForPowerStateChanges(
 						cancelCtx,
@@ -265,7 +266,7 @@ func ExampleCollector_WaitForUpdatesEx_errConcurrentCollector() {
 
 		waitForChanges := func(chanErr chan error) {
 			defer close(chanErr)
-			chanErr <- pc.WaitForUpdatesEx(ctx, waitOptions, onUpdatesFn)
+			chanErr <- pc.WaitForUpdatesEx(ctx, &waitOptions, onUpdatesFn)
 		}
 
 		// Start two goroutines that wait for changes, but only one will begin
@@ -296,7 +297,7 @@ func ExampleCollector_WaitForUpdatesEx_errConcurrentCollector() {
 
 		// The third WaitForUpdatesEx call should be able to successfully obtain
 		// the lock since the other two calls are completed.
-		if err := pc.WaitForUpdatesEx(ctx, waitOptions, onUpdatesFn); err != nil {
+		if err := pc.WaitForUpdatesEx(ctx, &waitOptions, onUpdatesFn); err != nil {
 			return fmt.Errorf(
 				"unexpected error from third call to WaitForUpdatesEx: %s", err)
 		}
@@ -311,3 +312,66 @@ func ExampleCollector_WaitForUpdatesEx_errConcurrentCollector() {
 	// WaitForUpdatesEx call returned ErrConcurrentCollector
 	// WaitForUpdatesEx call succeeded
 }
+
+func ExampleCollector_WaitForUpdatesEx_pagination() {
+	model := simulator.VPX()
+	model.Cluster = 3
+	model.Machine = 42
+
+	simulator.Run(func(ctx context.Context, c *vim25.Client) error {
+		pc := property.DefaultCollector(c)
+		m := view.NewManager(c)
+
+		// Note: both types can be collected with 1 ContainerView and 1 PropertyFilter,
+		// but we are creating 2 PropertyFilter for example purposes.
+		kinds := []string{"HostSystem", "VirtualMachine"}
+		for _, kind := range kinds {
+			v, err := m.CreateContainerView(ctx, c.ServiceContent.RootFolder, []string{kind}, true)
+			if err != nil {
+				return err
+			}
+
+			defer v.Destroy(ctx)
+
+			filter := new(property.WaitFilter).Add(v.Reference(), kind, []string{"name"}, v.TraversalSpec())
+
+			f, err := pc.CreateFilter(ctx, filter.CreateFilter)
+			if err != nil {
+				return err
+			}
+
+			defer f.Destroy(ctx)
+		}
+
+		options := &property.WaitOptions{
+			Options: &types.WaitOptions{MaxObjectUpdates: 50},
+		}
+
+		// Callback is invoked once for each FilterSet:
+		// 1st WaitForUpdatesEx call returns 2 FilterSet, 10 hosts + 40 vms
+		// Next 4 calls are 1 FilterSet of 50, 50 and 28 vms
+		callbacks := 0
+		objects := make(map[string]int)
+
+		err := pc.WaitForUpdatesEx(ctx, options, func(updates []types.ObjectUpdate) bool {
+			for _, update := range updates {
+				objects[update.Obj.Type]++
+			}
+			callbacks++
+			return options.Truncated == false
+		})
+		if err != nil {
+			return err
+		}
+
+		fmt.Printf("%d Callbacks\n", callbacks)
+		for _, kind := range kinds {
+			fmt.Printf("%d %s\n", objects[kind], kind)
+		}
+		return nil
+	}, model)
+	// Output:
+	// 5 Callbacks
+	// 10 HostSystem
+	// 168 VirtualMachine
+}
diff --git a/simulator/property_collector_test.go b/simulator/property_collector_test.go
index ce9bc264..edf14bc1 100644
--- a/simulator/property_collector_test.go
+++ b/simulator/property_collector_test.go
@@ -18,6 +18,7 @@ package simulator
 
 import (
 	"context"
+	"fmt"
 	"log"
 	"reflect"
 	"sync"
@@ -1766,3 +1767,70 @@ func TestUcFirst(t *testing.T) {
 		})
 	}
 }
+
+func TestPageUpdateSet(t *testing.T) {
+	max := defaultMaxObjectUpdates
+
+	sum := func(u []int) int {
+		total := 0
+		for i := range u {
+			total += u[i]
+		}
+		return total
+	}
+
+	sumUpdateSet := func(u *types.UpdateSet) int {
+		total := 0
+		for _, f := range u.FilterSet {
+			total += len(f.ObjectSet)
+		}
+		return total
+	}
+
+	tests := []struct {
+		filters int
+		objects []int
+	}{
+		{0, nil},
+		{1, []int{10}},
+		{1, []int{104}},
+		{3, []int{10, 0, 25}},
+		{1, []int{234, 21, 4}},
+		{2, []int{95, 32}},
+	}
+
+	for _, test := range tests {
+		name := fmt.Sprintf("%d filter %d objs", test.filters, sum(test.objects))
+		t.Run(name, func(t *testing.T) {
+			update := &types.UpdateSet{}
+
+			for i := 0; i < test.filters; i++ {
+				f := types.PropertyFilterUpdate{}
+				for j := 0; j < 156; j++ {
+					f.ObjectSet = append(f.ObjectSet, types.ObjectUpdate{})
+				}
+				update.FilterSet = append(update.FilterSet, f)
+			}
+
+			total := sumUpdateSet(update)
+			sum := 0
+
+			for {
+				pending := pageUpdateSet(update, max)
+				n := sumUpdateSet(update)
+				if n > max {
+					t.Fatalf("%d > %d", n, max)
+				}
+				sum += n
+				if pending == nil {
+					break
+				}
+				update = pending
+			}
+
+			if sum != total {
+				t.Fatalf("%d != %d", sum, total)
+			}
+		})
+	}
+}