Merge pull request #241 from embano1/retries

Add retries to processors

vmware-event-router/README.MD

@@ -24,21 +24,24 @@ to normalize events from the supported event `providers`. See
 
 **Event Delivery Guarantees:**
 
-- At-least-once event delivery for the [vCenter event provider](#provider-type-vcenter) (using checkpoints)
-- At-most-once event delivery ("stateless" mode)
+- At-least-once event delivery
+  -  with the [vCenter event provider](#provider-type-vcenter) option `checkpoint: true`
+- At-most-once event delivery
+  - with the [vCenter event provider](#provider-type-vcenter) option `checkpoint: false`
+  - with the [vcsim event provider](#provider-type-vcsim)
 
 **Current limitations:**
 
 - Only one event `provider` and one event `processor` can be configured at a
-  time
-- At-least-once event delivery semantics cannot be guaranteed if the event
-  router crashes **within seconds** right after having received some events but before creating the
-  first valid checkpoint
-- At-least-once event delivery semantics are only supported for the vCenter
-  event provider
-- At-least-once event delivery semantics currently only guard against crashes of
-  the event router. If an event cannot be delivered to an event processor,
-  currently no retries are performed/dead letter queues are not supported
+  time (see note below)
+- At-least-once event delivery semantics are not guaranteed if the event
+  router crashes **within seconds** right after startup and having received *n* events but before creating the
+  first valid checkpoint (current checkpoint interval is 5s)
+- If an event cannot be successfully delivered (retried) by an event `processor` it is
+  logged and discarded, i.e. there is currently no support for dead letter queues
+- Retries in the [OpenFaaS event processor](#processor-type-openfaas) are only
+  supported when running in synchronous mode, i.e. `async: false` (see this
+  OpenFaaS [issue](https://github.com/openfaas/nats-queue-worker/issues/84))
 
 > **Note:** It is possible though to run **multiple instances** of the event
 > router with different configurations to address multi-vCenter scenarios
@@ -525,6 +528,7 @@ the following script:
 - `kind`
 - `docker`
 
+
 ## Example Event Structure
 
 The following example for a `VmPoweredOnEvent` shows the event structure and

vmware-event-router/cmd/router/main.go

@@ -17,6 +17,8 @@ import (
 	config "github.com/vmware-samples/vcenter-event-broker-appliance/vmware-event-router/internal/config/v1alpha1"
 	"github.com/vmware-samples/vcenter-event-broker-appliance/vmware-event-router/internal/metrics"
 	"github.com/vmware-samples/vcenter-event-broker-appliance/vmware-event-router/internal/processor"
+	"github.com/vmware-samples/vcenter-event-broker-appliance/vmware-event-router/internal/processor/aws"
+	"github.com/vmware-samples/vcenter-event-broker-appliance/vmware-event-router/internal/processor/openfaas"
 	"github.com/vmware-samples/vcenter-event-broker-appliance/vmware-event-router/internal/provider"
 	"github.com/vmware-samples/vcenter-event-broker-appliance/vmware-event-router/internal/provider/vcenter"
 	"github.com/vmware-samples/vcenter-event-broker-appliance/vmware-event-router/internal/provider/vcsim"
@@ -121,15 +123,15 @@ func main() {
 	// set up event processor
 	switch cfg.EventProcessor.Type {
 	case config.ProcessorOpenFaaS:
-		proc, err = processor.NewOpenFaaSProcessor(ctx, cfg.EventProcessor.OpenFaaS, ms, processor.WithOpenFaaSVerbose(verbose))
+		proc, err = openfaas.NewProcessor(ctx, cfg.EventProcessor.OpenFaaS, ms, openfaas.WithVerbose(verbose))
 		if err != nil {
 			logger.Fatalf("could not connect to OpenFaaS: %v", err)
 		}
 
 		logger.Printf("connected to OpenFaaS gateway %q (async mode: %t)", cfg.EventProcessor.OpenFaaS.Address, cfg.EventProcessor.OpenFaaS.Async)
 
 	case config.ProcessorEventBridge:
-		proc, err = processor.NewEventBridgeProcessor(ctx, cfg.EventProcessor.EventBridge, ms, processor.WithAWSVerbose(verbose))
+		proc, err = aws.NewEventBridgeProcessor(ctx, cfg.EventProcessor.EventBridge, ms, aws.WithVerbose(verbose))
 		if err != nil {
 			logger.Fatalf("could not connect to AWS EventBridge: %v", err)
 		}
@@ -175,6 +177,7 @@ func main() {
 	eg.Go(func() error {
 		defer func() {
 			_ = prov.Shutdown(egCtx)
+			_ = proc.Shutdown(egCtx)
 		}()
 		return prov.Stream(egCtx, proc)
 	})

vmware-event-router/go.mod

@@ -4,17 +4,29 @@ go 1.14
 
 require (
 	github.com/alecthomas/jsonschema v0.0.0-20200530073317-71f438968921
-	github.com/aws/aws-sdk-go v1.27.3
+	github.com/avast/retry-go v2.6.1+incompatible
+	github.com/aws/aws-sdk-go v1.33.12
 	github.com/cloudevents/sdk-go/v2 v2.0.0-RC2
+	github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e // indirect
 	github.com/google/uuid v1.1.1
 	github.com/jpillora/backoff v1.0.0
 	github.com/onsi/ginkgo v1.12.2
 	github.com/onsi/gomega v1.10.1
-	github.com/openfaas-incubator/connector-sdk v0.0.0-20191214130609-df5d76475412
+	github.com/openfaas-incubator/connector-sdk v0.0.0-20200902074656-7f648543d4aa
 	github.com/openfaas/faas-provider v0.15.1
 	github.com/pkg/errors v0.9.1
 	github.com/vmware/govmomi v0.22.2
-	golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a
-	golang.org/x/sys v0.0.0-20200523222454-059865788121 // indirect
+	go.opencensus.io v0.22.3 // indirect
+	go.uber.org/zap v1.14.1 // indirect
+	golang.org/x/lint v0.0.0-20200302205851-738671d3881b // indirect
+	golang.org/x/net v0.0.0-20200707034311-ab3426394381 // indirect
+	golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208
+	golang.org/x/sys v0.0.0-20200728102440-3e129f6d46b1 // indirect
+	golang.org/x/text v0.3.3 // indirect
+	golang.org/x/tools v0.0.0-20200725200936-102e7d357031 // indirect
+	google.golang.org/protobuf v1.24.0 // indirect
 	gopkg.in/yaml.v2 v2.3.0
+	honnef.co/go/tools v0.0.1-2020.1.3 // indirect
 )
+
+replace github.com/openfaas-incubator/connector-sdk => github.com/embano1/connector-sdk v0.0.0-20201013145543-3487b96b0f91