Skip to content

Commit

Permalink
feat: Add task manager and collector
Browse files Browse the repository at this point in the history
This change introduces a task manager
(https://vdc-download.vmware.com/vmwb-repository/dcr-public/8946c1b6-2861-4c12-a45f-f14ae0d3b1b9/a5b8094c-c222-4307-9399-3b606a04af55/vim.TaskManager.html)
with support for creating a task history collector.

History collector is moved from object into its own package and changes
were made to the event package accordingly (collector logic is shared
between task and event manager).

BREAKING:

`event.Manager` does not embed `object.Common` anymore. Only the methods
`Client()` and `Reference()` are implemented.
`event.NewHistoryCollector()` is now unexported (to
`newHistoryCollector()`) as it was merely a helper and to comply with
the task manager implementation.

Closes: vmware#2497
Signed-off-by: Michael Gasch <[email protected]>
  • Loading branch information
Michael Gasch committed Sep 2, 2021
1 parent 6258af3 commit 397c8aa
Show file tree
Hide file tree
Showing 8 changed files with 341 additions and 93 deletions.
8 changes: 4 additions & 4 deletions event/history_collector.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,20 +19,20 @@ package event
import (
"context"

"github.com/vmware/govmomi/object"
"github.com/vmware/govmomi/history"
"github.com/vmware/govmomi/vim25"
"github.com/vmware/govmomi/vim25/methods"
"github.com/vmware/govmomi/vim25/mo"
"github.com/vmware/govmomi/vim25/types"
)

type HistoryCollector struct {
*object.HistoryCollector
*history.Collector
}

func NewHistoryCollector(c *vim25.Client, ref types.ManagedObjectReference) *HistoryCollector {
func newHistoryCollector(c *vim25.Client, ref types.ManagedObjectReference) *HistoryCollector {
return &HistoryCollector{
HistoryCollector: object.NewHistoryCollector(c, ref),
Collector: history.NewCollector(c, ref),
}
}

Expand Down
39 changes: 22 additions & 17 deletions event/manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@ import (
"reflect"
"sync"

"github.com/vmware/govmomi/object"
"github.com/vmware/govmomi/property"
"github.com/vmware/govmomi/vim25"
"github.com/vmware/govmomi/vim25/methods"
Expand All @@ -31,7 +30,8 @@ import (
)

type Manager struct {
object.Common
r types.ManagedObjectReference
c *vim25.Client

eventCategory map[string]string
eventCategoryMu *sync.Mutex
Expand All @@ -40,8 +40,8 @@ type Manager struct {

func NewManager(c *vim25.Client) *Manager {
m := Manager{
Common: object.NewCommon(c, *c.ServiceContent.EventManager),

r: c.ServiceContent.EventManager.Reference(),
c: c,
eventCategory: make(map[string]string),
eventCategoryMu: new(sync.Mutex),
maxObjects: 10,
Expand All @@ -50,28 +50,32 @@ func NewManager(c *vim25.Client) *Manager {
return &m
}

func (m Manager) Client() *vim25.Client {
return m.c
}

func (m Manager) CreateCollectorForEvents(ctx context.Context, filter types.EventFilterSpec) (*HistoryCollector, error) {
req := types.CreateCollectorForEvents{
This: m.Common.Reference(),
This: m.r,
Filter: filter,
}

res, err := methods.CreateCollectorForEvents(ctx, m.Client(), &req)
res, err := methods.CreateCollectorForEvents(ctx, m.c, &req)
if err != nil {
return nil, err
}

return NewHistoryCollector(m.Client(), res.Returnval), nil
return newHistoryCollector(m.c, res.Returnval), nil
}

func (m Manager) LogUserEvent(ctx context.Context, entity types.ManagedObjectReference, msg string) error {
req := types.LogUserEvent{
This: m.Common.Reference(),
This: m.r,
Entity: entity,
Msg: msg,
}

_, err := methods.LogUserEvent(ctx, m.Client(), &req)
_, err := methods.LogUserEvent(ctx, m.c, &req)
if err != nil {
return err
}
Expand All @@ -81,15 +85,15 @@ func (m Manager) LogUserEvent(ctx context.Context, entity types.ManagedObjectRef

func (m Manager) PostEvent(ctx context.Context, eventToPost types.BaseEvent, taskInfo ...types.TaskInfo) error {
req := types.PostEvent{
This: m.Common.Reference(),
This: m.r,
EventToPost: eventToPost,
}

if len(taskInfo) == 1 {
req.TaskInfo = &taskInfo[0]
}

_, err := methods.PostEvent(ctx, m.Client(), &req)
_, err := methods.PostEvent(ctx, m.c, &req)
if err != nil {
return err
}
Expand All @@ -99,11 +103,11 @@ func (m Manager) PostEvent(ctx context.Context, eventToPost types.BaseEvent, tas

func (m Manager) QueryEvents(ctx context.Context, filter types.EventFilterSpec) ([]types.BaseEvent, error) {
req := types.QueryEvents{
This: m.Common.Reference(),
This: m.r,
Filter: filter,
}

res, err := methods.QueryEvents(ctx, m.Client(), &req)
res, err := methods.QueryEvents(ctx, m.c, &req)
if err != nil {
return nil, err
}
Expand All @@ -113,11 +117,11 @@ func (m Manager) QueryEvents(ctx context.Context, filter types.EventFilterSpec)

func (m Manager) RetrieveArgumentDescription(ctx context.Context, eventTypeID string) ([]types.EventArgDesc, error) {
req := types.RetrieveArgumentDescription{
This: m.Common.Reference(),
This: m.r,
EventTypeId: eventTypeID,
}

res, err := methods.RetrieveArgumentDescription(ctx, m.Client(), &req)
res, err := methods.RetrieveArgumentDescription(ctx, m.c, &req)
if err != nil {
return nil, err
}
Expand All @@ -136,7 +140,7 @@ func (m Manager) eventCategoryMap(ctx context.Context) (map[string]string, error
var o mo.EventManager

ps := []string{"description.eventInfo"}
err := property.DefaultCollector(m.Client()).RetrieveOne(ctx, m.Common.Reference(), ps, &o)
err := property.DefaultCollector(m.c).RetrieveOne(ctx, m.r, ps, &o)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -171,7 +175,8 @@ func (m Manager) EventCategory(ctx context.Context, event types.BaseEvent) (stri
return eventCategory[class], nil
}

// Get the events from the specified object(s) and optionanlly tail the event stream
// Events gets the events from the specified object(s) and optionanlly tail the
// event stream
func (m Manager) Events(ctx context.Context, objects []types.ManagedObjectReference, pageSize int32, tail bool, force bool, f func(types.ManagedObjectReference, []types.BaseEvent) error, kind ...string) error {
// TODO: deprecated this method and add one that uses a single config struct, so we can extend further without breaking the method signature.
if len(objects) >= m.maxObjects && !force {
Expand Down
94 changes: 94 additions & 0 deletions examples/tasks/main.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
/*
Copyright (c) 2020 VMware, Inc. All Rights Reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package main

import (
"context"
"flag"
"time"

"github.com/kr/pretty"

"github.com/vmware/govmomi/task"

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

func main() {
// example use against vCenter with optional time filters:
// go run main.go -url $GOVMOMI_URL -insecure $GOVMOMI_INSECURE -b 8h -f
begin := flag.Duration("b", 10*time.Minute, "Begin time (filtered by started time)") // default BeginTime is 10min ago
end := flag.Duration("e", 0, "End time (filtered by started time)")
follow := flag.Bool("f", false, "Follow task stream")

examples.Run(func(ctx context.Context, c *vim25.Client) error {
m := task.NewManager(c)

ref := c.ServiceContent.RootFolder

now, err := methods.GetCurrentTime(ctx, c) // vCenter server time (UTC)
if err != nil {
return err
}

filter := types.TaskFilterSpec{
Entity: &types.TaskFilterSpecByEntity{
Entity: ref,
Recursion: types.TaskFilterSpecRecursionOptionAll,
},
Time: &types.TaskFilterSpecByTime{
TimeType: types.TaskFilterSpecTimeOptionStartedTime,
BeginTime: types.NewTime(now.Add(*begin * -1)),
},
}

if *end != 0 {
filter.Time.EndTime = types.NewTime(now.Add(*end * -1))
}

collector, err := m.CreateCollectorForTasks(ctx, filter)
if err != nil {
return err
}

defer collector.Destroy(ctx)

for {
tasks, err := collector.ReadNextTasks(ctx, 10)
if err != nil {
return err
}

if len(tasks) == 0 {
if *follow {
time.Sleep(time.Second)
continue
}
break
}

for i := range tasks {
pretty.Println(tasks[i])
}
}

return nil
})
}
4 changes: 4 additions & 0 deletions govc/test/vcsim.bats
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,10 @@ EOF

# compile + run examples against vcsim
for main in ../../examples/*/main.go ; do
# TODO: #2567
if [[ $main =~ "task" ]]; then
continue
fi
run go run "$main" -insecure -url "$GOVC_URL"
assert_success
done
Expand Down
91 changes: 91 additions & 0 deletions history/collector.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
/*
Copyright (c) 2015 VMware, Inc. All Rights Reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package history

import (
"context"

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

type Collector struct {
r types.ManagedObjectReference
c *vim25.Client
}

func NewCollector(c *vim25.Client, ref types.ManagedObjectReference) *Collector {
return &Collector{
r: ref,
c: c,
}
}

// Reference returns the managed object reference of this collector
func (c Collector) Reference() types.ManagedObjectReference {
return c.r
}

// Client returns the vim25 client used by this collector
func (c Collector) Client() *vim25.Client {
return c.c
}

// Properties wraps property.DefaultCollector().RetrieveOne() and returns
// properties for the specified managed object reference
func (c Collector) Properties(ctx context.Context, r types.ManagedObjectReference, ps []string, dst interface{}) error {
return property.DefaultCollector(c.c).RetrieveOne(ctx, r, ps, dst)
}

func (c Collector) Destroy(ctx context.Context) error {
req := types.DestroyCollector{
This: c.r,
}

_, err := methods.DestroyCollector(ctx, c.c, &req)
return err
}

func (c Collector) Reset(ctx context.Context) error {
req := types.ResetCollector{
This: c.r,
}

_, err := methods.ResetCollector(ctx, c.c, &req)
return err
}

func (c Collector) Rewind(ctx context.Context) error {
req := types.RewindCollector{
This: c.r,
}

_, err := methods.RewindCollector(ctx, c.c, &req)
return err
}

func (c Collector) SetPageSize(ctx context.Context, maxCount int32) error {
req := types.SetCollectorPageSize{
This: c.r,
MaxCount: maxCount,
}

_, err := methods.SetCollectorPageSize(ctx, c.c, &req)
return err
}
Loading

0 comments on commit 397c8aa

Please sign in to comment.