Add configurable timeout to influxdb input

closes #1773
This commit is contained in:
Cameron Sparr 2016-09-16 16:15:10 +01:00
parent 8d3285522c
commit 6648c101dd
2 changed files with 24 additions and 12 deletions

View File

@ -34,7 +34,8 @@
### Bugfixes ### Bugfixes
- [#1775](https://github.com/influxdata/telegraf/issues/1775): Prometheus output: Fix bug with multi-batch writes. - [#1775](https://github.com/influxdata/telegraf/issues/1775): Prometheus output: Fix bug with multi-batch writes.
- [#1738](https://github.com/influxdata/telegraf/issues/1738): Fix unmarshal of influxdb metrics with null tags - [#1738](https://github.com/influxdata/telegraf/issues/1738): Fix unmarshal of influxdb metrics with null tags.
- [#1773](https://github.com/influxdata/telegraf/issues/1773): Add configurable timeout to influxdb input plugin.
## v1.0 [2016-09-08] ## v1.0 [2016-09-08]

View File

@ -10,11 +10,16 @@ import (
"time" "time"
"github.com/influxdata/telegraf" "github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/plugins/inputs" "github.com/influxdata/telegraf/plugins/inputs"
) )
type InfluxDB struct { type InfluxDB struct {
URLs []string `toml:"urls"` URLs []string `toml:"urls"`
Timeout internal.Duration
client *http.Client
} }
func (*InfluxDB) Description() string { func (*InfluxDB) Description() string {
@ -32,6 +37,9 @@ func (*InfluxDB) SampleConfig() string {
urls = [ urls = [
"http://localhost:8086/debug/vars" "http://localhost:8086/debug/vars"
] ]
## http request & header timeout
timeout = "5s"
` `
} }
@ -39,6 +47,16 @@ func (i *InfluxDB) Gather(acc telegraf.Accumulator) error {
if len(i.URLs) == 0 { if len(i.URLs) == 0 {
i.URLs = []string{"http://localhost:8086/debug/vars"} i.URLs = []string{"http://localhost:8086/debug/vars"}
} }
if i.client == nil {
i.client = &http.Client{
Transport: &http.Transport{
ResponseHeaderTimeout: i.Timeout.Duration,
},
Timeout: i.Timeout.Duration,
}
}
errorChannel := make(chan error, len(i.URLs)) errorChannel := make(chan error, len(i.URLs))
var wg sync.WaitGroup var wg sync.WaitGroup
@ -104,15 +122,6 @@ type memstats struct {
GCCPUFraction float64 `json:"GCCPUFraction"` GCCPUFraction float64 `json:"GCCPUFraction"`
} }
var tr = &http.Transport{
ResponseHeaderTimeout: time.Duration(3 * time.Second),
}
var client = &http.Client{
Transport: tr,
Timeout: time.Duration(4 * time.Second),
}
// Gathers data from a particular URL // Gathers data from a particular URL
// Parameters: // Parameters:
// acc : The telegraf Accumulator to use // acc : The telegraf Accumulator to use
@ -127,7 +136,7 @@ func (i *InfluxDB) gatherURL(
shardCounter := 0 shardCounter := 0
now := time.Now() now := time.Now()
resp, err := client.Get(url) resp, err := i.client.Get(url)
if err != nil { if err != nil {
return err return err
} }
@ -248,6 +257,8 @@ func (i *InfluxDB) gatherURL(
func init() { func init() {
inputs.Add("influxdb", func() telegraf.Input { inputs.Add("influxdb", func() telegraf.Input {
return &InfluxDB{} return &InfluxDB{
Timeout: internal.Duration{Duration: time.Second * 5},
}
}) })
} }