Merge pull request #4 from chuckyz/allow_consul_stale

Allow stale requests from Consul

controller/consul.go

@@ -5,20 +5,25 @@ import (
 	"fmt"
 	"github.com/golang/glog"
 	"net/http"
+	"net/url"
 	"os"
 	"strings"
+	"time"
 )
 
 const (
-	consulNodeEnv  = "CONSUL_NODE"
-	matchTag       = "enable_gocast"
-	nodeUrl        = "/catalog/node"
-	healthCheckurl = "/health/checks"
+	consulNodeEnv        = "CONSUL_NODE"
+	allowStale           = "CONSUL_STALE"
+	matchTag             = "enable_gocast"
+	nodeURL              = "/catalog/node"
+	remoteHealthCheckurl = "/health/checks"
+	localHealthCheckurl  = "/agent/checks"
 )
 
 type ConsulMon struct {
-	addr string
-	node string
+	addr   string
+	node   string
+	client *http.Client
 }
 
 type ConsulServiceData struct {
@@ -43,13 +48,17 @@ func NewConsulMon(addr string) (*ConsulMon, error) {
 	if node == "" {
 		return nil, fmt.Errorf("%s env variable not set", consulNodeEnv)
 	}
-	return &ConsulMon{addr: addr, node: node}, nil
+	return &ConsulMon{addr: addr, node: node, client: &http.Client{Timeout: 10 * time.Second}}, nil
 }
 
 func (c *ConsulMon) queryServices() ([]*App, error) {
 	var apps []*App
-	addr := c.addr + fmt.Sprintf("%s/%s", nodeUrl, c.node)
-	resp, err := http.Get(addr)
+	var stale string
+	if os.Getenv(allowStale) == "true" {
+		stale = "stale"
+	}
+	addr := c.addr + fmt.Sprintf("%s/%s?%s", nodeURL, c.node, stale)
+	resp, err := c.client.Get(addr)
 	if err != nil {
 		return apps, err
 	}
@@ -97,10 +106,43 @@ func (c *ConsulMon) queryServices() ([]*App, error) {
 	return apps, nil
 }
 
-func (c *ConsulMon) healthCheck(service string) (bool, error) {
-	addr := c.addr + fmt.Sprintf("%s/%s", healthCheckurl, service)
-	resp, err := http.Get(addr)
+// healthCheckLocal queries a node's local consul agent to perform service healthchecks
+// This is the underlying api call: https://www.consul.io/api/agent/check.html
+func (c *ConsulMon) healthCheckLocal(service string) (bool, error) {
+	params := url.Values{}
+	params.Add("filter", "enable_gocast in ServiceTags")
+	addr := c.addr + fmt.Sprintf("%s?%s", localHealthCheckurl, params.Encode())
+	resp, err := c.client.Get(addr)
 	if err != nil {
+		glog.V(2).Infof("Error getting %s with %s", addr, err)
+		return false, err
+	}
+	defer resp.Body.Close()
+	var services map[string]interface{}
+	if err := json.NewDecoder(resp.Body).Decode(&services); err != nil {
+		return false, err
+	}
+	for _, sInfo := range services {
+		serviceInfo := sInfo.(map[string]interface{})
+		if serviceInfo["ServiceName"].(string) == service {
+			status := serviceInfo["Status"].(string)
+			if status == "passing" {
+				return true, nil
+			}
+			glog.V(2).Infof("Consul local healthcheck returned %s status", status)
+			return false, nil
+		}
+	}
+	return false, fmt.Errorf("No local healcheck info found for service %s on node %s in consul", service, c.node)
+}
+
+// healthCheckRemote queries the consul cluster's healthcheck endpoint to perform service healthchecks
+// This is the underlying api call: https://www.consul.io/api/health.html
+func (c *ConsulMon) healthCheckRemote(service string) (bool, error) {
+	addr := c.addr + fmt.Sprintf("%s/%s", remoteHealthCheckurl, service)
+	resp, err := c.client.Get(addr)
+	if err != nil {
+		glog.V(2).Infof("Error getting %s with %s", addr, err)
 		return false, err
 	}
 	defer resp.Body.Close()
@@ -108,16 +150,26 @@ func (c *ConsulMon) healthCheck(service string) (bool, error) {
 	if err := json.NewDecoder(resp.Body).Decode(&data); err != nil {
 		return false, err
 	}
+
 	for _, nodeInfo := range data {
 		n := nodeInfo.(map[string]interface{})
-		if n["Node"] == c.node {
+		if n["Node"].(string) == c.node {
 			if n["Status"].(string) == "passing" {
 				return true, nil
-			} else {
-				glog.V(2).Infof("Consul Healthcheck returned %s status", n["Status"].(string))
-				return false, nil
 			}
+			glog.V(2).Infof("Consul healthcheck returned %s status", n["Status"].(string))
+			return false, nil
 		}
 	}
 	return false, fmt.Errorf("No healcheck info found for node %s in consul", c.node)
 }
+
+// healthCheck determines if we should use the local agent
+// If the address contains "localhost", then it presumes that the local agent is to be used.
+func (c *ConsulMon) healthCheck(service string) (bool, error) {
+	usingLocalAgent := strings.Contains(c.addr, "localhost")
+	if usingLocalAgent {
+		return c.healthCheckLocal(service)
+	}
+	return c.healthCheckRemote(service)
+}

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/eapache/queue v1.1.0
 	github.com/fsnotify/fsnotify v1.4.7
 	github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b
-	github.com/golang/protobuf v0.0.0-20181022004443-7be363195599
+	github.com/golang/protobuf v1.2.0
 	github.com/hashicorp/hcl v1.0.0
 	github.com/influxdata/influxdb v1.6.4
 	github.com/konsorten/go-windows-terminal-sequences v1.0.1