Add tcp/udp check connection input plugin

plugins/inputs/all/all.go

@@ -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"

plugins/inputs/connection/README.md

@@ -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 connection -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
+
+```

plugins/inputs/connection/connection.go

@@ -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{}
+	})
+}