add ability to read redis from a socket (#1480)

* add ability to read redis from a socket

* update CHANGELOG
This commit is contained in:
Mark McKinstry 2016-07-18 07:03:39 -04:00 committed by Cameron Sparr
parent b4a6d9c647
commit 2d6c8767f7
2 changed files with 35 additions and 14 deletions

View File

@ -39,6 +39,7 @@ should now look like:
- [#1402](https://github.com/influxdata/telegraf/pull/1402): docker-machine/boot2docker no longer required for unit tests. - [#1402](https://github.com/influxdata/telegraf/pull/1402): docker-machine/boot2docker no longer required for unit tests.
- [#1350](https://github.com/influxdata/telegraf/pull/1350): cgroup input plugin. - [#1350](https://github.com/influxdata/telegraf/pull/1350): cgroup input plugin.
- [#1369](https://github.com/influxdata/telegraf/pull/1369): Add input plugin for consuming metrics from NSQD. - [#1369](https://github.com/influxdata/telegraf/pull/1369): Add input plugin for consuming metrics from NSQD.
- [#1369](https://github.com/influxdata/telegraf/pull/1480): add ability to read redis from a socket.
- [#1387](https://github.com/influxdata/telegraf/pull/1387): **Breaking Change** - Redis `role` tag renamed to `replication_role` to avoid global_tags override - [#1387](https://github.com/influxdata/telegraf/pull/1387): **Breaking Change** - Redis `role` tag renamed to `replication_role` to avoid global_tags override
- [#1437](https://github.com/influxdata/telegraf/pull/1437): Fetching Galera status metrics in MySQL - [#1437](https://github.com/influxdata/telegraf/pull/1437): Fetching Galera status metrics in MySQL
- [#1500](https://github.com/influxdata/telegraf/pull/1500): Aerospike plugin refactored to use official client lib. - [#1500](https://github.com/influxdata/telegraf/pull/1500): Aerospike plugin refactored to use official client lib.

View File

@ -25,6 +25,7 @@ var sampleConfig = `
## e.g. ## e.g.
## tcp://localhost:6379 ## tcp://localhost:6379
## tcp://:password@192.168.99.100 ## tcp://:password@192.168.99.100
## unix:///var/run/redis.sock
## ##
## If no servers are specified, then localhost is used as the host. ## If no servers are specified, then localhost is used as the host.
## If no port is specified, 6379 is used ## If no port is specified, 6379 is used
@ -80,12 +81,15 @@ var Tracking = map[string]string{
var ErrProtocolError = errors.New("redis protocol error") var ErrProtocolError = errors.New("redis protocol error")
const defaultPort = "6379"
// Reads stats from all configured servers accumulates stats. // Reads stats from all configured servers accumulates stats.
// Returns one of the errors encountered while gather stats (if any). // Returns one of the errors encountered while gather stats (if any).
func (r *Redis) Gather(acc telegraf.Accumulator) error { func (r *Redis) Gather(acc telegraf.Accumulator) error {
if len(r.Servers) == 0 { if len(r.Servers) == 0 {
url := &url.URL{ url := &url.URL{
Host: ":6379", Scheme: "tcp",
Host: ":6379",
} }
r.gatherServer(url, acc) r.gatherServer(url, acc)
return nil return nil
@ -96,6 +100,10 @@ func (r *Redis) Gather(acc telegraf.Accumulator) error {
var outerr error var outerr error
for _, serv := range r.Servers { for _, serv := range r.Servers {
if !strings.HasPrefix(serv, "tcp://") || !strings.HasPrefix(serv, "unix://") {
serv = "tcp://" + serv
}
u, err := url.Parse(serv) u, err := url.Parse(serv)
if err != nil { if err != nil {
return fmt.Errorf("Unable to parse to address '%s': %s", serv, err) return fmt.Errorf("Unable to parse to address '%s': %s", serv, err)
@ -105,6 +113,13 @@ func (r *Redis) Gather(acc telegraf.Accumulator) error {
u.Host = serv u.Host = serv
u.Path = "" u.Path = ""
} }
if u.Scheme == "tcp" {
_, _, err := net.SplitHostPort(u.Host)
if err != nil {
u.Host = u.Host + ":" + defaultPort
}
}
wg.Add(1) wg.Add(1)
go func(serv string) { go func(serv string) {
defer wg.Done() defer wg.Done()
@ -117,17 +132,17 @@ func (r *Redis) Gather(acc telegraf.Accumulator) error {
return outerr return outerr
} }
const defaultPort = "6379"
func (r *Redis) gatherServer(addr *url.URL, acc telegraf.Accumulator) error { func (r *Redis) gatherServer(addr *url.URL, acc telegraf.Accumulator) error {
_, _, err := net.SplitHostPort(addr.Host) var address string
if err != nil {
addr.Host = addr.Host + ":" + defaultPort
}
c, err := net.DialTimeout("tcp", addr.Host, defaultTimeout) if addr.Scheme == "unix" {
address = addr.Path
} else {
address = addr.Host
}
c, err := net.DialTimeout(addr.Scheme, address, defaultTimeout)
if err != nil { if err != nil {
return fmt.Errorf("Unable to connect to redis server '%s': %s", addr.Host, err) return fmt.Errorf("Unable to connect to redis server '%s': %s", address, err)
} }
defer c.Close() defer c.Close()
@ -155,12 +170,17 @@ func (r *Redis) gatherServer(addr *url.URL, acc telegraf.Accumulator) error {
c.Write([]byte("EOF\r\n")) c.Write([]byte("EOF\r\n"))
rdr := bufio.NewReader(c) rdr := bufio.NewReader(c)
// Setup tags for all redis metrics var tags map[string]string
host, port := "unknown", "unknown"
// If there's an error, ignore and use 'unknown' tags
host, port, _ = net.SplitHostPort(addr.Host)
tags := map[string]string{"server": host, "port": port}
if addr.Scheme == "unix" {
tags = map[string]string{"socket": addr.Path}
} else {
// Setup tags for all redis metrics
host, port := "unknown", "unknown"
// If there's an error, ignore and use 'unknown' tags
host, port, _ = net.SplitHostPort(addr.Host)
tags = map[string]string{"server": host, "port": port}
}
return gatherInfoOutput(rdr, acc, tags) return gatherInfoOutput(rdr, acc, tags)
} }