Improve Consul routing provider

Author: fwkz

README.md

@@ -36,6 +36,11 @@ Exposing dynamic backends on the static frontend ports is the bread-and-butter o
       - [routing.http.poll\_interval `duration`](#routinghttppoll_interval-duration)
       - [routing.http.poll\_timeout `duration`](#routinghttppoll_timeout-duration)
       - [routing.consul.address `string`](#routingconsuladdress-string)
+      - [routing.consul.token `string`](#routingconsultoken-string)
+      - [routing.consul.datacenter `string`](#routingconsuldatacenter-string)
+      - [routing.consul.namespace `string`](#routingconsulnamespace-string)
+      - [routing.consul.wait `duration`](#routingconsulwait-duration)
+      - [routing.consul.consistency\_mode `string`](#routingconsulconsistency_mode-string)
     - [Formats](#formats)
       - [TOML](#toml)
       - [YAML](#yaml)
@@ -277,12 +282,15 @@ Value of Go's `runtime.GOMAXPROCS()`
 
 _default: equals to `runtime.NumCPU()`_
 
+---
 #### routing.file.path `string`
 Location of [routing configuration file](#schema).
 #### routing.file.watch `bool`
 Subscribe to changes made to the routing configuration file which would give you the full power of dill's dynamic routing capabilities.
 
 _default: `true`_
+
+---
 #### routing.http.endpoint `string`
 Endpoint which [http provider](#http) will poll for routing configuration
 #### routing.http.poll_interval `duration`
@@ -293,8 +301,30 @@ _default: `5s`_
 Maximum time  [http provider](#http) will wait when fetching routing configuration
 
 _default: `5s`_
+
+---
 #### routing.consul.address `string`
 Consul address from which `dill` will fetch the updates and build the routing table.
+#### routing.consul.token `string`
+Token giving access to Consul API. Required ACLs `node:read,service:read`
+
+_Optional_
+#### routing.consul.datacenter `string`
+Defines what datacenter will be queried when building routing table.
+
+_Optional. If not provided `dill` uses Consul defaults._
+#### routing.consul.namespace `string`
+Defines what namespace will be queried when building routing table. Namespaces are available only for Consul Enterprise users. 
+
+_Optional. If not provided `dill` uses Consul defaults._
+#### routing.consul.wait `duration`
+Defines how long [blocking API query](https://developer.hashicorp.com/consul/api-docs/features/blocking) will wait for a potential change using long polling.
+
+_Optional. If not provided `dill` uses Consul defaults._
+#### routing.consul.consistency_mode `string`
+Defines what [consistency mode](https://developer.hashicorp.com/consul/api-docs/features/consistency) to use when `dill` fetches the updates.
+
+_Optional. Allowed values: `stale`, `consistent`, `leader`. If not provided `dill` uses Consul defaults._
 ### Formats
 Configuration is powered by [Viper](https://github.com/spf13/viper) so it's possible to use format that suits you best.
 

pkg/routing/consul/client.go

@@ -0,0 +1,50 @@
+package consul
+
+import (
+	"net/http"
+
+	log "github.com/sirupsen/logrus"
+
+	"golang.org/x/exp/slices"
+)
+
+var validConsistencyModes = []string{"stale", "consistent", "leader"}
+
+type consulConfig struct {
+	Address         string
+	Token           string
+	Datacenter      string
+	Namespace       string
+	Wait            string
+	ConsistencyMode string `mapstructure:"consistency_mode"`
+}
+
+func (c *consulConfig) Validate() {
+	if c.ConsistencyMode != "" && !slices.Contains(validConsistencyModes, c.ConsistencyMode) {
+		log.Fatal("Invalid Consul's consistency mode")
+	}
+}
+
+type httpClient struct {
+	client http.Client
+	config *consulConfig
+}
+
+func (c *httpClient) Do(req *http.Request) (*http.Response, error) {
+	if c.config.Token != "" {
+		req.Header.Add("X-Consul-Token", c.config.Token)
+	}
+
+	q := req.URL.Query()
+	if c.config.Datacenter != "" {
+		q.Add("dc", c.config.Datacenter)
+	}
+	if c.config.Namespace != "" {
+		q.Add("ns", c.config.Namespace)
+	}
+	if c.config.ConsistencyMode != "" {
+		q.Add(c.config.ConsistencyMode, "")
+	}
+	req.URL.RawQuery = q.Encode()
+	return c.client.Do(req)
+}

pkg/routing/consul/consul.go

@@ -5,10 +5,9 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"net/url"
 	"strconv"
 	"strings"
-
-	"github.com/spf13/viper"
 )
 
 type service struct {
@@ -44,31 +43,36 @@ func (s *service) Proxy() string {
 	return ""
 }
 
-func fetchHealthyServices(index int) ([]string, int, error) {
-	// TODO allow for stale reads
+func fetchHealthyServices(index int, consulClient *httpClient) ([]string, int, error) {
 	req, err := http.NewRequest(
 		"GET",
-		viper.GetString("consul.address")+"/v1/health/state/passing",
+		consulClient.config.Address+"/v1/health/state/passing",
 		nil,
 	)
 	if err != nil {
 		return nil, 0, err
 	}
-	q := req.URL.Query()
+	q := url.Values{}
 	q.Add("filter", "ServiceTags contains `dill`")
 	if index <= 0 {
 		index = 1
 	}
 	q.Add("index", strconv.Itoa(index))
+	if consulClient.config.Wait != "" {
+		q.Add("wait", consulClient.config.Wait)
+	}
 	req.URL.RawQuery = q.Encode()
-
-	c := &http.Client{}
-	res, err := c.Do(req)
+	res, err := consulClient.Do(req)
 	if err != nil {
 		return nil, 0, err
 	}
 	defer res.Body.Close()
 	data, _ := io.ReadAll(res.Body)
+
+	if res.StatusCode != 200 {
+		return nil, 0, fmt.Errorf("%d: %s", res.StatusCode, data)
+	}
+
 	var healthyServices []struct {
 		Name string `json:"ServiceName"`
 	}
@@ -93,20 +97,30 @@ func fetchHealthyServices(index int) ([]string, int, error) {
 	return unique, newIndex, nil
 }
 
-func fetchServiceDetails(name string) ([]service, error) {
-	res, err := http.Get(
-		fmt.Sprintf(
-			"%s/v1/health/service/%s?passing=true",
-			viper.GetString("consul.address"),
-			name,
-		),
+func fetchServiceDetails(name string, consulClient *httpClient) ([]service, error) {
+	req, err := http.NewRequest(
+		"GET",
+		consulClient.config.Address+"/v1/health/service/"+name,
+		nil,
 	)
 	if err != nil {
 		return nil, err
 	}
+	q := url.Values{}
+	q.Add("passing", "true")
+	req.URL.RawQuery = q.Encode()
+	res, err := consulClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+
 	defer res.Body.Close()
 	data, _ := io.ReadAll(res.Body)
 
+	if res.StatusCode != 200 {
+		return nil, fmt.Errorf("%d: %s", res.StatusCode, data)
+	}
+
 	var parsed []struct {
 		Node struct {
 			Address string `json:"Address"`

pkg/routing/consul/monitor.go

@@ -4,6 +4,7 @@ import (
 	"time"
 
 	log "github.com/sirupsen/logrus"
+	"github.com/spf13/viper"
 
 	"dill/pkg/proxy"
 )
@@ -13,9 +14,16 @@ var waitTime time.Duration = 5 * time.Second
 // MonitorServices fetches healthy services that was tagged as `dill`
 func MonitorServices(c chan<- *proxy.RoutingTable) {
 	log.Info("Starting service monitor")
+
+	cfg := consulConfig{}
+	viper.UnmarshalKey("routing.consul", &cfg)
+	cfg.Validate()
+
+	consulClient := &httpClient{config: &cfg}
+
 	index := 1
 	for {
-		services, newIndex, err := fetchHealthyServices(index)
+		services, newIndex, err := fetchHealthyServices(index, consulClient)
 		if err != nil {
 			log.WithField("error", err).Warning("Fetching healthy services failed")
 			time.Sleep(waitTime)
@@ -24,7 +32,7 @@ func MonitorServices(c chan<- *proxy.RoutingTable) {
 		index = newIndex
 		rt := &proxy.RoutingTable{Table: map[string][]proxy.Upstream{}, ConsulIndex: newIndex}
 		for _, s := range services {
-			details, err := fetchServiceDetails(s)
+			details, err := fetchServiceDetails(s, consulClient)
 			if err != nil {
 				log.WithFields(
 					log.Fields{"error": err, "service": s},