dom
/
telegraf
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Utilize timeout in net_response plugin. 

Also changing the net_response and http_response plugins to only accept
duration strings for their timeout parameters. This is a breaking config
file change.

closes #1214
This commit is contained in:
Cameron Sparr 2016-05-23 13:33:43 +01:00
parent c6699c36d3
commit c44ecf54a5
10 changed files with 118 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGELOG.md
View File

@ -2,6 +2,8 @@
### Release Notes
- net_response and http_response plugins timeouts will now accept duration
strings, ie, "2s" or "500ms".
- Input plugin Gathers will no longer be logged by default, but a Gather for
_each_ plugin will be logged in Debug mode.
- Debug mode will no longer print every point added to the accumulator. This
@ -24,6 +26,7 @@ to "stdout".
- [#1228](https://github.com/influxdata/telegraf/pull/1228): Fix service plugin host tag overwrite.
- [#1198](https://github.com/influxdata/telegraf/pull/1198): http_response: override request Host header properly
- [#1230](https://github.com/influxdata/telegraf/issues/1230): Fix Telegraf process hangup due to a single plugin hanging.
- [#1214](https://github.com/influxdata/telegraf/issues/1214): Use TCP timeout argument in net_response plugin.
## v0.13 [2016-05-11]

9
agent/agent.go
View File

@ -137,16 +137,15 @@ func (a *Agent) gatherer(
}
gatherWithTimeout(shutdown, input, acc, interval)
elapsed := time.Since(start)
if a.Config.Agent.Debug {
log.Printf("Input [%s] gathered metrics, (%s interval) in %s\n",
input.Name, interval, elapsed)
}
if outerr != nil {
return outerr
}
if a.Config.Agent.Debug {
log.Printf("Input [%s] gathered metrics, (%s interval) in %s\n",
input.Name, interval, elapsed)
}
select {
case <-shutdown:

14
etc/telegraf.conf
View File

@ -680,7 +680,7 @@
# ## Server address (default http://localhost)
# address = "http://github.com"
# ## Set response_timeout (default 5 seconds)
# response_timeout = 5
# response_timeout = "5s"
# ## HTTP Request Method
# method = "GET"
# ## Whether to follow redirects from the server (defaults to false)
@ -946,14 +946,15 @@
# protocol = "tcp"
# ## Server address (default localhost)
# address = "github.com:80"
# ## Set timeout (default 1.0 seconds)
# timeout = 1.0
# ## Set read timeout (default 1.0 seconds)
# read_timeout = 1.0
# ## Set timeout
# timeout = "1s"
#
# ## Optional string sent to the server
# # send = "ssh"
# ## Optional expected string in answer
# # expect = "ssh"
# ## Set read timeout (only used if expecting a response)
# read_timeout = "1s"
# # Read TCP metrics such as established, time wait and sockets counts.
@ -1144,6 +1145,9 @@
# ## user as argument for pgrep (ie, pgrep -u <user>)
# # user = "nginx"
#
# ## override for process_name
# ## This is optional; default is sourced from /proc/<pid>/status
# # process_name = "bar"
# ## Field name prefix
# prefix = ""
# ## comment this out if you want raw cpu_time stats

22
internal/internal.go
View File

@ -12,6 +12,7 @@ import (
"log"
"os"
"os/exec"
"strconv"
"strings"
"time"
"unicode"
@ -32,12 +33,25 @@ type Duration struct {
// UnmarshalTOML parses the duration from the TOML config file
func (d *Duration) UnmarshalTOML(b []byte) error {
dur, err := time.ParseDuration(string(b[1 : len(b)-1]))
if err != nil {
return err
var err error
// Parse string duration, ie, "1s"
d.Duration, err = time.ParseDuration(string(b[1 : len(b)-1]))
if err == nil {
return nil
}
d.Duration = dur
// First try parsing as integer seconds
sI, err := strconv.ParseInt(string(b), 10, 64)
if err == nil {
d.Duration = time.Second * time.Duration(sI)
return nil
}
// Second try parsing as float seconds
sF, err := strconv.ParseFloat(string(b), 64)
if err == nil {
d.Duration = time.Second * time.Duration(sF)
return nil
}
return nil
}

16
plugins/inputs/http_response/README.md
View File

@ -5,23 +5,23 @@ This input plugin will test HTTP/HTTPS connections.
### Configuration:
```
# List of UDP/TCP connections you want to check
# HTTP/HTTPS request given an address a method and a timeout
[[inputs.http_response]]
## Server address (default http://localhost)
address = "http://github.com"
## Set response_timeout (default 5 seconds)
response_timeout = 5
response_timeout = "5s"
## HTTP Request Method
method = "GET"
## HTTP Request Headers
[inputs.http_response.headers]
Host = github.com
## Whether to follow redirects from the server (defaults to false)
follow_redirects = true
## HTTP Request Headers (all values must be strings)
# [inputs.http_response.headers]
# Host = "github.com"
## Optional HTTP Request Body
body = '''
{'fake':'data'}
'''
# body = '''
# {'fake':'data'}
# '''
```
### Measurements & Fields:

13
plugins/inputs/http_response/http_response.go
View File

@ -9,6 +9,7 @@ import (
"time"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/plugins/inputs"
)
@ -17,7 +18,7 @@ type HTTPResponse struct {
Address string
Body string
Method string
ResponseTimeout int
ResponseTimeout internal.Duration
Headers map[string]string
FollowRedirects bool
}
@ -31,7 +32,7 @@ var sampleConfig = `
## Server address (default http://localhost)
address = "http://github.com"
## Set response_timeout (default 5 seconds)
response_timeout = 5
response_timeout = "5s"
## HTTP Request Method
method = "GET"
## Whether to follow redirects from the server (defaults to false)
@ -57,7 +58,7 @@ var ErrRedirectAttempted = errors.New("redirect")
// timeout period and can follow redirects if specified
func CreateHttpClient(followRedirects bool, ResponseTimeout time.Duration) *http.Client {
client := &http.Client{
Timeout: time.Second * ResponseTimeout,
Timeout: ResponseTimeout,
}
if followRedirects == false {
@ -73,7 +74,7 @@ func (h *HTTPResponse) HTTPGather() (map[string]interface{}, error) {
// Prepare fields
fields := make(map[string]interface{})
client := CreateHttpClient(h.FollowRedirects, time.Duration(h.ResponseTimeout))
client := CreateHttpClient(h.FollowRedirects, h.ResponseTimeout.Duration)
var body io.Reader
if h.Body != "" {
@ -113,8 +114,8 @@ func (h *HTTPResponse) HTTPGather() (map[string]interface{}, error) {
// Gather gets all metric fields and tags and returns any errors it encounters
func (h *HTTPResponse) Gather(acc telegraf.Accumulator) error {
// Set default values
if h.ResponseTimeout < 1 {
h.ResponseTimeout = 5
if h.ResponseTimeout.Duration < time.Second {
h.ResponseTimeout.Duration = time.Second * 5
}
// Check send and expected string
if h.Method == "" {

27
plugins/inputs/http_response/http_response_test.go
View File

@ -2,13 +2,16 @@ package http_response
import (
"fmt"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"io/ioutil"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/influxdata/telegraf/internal"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func setUpTestMux() http.Handler {
@ -61,7 +64,7 @@ func TestHeaders(t *testing.T) {
h := &HTTPResponse{
Address: ts.URL,
Method: "GET",
ResponseTimeout: 2,
ResponseTimeout: internal.Duration{Duration: time.Second * 2},
Headers: map[string]string{
"Content-Type": "application/json",
"Host": "Hello",
@ -85,7 +88,7 @@ func TestFields(t *testing.T) {
Address: ts.URL + "/good",
Body: "{ 'test': 'data'}",
Method: "GET",
ResponseTimeout: 20,
ResponseTimeout: internal.Duration{Duration: time.Second * 20},
Headers: map[string]string{
"Content-Type": "application/json",
},
@ -109,7 +112,7 @@ func TestRedirects(t *testing.T) {
Address: ts.URL + "/redirect",
Body: "{ 'test': 'data'}",
Method: "GET",
ResponseTimeout: 20,
ResponseTimeout: internal.Duration{Duration: time.Second * 20},
Headers: map[string]string{
"Content-Type": "application/json",
},
@ -126,7 +129,7 @@ func TestRedirects(t *testing.T) {
Address: ts.URL + "/badredirect",
Body: "{ 'test': 'data'}",
Method: "GET",
ResponseTimeout: 20,
ResponseTimeout: internal.Duration{Duration: time.Second * 20},
Headers: map[string]string{
"Content-Type": "application/json",
},
@ -145,7 +148,7 @@ func TestMethod(t *testing.T) {
Address: ts.URL + "/mustbepostmethod",
Body: "{ 'test': 'data'}",
Method: "POST",
ResponseTimeout: 20,
ResponseTimeout: internal.Duration{Duration: time.Second * 20},
Headers: map[string]string{
"Content-Type": "application/json",
},
@ -162,7 +165,7 @@ func TestMethod(t *testing.T) {
Address: ts.URL + "/mustbepostmethod",
Body: "{ 'test': 'data'}",
Method: "GET",
ResponseTimeout: 20,
ResponseTimeout: internal.Duration{Duration: time.Second * 20},
Headers: map[string]string{
"Content-Type": "application/json",
},
@ -180,7 +183,7 @@ func TestMethod(t *testing.T) {
Address: ts.URL + "/mustbepostmethod",
Body: "{ 'test': 'data'}",
Method: "head",
ResponseTimeout: 20,
ResponseTimeout: internal.Duration{Duration: time.Second * 20},
Headers: map[string]string{
"Content-Type": "application/json",
},
@ -203,7 +206,7 @@ func TestBody(t *testing.T) {
Address: ts.URL + "/musthaveabody",
Body: "{ 'test': 'data'}",
Method: "GET",
ResponseTimeout: 20,
ResponseTimeout: internal.Duration{Duration: time.Second * 20},
Headers: map[string]string{
"Content-Type": "application/json",
},
@ -219,7 +222,7 @@ func TestBody(t *testing.T) {
h = &HTTPResponse{
Address: ts.URL + "/musthaveabody",
Method: "GET",
ResponseTimeout: 20,
ResponseTimeout: internal.Duration{Duration: time.Second * 20},
Headers: map[string]string{
"Content-Type": "application/json",
},
@ -242,7 +245,7 @@ func TestTimeout(t *testing.T) {
Address: ts.URL + "/twosecondnap",
Body: "{ 'test': 'data'}",
Method: "GET",
ResponseTimeout: 1,
ResponseTimeout: internal.Duration{Duration: time.Second * 1},
Headers: map[string]string{
"Content-Type": "application/json",
},

35
plugins/inputs/net_response/README.md
View File

@ -6,41 +6,30 @@ It can also check response text.
### Configuration:
```
# List of UDP/TCP connections you want to check
[[inputs.net_response]]
protocol = "tcp"
# Server address (default IP localhost)
address = "github.com:80"
# Set timeout (default 1.0)
timeout = 1.0
# Set read timeout (default 1.0)
read_timeout = 1.0
# String sent to the server
send = "ssh"
# Expected string in answer
expect = "ssh"
[[inputs.net_response]]
protocol = "tcp"
address = ":80"
# TCP or UDP 'ping' given url and collect response time in seconds
[[inputs.net_response]]
protocol = "udp"
# Server address (default IP localhost)
## Protocol, must be "tcp" or "udp"
protocol = "tcp"
## Server address (default localhost)
address = "github.com:80"
# Set timeout (default 1.0)
timeout = 1.0
# Set read timeout (default 1.0)
read_timeout = 1.0
# String sent to the server
## Set timeout
timeout = "1s"
## Optional string sent to the server
send = "ssh"
# Expected string in answer
## Optional expected string in answer
expect = "ssh"
## Set read timeout (only used if expecting a response)
read_timeout = "1s"
[[inputs.net_response]]
protocol = "udp"
address = "localhost:161"
timeout = 2.0
timeout = "2s"
```
### Measurements & Fields:

67
plugins/inputs/net_response/net_response.go
View File

@ -9,14 +9,15 @@ import (
"time"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/plugins/inputs"
)
// NetResponses struct
type NetResponse struct {
Address string
Timeout float64
ReadTimeout float64
Timeout internal.Duration
ReadTimeout internal.Duration
Send string
Expect string
Protocol string
@ -31,29 +32,28 @@ var sampleConfig = `
protocol = "tcp"
## Server address (default localhost)
address = "github.com:80"
## Set timeout (default 1.0 seconds)
timeout = 1.0
## Set read timeout (default 1.0 seconds)
read_timeout = 1.0
## Set timeout
timeout = "1s"
## Optional string sent to the server
# send = "ssh"
## Optional expected string in answer
# expect = "ssh"
## Set read timeout (only used if expecting a response)
read_timeout = "1s"
`
func (_ *NetResponse) SampleConfig() string {
return sampleConfig
}
func (t *NetResponse) TcpGather() (map[string]interface{}, error) {
func (n *NetResponse) TcpGather() (map[string]interface{}, error) {
// Prepare fields
fields := make(map[string]interface{})
// Start Timer
start := time.Now()
// Resolving
tcpAddr, err := net.ResolveTCPAddr("tcp", t.Address)
// Connecting
conn, err := net.DialTCP("tcp", nil, tcpAddr)
conn, err := net.DialTimeout("tcp", n.Address, n.Timeout.Duration)
// Stop timer
responseTime := time.Since(start).Seconds()
// Handle error
@ -62,17 +62,16 @@ func (t *NetResponse) TcpGather() (map[string]interface{}, error) {
}
defer conn.Close()
// Send string if needed
if t.Send != "" {
msg := []byte(t.Send)
if n.Send != "" {
msg := []byte(n.Send)
conn.Write(msg)
conn.CloseWrite()
// Stop timer
responseTime = time.Since(start).Seconds()
}
// Read string if needed
if t.Expect != "" {
if n.Expect != "" {
// Set read timeout
conn.SetReadDeadline(time.Now().Add(time.Duration(t.ReadTimeout) * time.Second))
conn.SetReadDeadline(time.Now().Add(n.ReadTimeout.Duration))
// Prepare reader
reader := bufio.NewReader(conn)
tp := textproto.NewReader(reader)
@ -85,7 +84,7 @@ func (t *NetResponse) TcpGather() (map[string]interface{}, error) {
fields["string_found"] = false
} else {
// Looking for string in answer
RegEx := regexp.MustCompile(`.*` + t.Expect + `.*`)
RegEx := regexp.MustCompile(`.*` + n.Expect + `.*`)
find := RegEx.FindString(string(data))
if find != "" {
fields["string_found"] = true
@ -99,13 +98,13 @@ func (t *NetResponse) TcpGather() (map[string]interface{}, error) {
return fields, nil
}
func (u *NetResponse) UdpGather() (map[string]interface{}, error) {
func (n *NetResponse) UdpGather() (map[string]interface{}, error) {
// Prepare fields
fields := make(map[string]interface{})
// Start Timer
start := time.Now()
// Resolving
udpAddr, err := net.ResolveUDPAddr("udp", u.Address)
udpAddr, err := net.ResolveUDPAddr("udp", n.Address)
LocalAddr, err := net.ResolveUDPAddr("udp", "127.0.0.1:0")
// Connecting
conn, err := net.DialUDP("udp", LocalAddr, udpAddr)
@ -115,11 +114,11 @@ func (u *NetResponse) UdpGather() (map[string]interface{}, error) {
return nil, err
}
// Send string
msg := []byte(u.Send)
msg := []byte(n.Send)
conn.Write(msg)
// Read string
// Set read timeout
conn.SetReadDeadline(time.Now().Add(time.Duration(u.ReadTimeout) * time.Second))
conn.SetReadDeadline(time.Now().Add(n.ReadTimeout.Duration))
// Read
buf := make([]byte, 1024)
_, _, err = conn.ReadFromUDP(buf)
@ -130,7 +129,7 @@ func (u *NetResponse) UdpGather() (map[string]interface{}, error) {
return nil, err
} else {
// Looking for string in answer
RegEx := regexp.MustCompile(`.*` + u.Expect + `.*`)
RegEx := regexp.MustCompile(`.*` + n.Expect + `.*`)
find := RegEx.FindString(string(buf))
if find != "" {
fields["string_found"] = true
@ -142,28 +141,28 @@ func (u *NetResponse) UdpGather() (map[string]interface{}, error) {
return fields, nil
}
func (c *NetResponse) Gather(acc telegraf.Accumulator) error {
func (n *NetResponse) Gather(acc telegraf.Accumulator) error {
// Set default values
if c.Timeout == 0 {
c.Timeout = 1.0
if n.Timeout.Duration == 0 {
n.Timeout.Duration = time.Second
}
if c.ReadTimeout == 0 {
c.ReadTimeout = 1.0
if n.ReadTimeout.Duration == 0 {
n.ReadTimeout.Duration = time.Second
}
// Check send and expected string
if c.Protocol == "udp" && c.Send == "" {
if n.Protocol == "udp" && n.Send == "" {
return errors.New("Send string cannot be empty")
}
if c.Protocol == "udp" && c.Expect == "" {
if n.Protocol == "udp" && n.Expect == "" {
return errors.New("Expected string cannot be empty")
}
// Prepare host and port
host, port, err := net.SplitHostPort(c.Address)
host, port, err := net.SplitHostPort(n.Address)
if err != nil {
return err
}
if host == "" {
c.Address = "localhost:" + port
n.Address = "localhost:" + port
}
if port == "" {
return errors.New("Bad port")
@ -172,11 +171,11 @@ func (c *NetResponse) Gather(acc telegraf.Accumulator) error {
tags := map[string]string{"server": host, "port": port}
var fields map[string]interface{}
// Gather data
if c.Protocol == "tcp" {
fields, err = c.TcpGather()
if n.Protocol == "tcp" {
fields, err = n.TcpGather()
tags["protocol"] = "tcp"
} else if c.Protocol == "udp" {
fields, err = c.UdpGather()
} else if n.Protocol == "udp" {
fields, err = n.UdpGather()
tags["protocol"] = "udp"
} else {
return errors.New("Bad protocol")

16
plugins/inputs/net_response/net_response_test.go
View File

@ -5,7 +5,9 @@ import (
"regexp"
"sync"
"testing"
"time"
"github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/assert"
@ -35,7 +37,7 @@ func TestTCPError(t *testing.T) {
// Error
err1 := c.Gather(&acc)
require.Error(t, err1)
assert.Equal(t, "dial tcp 127.0.0.1:9999: getsockopt: connection refused", err1.Error())
assert.Contains(t, err1.Error(), "getsockopt: connection refused")
}
func TestTCPOK1(t *testing.T) {
@ -46,8 +48,8 @@ func TestTCPOK1(t *testing.T) {
Address: "127.0.0.1:2004",
Send: "test",
Expect: "test",
ReadTimeout: 3.0,
Timeout: 1.0,
ReadTimeout: internal.Duration{Duration: time.Second * 3},
Timeout: internal.Duration{Duration: time.Second},
Protocol: "tcp",
}
// Start TCP server
@ -86,8 +88,8 @@ func TestTCPOK2(t *testing.T) {
Address: "127.0.0.1:2004",
Send: "test",
Expect: "test2",
ReadTimeout: 3.0,
Timeout: 1.0,
ReadTimeout: internal.Duration{Duration: time.Second * 3},
Timeout: internal.Duration{Duration: time.Second},
Protocol: "tcp",
}
// Start TCP server
@ -141,8 +143,8 @@ func TestUDPOK1(t *testing.T) {
Address: "127.0.0.1:2004",
Send: "test",
Expect: "test",
ReadTimeout: 3.0,
Timeout: 1.0,
ReadTimeout: internal.Duration{Duration: time.Second * 3},
Timeout: internal.Duration{Duration: time.Second},
Protocol: "udp",
}
// Start UDP server