Close idle connections in influxdb outputs when reloading (#5912)

internal/http_go1.11.go

@@ -0,0 +1,15 @@
+// +build !go1.12
+
+package internal
+
+import "net/http"
+
+func CloseIdleConnections(c *http.Client) {
+	type closeIdler interface {
+		CloseIdleConnections()
+	}
+
+	if tr, ok := c.Transport.(closeIdler); ok {
+		tr.CloseIdleConnections()
+	}
+}

internal/http_go1.12.go

@@ -0,0 +1,9 @@
+// +build go1.12
+
+package internal
+
+import "net/http"
+
+func CloseIdleConnections(c *http.Client) {
+	c.CloseIdleConnections()
+}

plugins/outputs/influxdb/http.go

@@ -448,3 +448,7 @@ func makeQueryURL(loc *url.URL) (string, error) {
 	}
 	return u.String(), nil
 }
+
+func (c *httpClient) Close() {
+	internal.CloseIdleConnections(c.client)
+}

plugins/outputs/influxdb/influxdb.go

@@ -27,6 +27,7 @@ type Client interface {
 	CreateDatabase(ctx context.Context, database string) error
 	Database() string
 	URL() string
+	Close()
 }
 
 // InfluxDB struct is the primary data structure for the plugin
@@ -183,6 +184,9 @@ func (i *InfluxDB) Connect() error {
 }
 
 func (i *InfluxDB) Close() error {
+	for _, client := range i.clients {
+		client.Close()
+	}
 	return nil
 }
 

plugins/outputs/influxdb/influxdb_test.go

@@ -19,6 +19,7 @@ type MockClient struct {
 	WriteF          func(context.Context, []telegraf.Metric) error
 	CreateDatabaseF func(ctx context.Context, database string) error
 	DatabaseF       func() string
+	CloseF          func()
 }
 
 func (c *MockClient) URL() string {
@@ -37,6 +38,10 @@ func (c *MockClient) Database() string {
 	return c.DatabaseF()
 }
 
+func (c *MockClient) Close() {
+	c.CloseF()
+}
+
 func TestDeprecatedURLSupport(t *testing.T) {
 	var actual *influxdb.UDPConfig
 	output := influxdb.InfluxDB{

plugins/outputs/influxdb/udp.go

@@ -136,3 +136,6 @@ func scanLines(data []byte, atEOF bool) (advance int, token []byte, err error) {
 	}
 	return 0, nil, nil
 }
+
+func (c *udpClient) Close() {
+}

plugins/outputs/influxdb_v2/http.go

@@ -307,3 +307,7 @@ func makeWriteURL(loc url.URL, org, bucket string) (string, error) {
 	loc.RawQuery = params.Encode()
 	return loc.String(), nil
 }
+
+func (c *httpClient) Close() {
+	internal.CloseIdleConnections(c.client)
+}

plugins/outputs/influxdb_v2/influxdb.go

@@ -74,6 +74,7 @@ type Client interface {
 	Write(context.Context, []telegraf.Metric) error
 
 	URL() string // for logging
+	Close()
 }
 
 type InfluxDB struct {
@@ -137,6 +138,9 @@ func (i *InfluxDB) Connect() error {
 }
 
 func (i *InfluxDB) Close() error {
+	for _, client := range i.clients {
+		client.Close()
+	}
 	return nil
 }