Repository: allegro/marathon-consul
Pull request #115 (2016-09-07 07:21:14)

Sync support for multi-registrations
In multi-registration scenario, some, but not all, service registrations may fail. Sync should check if all registrations are in Consul and register the missing ones.

Change base to *develop* once #114 is merged.

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

Coverage decreased (-0.9%) to 78.915% when pulling **8e6cea863f0e8f92426c7ce0a2e6cdbab2f6801b on sync-multiple-registrations** into **741d90effd071b20b30b88c204bdd66479249e05 on multiple-registrations**.

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

Coverage decreased (-0.4%) to 79.423% when pulling **e562b50c9686df0f4a5a34c6a6f8f1f65ae4c36c on sync-multiple-registrations** into **741d90effd071b20b30b88c204bdd66479249e05 on multiple-registrations**.

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

Coverage decreased (-0.4%) to 79.423% when pulling **40d62580cebf6313a10c4397c52c635299f73052 on sync-multiple-registrations** into **741d90effd071b20b30b88c204bdd66479249e05 on multiple-registrations**.

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

Coverage decreased (-0.4%) to 79.423% when pulling **e3a1ae8e85e5ed16aed15ec7dded48fe482c540d on sync-multiple-registrations** into **cb97fdedd96fc3b363751002c91b38ced8e6e95b on develop**.


Patch:
diff --git a/apps/app.go b/apps/app.go
index 34ba744..a352789 100644
--- a/apps/app.go
+++ b/apps/app.go
@@ -119,6 +119,19 @@ type RegistrationIntent struct {
 	Tags []string
 }
 
+func (app *App) RegistrationIntentsNumber() int {
+	if !app.IsConsulApp() {
+		return 0
+	}
+
+	definitions := app.findConsulPortDefinitions()
+	if len(definitions) == 0 {
+		return 1
+	}
+
+	return len(definitions)
+}
+
 func (app *App) RegistrationIntents(task *Task, nameSeparator string) []*RegistrationIntent {
 	commonTags := labelsToTags(app.Labels)
 
diff --git a/consul/consul_stub.go b/consul/consul_stub.go
index 99b245e..d914161 100644
--- a/consul/consul_stub.go
+++ b/consul/consul_stub.go
@@ -109,6 +109,11 @@ func (c *ConsulStub) RegisterWithoutMarathonTaskTag(task *apps.Task, app *apps.A
 	}
 }
 
+func (c *ConsulStub) RegisterOnlyFirstRegistrationIntent(task *apps.Task, app *apps.App) {
+	serviceRegistrations, _ := c.consul.marathonTaskToConsulServices(task, app)
+	c.services[service.ServiceId(serviceRegistrations[0].ID)] = serviceRegistrations[0]
+}
+
 func (c *ConsulStub) ServiceNames(app *apps.App) []string {
 	return c.consul.ServiceNames(app)
 }
diff --git a/sync/sync.go b/sync/sync.go
index 57f7746..000bf7b 100644
--- a/sync/sync.go
+++ b/sync/sync.go
@@ -143,14 +143,19 @@ func (s *Sync) deregisterConsulServicesNotFoundInMarathon(marathonApps []*apps.A
 }
 
 func (s *Sync) registerAppTasksNotFoundInConsul(marathonApps []*apps.App, services []*service.Service) {
-	registeredTaskIds := s.taskIdsInConsulServices(services)
+	registrationsUnderTaskIds := s.taskIdsInConsulServices(services)
 	for _, app := range marathonApps {
 		if !app.IsConsulApp() {
 			log.WithField("Id", app.ID).Debug("Not a Consul app, skipping registration")
 			continue
 		}
+		expectedRegistrations := app.RegistrationIntentsNumber()
 		for _, task := range app.Tasks {
-			if _, ok := registeredTaskIds[task.ID]; !ok {
+			if registrations := registrationsUnderTaskIds[task.ID]; registrations != expectedRegistrations {
+				if registrations != 0 {
+					log.WithField("Id", task.ID).WithField("HasRegistrations", registrations).
+					WithField("ExpectedRegistrations", expectedRegistrations).Info("Registering missing service registrations")
+				}
 				if task.IsHealthy() {
 					err := s.serviceRegistry.Register(&task, app)
 					if err != nil {
@@ -166,15 +171,14 @@ func (s *Sync) registerAppTasksNotFoundInConsul(marathonApps []*apps.App, servic
 	}
 }
 
-func (s *Sync) taskIdsInConsulServices(services []*service.Service) map[apps.TaskId]struct{} {
-	servicesSet := make(map[apps.TaskId]struct{})
-	var exists struct{}
+func (s *Sync) taskIdsInConsulServices(services []*service.Service) map[apps.TaskId]int {
+	serviceCounters := make(map[apps.TaskId]int)
 	for _, service := range services {
 		if taskId, err := service.TaskId(); err == nil {
-			servicesSet[taskId] = exists
+			serviceCounters[taskId] += 1
 		}
 	}
-	return servicesSet
+	return serviceCounters
 }
 
 func (s *Sync) marathonTaskIdsSet(marathonApps []*apps.App) map[apps.TaskId]struct{} {
diff --git a/utils/apps.go b/utils/apps.go
index 04aac94..30e9d42 100644
--- a/utils/apps.go
+++ b/utils/apps.go
@@ -8,24 +8,32 @@ import (
 )
 
 func ConsulApp(name string, instances int) *apps.App {
-	return app(name, instances, true, 0)
+	return app(name, instances, 1, true, 0)
 }
 
 func ConsulAppWithUnhealthyInstances(name string, instances int, unhealthyInstances int) *apps.App {
-	return app(name, instances, true, unhealthyInstances)
+	return app(name, instances, 1, true, unhealthyInstances)
+}
+
+func ConsulAppMultipleRegistrations(name string, instances int, registrations int) *apps.App {
+	return app(name, instances, registrations, true, 0)
 }
 
 func NonConsulApp(name string, instances int) *apps.App {
-	return app(name, instances, false, 0)
+	return app(name, instances, 1, false, 0)
 }
 
-func app(name string, instances int, consul bool, unhealthyInstances int) *apps.App {
+func app(name string, instances int, registrationsPerInstance int, consul bool, unhealthyInstances int) *apps.App {
 	var appTasks []apps.Task
 	for i := 0; i < instances; i++ {
+		var ports []int
+		for j := 1; j <= registrationsPerInstance; j++ {
+			ports = append(ports, 8080 + (i * j) + j - 1)
+		}
 		task := apps.Task{
 			AppID: apps.AppId(name),
 			ID:    apps.TaskId(fmt.Sprintf("%s.%d", strings.Replace(strings.Trim(name, "/"), "/", "_", -1), i)),
-			Ports: []int{8080 + i},
+			Ports: ports,
 			Host:  "localhost",
 		}
 		if unhealthyInstances > 0 {
@@ -45,9 +53,20 @@ func app(name string, instances int, consul bool, unhealthyInstances int) *apps.
 		labels[apps.MARATHON_CONSUL_LABEL] = "true"
 	}
 
-	return &apps.App{
+	app := &apps.App{
 		ID:     apps.AppId(name),
 		Tasks:  appTasks,
 		Labels: labels,
 	}
+
+	if registrationsPerInstance > 1 {
+		for i := 0; i < registrationsPerInstance; i++ {
+			app.PortDefinitions = append(app.PortDefinitions, apps.PortDefinition{
+				Port: 0,
+				Labels: map[string]string{"consul": ""},
+			})
+		}
+	}
+
+	return app
 }


Test patch:
diff --git a/apps/app_test.go b/apps/app_test.go
index 060031c..6d9e57c 100644
--- a/apps/app_test.go
+++ b/apps/app_test.go
@@ -527,3 +527,67 @@ func TestHasSameConsulNamesAs_DifferentConfigsDifferentNumberOfRegitrationsWithP
 	// expect
 	assert.False(t, app.HasSameConsulNamesAs(other))
 }
+
+func TestRegistrationIntentsNumber_NotConsulApp(t *testing.T) {
+	t.Parallel()
+
+	// given
+	app := &App{
+		ID: "id",
+	}
+
+	// expect
+	assert.Equal(t, 0, app.RegistrationIntentsNumber())
+}
+
+func TestRegistrationIntentsNumber_NoPortDefinitions(t *testing.T) {
+	t.Parallel()
+
+	// given
+	app := &App{
+		ID: "id",
+		Labels: map[string]string{"consul": ""},
+	}
+
+	// expect
+	assert.Equal(t, 1, app.RegistrationIntentsNumber())
+}
+
+func TestRegistrationIntentsNumber_SinglePortDefinitions(t *testing.T) {
+	t.Parallel()
+
+	// given
+	app := &App{
+		ID:     "id",
+		Labels: map[string]string{"consul": ""},
+		PortDefinitions: []PortDefinition{
+			PortDefinition{
+				Labels: map[string]string{"consul": ""},
+			},
+		},
+	}
+
+	// expect
+	assert.Equal(t, 1, app.RegistrationIntentsNumber())
+}
+
+func TestRegistrationIntentsNumber_MultiplePortDefinitions(t *testing.T) {
+	t.Parallel()
+
+	// given
+	app := &App{
+		ID:     "id",
+		Labels: map[string]string{"consul": ""},
+		PortDefinitions: []PortDefinition{
+			PortDefinition{
+				Labels: map[string]string{"consul": ""},
+			},
+			PortDefinition{
+				Labels: map[string]string{"consul": ""},
+			},
+		},
+	}
+
+	// expect
+	assert.Equal(t, 2, app.RegistrationIntentsNumber())
+}
diff --git a/sync/sync_test.go b/sync/sync_test.go
index 9bd0e58..56bd41b 100644
--- a/sync/sync_test.go
+++ b/sync/sync_test.go
@@ -326,6 +326,28 @@ func TestSync_WithRegisteringProblems(t *testing.T) {
 	assert.Len(t, services, 2)
 }
 
+func TestSync_ShouldRegisterMissingRegistrationInMultiregistrationScenario(t *testing.T) {
+	t.Parallel()
+	// given
+	app := ConsulAppMultipleRegistrations("/test/app", 1, 2)
+	marathon := marathon.MarathonerStubForApps(app)
+	consul := consul.NewConsulStub()
+
+	consul.RegisterOnlyFirstRegistrationIntent(&app.Tasks[0], app)
+	services, _ := consul.GetAllServices()
+	assert.Len(t, services, 1)
+
+	sync := newSyncWithDefaultConfig(marathon, consul)
+
+	// when
+	err := sync.SyncServices()
+
+	// then
+	services, _ = consul.GetAllServices()
+	assert.NoError(t, err)
+	assert.Len(t, services, 2)
+}
+
 func TestSync_WithDeregisteringProblems(t *testing.T) {
 	t.Parallel()
 	// given