Skip to content

Commit

Permalink
vcsim: add support for PropertyCollector incremental updates
Browse files Browse the repository at this point in the history
Prior to this change, clients could only call WaitForUpdatesEx once,
which would return the current state.  The next invocation with Version
given would fail as unsupported.  Clients can now use WaitForUpdatesEx
to wait for multiple updates.

The basic idea is that methods should now update object fields using
the Registry.UpdateObject method instead of setting fields directly.
UpdateObject uses the same PropertyChange structure that is sent as
part of the WaitForUpdatesEx response.  UpdateObject will also apply
the PropertyChange(s) to the local object, removing the need to set
fields directly.

- govc commands now use Client.Flag.WithCancel to catch SIGINT,
  allowing proper cleanup of collectors

- Add vcsim ListView support to ViewManager

Fixes vmware#922
  • Loading branch information
dougm committed Jul 10, 2018
1 parent 503af4e commit 2511206
Show file tree
Hide file tree
Showing 20 changed files with 716 additions and 152 deletions.
2 changes: 1 addition & 1 deletion Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ install:
go install -v github.com/vmware/govmomi/vcsim

go-test:
go test -race -v $(TEST_OPTS) ./...
go test -timeout 5m -race -v $(TEST_OPTS) ./...

govc-test: install
(cd govc/test && ./vendor/github.com/sstephenson/bats/libexec/bats -t .)
Expand Down
31 changes: 14 additions & 17 deletions govc/events/command.go
Original file line number Diff line number Diff line change
Expand Up @@ -164,24 +164,21 @@ func (cmd *events) Run(ctx context.Context, f *flag.FlagSet) error {
return err
}

if len(objs) > 0 {
// need an event manager
m := event.NewManager(c)

// get the event stream
err = m.Events(ctx, objs, cmd.Max, cmd.Tail, cmd.Force, func(obj types.ManagedObjectReference, ee []types.BaseEvent) error {
var o *types.ManagedObjectReference
if len(objs) > 1 {
o = &obj
}
if len(objs) == 0 {
return nil
}

return cmd.printEvents(ctx, o, ee, m)
}, cmd.Kind...)
m := event.NewManager(c)

if err != nil {
return err
}
}
return cmd.WithCancel(ctx, func(wctx context.Context) error {
return m.Events(wctx, objs, cmd.Max, cmd.Tail, cmd.Force,
func(obj types.ManagedObjectReference, ee []types.BaseEvent) error {
var o *types.ManagedObjectReference
if len(objs) > 1 {
o = &obj
}

return nil
return cmd.printEvents(ctx, o, ee, m)
}, cmd.Kind...)
})
}
28 changes: 28 additions & 0 deletions govc/flags/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,10 @@ import (
"net/http"
"net/url"
"os"
"os/signal"
"path/filepath"
"strings"
"syscall"
"time"

"github.com/vmware/govmomi/session"
Expand Down Expand Up @@ -625,3 +627,29 @@ func (flag *ClientFlag) Environ(extra bool) []string {

return env
}

// WithCancel calls the given function, returning when complete or canceled via SIGINT.
func (flag *ClientFlag) WithCancel(ctx context.Context, f func(context.Context) error) error {
wctx, cancel := context.WithCancel(ctx)
defer cancel()

done := make(chan bool)
var werr error

go func() {
defer close(done)
werr = f(wctx)
}()

sig := make(chan os.Signal)
signal.Notify(sig, syscall.SIGINT)

select {
case <-sig:
cancel()
<-done // Wait for f() to complete
case <-done:
}

return werr
}
60 changes: 35 additions & 25 deletions govc/object/collect.go
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import (
"reflect"
"strings"
"text/tabwriter"
"time"

"github.com/vmware/govmomi/govc/cli"
"github.com/vmware/govmomi/govc/flags"
Expand All @@ -47,6 +48,7 @@ type collect struct {
dump bool
n int
kind kinds
wait time.Duration

filter property.Filter
obj string
Expand All @@ -65,6 +67,7 @@ func (cmd *collect) Register(ctx context.Context, f *flag.FlagSet) {
f.StringVar(&cmd.raw, "R", "", "Raw XML encoded CreateFilter request")
f.IntVar(&cmd.n, "n", 0, "Wait for N property updates")
f.Var(&cmd.kind, "type", "Resource type. If specified, MOID is used for a container view root")
f.DurationVar(&cmd.wait, "wait", 0, "Max wait time for updates")
}

func (cmd *collect) Usage() string {
Expand Down Expand Up @@ -388,41 +391,48 @@ func (cmd *collect) Run(ctx context.Context, f *flag.FlagSet) error {
entered := false
hasFilter := len(cmd.filter) != 0

return property.WaitForUpdates(ctx, p, filter, func(updates []types.ObjectUpdate) bool {
matches := 0
if cmd.wait != 0 {
filter.Options = &types.WaitOptions{
MaxWaitSeconds: types.NewInt32(int32(cmd.wait.Seconds())),
}
}

for _, update := range updates {
if entered && update.Kind == types.ObjectUpdateKindEnter {
// on the first update we only get kind "enter"
// if a new object is added, the next update with have both "enter" and "modify".
continue
}
return cmd.WithCancel(ctx, func(wctx context.Context) error {
return property.WaitForUpdates(wctx, p, filter, func(updates []types.ObjectUpdate) bool {
matches := 0
for _, update := range updates {
if entered && update.Kind == types.ObjectUpdateKindEnter {
// on the first update we only get kind "enter"
// if a new object is added, the next update with have both "enter" and "modify".
continue
}

c := &change{cmd, update}
c := &change{cmd, update}

if hasFilter {
if cmd.match(update) {
matches++
} else {
continue
if hasFilter {
if cmd.match(update) {
matches++
} else {
continue
}
}

_ = cmd.WriteResult(c)
}

_ = cmd.WriteResult(c)
}
entered = true

entered = true
if hasFilter {
if matches > 0 {
return true
}

if hasFilter {
if matches > 0 {
return true
return false
}

return false
}

cmd.n--
cmd.n--

return cmd.n == -1
return cmd.n == -1 && cmd.wait == 0
})
})
}
6 changes: 6 additions & 0 deletions property/collector.go
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,12 @@ func (p *Collector) WaitForUpdates(ctx context.Context, v string) (*types.Update
return res.Returnval, nil
}

func (p *Collector) CancelWaitForUpdates(ctx context.Context) error {
req := &types.CancelWaitForUpdates{This: p.Reference()}
_, err := methods.CancelWaitForUpdates(ctx, p.roundTripper, req)
return err
}

func (p *Collector) RetrieveProperties(ctx context.Context, req types.RetrieveProperties) (*types.RetrievePropertiesResponse, error) {
req.This = p.Reference()
return methods.RetrieveProperties(ctx, p.roundTripper, &req)
Expand Down
30 changes: 23 additions & 7 deletions property/wait.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,12 +19,14 @@ package property
import (
"context"

"github.com/vmware/govmomi/vim25/methods"
"github.com/vmware/govmomi/vim25/types"
)

// WaitFilter provides helpers to construct a types.CreateFilter for use with property.Wait
type WaitFilter struct {
types.CreateFilter
Options *types.WaitOptions
}

// Add a new ObjectSpec and PropertySpec to the WaitFilter
Expand Down Expand Up @@ -75,6 +77,7 @@ func Wait(ctx context.Context, c *Collector, obj types.ManagedObjectReference, p
// creates a new property collector and calls CreateFilter. A new property
// collector is required because filters can only be added, not removed.
//
// If the Context is canceled, a call to CancelWaitForUpdates() is made and its error value is returned.
// The newly created collector is destroyed before this function returns (both
// in case of success or error).
//
Expand All @@ -85,28 +88,41 @@ func WaitForUpdates(ctx context.Context, c *Collector, filter *WaitFilter, f fun
}

// Attempt to destroy the collector using the background context, as the
// specified context may have timed out or have been cancelled.
// specified context may have timed out or have been canceled.
defer p.Destroy(context.Background())

err = p.CreateFilter(ctx, filter.CreateFilter)
if err != nil {
return err
}

for version := ""; ; {
res, err := p.WaitForUpdates(ctx, version)
req := types.WaitForUpdatesEx{
This: p.Reference(),
Options: filter.Options,
}

for {
res, err := methods.WaitForUpdatesEx(ctx, p.roundTripper, &req)
if err != nil {
if ctx.Err() == context.Canceled {
werr := p.CancelWaitForUpdates(context.Background())
return werr
}
return err
}

// Retry if the result came back empty
if res == nil {
set := res.Returnval
if set == nil {
if req.Options != nil && req.Options.MaxWaitSeconds != nil {
return nil // WaitOptions.MaxWaitSeconds exceeded
}
// Retry if the result came back empty
continue
}

version = res.Version
req.Version = set.Version

for _, fs := range res.FilterSet {
for _, fs := range set.FilterSet {
if f(fs.ObjectSet) {
return nil
}
Expand Down
6 changes: 4 additions & 2 deletions scripts/debug-xmlformat.sh
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@ for file in *.req.xml; do
header Request "$file" "${base}.req.headers"
xmlformat < "$file"
file="${base}.res.xml"
header Response "$file" "${base}.res.headers"
xmlformat < "$file"
if [ -e "$file" ] ; then
header Response "$file" "${base}.res.headers"
xmlformat < "$file"
fi
done
7 changes: 4 additions & 3 deletions simulator/event_manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -157,14 +157,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.Next()
m.page = m.page.Prev()
m.page.Value = req.EventToPost
m.formatMessage(req.EventToPost)

for _, c := range m.collectors {
if c.eventMatches(req.EventToPost) {
c.page = c.page.Next()
c.page.Value = event
c.page = c.page.Prev()
c.page.Value = req.EventToPost
Map.Update(c, []types.PropertyChange{{Name: "latestPage", Val: c.GetLatestPage()}})
}
}

Expand Down
Loading

0 comments on commit 2511206

Please sign in to comment.