Allow to configure startup consul agent (#222)

In some cases consul agents cache could be empty.
This chagne allow providing consul agent hostname to
initially populate the cache.

Fixes: #191

README.md

@@ -176,7 +176,10 @@ for more details.
       node leadership detection (should be set to cluster-wide node name), while `marathon-location` is used for
       connection purpose (may be set to `localhost`)
     - On every node, `sync-force` parameter should be set to `true`
-
+- If marathon-consul fails on startup sync and you see following error
+`"Can't get Consul services: No Consul client available in agents cache"`
+ it may be caused by empty consul agents cache. If this occurs try configuring
+ `--consul-local-agent-host` to Consul Master or Consul agent.
 
 ### Options
 
@@ -188,6 +191,7 @@ consul-auth-password        |                 | The basic authentication passwor
 consul-auth-username        |                 | The basic authentication username
 consul-enable-tag-override  | `false`         | Disable the anti-entropy feature for all services
 consul-ignored-healthchecks |                 | A comma separated blacklist of Marathon health check types that will not be migrated to Consul, e.g. command,tcp
+consul-local-agent-host     |                 | Consul Agent hostname or IP that should be used for startup sync
 consul-name-separator       | `.`             | Separator used to create default service name for Consul
 consul-get-services-retry   | `3`             | Number of retries on failure when performing requests to Consul. Each retry uses different cached agent
 consul-max-agent-failures   | `3`             | Max number of consecutive request failures for agent before removal from cache

config/config.go

@@ -78,6 +78,7 @@ func (config *Config) parseFlags() {
 	flag.StringVar(&config.Consul.ConsulNameSeparator, "consul-name-separator", ".", "Separator used to create default service name for Consul")
 	flag.StringVar(&config.Consul.IgnoredHealthChecks, "consul-ignored-healthchecks", "", "A comma separated blacklist of Marathon health check types that will not be migrated to Consul, e.g. command,tcp")
 	flag.BoolVar(&config.Consul.EnableTagOverride, "consul-enable-tag-override", false, "Disable the anti-entropy feature for all services")
+	flag.StringVar(&config.Consul.LocalAgentHost, "consul-local-agent-host", "", "Consul Agent hostname or IP that should be used for startup sync")
 
 	// Web
 	flag.StringVar(&config.Web.Listen, "listen", ":4000", "Accept connections at this address")

config/config_test.go

@@ -104,6 +104,7 @@ func TestConfig_ShouldBeMergedWithFileDefaultsAndFlags(t *testing.T) {
 			AgentFailuresTolerance: 3,
 			ConsulNameSeparator:    ".",
 			EnableTagOverride:      false,
+			LocalAgentHost:         "",
 		},
 		Web: web.Config{
 			Listen:       ":4000",

consul/agents.go

@@ -36,11 +36,35 @@ func NewAgents(config *Config) *ConcurrentAgents {
 		},
 		Timeout: config.Timeout.Duration,
 	}
-	return &ConcurrentAgents{
+
+	agents := &ConcurrentAgents{
 		agents: make(map[string]*Agent),
 		config: config,
 		client: client,
 	}
+	if config.LocalAgentHost != "" {
+		agent, err := agents.GetAgent(config.LocalAgentHost)
+		if err != nil {
+			log.WithError(err).WithField("agent", config.LocalAgentHost).Fatal(
+				"Cannot connect with consul agent. Check if configuration is valid.")
+		}
+
+		// Get all agents from current DC and store them in cache
+		nodes, _, err := agent.Catalog().Nodes(nil)
+		if err != nil {
+			log.WithError(err).WithField("agent", config.LocalAgentHost).Warn(
+				"Cannot obtain agents from local consul agent.")
+			return agents
+		}
+		for _, node := range nodes {
+			_, err := agents.GetAgent(node.Address)
+			if err != nil {
+				log.WithError(err).WithField("agent", node.Address).Warn(
+					"Cannot connect with consul agent. Check if configuration is valid.")
+			}
+		}
+	}
+	return agents
 }
 
 func (a *ConcurrentAgents) GetAnyAgent() (*Agent, error) {

consul/agents_test.go

@@ -20,6 +20,16 @@ func TestGetAgent(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestPopulateAgentsCacheWithLocalAgent(t *testing.T) {
+	t.Parallel()
+	// when
+	agents := NewAgents(&Config{LocalAgentHost: "127.0.0.1"})
+
+	// then
+	assert.Len(t, agents.agents, 1)
+	assert.NotNil(t, agents.agents["127.0.0.1"])
+}
+
 func TestGetAnyAgent_SingleAgentAvailable(t *testing.T) {
 	t.Parallel()
 	// given

consul/config.go

@@ -17,6 +17,7 @@ type Config struct {
 	ConsulNameSeparator    string
 	IgnoredHealthChecks    string
 	EnableTagOverride      bool
+	LocalAgentHost         string
 }
 
 type Auth struct {

consul/consul_test_server.go

@@ -62,7 +62,16 @@ func ClientAtServer(server *testutil.TestServer) *Consul {
 }
 
 func FailingClient() *Consul {
-	return consulClientAtAddress("127.5.5.5", 5555)
+	host, port := "192.0.2.5", 5555
+	config := Config{
+		Port:                fmt.Sprintf("%d", port),
+		ConsulNameSeparator: ".",
+		EnableTagOverride:   true,
+	}
+	consul := New(config)
+	// initialize the agents cache with a single client pointing at provided location
+	consul.AddAgent(host)
+	return consul
 }
 
 func consulClientAtAddress(host string, port int) *Consul {
@@ -71,6 +80,7 @@ func consulClientAtAddress(host string, port int) *Consul {
 		Port:                fmt.Sprintf("%d", port),
 		ConsulNameSeparator: ".",
 		EnableTagOverride:   true,
+		LocalAgentHost:      host,
 	}
 	consul := New(config)
 	// initialize the agents cache with a single client pointing at provided location

debian/config.json

@@ -17,7 +17,8 @@
     "AgentFailuresTolerance": 3,
     "RequestRetries": 5,
     "IgnoredHealthChecks": "",
-    "EnableTagOverride": false
+    "EnableTagOverride": false,
+    "LocalAgentHost": ""
   },
   "Web": {
     "Listen": ":4000",