Repository: allegro/marathon-consul
Pull request #211 (2017-03-22 13:31:08)

Move sync.Leader to marathon module
Mentioned: https://github.com/allegro/marathon-consul/pull/187#discussion_r103667851


Patch:
diff --git a/README.md b/README.md
index 087d181..17816c1 100644
--- a/README.md
+++ b/README.md
@@ -207,6 +207,7 @@ marathon-protocol           | `http`          | Marathon protocol (http or https
 marathon-ssl-verify         | `true`          | Verify certificates when connecting via SSL
 marathon-timeout            | `30s`           | Time limit for requests made by the Marathon HTTP client. A Timeout of zero means no timeout
 marathon-username           |                 | Marathon username for basic auth
+marathon-leader             |                 | Marathon cluster-wide node name (defaults to <hostname>:8080), the some leader specific calls will be made only if the specified node is the current Marathon-leader")
 metrics-interval            | `30s`           | Metrics reporting interval
 metrics-location            |                 | Graphite URL (used when metrics-target is set to graphite)
 metrics-prefix              | `default`       | Metrics prefix (default is resolved to <hostname>.<app_name>
diff --git a/config/config.go b/config/config.go
index 7ace9ba..3d85104 100644
--- a/config/config.go
+++ b/config/config.go
@@ -91,7 +91,6 @@ func (config *Config) parseFlags() {
 	// Sync
 	flag.BoolVar(&config.Sync.Enabled, "sync-enabled", true, "Enable Marathon-consul scheduled sync")
 	flag.DurationVar(&config.Sync.Interval.Duration, "sync-interval", 15*time.Minute, "Marathon-consul sync interval")
-	flag.StringVar(&config.Sync.Leader, "sync-leader", "", "Marathon cluster-wide node name (defaults to <hostname>:8080), the sync will run only if the specified node is the current Marathon-leader")
 	flag.BoolVar(&config.Sync.Force, "sync-force", false, "Force leadership-independent Marathon-consul sync (run always)")
 
 	// Marathon
@@ -99,6 +98,7 @@ func (config *Config) parseFlags() {
 	flag.StringVar(&config.Marathon.Protocol, "marathon-protocol", "http", "Marathon protocol (http or https)")
 	flag.StringVar(&config.Marathon.Username, "marathon-username", "", "Marathon username for basic auth")
 	flag.StringVar(&config.Marathon.Password, "marathon-password", "", "Marathon password for basic auth")
+	flag.StringVar(&config.Marathon.Leader, "marathon-leader", "", "Marathon cluster-wide node name (defaults to <hostname>:8080), the some leader specific calls will be made only if the specified node is the current Marathon-leader")
 	flag.BoolVar(&config.Marathon.VerifySsl, "marathon-ssl-verify", true, "Verify certificates when connecting via SSL")
 	flag.DurationVar(&config.Marathon.Timeout.Duration, "marathon-timeout", 30*time.Second, "Time limit for requests made by the Marathon HTTP client. A Timeout of zero means no timeout")
 
diff --git a/main.go b/main.go
index 3e25410..0c93618 100644
--- a/main.go
+++ b/main.go
@@ -36,7 +36,7 @@ func main() {
 
 	consulInstance := consul.New(config.Consul)
 	// TODO(tz) - move Leader from sync module to highest level config, access like config.Leader
-	remote, err := marathon.New(config.Marathon, config.Sync.Leader)
+	remote, err := marathon.New(config.Marathon)
 	if err != nil {
 		log.Fatal(err.Error())
 	}
diff --git a/marathon/config.go b/marathon/config.go
index b40d6c8..d4881f8 100644
--- a/marathon/config.go
+++ b/marathon/config.go
@@ -7,6 +7,7 @@ type Config struct {
 	Protocol  string
 	Username  string
 	Password  string
+	Leader    string
 	VerifySsl bool
 	Timeout   time.Interval
 }
diff --git a/marathon/marathon.go b/marathon/marathon.go
index 93bed15..8121ab0 100644
--- a/marathon/marathon.go
+++ b/marathon/marathon.go
@@ -7,6 +7,7 @@ import (
 	"io/ioutil"
 	"net/http"
 	"net/url"
+	"os"
 	"strings"
 	"time"
 
@@ -15,13 +16,15 @@ import (
 	"github.com/allegro/marathon-consul/metrics"
 )
 
+var hostname = os.Hostname
+
 type Marathoner interface {
 	ConsulApps() ([]*apps.App, error)
 	App(apps.AppID) (*apps.App, error)
 	Tasks(apps.AppID) ([]apps.Task, error)
 	Leader() (string, error)
 	EventStream([]string, int, int) (*Streamer, error)
-	AmILeader() (bool, error)
+	IsLeader() (bool, error)
 }
 
 type Marathon struct {
@@ -36,7 +39,7 @@ type LeaderResponse struct {
 	Leader string `json:"leader"`
 }
 
-func New(config Config, leader string) (*Marathon, error) {
+func New(config Config) (*Marathon, error) {
 	var auth *url.Userinfo
 	if len(config.Username) == 0 && len(config.Password) == 0 {
 		auth = nil
@@ -53,8 +56,8 @@ func New(config Config, leader string) (*Marathon, error) {
 	return &Marathon{
 		Location: config.Location,
 		Protocol: config.Protocol,
+		MyLeader: config.Leader,
 		Auth:     auth,
-		MyLeader: leader,
 		client: &http.Client{
 			Transport: transport,
 			Timeout:   config.Timeout.Duration,
@@ -142,7 +145,7 @@ func (m Marathon) leaderPoll() error {
 	retries := 5
 	i := 0
 	for range pollTicker {
-		leading, err := m.AmILeader()
+		leading, err := m.IsLeader()
 		if err != nil {
 			if i >= retries {
 				return fmt.Errorf("Failed to get a leader after %d retries", i)
@@ -244,7 +247,22 @@ func (m Marathon) urlWithQuery(path string, params params) string {
 	return marathon.String()
 }
 
-func (m Marathon) AmILeader() (bool, error) {
+func (m *Marathon) IsLeader() (bool, error) {
+	if m.MyLeader == "" {
+		if err := m.resolveHostname(); err != nil {
+			return false, fmt.Errorf("Could not resolve hostname: %v", err)
+		}
+	}
 	leader, err := m.Leader()
 	return m.MyLeader == leader, err
 }
+
+func (m *Marathon) resolveHostname() error {
+	hostname, err := hostname()
+	if err != nil {
+		return err
+	}
+	m.MyLeader = fmt.Sprintf("%s:8080", hostname)
+	log.WithField("Leader", m.MyLeader).Info("Marathon Leader mode set to resolved hostname")
+	return nil
+}
diff --git a/marathon/marathon_stub.go b/marathon/marathon_stub.go
index 97f272b..257cde8 100644
--- a/marathon/marathon_stub.go
+++ b/marathon/marathon_stub.go
@@ -11,6 +11,7 @@ type MarathonerStub struct {
 	AppsStub       []*apps.App
 	AppStub        map[apps.AppID]*apps.App
 	TasksStub      map[apps.AppID][]apps.Task
+	MyLeader       string
 	leader         string
 	interactionsMu sync.RWMutex
 	interactions   bool
@@ -46,8 +47,8 @@ func (m *MarathonerStub) EventStream([]string, int, int) (*Streamer, error) {
 	return &Streamer{}, nil
 }
 
-func (m *MarathonerStub) AmILeader() (bool, error) {
-	return false, nil
+func (m *MarathonerStub) IsLeader() (bool, error) {
+	return m.leader == m.MyLeader, nil
 }
 
 func (m *MarathonerStub) Interactions() bool {
@@ -62,9 +63,10 @@ func (m *MarathonerStub) noteInteraction() {
 	m.interactions = true
 }
 
-func MarathonerStubWithLeaderForApps(leader string, args ...*apps.App) *MarathonerStub {
+func MarathonerStubWithLeaderForApps(leader, myLeader string, args ...*apps.App) *MarathonerStub {
 	stub := MarathonerStubForApps(args...)
 	stub.leader = leader
+	stub.MyLeader = myLeader
 	return stub
 }
 
@@ -86,6 +88,7 @@ func MarathonerStubForApps(args ...*apps.App) *MarathonerStub {
 		AppsStub:  args,
 		AppStub:   appsMap,
 		TasksStub: tasksMap,
+		MyLeader:  "localhost:8080",
 		leader:    "localhost:8080",
 	}
 }
diff --git a/sse/sse.go b/sse/sse.go
index 94f3909..7291f10 100644
--- a/sse/sse.go
+++ b/sse/sse.go
@@ -53,7 +53,7 @@ func leaderGuard(s *marathon.Streamer, m marathon.Marathoner) chan<- events.Stop
 		for {
 			select {
 			case <-ticker:
-				if iAMLeader, err := m.AmILeader(); !iAMLeader && err != nil {
+				if iAMLeader, err := m.IsLeader(); !iAMLeader && err != nil {
 					// Leader changed, not revocerable.
 					s.Stop()
 					log.Error("Tearing down SSE stream, marathon leader changed.")
diff --git a/sync/sync.go b/sync/sync.go
index 899855a..b773433 100644
--- a/sync/sync.go
+++ b/sync/sync.go
@@ -2,7 +2,6 @@ package sync
 
 import (
 	"fmt"
-	"os"
 	"time"
 
 	log "github.com/Sirupsen/logrus"
@@ -94,31 +93,16 @@ func (s *Sync) shouldPerformSync() (bool, error) {
 		log.Debug("Forcing sync")
 		return true, nil
 	}
-	leader, err := s.marathon.Leader()
+	leading, err := s.marathon.IsLeader()
 	if err != nil {
 		return false, fmt.Errorf("Could not get Marathon leader: %v", err)
 	}
-	if s.config.Leader == "" {
-		if err = s.resolveHostname(); err != nil {
-			return false, fmt.Errorf("Could not resolve hostname: %v", err)
-		}
-	}
-	if leader != s.config.Leader {
-		log.WithField("Leader", leader).WithField("Node", s.config.Leader).Debug("Node is not a leader, skipping sync")
-		return false, nil
-	}
-	log.WithField("Node", s.config.Leader).Debug("Node has leadership")
-	return true, nil
-}
-
-func (s *Sync) resolveHostname() error {
-	hostname, err := os.Hostname()
-	if err != nil {
-		return err
+	if leading {
+		log.Debug("Node has leadership")
+		return true, nil
 	}
-	s.config.Leader = fmt.Sprintf("%s:8080", hostname)
-	log.WithField("Leader", s.config.Leader).Info("Marathon-consul sync leader mode set to resolved hostname")
-	return nil
+	log.Debug("Node is not a leader, skipping sync")
+	return false, nil
 }
 
 func (s *Sync) deregisterConsulServicesNotFoundInMarathon(marathonApps []*apps.App, services []*service.Service) (deregisterCount int, errorCount int) {


Test patch:
diff --git a/marathon/marathon_stub_test.go b/marathon/marathon_stub_test.go
index 5b5a0de..e9413ee 100644
--- a/marathon/marathon_stub_test.go
+++ b/marathon/marathon_stub_test.go
@@ -11,7 +11,7 @@ import (
 func TestMarathonStub(t *testing.T) {
 	t.Parallel()
 	// given
-	m := marathon.MarathonerStubWithLeaderForApps("some.host:1234", utils.ConsulApp("/test/app", 3))
+	m := marathon.MarathonerStubWithLeaderForApps("some.host:1234", "some.host:1234", utils.ConsulApp("/test/app", 3))
 	// then
 	assert.False(t, m.Interactions())
 	// when
diff --git a/marathon/marathon_test.go b/marathon/marathon_test.go
index 74976ac..0e29b40 100644
--- a/marathon/marathon_test.go
+++ b/marathon/marathon_test.go
@@ -17,7 +17,7 @@ func TestMarathon_AppsWhenMarathonReturnEmptyList(t *testing.T) {
 	defer server.Close()
 
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"})
 	m.client.Transport = transport
 	// when
 	apps, err := m.ConsulApps()
@@ -29,7 +29,7 @@ func TestMarathon_AppsWhenMarathonReturnEmptyList(t *testing.T) {
 func TestMarathon_AppsWhenConfigIsWrong(t *testing.T) {
 	t.Parallel()
 	// given
-	m, _ := New(Config{Location: "not::valid/location", Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: "not::valid/location", Protocol: "HTTP"})
 	// when
 	apps, err := m.ConsulApps()
 	//then
@@ -43,7 +43,7 @@ func TestMarathon_AppsWhenServerIsNotResponding(t *testing.T) {
 	}
 	t.Parallel()
 	// given
-	m, _ := New(Config{Location: "unknown:22", Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: "unknown:22", Protocol: "HTTP"})
 	// when
 	apps, err := m.ConsulApps()
 	//then
@@ -62,7 +62,7 @@ func TestMarathon_AppsWhenMarathonConnectionFailedShouldNotRetry(t *testing.T) {
 	defer server.Close()
 
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"})
 	m.client.Transport = transport
 	// when
 	apps, err := m.ConsulApps()
@@ -83,7 +83,7 @@ func TestMarathon_TasksWhenMarathonConnectionFailedShouldNotRetry(t *testing.T)
 	defer server.Close()
 
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"})
 	m.client.Transport = transport
 	// when
 	tasks, err := m.Tasks("/app/id")
@@ -104,7 +104,7 @@ func TestMarathon_AppWhenMarathonConnectionFailedShouldNotRetry(t *testing.T) {
 	defer server.Close()
 
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"})
 	m.client.Transport = transport
 	// when
 	app, err := m.App("/app/id")
@@ -121,7 +121,7 @@ func TestMarathon_AppsWhenMarathonReturnEmptyResponse(t *testing.T) {
 	defer server.Close()
 
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"})
 	m.client.Transport = transport
 	// when
 	apps, err := m.ConsulApps()
@@ -137,7 +137,7 @@ func TestMarathon_AppsWhenMarathonReturnMalformedJsonResponse(t *testing.T) {
 	defer server.Close()
 
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"})
 	m.client.Transport = transport
 	// when
 	app, err := m.App("/test/app")
@@ -153,7 +153,7 @@ func TestMarathon_AppWhenMarathonReturnEmptyApp(t *testing.T) {
 	defer server.Close()
 
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"})
 	m.client.Transport = transport
 	// when
 	app, err := m.App("/test/app")
@@ -169,7 +169,7 @@ func TestMarathon_AppWhenMarathonReturnEmptyResponse(t *testing.T) {
 	defer server.Close()
 
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"})
 	m.client.Transport = transport
 	// when
 	app, err := m.App("/test/app")
@@ -185,7 +185,7 @@ func TestMarathon_AppWhenMarathonReturnMalformedJsonResponse(t *testing.T) {
 	defer server.Close()
 
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"})
 	m.client.Transport = transport
 	// when
 	apps, err := m.ConsulApps()
@@ -207,7 +207,7 @@ func TestMarathon_TasksWhenMarathonReturnEmptyList(t *testing.T) {
 	}]}`)
 	defer server.Close()
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"})
 	m.client.Transport = transport
 	// when
 	tasks, err := m.Tasks("//test/app")
@@ -222,7 +222,7 @@ func TestMarathon_TasksWhenMarathonReturnEmptyResponse(t *testing.T) {
 	server, transport := stubServer("/v2/apps/test/app/tasks", ``)
 	defer server.Close()
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"})
 	m.client.Transport = transport
 	// when
 	tasks, err := m.Tasks("/test/app")
@@ -237,7 +237,7 @@ func TestMarathon_TasksWhenMarathonReturnMalformedJsonResponse(t *testing.T) {
 	server, transport := stubServer("/v2/apps/test/app/tasks", ``)
 	defer server.Close()
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"})
 	m.client.Transport = transport
 	// when
 	tasks, err := m.Tasks("/test/app")
@@ -251,7 +251,7 @@ func TestConfig_transport(t *testing.T) {
 	// given
 	config := Config{VerifySsl: false}
 	// when
-	marathon, _ := New(config, "")
+	marathon, _ := New(config)
 	// then
 	transport, ok := marathon.client.Transport.(*http.Transport)
 	assert.True(t, ok)
@@ -263,7 +263,7 @@ func TestUrl_WithoutAuth(t *testing.T) {
 	// given
 	config := Config{Location: "localhost:8080", Protocol: "http"}
 	// when
-	m, _ := New(config, "")
+	m, _ := New(config)
 	// then
 	assert.Equal(t, "http://localhost:8080/v2/apps", m.url("/v2/apps"))
 }
@@ -273,7 +273,7 @@ func TestUrl_WithAuth(t *testing.T) {
 	// given
 	config := Config{Location: "localhost:8080", Protocol: "http", Username: "peter", Password: "parker"}
 	// when
-	m, _ := New(config, "")
+	m, _ := New(config)
 	// then
 	assert.Equal(t, "http://peter:parker@localhost:8080/v2/apps", m.url("/v2/apps"))
 }
@@ -285,7 +285,7 @@ func TestLeader_SuccessfulResponse(t *testing.T) {
 	server, transport := stubServer("/v2/leader", `{"leader": "some.leader.host:8081"}`)
 	defer server.Close()
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"})
 	m.client.Transport = transport
 
 	// when
@@ -303,7 +303,7 @@ func TestLeader_ErrorOnMalformedJsonResponse(t *testing.T) {
 	server, transport := stubServer("/v2/leader", "{")
 	defer server.Close()
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"})
 	m.client.Transport = transport
 
 	// when
@@ -325,7 +325,7 @@ func TestLeader_NotRetryOnFailingResponse(t *testing.T) {
 	})
 	defer server.Close()
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"})
 	m.client.Transport = transport
 
 	// when
@@ -344,7 +344,7 @@ func TestLeaderPoll_PassingRunningOnLeader(t *testing.T) {
 	server, transport := stubServer("/v2/leader", `{"leader": "this.leader:8080"}`)
 	defer server.Close()
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "this.leader:8080")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP", Leader: "this.leader:8080"})
 	m.client.Transport = transport
 
 	// when
@@ -354,42 +354,65 @@ func TestLeaderPoll_PassingRunningOnLeader(t *testing.T) {
 	assert.NoError(t, err)
 }
 
-func TestAMILeader_PassingRunningOnLeader(t *testing.T) {
+func TestIsLeader_PassingRunningOnLeader(t *testing.T) {
 	t.Parallel()
 
 	// given
 	server, transport := stubServer("/v2/leader", `{"leader": "this.leader:8080"}`)
 	defer server.Close()
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "this.leader:8080")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP", Leader: "this.leader:8080"})
 	m.client.Transport = transport
 
 	// when
-	leading, err := m.AmILeader()
+	leading, err := m.IsLeader()
 
 	//then
 	assert.True(t, leading)
 	assert.NoError(t, err)
 }
 
-func TestAMILeader_NotPassingNotRunningOnLeader(t *testing.T) {
+func TestIsLeader_NotPassingNotRunningOnLeader(t *testing.T) {
 	t.Parallel()
 
 	// given
 	server, transport := stubServer("/v2/leader", `{"leader": "other.leader:8080"}`)
 	defer server.Close()
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "this.leader:8080")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP", Leader: "this.leader:8080"})
 	m.client.Transport = transport
 
 	// when
-	leading, err := m.AmILeader()
+	leading, err := m.IsLeader()
 
 	//then
 	assert.False(t, leading)
 	assert.NoError(t, err)
 }
 
+func TestIsLeader_NotPassingMyLeaderIsEmptyStringShouldFillItWithLocalHostname(t *testing.T) {
+	t.Parallel()
+
+	// given
+	server, transport := stubServer("/v2/leader", `{"leader": "other.leader:8080"}`)
+	defer server.Close()
+	url, _ := url.Parse(server.URL)
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP", Leader: ""})
+	m.client.Transport = transport
+	myHostname := "localhost"
+	hostname = func() (string, error) {
+		return myHostname, nil
+	}
+
+	// when
+	leading, err := m.IsLeader()
+
+	//then
+	assert.Equal(t, m.MyLeader, fmt.Sprintf("%s:8080", myHostname))
+	assert.False(t, leading)
+	assert.NoError(t, err)
+}
+
 func TestEventStream_PassingStreamerCreated(t *testing.T) {
 	t.Parallel()
 
@@ -397,7 +420,7 @@ func TestEventStream_PassingStreamerCreated(t *testing.T) {
 	server, transport := stubServer("/v2/leader", `{"leader": "this.leader:8080"}`)
 	defer server.Close()
 	url, _ := url.Parse(server.URL)
-	m, _ := New(Config{Location: url.Host, Protocol: "HTTP"}, "this.leader:8080")
+	m, _ := New(Config{Location: url.Host, Protocol: "HTTP", Leader: "this.leader:8080"})
 	m.client.Transport = transport
 
 	// when
@@ -412,7 +435,7 @@ func TestUrlWithQuery_NoProxyMarathon(t *testing.T) {
 	t.Parallel()
 
 	// given
-	m, _ := New(Config{Location: "localhost:8080", Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: "localhost:8080", Protocol: "HTTP"})
 	// when
 	path := m.urlWithQuery("/testpath", params{})
 
@@ -424,7 +447,7 @@ func TestUrlWithQuery_ProxyMarathon(t *testing.T) {
 	t.Parallel()
 
 	// given
-	m, _ := New(Config{Location: "localhost:8080/proxy/url/segments", Protocol: "HTTP"}, "")
+	m, _ := New(Config{Location: "localhost:8080/proxy/url/segments", Protocol: "HTTP"})
 	// when
 	path := m.urlWithQuery("/testpath", params{})
 
diff --git a/sync/error_marathon_stub_test.go b/sync/error_marathon_stub_test.go
index ed06fec..fbab4ec 100644
--- a/sync/error_marathon_stub_test.go
+++ b/sync/error_marathon_stub_test.go
@@ -30,6 +30,6 @@ func (m errorMarathon) EventStream([]string, int, int) (*marathon.Streamer, erro
 	return &marathon.Streamer{}, errors.New("Error")
 }
 
-func (m errorMarathon) AmILeader() (bool, error) {
+func (m errorMarathon) IsLeader() (bool, error) {
 	return false, errors.New("Error")
 }
diff --git a/sync/sync_test.go b/sync/sync_test.go
index 0bb8d46..445c49b 100644
--- a/sync/sync_test.go
+++ b/sync/sync_test.go
@@ -2,7 +2,6 @@ package sync
 
 import (
 	"fmt"
-	"os"
 	"sync"
 	"testing"
 	"time"
@@ -23,12 +22,11 @@ func TestSyncJob_ShouldSyncOnLeadership(t *testing.T) {
 	t.Parallel()
 	// given
 	app := ConsulApp("app1", 1)
-	marathon := marathon.MarathonerStubWithLeaderForApps("current.leader:8080", app)
+	marathon := marathon.MarathonerStubWithLeaderForApps("current.leader:8080", "current.leader:8080", app)
 	services := newConsulServicesMock()
 	sync := New(Config{
 		Enabled:  true,
 		Interval: timeutil.Interval{Duration: 10 * time.Millisecond},
-		Leader:   "current.leader:8080",
 	}, marathon, services, noopSyncStartedListener)
 
 	// when
@@ -43,12 +41,11 @@ func TestSyncJob_ShouldNotSyncWhenDisabled(t *testing.T) {
 	t.Parallel()
 	// given
 	app := ConsulApp("app1", 1)
-	marathon := marathon.MarathonerStubWithLeaderForApps("current.leader:8080", app)
+	marathon := marathon.MarathonerStubWithLeaderForApps("current.leader:8080", "current.leader:8080", app)
 	services := newConsulServicesMock()
 	sync := New(Config{
 		Enabled:  false,
 		Interval: timeutil.Interval{Duration: 10 * time.Millisecond},
-		Leader:   "current.leader:8080",
 	}, marathon, services, noopSyncStartedListener)
 
 	// when
@@ -59,31 +56,11 @@ func TestSyncJob_ShouldNotSyncWhenDisabled(t *testing.T) {
 	assert.Equal(t, 0, services.RegistrationsCount(app.Tasks[0].ID.String()))
 }
 
-func TestSyncJob_ShouldDefaultLeaderConfigurationToResolvedHostname(t *testing.T) {
-	t.Parallel()
-	// given
-	hostname, _ := os.Hostname()
-	app := ConsulApp("app1", 1)
-	marathon := marathon.MarathonerStubWithLeaderForApps(fmt.Sprintf("%s:8080", hostname), app)
-	services := newConsulServicesMock()
-	sync := New(Config{
-		Enabled:  true,
-		Interval: timeutil.Interval{Duration: 10 * time.Millisecond},
-	}, marathon, services, noopSyncStartedListener)
-
-	// when
-	sync.StartSyncServicesJob()
-
-	// then
-	<-time.After(15 * time.Millisecond)
-	assert.Equal(t, 2, services.RegistrationsCount(app.Tasks[0].ID.String()))
-}
-
 func TestSyncServices_ShouldNotSyncOnNoForceNorLeaderSpecified(t *testing.T) {
 	t.Parallel()
 	// given
 	app := ConsulApp("app1", 1)
-	marathon := marathon.MarathonerStubWithLeaderForApps("localhost:8080", app)
+	marathon := marathon.MarathonerStubWithLeaderForApps("localhost:8080", "", app)
 	services := newConsulServicesMock()
 	sync := New(Config{}, marathon, services, noopSyncStartedListener)
 
@@ -98,9 +75,9 @@ func TestSyncServices_ShouldNotSyncOnNoLeadership(t *testing.T) {
 	t.Parallel()
 	// given
 	app := ConsulApp("app1", 1)
-	marathon := marathon.MarathonerStubWithLeaderForApps("leader:8080", app)
+	marathon := marathon.MarathonerStubWithLeaderForApps("leader:8080", "different.node:8090", app)
 	services := newConsulServicesMock()
-	sync := New(Config{Leader: "different.node:8090"}, marathon, services, noopSyncStartedListener)
+	sync := New(Config{}, marathon, services, noopSyncStartedListener)
 
 	// when
 	err := sync.SyncServices()
@@ -114,9 +91,9 @@ func TestSyncServices_ShouldSyncOnForceWithoutLeadership(t *testing.T) {
 	t.Parallel()
 	// given
 	app := ConsulApp("app1", 1)
-	marathon := marathon.MarathonerStubWithLeaderForApps("leader:8080", app)
+	marathon := marathon.MarathonerStubWithLeaderForApps("leader:8080", "different.node:8090", app)
 	services := newConsulServicesMock()
-	sync := New(Config{Leader: "different.node:8090", Force: true}, marathon, services, noopSyncStartedListener)
+	sync := New(Config{Force: true}, marathon, services, noopSyncStartedListener)
 
 	// when
 	err := sync.SyncServices()
@@ -479,8 +456,8 @@ func TestSync_AddingAgentsFromMarathonTasks(t *testing.T) {
 	app := ConsulApp("serviceA", 2)
 	app.Tasks[0].Host = consulServer.Config.Bind
 	app.Tasks[1].Host = consulServer.Config.Bind
-	marathon := marathon.MarathonerStubWithLeaderForApps("localhost:8080", app)
-	sync := New(Config{Leader: "localhost:8080"}, marathon, consulInstance, consulInstance.AddAgentsFromApps)
+	marathon := marathon.MarathonerStubWithLeaderForApps("localhost:8080", "localhost:8080", app)
+	sync := New(Config{}, marathon, consulInstance, consulInstance.AddAgentsFromApps)
 
 	// when
 	err := sync.SyncServices()