Local Agent used for service list reads (#276)

If defined, the local consul agent is used to read list of services.
Otherwise, a random agent is picked, but only from the cache of agents
already syncing registrations.
This change prevents situations, where contacting a random agent in the
cluster fails, because it has no ACL-DC configured.

Author: Dariusz Jedrzejczyk

README.md

@@ -191,7 +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-local-agent-host     |                 | Consul Agent hostname or IP that should be used for startup sync and service listing operations
 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

consul/agents.go

@@ -10,20 +10,21 @@ import (
 	log "github.com/Sirupsen/logrus"
 	"github.com/allegro/marathon-consul/metrics"
 	"github.com/allegro/marathon-consul/utils"
-	consulapi "github.com/hashicorp/consul/api"
 )
 
 type Agents interface {
-	GetAgent(agentAddress string) (agent *consulapi.Client, err error)
+	GetAgent(agentAddress string) (agent *Agent, err error)
+	GetLocalAgent() (agent *Agent, err error)
 	GetAnyAgent() (agent *Agent, err error)
 	RemoveAgent(agentAddress string)
 }
 
 type ConcurrentAgents struct {
-	agents map[string]*Agent
-	config *Config
-	lock   sync.Mutex
-	client *http.Client
+	localAgent *Agent
+	agents     map[string]*Agent
+	config     *Config
+	lock       sync.Mutex
+	client     *http.Client
 }
 
 func NewAgents(config *Config) *ConcurrentAgents {
@@ -48,21 +49,7 @@ func NewAgents(config *Config) *ConcurrentAgents {
 			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.")
-			}
-		}
+		agents.localAgent = agent
 	}
 	return agents
 }
@@ -78,6 +65,13 @@ func (a *ConcurrentAgents) GetAnyAgent() (*Agent, error) {
 	return nil, errors.New("No Consul client available in agents cache")
 }
 
+func (a *ConcurrentAgents) GetLocalAgent() (*Agent, error) {
+	if a.localAgent == nil {
+		return nil, errors.New("No local consul agent defined")
+	}
+	return a.localAgent, nil
+}
+
 func (a *ConcurrentAgents) getRandomAgentIPAddress() string {
 	ipAddresses := []string{}
 	for ipAddress := range a.agents {
@@ -101,7 +95,7 @@ func (a *ConcurrentAgents) RemoveAgent(agentAddress string) {
 	}
 }
 
-func (a *ConcurrentAgents) GetAgent(agentAddress string) (*consulapi.Client, error) {
+func (a *ConcurrentAgents) GetAgent(agentAddress string) (*Agent, error) {
 	a.lock.Lock()
 	defer a.lock.Unlock()
 
@@ -112,7 +106,7 @@ func (a *ConcurrentAgents) GetAgent(agentAddress string) (*consulapi.Client, err
 	ipAddress := IP.String()
 
 	if agent, ok := a.agents[ipAddress]; ok {
-		return agent.Client, nil
+		return agent, nil
 	}
 
 	newAgent, err := a.createAgent(ipAddress)
@@ -121,7 +115,7 @@ func (a *ConcurrentAgents) GetAgent(agentAddress string) (*consulapi.Client, err
 	}
 	a.addAgent(ipAddress, newAgent)
 
-	return newAgent.Client, nil
+	return newAgent, nil
 }
 
 func (a *ConcurrentAgents) addAgent(agentHost string, agent *Agent) {

consul/consul.go

@@ -38,14 +38,18 @@ func (c *Consul) GetServices(name string) ([]*service.Service, error) {
 
 func (c *Consul) getServicesUsingProviderWithRetriesOnAgentFailure(provide ServicesProvider) ([]*service.Service, error) {
 	for retry := uint32(0); retry <= c.config.RequestRetries; retry++ {
-		agent, err := c.agents.GetAnyAgent()
+		localAgent, err := c.agents.GetLocalAgent()
+		agent := localAgent
+		if err != nil {
+			agent, err = c.agents.GetAnyAgent()
+		}
 		if err != nil {
 			return nil, err
 		}
 		if services, err := provide(agent.Client); err != nil {
 			log.WithError(err).WithField("Address", agent.IP).
-				Error("An error occurred getting services from Consul, retrying with another agent")
-			if failures := agent.IncFailures(); failures > c.config.AgentFailuresTolerance {
+				Error("An error occurred getting services from Consul, retrying locally or with another agent")
+			if failures := agent.IncFailures(); agent != localAgent && failures > c.config.AgentFailuresTolerance {
 				log.WithField("Address", agent.IP).WithField("Failures", failures).
 					Warn("Removing agent due to too many failures")
 				c.agents.RemoveAgent(agent.IP)
@@ -181,6 +185,7 @@ func (c *Consul) register(service *consulapi.AgentServiceRegistration) error {
 	if err != nil {
 		return err
 	}
+	client := agent.Client
 	fields := log.Fields{
 		"Name":              service.Name,
 		"Id":                service.ID,
@@ -191,7 +196,7 @@ func (c *Consul) register(service *consulapi.AgentServiceRegistration) error {
 	}
 	log.WithFields(fields).Info("Registering")
 
-	err = agent.Agent().ServiceRegister(service)
+	err = client.Agent().ServiceRegister(service)
 	if err != nil {
 		log.WithError(err).WithFields(fields).Error("Unable to register")
 	}
@@ -265,10 +270,10 @@ func (c *Consul) deregister(toDeregister *service.Service) error {
 	if err != nil {
 		return err
 	}
-
+	client := agent.Client
 	log.WithField("Id", toDeregister.ID).WithField("Address", toDeregister.RegisteringAgentAddress).Info("Deregistering")
 
-	err = agent.Agent().ServiceDeregister(toDeregister.ID.String())
+	err = client.Agent().ServiceDeregister(toDeregister.ID.String())
 	if err != nil {
 		log.WithError(err).WithField("Id", toDeregister.ID).WithField("Address", toDeregister.RegisteringAgentAddress).Error("Unable to deregister")
 	}