Merge branch 'reuse-transport'

CHANGELOG.md

@@ -9,6 +9,7 @@
 
 - [#2749](https://github.com/influxdata/telegraf/pull/2749): Fixed sqlserver input to work with case sensitive server collation.
 - [#2716](https://github.com/influxdata/telegraf/pull/2716): Systemd does not see all shutdowns as failures
+- [#2782](https://github.com/influxdata/telegraf/pull/2782): Reuse transports in input plugins
 
 ## v1.3 [unreleased]
 

plugins/inputs/apache/apache.go

@@ -29,6 +29,8 @@ type Apache struct {
 	SSLKey string `toml:"ssl_key"`
 	// Use SSL but skip chain & host verification
 	InsecureSkipVerify bool
+
+	client *http.Client
 }
 
 var sampleConfig = `
@@ -66,6 +68,14 @@ func (n *Apache) Gather(acc telegraf.Accumulator) error {
 		n.ResponseTimeout.Duration = time.Second * 5
 	}
 
+	if n.client == nil {
+		client, err := n.createHttpClient()
+		if err != nil {
+			return err
+		}
+		n.client = client
+	}
+
 	var wg sync.WaitGroup
 	wg.Add(len(n.Urls))
 	for _, u := range n.Urls {
@@ -85,31 +95,24 @@ func (n *Apache) Gather(acc telegraf.Accumulator) error {
 	return nil
 }
 
-func (n *Apache) gatherUrl(addr *url.URL, acc telegraf.Accumulator) error {
-
-	var tr *http.Transport
-
-	if addr.Scheme == "https" {
-		tlsCfg, err := internal.GetTLSConfig(
-			n.SSLCert, n.SSLKey, n.SSLCA, n.InsecureSkipVerify)
-		if err != nil {
-			return err
-		}
-		tr = &http.Transport{
-			ResponseHeaderTimeout: time.Duration(3 * time.Second),
-			TLSClientConfig:       tlsCfg,
-		}
-	} else {
-		tr = &http.Transport{
-			ResponseHeaderTimeout: time.Duration(3 * time.Second),
-		}
+func (n *Apache) createHttpClient() (*http.Client, error) {
+	tlsCfg, err := internal.GetTLSConfig(
+		n.SSLCert, n.SSLKey, n.SSLCA, n.InsecureSkipVerify)
+	if err != nil {
+		return nil, err
 	}
 
 	client := &http.Client{
-		Transport: tr,
-		Timeout:   n.ResponseTimeout.Duration,
+		Transport: &http.Transport{
+			TLSClientConfig: tlsCfg,
+		},
+		Timeout: n.ResponseTimeout.Duration,
 	}
 
+	return client, nil
+}
+
+func (n *Apache) gatherUrl(addr *url.URL, acc telegraf.Accumulator) error {
 	req, err := http.NewRequest("GET", addr.String(), nil)
 	if err != nil {
 		return fmt.Errorf("error on new request to %s : %s\n", addr.String(), err)
@@ -119,7 +122,7 @@ func (n *Apache) gatherUrl(addr *url.URL, acc telegraf.Accumulator) error {
 		req.SetBasicAuth(n.Username, n.Password)
 	}
 
-	resp, err := client.Do(req)
+	resp, err := n.client.Do(req)
 	if err != nil {
 		return fmt.Errorf("error on request to %s : %s\n", addr.String(), err)
 	}

plugins/inputs/http_response/http_response.go

@@ -25,7 +25,6 @@ type HTTPResponse struct {
 	Headers             map[string]string
 	FollowRedirects     bool
 	ResponseStringMatch string
-	compiledStringMatch *regexp.Regexp
 
 	// Path to CA file
 	SSLCA string `toml:"ssl_ca"`
@@ -35,6 +34,9 @@ type HTTPResponse struct {
 	SSLKey string `toml:"ssl_key"`
 	// Use SSL but skip chain & host verification
 	InsecureSkipVerify bool
+
+	compiledStringMatch *regexp.Regexp
+	client              *http.Client
 }
 
 // Description returns the plugin Description
@@ -88,13 +90,12 @@ func (h *HTTPResponse) createHttpClient() (*http.Client, error) {
 	if err != nil {
 		return nil, err
 	}
-	tr := &http.Transport{
-		ResponseHeaderTimeout: h.ResponseTimeout.Duration,
-		TLSClientConfig:       tlsCfg,
-	}
 	client := &http.Client{
-		Transport: tr,
-		Timeout:   h.ResponseTimeout.Duration,
+		Transport: &http.Transport{
+			DisableKeepAlives: true,
+			TLSClientConfig:   tlsCfg,
+		},
+		Timeout: h.ResponseTimeout.Duration,
 	}
 
 	if h.FollowRedirects == false {
@@ -106,15 +107,10 @@ func (h *HTTPResponse) createHttpClient() (*http.Client, error) {
 }
 
 // HTTPGather gathers all fields and returns any errors it encounters
-func (h *HTTPResponse) HTTPGather() (map[string]interface{}, error) {
+func (h *HTTPResponse) httpGather() (map[string]interface{}, error) {
 	// Prepare fields
 	fields := make(map[string]interface{})
 
-	client, err := h.createHttpClient()
-	if err != nil {
-		return nil, err
-	}
-
 	var body io.Reader
 	if h.Body != "" {
 		body = strings.NewReader(h.Body)
@@ -133,7 +129,7 @@ func (h *HTTPResponse) HTTPGather() (map[string]interface{}, error) {
 
 	// Start Timer
 	start := time.Now()
-	resp, err := client.Do(request)
+	resp, err := h.client.Do(request)
 	if err != nil {
 		if h.FollowRedirects {
 			return nil, err
@@ -145,6 +141,11 @@ func (h *HTTPResponse) HTTPGather() (map[string]interface{}, error) {
 			return nil, err
 		}
 	}
+	defer func() {
+		io.Copy(ioutil.Discard, resp.Body)
+		resp.Body.Close()
+	}()
+
 	fields["response_time"] = time.Since(start).Seconds()
 	fields["http_response_code"] = resp.StatusCode
 
@@ -202,8 +203,17 @@ func (h *HTTPResponse) Gather(acc telegraf.Accumulator) error {
 	// Prepare data
 	tags := map[string]string{"server": h.Address, "method": h.Method}
 	var fields map[string]interface{}
+
+	if h.client == nil {
+		client, err := h.createHttpClient()
+		if err != nil {
+			return err
+		}
+		h.client = client
+	}
+
 	// Gather data
-	fields, err = h.HTTPGather()
+	fields, err = h.httpGather()
 	if err != nil {
 		return err
 	}