Close HTTP2 connections on timeout in influxdb outputs (#7517)

influxdata · May 19, 2020 · edd8338 · edd8338
1 parent 443ac6d
commit edd8338
Show file tree

Hide file tree

Showing 5 changed files with 20 additions and 30 deletions.
diff --git a/internal/http.go b/internal/http.go
@@ -4,6 +4,7 @@ import (
 	"crypto/subtle"
 	"net"
 	"net/http"
+	"net/url"
 )
 
 type BasicAuthErrorFunc func(rw http.ResponseWriter)
@@ -95,3 +96,13 @@ func (h *ipRangeHandler) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
 
 	h.onError(rw, http.StatusForbidden)
 }
+
+func OnClientError(client *http.Client, err error) {
+	// Close connection after a timeout error. If this is a HTTP2
+	// connection this ensures that next interval a new connection will be
+	// used and name lookup will be performed.
+	//   https://github.com/golang/go/issues/36026
+	if err, ok := err.(*url.Error); ok && err.Timeout() {
+		client.CloseIdleConnections()
+	}
+}
diff --git a/internal/http_go1.11.go b/internal/http_go1.11.go
diff --git a/internal/http_go1.12.go b/internal/http_go1.12.go
diff --git a/plugins/outputs/influxdb/http.go b/plugins/outputs/influxdb/http.go
@@ -209,6 +209,7 @@ func (c *httpClient) CreateDatabase(ctx context.Context, database string) error
 
 	resp, err := c.client.Do(req.WithContext(ctx))
 	if err != nil {
+		internal.OnClientError(c.client, err)
 		return err
 	}
 	defer resp.Body.Close()
@@ -311,7 +312,7 @@ func (c *httpClient) Write(ctx context.Context, metrics []telegraf.Metric) error
 }
 
 func (c *httpClient) writeBatch(ctx context.Context, db, rp string, metrics []telegraf.Metric) error {
-	url, err := makeWriteURL(c.config.URL, db, rp, c.config.Consistency)
+	loc, err := makeWriteURL(c.config.URL, db, rp, c.config.Consistency)
 	if err != nil {
 		return err
 	}
@@ -322,13 +323,14 @@ func (c *httpClient) writeBatch(ctx context.Context, db, rp string, metrics []te
 	}
 	defer reader.Close()
 
-	req, err := c.makeWriteRequest(url, reader)
+	req, err := c.makeWriteRequest(loc, reader)
 	if err != nil {
 		return err
 	}
 
 	resp, err := c.client.Do(req.WithContext(ctx))
 	if err != nil {
+		internal.OnClientError(c.client, err)
 		return err
 	}
 	defer resp.Body.Close()
@@ -505,5 +507,5 @@ func makeQueryURL(loc *url.URL) (string, error) {
 }
 
 func (c *httpClient) Close() {
-	internal.CloseIdleConnections(c.client)
+	c.client.CloseIdleConnections()
 }
diff --git a/plugins/outputs/influxdb_v2/http.go b/plugins/outputs/influxdb_v2/http.go
@@ -210,7 +210,7 @@ func (c *httpClient) Write(ctx context.Context, metrics []telegraf.Metric) error
 }
 
 func (c *httpClient) writeBatch(ctx context.Context, bucket string, metrics []telegraf.Metric) error {
-	url, err := makeWriteURL(*c.url, c.Organization, bucket)
+	loc, err := makeWriteURL(*c.url, c.Organization, bucket)
 	if err != nil {
 		return err
 	}
@@ -221,13 +221,14 @@ func (c *httpClient) writeBatch(ctx context.Context, bucket string, metrics []te
 	}
 	defer reader.Close()
 
-	req, err := c.makeWriteRequest(url, reader)
+	req, err := c.makeWriteRequest(loc, reader)
 	if err != nil {
 		return err
 	}
 
 	resp, err := c.client.Do(req.WithContext(ctx))
 	if err != nil {
+		internal.OnClientError(c.client, err)
 		return err
 	}
 	defer resp.Body.Close()
@@ -347,5 +348,5 @@ func makeWriteURL(loc url.URL, org, bucket string) (string, error) {
 }
 
 func (c *httpClient) Close() {
-	internal.CloseIdleConnections(c.client)
+	c.client.CloseIdleConnections()
 }