Merge pull request #219 from influxdata/nc-issue#187

Retry connecting to InfluxDB on startup

CHANGELOG.md

@@ -25,6 +25,7 @@ There was a breaking change to the `define` command, see [#173](https://github.c
 - [#208](https://github.com/influxdata/kapacitor/issues/208): Add default stats dbrp to default subscription excludes.
 - [#203](https://github.com/influxdata/kapacitor/issues/203): Fix hang when deleteing invalid batch task.
 - [#182](https://github.com/influxdata/kapacitor/issues/182): Fix missing/incorrect Content-Type headers for various HTTP endpoints.
+- [#187](https://github.com/influxdata/kapacitor/issues/187): Retry connecting to InfluxDB on startup for up to 5 minutes by default.
 
 ## v0.10.0 [2016-01-26]
 

services/influxdb/config.go

@@ -6,18 +6,25 @@ import (
 
 	"github.com/influxdata/kapacitor/services/stats"
 	"github.com/influxdata/kapacitor/services/udp"
+	"github.com/influxdb/influxdb/toml"
+)
+
+const (
+	// Maximum time to try and connect to InfluxDB during startup.
+	DefaultStartUpTimeout = time.Minute * 5
 )
 
 type Config struct {
 	Enabled               bool                `toml:"enabled"`
 	URLs                  []string            `toml:"urls"`
 	Username              string              `toml:"username"`
 	Password              string              `toml:"password"`
-	Timeout               time.Duration       `toml:"timeout"`
+	Timeout               toml.Duration       `toml:"timeout"`
 	Subscriptions         map[string][]string `toml:"subscriptions"`
 	ExcludedSubscriptions map[string][]string `toml:"excluded-subscriptions"`
 	UDPBuffer             int                 `toml:"udp-buffer"`
 	UDPReadBuffer         int                 `toml:"udp-read-buffer"`
+	StartUpTimeout        toml.Duration       `toml:"startup-timeout"`
 }
 
 func NewConfig() Config {
@@ -30,7 +37,8 @@ func NewConfig() Config {
 		ExcludedSubscriptions: map[string][]string{
 			stats.DefaultDatabse: []string{stats.DefaultRetentionPolicy},
 		},
-		UDPBuffer: udp.DefaultBuffer,
+		UDPBuffer:      udp.DefaultBuffer,
+		StartUpTimeout: toml.Duration(DefaultStartUpTimeout),
 	}
 }
 

services/influxdb/service.go

@@ -6,27 +6,31 @@ import (
 	"net"
 	"net/url"
 	"strings"
+	"time"
 
+	"github.com/cenkalti/backoff"
 	"github.com/influxdata/kapacitor"
 	"github.com/influxdata/kapacitor/services/udp"
 	"github.com/influxdb/influxdb/client"
 	"github.com/influxdb/influxdb/cluster"
 )
 
 const (
+	// The name to give to all subscriptions
 	subName = "kapacitor"
 )
 
 // Handles requests to write or read from an InfluxDB cluster
 type Service struct {
-	configs       []client.Config
-	i             int
-	configSubs    map[subEntry]bool
-	exConfigSubs  map[subEntry]bool
-	hostname      string
-	logger        *log.Logger
-	udpBuffer     int
-	udpReadBuffer int
+	configs        []client.Config
+	i              int
+	configSubs     map[subEntry]bool
+	exConfigSubs   map[subEntry]bool
+	hostname       string
+	logger         *log.Logger
+	udpBuffer      int
+	udpReadBuffer  int
+	startupTimeout time.Duration
 
 	PointsWriter interface {
 		WritePoints(p *cluster.WritePointsRequest) error
@@ -61,7 +65,7 @@ func NewService(c Config, hostname string, l *log.Logger) *Service {
 			Username:  c.Username,
 			Password:  c.Password,
 			UserAgent: "Kapacitor",
-			Timeout:   c.Timeout,
+			Timeout:   time.Duration(c.Timeout),
 		}
 	}
 	subs := make(map[subEntry]bool, len(c.Subscriptions))
@@ -79,13 +83,14 @@ func NewService(c Config, hostname string, l *log.Logger) *Service {
 		}
 	}
 	return &Service{
-		configs:       configs,
-		configSubs:    subs,
-		exConfigSubs:  exSubs,
-		hostname:      hostname,
-		logger:        l,
-		udpBuffer:     c.UDPBuffer,
-		udpReadBuffer: c.UDPReadBuffer,
+		configs:        configs,
+		configSubs:     subs,
+		exConfigSubs:   exSubs,
+		hostname:       hostname,
+		logger:         l,
+		udpBuffer:      c.UDPBuffer,
+		udpReadBuffer:  c.UDPReadBuffer,
+		startupTimeout: time.Duration(c.StartUpTimeout),
 	}
 }
 
@@ -111,7 +116,6 @@ func (s *Service) Addr() string {
 }
 
 func (s *Service) NewClient() (c *client.Client, err error) {
-
 	tries := 0
 	for tries < len(s.configs) {
 		tries++
@@ -131,7 +135,21 @@ func (s *Service) NewClient() (c *client.Client, err error) {
 }
 
 func (s *Service) linkSubscriptions() error {
-	cli, err := s.NewClient()
+	s.logger.Println("I! linking subscriptions")
+	b := backoff.NewExponentialBackOff()
+	b.MaxElapsedTime = s.startupTimeout
+	ticker := backoff.NewTicker(b)
+	var err error
+	var cli *client.Client
+	for range ticker.C {
+		cli, err = s.NewClient()
+		if err != nil {
+			s.logger.Println("D! failed to connect to InfluxDB, retrying... ", err)
+			continue
+		}
+		ticker.Stop()
+		break
+	}
 	if err != nil {
 		return err
 	}