Repository: allegro/marathon-consul
Pull request #255 (2017-07-31 14:18:07)

Allow configuration of SSE recover retries.
Fixes: #246

coveralls: 
[![Coverage Status](https://coveralls.io/builds/12623245/badge)](https://coveralls.io/builds/12623245)

Coverage increased (+0.01%) to 89.407% when pulling **4c2d1f925d653a241b4436d9554b152762f9662d on expose_sse_retries** into **b1c1ec5b1e30a40a7a531d220cfa49036ef5831c on master**.


Patch:
diff --git a/README.md b/README.md
index c9f4f95..8edeb03 100644
--- a/README.md
+++ b/README.md
@@ -224,6 +224,8 @@ sentry-dsn                  |                 | Sentry DSN. If it's not set sent
 sentry-env                  |                 | Sentry environment
 sentry-level                | `error`         | Sentry alerting level (info|warning|error|fatal|panic)
 sentry-timeout              | `1s`            | Sentry hook initialization timeout
+sse-retries                 | `0`             | Number of times to recover SSE stream.
+sse-retry-backoff           | `0s`            | Configuration of initial time between retries to recover SSE stream.
 sync-enabled                | `true`          | Enable Marathon-consul scheduled sync
 sync-force                  | `false`         | Force leadership-independent Marathon-consul sync (run always)
 sync-interval               | `15m0s`         | Marathon-consul sync interval
diff --git a/config/config.go b/config/config.go
index c526a94..6376baa 100644
--- a/config/config.go
+++ b/config/config.go
@@ -87,6 +87,8 @@ func (config *Config) parseFlags() {
 	flag.Int64Var(&config.Web.MaxEventSize, "event-max-size", 4096, "Maximum size of event to process (bytes)")
 
 	// SSE
+	flag.IntVar(&config.SSE.Retries, "sse-retries", 0, "Number of times to recover SSE stream.")
+	flag.DurationVar(&config.SSE.RetryBackoff.Duration, "sse-retry-backoff", 0, "Configuration of initial time between retries to recover SSE stream.")
 
 	// Sync
 	flag.BoolVar(&config.Sync.Enabled, "sync-enabled", true, "Enable Marathon-consul scheduled sync")
diff --git a/debian/config.json b/debian/config.json
index 62abc2a..405a528 100644
--- a/debian/config.json
+++ b/debian/config.json
@@ -27,6 +27,8 @@
     "MaxEventSize": 4096
   },
   "SSE": {
+    "Retries": 0,
+    "RetryBackoff": "0s"
   },
   "Sync": {
     "Enabled": true,
diff --git a/marathon/marathon.go b/marathon/marathon.go
index a22187d..d83119e 100644
--- a/marathon/marathon.go
+++ b/marathon/marathon.go
@@ -23,7 +23,7 @@ type Marathoner interface {
 	App(apps.AppID) (*apps.App, error)
 	Tasks(apps.AppID) ([]apps.Task, error)
 	Leader() (string, error)
-	EventStream([]string, int, int) (*Streamer, error)
+	EventStream([]string, int, time.Duration) (*Streamer, error)
 	IsLeader() (bool, error)
 }
 
@@ -113,7 +113,7 @@ func (m Marathon) Leader() (string, error) {
 
 // EventStream method creates Streamer handler which is configured based on marathon
 // client and credentials.
-func (m Marathon) EventStream(desiredEvents []string, retries, retryBackoff int) (*Streamer, error) {
+func (m Marathon) EventStream(desiredEvents []string, retries int, retryBackoff time.Duration) (*Streamer, error) {
 	subURL := m.urlWithQuery("/v2/events", params{"event_type": desiredEvents})
 
 	// Before creating actual streamer, this function blocks until configured leader for this receiver is elected.
diff --git a/marathon/marathon_stub.go b/marathon/marathon_stub.go
index 257cde8..542bc12 100644
--- a/marathon/marathon_stub.go
+++ b/marathon/marathon_stub.go
@@ -3,6 +3,7 @@ package marathon
 import (
 	"errors"
 	"sync"
+	"time"
 
 	"github.com/allegro/marathon-consul/apps"
 )
@@ -43,7 +44,7 @@ func (m *MarathonerStub) Leader() (string, error) {
 	return m.leader, nil
 }
 
-func (m *MarathonerStub) EventStream([]string, int, int) (*Streamer, error) {
+func (m *MarathonerStub) EventStream([]string, int, time.Duration) (*Streamer, error) {
 	return &Streamer{}, nil
 }
 
diff --git a/marathon/streamer.go b/marathon/streamer.go
index a5d96e2..5ad3b66 100644
--- a/marathon/streamer.go
+++ b/marathon/streamer.go
@@ -19,7 +19,7 @@ type Streamer struct {
 	password     string
 	subURL       string
 	retries      int
-	retryBackoff int
+	retryBackoff time.Duration
 	noRecover    bool
 }
 
@@ -65,8 +65,7 @@ func (s *Streamer) Recover() error {
 	err := s.Start()
 	i := 0
 	for ; err != nil && i <= s.retries; err = s.Start() {
-		seconds := time.Duration(i * s.retryBackoff)
-		time.Sleep(seconds * time.Second)
+		time.Sleep(s.retryBackoff)
 		i++
 	}
 	if err != nil {
diff --git a/sse/config.go b/sse/config.go
index c19e323..86779a5 100644
--- a/sse/config.go
+++ b/sse/config.go
@@ -1,6 +1,8 @@
 package sse
 
+import "github.com/allegro/marathon-consul/time"
+
 type Config struct {
 	Retries      int
-	RetryBackoff int
+	RetryBackoff time.Interval
 }
diff --git a/sse/sse_handler.go b/sse/sse_handler.go
index a0eaf5a..1b44873 100644
--- a/sse/sse_handler.go
+++ b/sse/sse_handler.go
@@ -25,7 +25,7 @@ func newSSEHandler(eventQueue chan events.Event, service marathon.Marathoner, ma
 	streamer, err := service.EventStream(
 		[]string{events.StatusUpdateEventType, events.HealthStatusChangedEventType},
 		config.Retries,
-		config.RetryBackoff,
+		config.RetryBackoff.Duration,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("Unable to start Streamer: %s", err)


Test patch:
diff --git a/sync/error_marathon_stub_test.go b/sync/error_marathon_stub_test.go
index fbab4ec..e114fa0 100644
--- a/sync/error_marathon_stub_test.go
+++ b/sync/error_marathon_stub_test.go
@@ -2,6 +2,7 @@ package sync
 
 import (
 	"errors"
+	"time"
 
 	"github.com/allegro/marathon-consul/apps"
 	"github.com/allegro/marathon-consul/marathon"
@@ -26,7 +27,7 @@ func (m errorMarathon) Leader() (string, error) {
 	return "", errors.New("Error")
 }
 
-func (m errorMarathon) EventStream([]string, int, int) (*marathon.Streamer, error) {
+func (m errorMarathon) EventStream([]string, int, time.Duration) (*marathon.Streamer, error) {
 	return &marathon.Streamer{}, errors.New("Error")
 }