Switch to latest connector-SDK release

The connector now leverages the latest connector-sdk features such as
allowing multiple event topic subscriptions (delimited with ",") and
printing invokation responses via the `controller.Subscribe` interface
(implemented by `events.NewEventReceiver()`.

The controller uses a 10 second `RebuildInterval` (sync function
subscriptions) and 15 second `UpstreamTimeout` for invoking functions.
The controller uses asynchronous invocation mode to not block on
slow/long-running functions.

Function comments are line-wrapped.
Updates to Gopkg.toml to use the latest releases for imported packages:

- connector-sdk 0.5.3
- openfaas-cloud 0.11.10
- govmomi 0.21.0

Build successfully tested against VMware vCenter 6.7U3 and OpenFaaS faas-netes commit b14f727.
README updated covering the recent changes.

Signed-off-by: Michael Gasch <mgasch@vmware.com>

.gitignore

@@ -1 +1,2 @@
 vcenter-connector
+.vscode

Gopkg.toml

@@ -4,12 +4,12 @@
 
 [[constraint]]
   name = "github.com/openfaas-incubator/connector-sdk"
-  version = "0.2.0"
+  version = "0.5.3"
 
 [[constraint]]
   name = "github.com/vmware/govmomi"
-  version = "0.19.0"
+  version = "0.21.0"
 
 [[constraint]]
   name = "github.com/openfaas/openfaas-cloud"
-  version = "0.9.4"
+  version = "0.11.10"

README.md

@@ -10,9 +10,7 @@ With this project your functions can subscribe to events generated by the change
 
 ## Status
 
-This project uses the [OpenFaaS Connector SDK](https://github.com/openfaas-incubator/connector-sdk).
-
-vCenter credentials are stored in Kubernetes secrets.
+This project uses the [OpenFaaS Connector SDK](https://github.com/openfaas-incubator/connector-sdk). vCenter credentials are stored using Kubernetes [secrets](https://kubernetes.io/docs/concepts/configuration/secret/).
 
 ## Supported Events
 
@@ -26,6 +24,16 @@ The following event types (incl. their subtypes) are supported and can be used t
 
 For further details and naming see the [vSphere Web Services API](https://code.vmware.com/apis/358/vsphere#/doc/vim.event.Event.html) documentation.
 
+A function can be subscribed to multiple events using a comma-delimited syntax in its stack configuration:
+
+```yaml
+[...]
+annotations:
+    topic: "drs.vm.powered.on,vm.powered.off"
+```
+
+> **Note:** Wildcards for event subscriptions, e.g. "`vm.powered.*`", are **not** supported.
+
 ## Credentials
 
 ### Credentials within Kubernetes
@@ -73,7 +81,7 @@ The default path is `/var/openfaas/secrets/` which can be overridden by setting
 
 ### VEBA
 
-VMware have released an appliance which packages [OpenFaaS](https://github.com/openfaas/faas) and the OpenFaaS vcenter-connector (this repository) and called it "VEBA". The appliance uses [Photon OS](https://vmware.github.io/photon/).
+VMware have released an appliance which packages [OpenFaaS](https://github.com/openfaas/faas) and the OpenFaaS vcenter-connector (this repository) and called it the "vCenter Event Broker Appliance". The appliance uses [Photon OS](https://vmware.github.io/photon/).
 
 * [Audit VM configuration changes using the vCenter Event Broker by Dennis Zimmer](https://itnext.io/audit-vm-configuration-changes-using-the-vcenter-event-broker-45f9f5ba21f2)
 

main.go

@@ -11,11 +11,15 @@ import (
 
 	"github.com/openfaas-incubator/vcenter-connector/pkg/events"
 
-	"github.com/openfaas-incubator/connector-sdk/types"
+	ofsdk "github.com/openfaas-incubator/connector-sdk/types"
 	"github.com/openfaas/faas-provider/auth"
 	"github.com/openfaas/openfaas-cloud/sdk"
 )
 
+const (
+	topicDelimiter = ","
+)
+
 func main() {
 	var gatewayURL string
 	var vcenterURL string
@@ -26,7 +30,7 @@ func main() {
 
 	var insecure bool
 
-	// TODO: add secrets management, verbosity level
+	// TODO: add option to configure log verbosity
 	flag.StringVar(&gatewayURL, "gateway", "http://127.0.0.1:8080", "URL for OpenFaaS gateway")
 	flag.StringVar(&vcenterURL, "vcenter", "http://127.0.0.1:8989/sdk", "URL for vCenter")
 	flag.StringVar(&vcUser, "vc-user", "", "User to connect to vCenter")
@@ -83,15 +87,18 @@ func main() {
 	}
 
 	// OpenFaaS connector SDK controller configuration
-	ofconfig := types.ControllerConfig{
-		GatewayURL:      gatewayURL,
-		PrintResponse:   false,
-		RebuildInterval: time.Second * 10,
-		UpstreamTimeout: time.Second * 15,
+	ofconfig := ofsdk.ControllerConfig{
+		GatewayURL:               gatewayURL,
+		TopicAnnotationDelimiter: topicDelimiter,
+		RebuildInterval:          time.Second * 10,
+		UpstreamTimeout:          time.Second * 15,
+		AsyncFunctionInvocation:  true, // don't block when invoking long-running/heavy functions, higher throughput
+		PrintSync:                true,
 	}
 
-	// get OpenFaaS connector controller
-	ofcontroller := types.NewController(credentials, &ofconfig)
+	ofcontroller := ofsdk.NewController(credentials, &ofconfig)
+	responseHandler := events.NewEventReceiver()
+	ofcontroller.Subscribe(responseHandler)
 	ofcontroller.BeginMapBuilder()
 
 	ctx, cancel := context.WithCancel(context.Background())

pkg/events/events.go

@@ -10,7 +10,7 @@ import (
 	"time"
 	"unicode"
 
-	"github.com/openfaas-incubator/connector-sdk/types"
+	ofsdk "github.com/openfaas-incubator/connector-sdk/types"
 	"github.com/pkg/errors"
 	"github.com/vmware/govmomi"
 	"github.com/vmware/govmomi/event"
@@ -19,10 +19,13 @@ import (
 	vtypes "github.com/vmware/govmomi/vim25/types"
 )
 
-// OutboundEvent is the JSON object sent to subscribed functions
-// If the ManagedObjectReference for an event cannot be retrieved, it will be nil and thus not marshaled into the JSON OutboundEvent
-// It's the receivers responsibility to check whether managedObjectReference key is present in the JSON message payload
-// ObjectName is the name of the object as it appears in vCenter - uniqueness is only guaranteed at the folder level, if applicable, where this object resides
+// OutboundEvent is the JSON object sent to subscribed functions If the
+// ManagedObjectReference for an event cannot be retrieved, it will be nil and
+// thus not marshaled into the JSON OutboundEvent It's the receivers
+// responsibility to check whether managedObjectReference key is present in the
+// JSON message payload ObjectName is the name of the object as it appears in
+// vCenter - uniqueness is only guaranteed at the folder level, if applicable,
+// where this object resides
 type OutboundEvent struct {
 	Topic    string `json:"topic,omitempty"`
 	Category string `json:"category,omitempty"`
@@ -34,6 +37,25 @@ type OutboundEvent struct {
 	ManagedObjectReference *vtypes.ManagedObjectReference `json:"managedObjectReference,omitempty"`
 }
 
+// EventReceiver implements ResponseSubscriber to validate function invocation
+// and return status
+type EventReceiver struct{}
+
+// Response prints status information for each function invokation
+func (e *EventReceiver) Response(res ofsdk.InvokerResponse) {
+	if res.Error != nil {
+		log.Printf("function %s for topic %s returned status %d with error: %v", res.Function, res.Topic, res.Status, res.Error)
+	}
+	log.Printf("successfully invoked function %s for topic %s", res.Function, res.Topic)
+}
+
+// NewEventReceiver returns an EventReceiver which implements the
+// ResponseSubscriber interface to print status information for each function
+// invokation
+func NewEventReceiver() *EventReceiver {
+	return &EventReceiver{}
+}
+
 // NewVCenterClient returns a govmomi.Client to connect to vCenter
 func NewVCenterClient(ctx context.Context, user string, pass string, vcenterURL string, insecure bool) (*govmomi.Client, error) {
 	u, err := soap.ParseURL(vcenterURL)
@@ -46,7 +68,7 @@ func NewVCenterClient(ctx context.Context, user string, pass string, vcenterURL
 }
 
 // Stream is the main logic, blocking to receive and handle events from vCenter
-func Stream(ctx context.Context, c *vim25.Client, controller *types.Controller) error {
+func Stream(ctx context.Context, c *vim25.Client, controller ofsdk.Controller) error {
 	// create event manager to consume events from vCenter
 	m := event.NewManager(c)
 
@@ -66,8 +88,9 @@ func Stream(ctx context.Context, c *vim25.Client, controller *types.Controller)
 	return nil
 }
 
-// makeRecv returns a event handler function called by the event manager on each event
-func makeRecv(controller *types.Controller, m *event.Manager, source string) func(managedObjectReference vtypes.ManagedObjectReference, baseEvent []vtypes.BaseEvent) error {
+// makeRecv returns a event handler function called by the event manager on each
+// event
+func makeRecv(controller ofsdk.Controller, m *event.Manager, source string) func(managedObjectReference vtypes.ManagedObjectReference, baseEvent []vtypes.BaseEvent) error {
 	return func(managedObjectReference vtypes.ManagedObjectReference, baseEvent []vtypes.BaseEvent) error {
 		log.Printf("Object %v", managedObjectReference)
 
@@ -174,14 +197,16 @@ func getObjectNameAndMoref(event vtypes.BaseEvent) (string, *vtypes.ManagedObjec
 	return objName, ref
 }
 
-// convertToTopic converts an event type to an OpenFaaS subscriber topic, e.g. "VmPoweredOnEvent" to "vm.powered.on"
+// convertToTopic converts an event type to an OpenFaaS subscriber topic, e.g.
+// "VmPoweredOnEvent" to "vm.powered.on"
 func convertToTopic(eventType string) string {
 	eventType = strings.Replace(eventType, "Event", "", -1)
 	return camelCaseToLowerSeparated(eventType, ".")
 }
 
 // From https://github.com/vmware/dispatch/blob/master/pkg/utils/str_convert.go
-// camelCaseToLowerSeparated converts a camel cased string to a multi-word string delimited by the specified separator
+// camelCaseToLowerSeparated converts a camel cased string to a multi-word
+// string delimited by the specified separator
 func camelCaseToLowerSeparated(src string, sep string) string {
 	var words []string
 	var word []rune