Issue 558 - Add Polling Interval From Fabio to Consul to Fabio Config (…

…#572)

* consul: refactor service monitor

Refactor the set of functions which watch the consul state
and generate the route commands into a set of objects to make
them testable and extendable.

* consul: move build route command logic to separate object

... and finally add some tests.

* consul: fetch route updates concurrently

The code which updates the routing table from consul was using a
single go routine to fetch data from consul. This can be a slow
process if consul has lots of registered services.

This patch adds an option `registry.consul.serviceMonitors`
to increase the concurrency for the route updates.

* issue - 558 updated to include custom http.Transport

* added global polling interval for issue 558

* issue 558 updates and doco adds

* rebase to master

* fixed type in pollinterval config

Co-authored-by: Frank Schröder <frank.schroeder@gmail.com>

config/config.go

@@ -155,6 +155,7 @@ type Consul struct {
 	ChecksRequired     string
 	ServiceMonitors    int
 	TLS                ConsulTlS
+	PollInterval       time.Duration
 }
 
 type Custom struct {
@@ -163,7 +164,7 @@ type Custom struct {
 	QueryParams        string
 	Scheme             string
 	CheckTLSSkipVerify bool
-	PollingInterval    time.Duration
+	PollInterval       time.Duration
 	NoRouteHTML        string
 	Timeout            time.Duration
 }

config/default.go

@@ -65,12 +65,13 @@ var defaultConfig = &Config{
 			CheckTimeout:    3 * time.Second,
 			CheckScheme:     "http",
 			ChecksRequired:  "one",
+			PollInterval:    0,
 		},
 		Custom: Custom{
 			Host:               "",
 			Scheme:             "https",
 			CheckTLSSkipVerify: false,
-			PollingInterval:    5,
+			PollInterval:       5,
 			NoRouteHTML:        "",
 			Timeout:            10,
 			Path:               "",

config/load.go

@@ -191,6 +191,7 @@ func load(cmdline, environ, envprefix []string, props *properties.Properties) (c
 	f.StringVar(&obsoleteStr, "registry.consul.register.checkDeregisterCriticalServiceAfter", "", "This option is deprecated and has no effect.")
 	f.StringVar(&cfg.Registry.Consul.ChecksRequired, "registry.consul.checksRequired", defaultConfig.Registry.Consul.ChecksRequired, "number of checks which must pass: one or all")
 	f.IntVar(&cfg.Registry.Consul.ServiceMonitors, "registry.consul.serviceMonitors", defaultConfig.Registry.Consul.ServiceMonitors, "concurrency for route updates")
+	f.DurationVar(&cfg.Registry.Consul.PollInterval, "registry.consul.pollinterval", defaultConfig.Registry.Consul.PollInterval, "poll interval for route updates")
 	f.IntVar(&cfg.Runtime.GOGC, "runtime.gogc", defaultConfig.Runtime.GOGC, "sets runtime.GOGC")
 	f.IntVar(&cfg.Runtime.GOMAXPROCS, "runtime.gomaxprocs", defaultConfig.Runtime.GOMAXPROCS, "sets runtime.GOMAXPROCS")
 	f.StringVar(&cfg.UI.Access, "ui.access", defaultConfig.UI.Access, "access mode, one of [ro, rw]")
@@ -214,7 +215,7 @@ func load(cmdline, environ, envprefix []string, props *properties.Properties) (c
 	f.StringVar(&cfg.Registry.Custom.NoRouteHTML, "registry.custom.noroutehtml", defaultConfig.Registry.Custom.NoRouteHTML, "path to file for HTML returned when no route is found")
 	f.BoolVar(&cfg.Registry.Custom.CheckTLSSkipVerify, "registry.custom.checkTLSSkipVerify", defaultConfig.Registry.Custom.CheckTLSSkipVerify, "custom back end check TLS verification")
 	f.DurationVar(&cfg.Registry.Custom.Timeout, "registry.custom.timeout", defaultConfig.Registry.Custom.Timeout, "timeout for API request to custom back end")
-	f.DurationVar(&cfg.Registry.Custom.PollingInterval, "registry.custom.pollinginterval", defaultConfig.Registry.Custom.PollingInterval, "polling interval for API request to custom back end")
+	f.DurationVar(&cfg.Registry.Custom.PollInterval, "registry.custom.pollinterval", defaultConfig.Registry.Custom.PollInterval, "poll interval for API request to custom back end")
 	f.StringVar(&cfg.Registry.Custom.Path, "registry.custom.path", defaultConfig.Registry.Custom.Path, "custom back end path in the URL")
 	f.StringVar(&cfg.Registry.Custom.QueryParams, "registry.custom.queryparams", defaultConfig.Registry.Custom.QueryParams, "custom back end query parameters in the URL")
 

config/load_test.go

@@ -704,9 +704,9 @@ func TestLoad(t *testing.T) {
 			},
 		},
 		{
-			args: []string{"-registry.custom.pollinginterval", "5s"},
+			args: []string{"-registry.custom.pollinterval", "5s"},
 			cfg: func(cfg *Config) *Config {
-				cfg.Registry.Custom.PollingInterval = 5 * time.Second
+				cfg.Registry.Custom.PollInterval = 5 * time.Second
 				return cfg
 			},
 		},
@@ -724,6 +724,13 @@ func TestLoad(t *testing.T) {
 				return cfg
 			},
 		},
+		{
+			args: []string{"-registry.consul.pollinterval", "5s"},
+			cfg: func(cfg *Config) *Config {
+				cfg.Registry.Consul.PollInterval = 5 * time.Second
+				return cfg
+			},
+		},
 		{
 			args: []string{"-log.access.format", "foobar"},
 			cfg: func(cfg *Config) *Config {

docs/content/ref/registry.consul.pollInterval.md

@@ -0,0 +1,14 @@
+---
+title: "registry.consul.pollInterval"
+---
+
+
+`registry.consul.pollInterval` configures the poll interval
+for route updates. If Poll interval is set to 0 the updates will
+be disabled and fall back to blocking queries.  Other values can
+be any time definition. e.g. `1s, 100ms`
+
+
+The default is
+
+    registry.consul.pollInterval = 0

docs/content/ref/registry.custom.pollinginterval.md

@@ -1,9 +1,9 @@
 ---
-title: "registry.custom.pollinginterval"
+title: "registry.custom.pollinterval"
 ---
 
-`registry.custom.pollinginterval` is the length of time between API calls
+`registry.custom.pollinterval` is the length of time between API calls
 
 The default is
 
-    registry.custom.pollinginterval = 10s
+    registry.custom.pollinterval = 10s

fabio.properties

@@ -890,6 +890,14 @@
 #
 # registry.consul.serviceMonitors = 1
 
+# registry.consul.pollInterval configures the poll interval
+# for route updates. If Poll interval is set to 0 the updates will
+# be disabled and fall back to blocking queries.  Other values can
+# be any time definition. e.g. 1s, 100ms
+#
+# The default is
+# registry.consul.pollInterval = 0
+
 
 # registry.custom.host configures the host:port for fabio to make the API call
 #
@@ -920,11 +928,11 @@
 # registry.custom.timeout = 5s
 
 
-# registry.custom.pollinginterval is the length of time between API calls
+# registry.custom.pollinterval is the length of time between API calls
 #
 # The default is
 #
-#registry.custom.pollinginterval = 10s
+#registry.custom.pollinterval = 10s
 
 
 # registry.custom.path is the path used in the custom back end API Call

registry/consul/service.go

@@ -32,8 +32,15 @@ func NewServiceMonitor(client *api.Client, config *config.Consul, dc string) *Se
 // configuration to the updates channel on every change.
 func (w *ServiceMonitor) Watch(updates chan string) {
 	var lastIndex uint64
+	var q *api.QueryOptions
 	for {
-		q := &api.QueryOptions{RequireConsistent: true, WaitIndex: lastIndex}
+		if w.config.PollInterval != 0 {
+			q = &api.QueryOptions{RequireConsistent: true}
+			time.Sleep(w.config.PollInterval)
+		} else {
+			q = &api.QueryOptions{RequireConsistent: true, WaitIndex: lastIndex}
+		}
+
 		checks, meta, err := w.client.Health().State("any", q)
 		if err != nil {
 			log.Printf("[WARN] consul: Error fetching health state. %v", err)
@@ -54,7 +61,7 @@ func (w *ServiceMonitor) Watch(updates chan string) {
 	}
 }
 
-// makeCconfig determines which service instances have passing health checks
+// makeConfig determines which service instances have passing health checks
 // and then finds the ones which have tags with the right prefix to build the config from.
 func (w *ServiceMonitor) makeConfig(checks []*api.HealthCheck) string {
 	// map service name to list of service passing for which the health check is ok

registry/custom/custom.go

@@ -48,21 +48,21 @@ func customRoutes(cfg *config.Custom, ch chan string) {
 		resp, err := client.Do(req)
 		if err != nil {
 			ch <- fmt.Sprintf("Error Sending HTTPs Request To Custom be - %s -%s", URL, err.Error())
-			time.Sleep(cfg.PollingInterval)
+			time.Sleep(cfg.PollInterval)
 			continue
 		}
 
 		if resp.StatusCode != 200 {
 			ch <- fmt.Sprintf("Error Non-200 return (%v) from  -%s", resp.StatusCode, URL)
-			time.Sleep(cfg.PollingInterval)
+			time.Sleep(cfg.PollInterval)
 			continue
 		}
 		log.Printf("[DEBUG] Custom Registry begin decoding json %s \n", time.Now())
 		decoder := json.NewDecoder(resp.Body)
 		err = decoder.Decode(&Routes)
 		if err != nil {
 			ch <- fmt.Sprintf("Error decoding request - %s -%s", URL, err.Error())
-			time.Sleep(cfg.PollingInterval)
+			time.Sleep(cfg.PollInterval)
 			continue
 		}
 
@@ -75,7 +75,7 @@ func customRoutes(cfg *config.Custom, ch chan string) {
 		route.SetTable(t)
 		log.Printf("[DEBUG] Custom Registry table set complete %s \n", time.Now())
 		ch <- "OK"
-		time.Sleep(cfg.PollingInterval)
+		time.Sleep(cfg.PollInterval)
 
 	}
 

registry/custom/custom_test.go

@@ -18,7 +18,7 @@ func TestCustomRoutes(t *testing.T) {
 		Path:               "test",
 		Scheme:             "http",
 		CheckTLSSkipVerify: false,
-		PollingInterval:    3 * time.Second,
+		PollInterval:       3 * time.Second,
 		Timeout:            3 * time.Second,
 	}
 

route/table.go

@@ -363,7 +363,7 @@ func (t Table) matchingHostNoGlob(req *http.Request) (hosts []string) {
 
 	for pattern := range t {
 		normpat := normalizeHost(pattern, req.TLS != nil)
-		if  normpat == host {
+		if normpat == host {
 			hosts = append(hosts, strings.ToLower(pattern))
 			return
 		}