Repository: allegro/marathon-consul
Pull request #146 (2016-11-30 16:34:07)

Remove redundant ticker return in sync

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

Coverage decreased (-0.01%) to 82.514% when pulling **ab3426aa33b641c026df620179cd3f7c5e7c4526 on cleanup_sync_service** into **cf1abbb4aef154b4e048730debe76e34444e6c56 on master**.


Patch:
diff --git a/events/events.go b/events/events.go
index 3b23cea..b433f34 100644
--- a/events/events.go
+++ b/events/events.go
@@ -3,27 +3,42 @@ package events
 import (
 	"encoding/json"
 	"errors"
-
-	"github.com/allegro/marathon-consul/apps"
+	"strings"
+	"time"
 )
 
-type Event interface {
-	Apps() []*apps.App
-	GetType() string
+type Timestamp struct {
+	time.Time
+}
+
+func (t *Timestamp) UnmarshalJSON(b []byte) (err error) {
+	s := strings.Trim(string(b), "\"")
+	if s == "null" {
+		t.Time = time.Time{}
+		return
+	}
+	t.Time, err = time.Parse(time.RFC3339Nano, s)
+	return
+}
+
+func (t *Timestamp) String() string {
+	return t.Format(time.RFC3339Nano)
 }
 
-type baseEvent struct {
-	Type string `json:"eventType"`
+type Event struct {
+	Type      string    `json:"eventType"`
+	Timestamp Timestamp `json:"timestamp"`
 }
 
-func EventType(jsonBlob []byte) (string, error) {
-	event := baseEvent{}
+func ParseEvent(jsonBlob []byte) (Event, error) {
+	event := Event{}
 	err := json.Unmarshal(jsonBlob, &event)
 	if err != nil {
-		return "", err
+		return Event{}, err
 	} else if event.Type == "" {
-		return "", errors.New("no event")
-	} else {
-		return event.Type, nil
+		return Event{}, errors.New("Missing event type")
+	} else if event.Timestamp.Unix() == (time.Time{}).Unix() {
+		return Event{}, errors.New("Missing timestamp")
 	}
+	return event, nil
 }
diff --git a/main.go b/main.go
index 405f43e..086f35b3 100644
--- a/main.go
+++ b/main.go
@@ -34,8 +34,7 @@ func main() {
 		log.Fatal(err.Error())
 	}
 
-	sync := sync.New(config.Sync, remote, consulInstance, consulInstance.AddAgentsFromApps)
-	go sync.StartSyncServicesJob()
+	sync.New(config.Sync, remote, consulInstance, consulInstance.AddAgentsFromApps).StartSyncServicesJob()
 
 	handler, stop := web.NewHandler(config.Web, remote, consulInstance)
 	defer stop()
diff --git a/marathon/marathon.go b/marathon/marathon.go
index 0c78418..ed4c6fa 100644
--- a/marathon/marathon.go
+++ b/marathon/marathon.go
@@ -113,6 +113,7 @@ func (m Marathon) get(url string) ([]byte, error) {
 		return nil, err
 	}
 	request.Header.Add("Accept", "application/json")
+	request.Header.Set("User-Agent", "Marathon-Consul")
 
 	log.WithFields(log.Fields{
 		"Uri":      request.URL.RequestURI(),
diff --git a/sync/sync.go b/sync/sync.go
index 394f25b..174362b 100644
--- a/sync/sync.go
+++ b/sync/sync.go
@@ -25,10 +25,10 @@ func New(config Config, marathon marathon.Marathoner, serviceRegistry service.Se
 	return &Sync{config, marathon, serviceRegistry, syncStartedListener}
 }
 
-func (s *Sync) StartSyncServicesJob() *time.Ticker {
+func (s *Sync) StartSyncServicesJob() {
 	if !s.config.Enabled {
 		log.Info("Marathon-consul sync disabled")
-		return nil
+		return
 	}
 
 	log.WithFields(log.Fields{
@@ -42,14 +42,13 @@ func (s *Sync) StartSyncServicesJob() *time.Ticker {
 		if err := s.SyncServices(); err != nil {
 			log.WithError(err).Error("An error occured while performing sync")
 		}
-		for {
-			<-ticker.C
+		for range ticker.C {
 			if err := s.SyncServices(); err != nil {
 				log.WithError(err).Error("An error occured while performing sync")
 			}
 		}
 	}()
-	return ticker
+	return
 }
 
 func (s *Sync) SyncServices() error {
diff --git a/web/web_handler.go b/web/web_handler.go
index 788e44d..b4b01b5 100644
--- a/web/web_handler.go
+++ b/web/web_handler.go
@@ -48,21 +48,23 @@ func (h *EventHandler) Handle(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 
-		eventType, err := events.EventType(body)
+		e, err := events.ParseEvent(body)
 		if err != nil {
 			drop(err, w)
 			return
 		}
 
-		log.WithField("EventType", eventType).Debug("Received event")
-		metrics.Mark("events.requests." + eventType)
+		metrics.Mark("events.requests." + e.Type)
+		delay := time.Now().Unix() - e.Timestamp.Unix()
+		metrics.UpdateGauge("events.requests.delay.current", delay)
+		log.WithFields(log.Fields{"EventType": e.Type, "OriginalTimestamp": e.Timestamp.String()}).Debug("Received event")
 
-		if eventType != statusUpdateEventType && eventType != healthStatusChangedEventType {
-			drop(fmt.Errorf("%s is not supported", eventType), w)
+		if e.Type != statusUpdateEventType && e.Type != healthStatusChangedEventType {
+			drop(fmt.Errorf("%s is not supported", e.Type), w)
 			return
 		}
 
-		h.eventQueue <- event{eventType: eventType, body: body, timestamp: time.Now()}
+		h.eventQueue <- event{eventType: e.Type, body: body, timestamp: time.Now()}
 		accept(w)
 
 	})


Test patch:
diff --git a/events/events_test.go b/events/events_test.go
index 7c43856..3d0f470 100644
--- a/events/events_test.go
+++ b/events/events_test.go
@@ -8,24 +8,37 @@ import (
 
 func TestEventTypeValid(t *testing.T) {
 	t.Parallel()
-
-	out, err := EventType([]byte(`{"eventType":"test"}`))
+	out, err := ParseEvent([]byte(`{"eventType":"test","timestamp": "2014-03-01T23:29:30.158Z"}`))
 	assert.Nil(t, err)
-	assert.Equal(t, out, "test")
+	assert.Equal(t, "2014-03-01T23:29:30.158Z", out.Timestamp.String())
 }
 
 func TestEventTypeInvalid(t *testing.T) {
 	t.Parallel()
 
-	out, err := EventType([]byte(`{}`))
+	out, err := ParseEvent([]byte(`{}`))
+	assert.Error(t, err)
+	assert.Equal(t, out, Event{})
+}
+
+func TestTimestampInvalid(t *testing.T) {
+	t.Parallel()
+	out, err := ParseEvent([]byte(`{"eventType":"test","timestamp": "invalid"}`))
 	assert.Error(t, err)
-	assert.Equal(t, out, "")
+	assert.Equal(t, out, Event{})
+}
+
+func TestMissingTimestamp(t *testing.T) {
+	t.Parallel()
+	out, err := ParseEvent([]byte(`{"eventType":"test"}`))
+	assert.EqualError(t, err, "Missing timestamp")
+	assert.Equal(t, out, Event{})
 }
 
 func TestEventTypeInvalidJson(t *testing.T) {
 	t.Parallel()
 
-	out, err := EventType([]byte(`not a json`))
+	out, err := ParseEvent([]byte(`not a json`))
 	assert.Error(t, err)
-	assert.Equal(t, out, "")
+	assert.Equal(t, out, Event{})
 }
diff --git a/sync/sync_test.go b/sync/sync_test.go
index 11c87b3..043d8de 100644
--- a/sync/sync_test.go
+++ b/sync/sync_test.go
@@ -31,11 +31,10 @@ func TestSyncJob_ShouldSyncOnLeadership(t *testing.T) {
 	}, marathon, services, noopSyncStartedListener)
 
 	// when
-	ticker := sync.StartSyncServicesJob()
+	sync.StartSyncServicesJob()
 
 	// then
 	<-time.After(15 * time.Millisecond)
-	ticker.Stop()
 	assert.Equal(t, 2, services.RegistrationsCount(app.Tasks[0].ID.String()))
 }
 
@@ -52,10 +51,9 @@ func TestSyncJob_ShouldNotSyncWhenDisabled(t *testing.T) {
 	}, marathon, services, noopSyncStartedListener)
 
 	// when
-	ticker := sync.StartSyncServicesJob()
+	sync.StartSyncServicesJob()
 
 	// then
-	assert.Nil(t, ticker)
 	<-time.After(15 * time.Millisecond)
 	assert.Equal(t, 0, services.RegistrationsCount(app.Tasks[0].ID.String()))
 }
@@ -73,11 +71,10 @@ func TestSyncJob_ShouldDefaultLeaderConfigurationToResolvedHostname(t *testing.T
 	}, marathon, services, noopSyncStartedListener)
 
 	// when
-	ticker := sync.StartSyncServicesJob()
+	sync.StartSyncServicesJob()
 
 	// then
 	<-time.After(15 * time.Millisecond)
-	ticker.Stop()
 	assert.Equal(t, 2, services.RegistrationsCount(app.Tasks[0].ID.String()))
 }
 
@@ -90,10 +87,9 @@ func TestSyncServices_ShouldNotSyncOnNoForceNorLeaderSpecified(t *testing.T) {
 	sync := New(Config{}, marathon, services, noopSyncStartedListener)
 
 	// when
-	ticker := sync.StartSyncServicesJob()
+	sync.StartSyncServicesJob()
 
 	// then
-	assert.Nil(t, ticker)
 	assert.Zero(t, services.RegistrationsCount(app.Tasks[0].ID.String()))
 }