Skip to content

Commit

Permalink
add ability to read redis from a socket
Browse files Browse the repository at this point in the history
  • Loading branch information
mmckinst committed Jul 14, 2016
1 parent c046232 commit 6c76449
Showing 1 changed file with 34 additions and 14 deletions.
48 changes: 34 additions & 14 deletions plugins/inputs/redis/redis.go
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ var sampleConfig = `
## e.g.
## tcp://localhost:6379
## tcp://:[email protected]
## unix:///var/run/redis.sock
##
## If no servers are specified, then localhost is used as the host.
## If no port is specified, 6379 is used
Expand Down Expand Up @@ -79,12 +80,15 @@ var Tracking = map[string]string{

var ErrProtocolError = errors.New("redis protocol error")

const defaultPort = "6379"

// Reads stats from all configured servers accumulates stats.
// Returns one of the errors encountered while gather stats (if any).
func (r *Redis) Gather(acc telegraf.Accumulator) error {
if len(r.Servers) == 0 {
url := &url.URL{
Host: ":6379",
Scheme: "tcp",
Host: ":6379",
}
r.gatherServer(url, acc)
return nil
Expand All @@ -95,6 +99,10 @@ func (r *Redis) Gather(acc telegraf.Accumulator) error {
var outerr error

for _, serv := range r.Servers {
if !strings.HasPrefix(serv, "tcp://") || !strings.HasPrefix(serv, "unix://") {
serv = "tcp://" + serv
}

u, err := url.Parse(serv)
if err != nil {
return fmt.Errorf("Unable to parse to address '%s': %s", serv, err)
Expand All @@ -104,6 +112,13 @@ func (r *Redis) Gather(acc telegraf.Accumulator) error {
u.Host = serv
u.Path = ""
}
if u.Scheme == "tcp" {
_, _, err := net.SplitHostPort(u.Host)
if err != nil {
u.Host = u.Host + ":" + defaultPort
}
}

wg.Add(1)
go func(serv string) {
defer wg.Done()
Expand All @@ -116,17 +131,17 @@ func (r *Redis) Gather(acc telegraf.Accumulator) error {
return outerr
}

const defaultPort = "6379"

func (r *Redis) gatherServer(addr *url.URL, acc telegraf.Accumulator) error {
_, _, err := net.SplitHostPort(addr.Host)
if err != nil {
addr.Host = addr.Host + ":" + defaultPort
}
var address string

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 {
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()

Expand Down Expand Up @@ -154,12 +169,17 @@ func (r *Redis) gatherServer(addr *url.URL, acc telegraf.Accumulator) error {
c.Write([]byte("EOF\r\n"))
rdr := bufio.NewReader(c)

// 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}
var tags map[string]string

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)
}

Expand Down

0 comments on commit 6c76449

Please sign in to comment.