Skip to content

Commit

Permalink
Add tcp/udp check connection input plugin
Browse files Browse the repository at this point in the history
  • Loading branch information
titilambert committed Feb 6, 2016
1 parent 9a59512 commit 27d22e1
Show file tree
Hide file tree
Showing 4 changed files with 483 additions and 0 deletions.
1 change: 1 addition & 0 deletions plugins/inputs/all/all.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
_ "github.com/influxdata/telegraf/plugins/inputs/aerospike"
_ "github.com/influxdata/telegraf/plugins/inputs/apache"
_ "github.com/influxdata/telegraf/plugins/inputs/bcache"
_ "github.com/influxdata/telegraf/plugins/inputs/connection"
_ "github.com/influxdata/telegraf/plugins/inputs/couchdb"
_ "github.com/influxdata/telegraf/plugins/inputs/disque"
_ "github.com/influxdata/telegraf/plugins/inputs/docker"
Expand Down
66 changes: 66 additions & 0 deletions plugins/inputs/connection/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
# Example Input Plugin

The input plugin test UDP/TCP connections response time.
It can also check response text.

### Configuration:

```
# List of UDP/TCP connections you want to check
[[inputs.connection]]
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.connection]]
protocol = "tcp"
address = ":80"
[[inputs.connection]]
protocol = "udp"
# 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.connection]]
protocol = "udp"
address = "localhost:161"
timeout = 2.0
```

### Measurements & Fields:

- connection
- response_time (float, seconds)
- string_found (bool) # Only if "expected: option is set

### Tags:

- All measurements have the following tags:
- host
- port
- protocol

### Example Output:

```
$ ./telegraf -config telegraf.conf -input-filter example -test
connection,host=127.0.0.1,port=22,protocol=tcp response_time=0.18070360500000002,string_found=true 1454785464182527094
connection,host=127.0.0.1,port=2222,protocol=tcp response_time=1.090124776,string_found=false 1454784433658942325
```
218 changes: 218 additions & 0 deletions plugins/inputs/connection/connection.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,218 @@
package connection

import (
"bufio"
"errors"
"net"
"net/textproto"
"regexp"
"time"

"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/plugins/inputs"
)

// Connections struct
type Connection struct {
Address string
Timeout float64
ReadTimeout float64
Send string
Expect string
Protocol string
}

func (_ *Connection) Description() string {
return "Ping given url(s) and return statistics"
}

var sampleConfig = `
[[inputs.connection]]
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.connection]]
protocol = "tcp"
address = ":80"
[[inputs.connection]]
protocol = "udp"
# 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.connection]]
protocol = "udp"
address = "localhost:161"
timeout = 2.0
`

func (_ *Connection) SampleConfig() string {
return sampleConfig
}

func (t *Connection) 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)
// Stop timer
responseTime := time.Since(start).Seconds()
// Handle error
if err != nil {
return nil, err
}
defer conn.Close()
// Send string if needed
if t.Send != "" {
msg := []byte(t.Send)
conn.Write(msg)
conn.CloseWrite()
// Stop timer
responseTime = time.Since(start).Seconds()
}
// Read string if needed
if t.Expect != "" {
// Set read timeout
conn.SetReadDeadline(time.Now().Add(time.Duration(t.ReadTimeout) * time.Second))
// Prepare reader
reader := bufio.NewReader(conn)
tp := textproto.NewReader(reader)
// Read
data, err := tp.ReadLine()
// Stop timer
responseTime = time.Since(start).Seconds()
// Handle error
if err != nil {
fields["string_found"] = false
} else {
// Looking for string in answer
RegEx := regexp.MustCompile(`.*` + t.Expect + `.*`)
find := RegEx.FindString(string(data))
if find != "" {
fields["string_found"] = true
} else {
fields["string_found"] = false
}
}

}
fields["response_time"] = responseTime
return fields, nil
}

func (u *Connection) 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)
LocalAddr, err := net.ResolveUDPAddr("udp", "127.0.0.1:0")
// Connecting
conn, err := net.DialUDP("udp", LocalAddr, udpAddr)
defer conn.Close()
// Handle error
if err != nil {
return nil, err
}
// Send string
msg := []byte(u.Send)
conn.Write(msg)
// Read string
// Set read timeout
conn.SetReadDeadline(time.Now().Add(time.Duration(u.ReadTimeout) * time.Second))
// Read
buf := make([]byte, 1024)
_, _, err = conn.ReadFromUDP(buf)
// Stop timer
responseTime := time.Since(start).Seconds()
// Handle error
if err != nil {
return nil, err
} else {
// Looking for string in answer
RegEx := regexp.MustCompile(`.*` + u.Expect + `.*`)
find := RegEx.FindString(string(buf))
if find != "" {
fields["string_found"] = true
} else {
fields["string_found"] = false
}
}
fields["response_time"] = responseTime
return fields, nil
}

func (c *Connection) Gather(acc telegraf.Accumulator) error {
// Set default values
if c.Timeout == 0 {
c.Timeout = 1.0
}
if c.ReadTimeout == 0 {
c.ReadTimeout = 1.0
}
// Check send and expected string
if c.Protocol == "udp" && c.Send == "" {
return errors.New("Send string cannot be empty")
}
if c.Protocol == "udp" && c.Expect == "" {
return errors.New("Expected string cannot be empty")
}
// Prepare host and port
host, port, err := net.SplitHostPort(c.Address)
if err != nil {
return err
}
if host == "" {
c.Address = "localhost:" + port
}
if port == "" {
return errors.New("Bad port")
}
// Prepare data
tags := map[string]string{"host": host, "port": port}
var fields map[string]interface{}
// Gather data
if c.Protocol == "tcp" {
fields, err = c.TcpGather()
tags["protocol"] = "tcp"
} else if c.Protocol == "udp" {
fields, err = c.UdpGather()
tags["protocol"] = "udp"
} else {
return errors.New("Bad protocol")
}
if err != nil {
return err
}
// Add metrics
acc.AddFields("connection", fields, tags)
return nil
}

func init() {
inputs.Add("connection", func() telegraf.Input {
return &Connection{}
})
}
Loading

0 comments on commit 27d22e1

Please sign in to comment.