Merge pull request #4443 from simonpasquier/fix-consul-connections-leak

discovery/consul: close idle connections on stop

discovery/consul/consul.go

@@ -147,7 +147,6 @@ func init() {
 // and updates them via watches.
 type Discovery struct {
 	client           *consul.Client
-	clientConf       *consul.Config
 	clientDatacenter string
 	tagSeparator     string
 	watchedServices  []string // Set of services which will be discovered.
@@ -155,6 +154,7 @@ type Discovery struct {
 	watchedNodeMeta  map[string]string
 	allowStale       bool
 	refreshInterval  time.Duration
+	finalizer        func()
 	logger           log.Logger
 }
 
@@ -169,6 +169,7 @@ func NewDiscovery(conf *SDConfig, logger log.Logger) (*Discovery, error) {
 		return nil, err
 	}
 	transport := &http.Transport{
+		IdleConnTimeout: 5 * time.Duration(conf.RefreshInterval),
 		TLSClientConfig: tls,
 		DialContext: conntrack.NewDialContextFunc(
 			conntrack.DialWithTracing(),
@@ -197,14 +198,14 @@ func NewDiscovery(conf *SDConfig, logger log.Logger) (*Discovery, error) {
 	}
 	cd := &Discovery{
 		client:           client,
-		clientConf:       clientConf,
 		tagSeparator:     conf.TagSeparator,
 		watchedServices:  conf.Services,
 		watchedTag:       conf.ServiceTag,
 		watchedNodeMeta:  conf.NodeMeta,
 		allowStale:       conf.AllowStale,
 		refreshInterval:  time.Duration(conf.RefreshInterval),
-		clientDatacenter: clientConf.Datacenter,
+		clientDatacenter: conf.Datacenter,
+		finalizer:        transport.CloseIdleConnections,
 		logger:           logger,
 	}
 	return cd, nil
@@ -298,6 +299,9 @@ func (d *Discovery) initialize(ctx context.Context) {
 
 // Run implements the Discoverer interface.
 func (d *Discovery) Run(ctx context.Context, ch chan<- []*targetgroup.Group) {
+	if d.finalizer != nil {
+		defer d.finalizer()
+	}
 	d.initialize(ctx)
 
 	if len(d.watchedServices) == 0 || d.watchedTag != "" {