Add powerdns input plugin

plugins/inputs/all/all.go

@@ -26,6 +26,7 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/inputs/phpfpm"
 	_ "github.com/influxdata/telegraf/plugins/inputs/ping"
 	_ "github.com/influxdata/telegraf/plugins/inputs/postgresql"
+	_ "github.com/influxdata/telegraf/plugins/inputs/powerdns"
 	_ "github.com/influxdata/telegraf/plugins/inputs/procstat"
 	_ "github.com/influxdata/telegraf/plugins/inputs/prometheus"
 	_ "github.com/influxdata/telegraf/plugins/inputs/puppetagent"

plugins/inputs/powerdns/README.md

@@ -0,0 +1,68 @@
+# PowerDNS Input Plugin
+
+The powerdns plugin gathers metrics about PowerDNS using unix socket.
+
+### Configuration:
+
+```
+# Description
+[[inputs.powerdns]]
+  # An array of sockets to gather stats about.
+  # Specify a path to unix socket.
+  #
+  # If no servers are specified, then '/var/run/pdns.controlsocket' is used as the path.
+  unix_sockets = ["/var/run/pdns.controlsocket"]
+```
+
+### Measurements & Fields:
+
+- powerdns
+  - corrupt-packets
+  - deferred-cache-inserts
+  - deferred-cache-lookup
+  - dnsupdate-answers
+  - dnsupdate-changes
+  - dnsupdate-queries
+  - dnsupdate-refused
+  - packetcache-hit
+  - packetcache-miss
+  - packetcache-size
+  - query-cache-hit
+  - query-cache-miss
+  - rd-queries
+  - recursing-answers
+  - recursing-questions
+  - recursion-unanswered
+  - security-status
+  - servfail-packets
+  - signatures
+  - tcp-answers
+  - tcp-queries
+  - timedout-packets
+  - udp-answers
+  - udp-answers-bytes
+  - udp-do-queries
+  - udp-queries
+  - udp4-answers
+  - udp4-queries
+  - udp6-answers
+  - udp6-queries
+  - key-cache-size
+  - latency
+  - meta-cache-size
+  - qsize-q
+  - signature-cache-size
+  - sys-msec
+  - uptime
+  - user-msec
+
+### Tags:
+
+- tags: `server=socket`
+
+### Example Output:
+
+```
+$ ./telegraf -config telegraf.conf -input-filter powerdns -test
+> powerdns,server=/var/run/pdns.controlsocket corrupt-packets=0i,deferred-cache-inserts=0i,deferred-cache-lookup=0i,dnsupdate-answers=0i,dnsupdate-changes=0i,dnsupdate-queries=0i,dnsupdate-refused=0i,key-cache-size=0i,latency=26i,meta-cache-size=0i,packetcache-hit=0i,packetcache-miss=1i,packetcache-size=0i,qsize-q=0i,query-cache-hit=0i,query-cache-miss=6i,rd-queries=1i,recursing-answers=0i,recursing-questions=0i,recursion-unanswered=0i,security-status=3i,servfail-packets=0i,signature-cache-size=0i,signatures=0i,sys-msec=4349i,tcp-answers=0i,tcp-queries=0i,timedout-packets=0i,udp-answers=1i,udp-answers-bytes=50i,udp-do-queries=0i,udp-queries=0i,udp4-answers=1i,udp4-queries=1i,udp6-answers=0i,udp6-queries=0i,uptime=166738i,user-msec=3036i 1454078624932715706
+```

plugins/inputs/powerdns/powerdns.go

@@ -0,0 +1,125 @@
+package powerdns
+
+import (
+  "net"
+  "fmt"
+  "bufio"
+  "io"
+  "time"
+  "strings"
+  "strconv"
+
+  "github.com/influxdata/telegraf/plugins/inputs"
+)
+
+type Powerdns struct {
+  UnixSockets []string
+}
+
+var sampleConfig = `
+  # An array of sockets to gather stats about.
+  # Specify a path to unix socket.
+  #
+  # If no servers are specified, then '/var/run/pdns.controlsocket' is used as the path.
+  unix_sockets = ["/var/run/pdns.controlsocket"]
+`
+
+var defaultTimeout = 5 * time.Second
+
+func (p *Powerdns) SampleConfig() string {
+  return sampleConfig
+}
+
+func (p *Powerdns) Description() string {
+  return "Read metrics from one or many PowerDNS servers"
+}
+
+func (p *Powerdns) Gather(acc inputs.Accumulator) error {
+  if len(p.UnixSockets) == 0 {
+    return p.gatherServer("/var/run/pdns.controlsocket", acc)
+  }
+
+  for _, serverSocket := range p.UnixSockets {
+    if err := p.gatherServer(serverSocket, acc); err != nil {
+      return err
+    }
+  }
+
+  return nil
+}
+
+func (p *Powerdns) gatherServer(address string, acc inputs.Accumulator) error {
+  conn, err := net.DialTimeout("unix", address, defaultTimeout)
+  if err != nil {
+    return err
+  }
+
+  defer conn.Close()
+
+  conn.SetDeadline(time.Now().Add(defaultTimeout))
+
+  // Read and write buffer
+  rw := bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn))
+
+  // Send command
+  if _, err := fmt.Fprint(conn, "show * \n"); err != nil {
+    return nil
+  }
+  if err := rw.Flush(); err != nil {
+    return err
+  }
+
+  // Read data
+  buf := make([]byte, 0, 4096)
+  tmp := make([]byte, 1024)
+  for {
+    n, err := rw.Read(tmp)
+    if err != nil {
+      if err != io.EOF {
+        return err
+      }
+
+      break
+    }
+    buf = append(buf, tmp[:n]...)
+  }
+
+  metrics := string(buf)
+
+  // Process data
+  fields, err := parseResponse(metrics)
+  if err != nil {
+    return err
+  }
+
+  // Add server socket as a tag
+	tags := map[string]string{"server": address}
+
+  acc.AddFields("powerdns", fields, tags)
+
+  return nil
+}
+
+func parseResponse(metrics string) (map[string]interface{}, error) {
+  values := make(map[string]interface{})
+
+  s := strings.Split(metrics, ",")
+
+  for _, metric := range s[:len(s)-1] {
+    m := strings.Split(metric, "=")
+
+    i, err := strconv.ParseInt(m[1], 10, 64)
+    if err != nil {
+      return values, err
+    }
+    values[m[0]] = i
+  }
+
+  return values, nil
+}
+
+func init() {
+	inputs.Add("powerdns", func() inputs.Input {
+		return &Powerdns{}
+	})
+}

plugins/inputs/powerdns/powerdns_test.go

@@ -0,0 +1,137 @@
+package powerdns
+
+import (
+	"testing"
+  "encoding/binary"
+  "net"
+  "crypto/rand"
+  "fmt"
+
+	"github.com/influxdata/telegraf/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type statServer struct{}
+var metrics = "corrupt-packets=0,deferred-cache-inserts=0,deferred-cache-lookup=0,dnsupdate-answers=0,dnsupdate-changes=0,dnsupdate-queries=0,dnsupdate-refused=0,packetcache-hit=0,packetcache-miss=1,packetcache-size=0,query-cache-hit=0,query-cache-miss=6,rd-queries=1,recursing-answers=0,recursing-questions=0,recursion-unanswered=0,security-status=3,servfail-packets=0,signatures=0,tcp-answers=0,tcp-queries=0,timedout-packets=0,udp-answers=1,udp-answers-bytes=50,udp-do-queries=0,udp-queries=0,udp4-answers=1,udp4-queries=1,udp6-answers=0,udp6-queries=0,key-cache-size=0,latency=26,meta-cache-size=0,qsize-q=0,signature-cache-size=0,sys-msec=2889,uptime=86317,user-msec=2167,"
+
+func (s statServer) serverSocket(l net.Listener) {
+
+  for {
+    conn, err := l.Accept()
+    if err != nil {
+      return
+    }
+
+    go func(c net.Conn) {
+      buf := make([]byte, 1024)
+      n, _ := c.Read(buf)
+
+      data := buf[:n]
+      if string(data) == "show * \n" {
+        c.Write([]byte(metrics))
+        c.Close()
+      }
+    }(conn)
+  }
+}
+
+func TestMemcachedGeneratesMetrics(t *testing.T) {
+  // We create a fake server to return test data
+  var randomNumber int64
+	binary.Read(rand.Reader, binary.LittleEndian, &randomNumber)
+  socket, err := net.Listen("unix", fmt.Sprintf("/tmp/pdns%d.controlsocket", randomNumber))
+  if err != nil {
+    t.Fatal("Cannot initalize server on port ")
+  }
+
+  defer socket.Close()
+
+  s := statServer{}
+  go s.serverSocket(socket)
+
+  p := &Powerdns{
+		UnixSockets: []string{fmt.Sprintf("/tmp/pdns%d.controlsocket", randomNumber)},
+	}
+
+  var acc testutil.Accumulator
+
+  err = p.Gather(&acc)
+	require.NoError(t, err)
+
+  intMetrics := []string{"corrupt-packets", "deferred-cache-inserts",
+    "deferred-cache-lookup", "dnsupdate-answers", "dnsupdate-changes",
+    "dnsupdate-queries", "dnsupdate-refused", "packetcache-hit",
+    "packetcache-miss", "packetcache-size", "query-cache-hit", "query-cache-miss",
+    "rd-queries", "recursing-answers", "recursing-questions",
+    "recursion-unanswered", "security-status", "servfail-packets", "signatures",
+    "tcp-answers", "tcp-queries", "timedout-packets", "udp-answers",
+    "udp-answers-bytes", "udp-do-queries", "udp-queries", "udp4-answers",
+    "udp4-queries", "udp6-answers", "udp6-queries", "key-cache-size", "latency",
+    "meta-cache-size", "qsize-q", "signature-cache-size", "sys-msec", "uptime", "user-msec"}
+
+	for _, metric := range intMetrics {
+		assert.True(t, acc.HasIntField("powerdns", metric), metric)
+	}
+}
+
+func TestPowerdnsParseMetrics(t *testing.T) {
+  values, err := parseResponse(metrics)
+  require.NoError(t, err, "Error parsing memcached response")
+
+  tests := []struct {
+		key   string
+		value int64
+	}{
+		{"corrupt-packets", 0},
+		{"deferred-cache-inserts", 0},
+		{"deferred-cache-lookup", 0},
+    {"dnsupdate-answers", 0},
+    {"dnsupdate-changes", 0},
+    {"dnsupdate-queries", 0},
+    {"dnsupdate-refused", 0},
+    {"packetcache-hit", 0},
+    {"packetcache-miss", 1},
+    {"packetcache-size", 0},
+    {"query-cache-hit", 0},
+    {"query-cache-miss", 6},
+    {"rd-queries", 1},
+    {"recursing-answers", 0},
+    {"recursing-questions", 0},
+    {"recursion-unanswered", 0},
+    {"security-status", 3},
+    {"servfail-packets", 0},
+    {"signatures", 0},
+    {"tcp-answers", 0},
+    {"tcp-queries", 0},
+    {"timedout-packets", 0},
+    {"udp-answers", 1},
+    {"udp-answers-bytes", 50},
+    {"udp-do-queries", 0},
+    {"udp-queries", 0},
+    {"udp4-answers", 1},
+    {"udp4-queries", 1},
+    {"udp6-answers", 0},
+    {"udp6-queries", 0},
+    {"key-cache-size", 0},
+    {"latency", 26},
+    {"meta-cache-size", 0},
+    {"qsize-q" ,0},
+    {"signature-cache-size", 0},
+    {"sys-msec", 2889},
+    {"uptime", 86317},
+    {"user-msec", 2167},
+  }
+
+	for _, test := range tests {
+		value, ok := values[test.key]
+		if !ok {
+			t.Errorf("Did not find key for metric %s in values", test.key)
+			continue
+		}
+		if value != test.value {
+			t.Errorf("Metric: %s, Expected: %s, actual: %s",
+				test.key, test.value, value)
+		}
+	}
+}