Adding redis cluster support (oliver006#531)

Adding redis cluster support

README.md

@@ -154,6 +154,7 @@ redis-only-metrics          | REDIS_EXPORTER_REDIS_ONLY_METRICS          | Wheth
 include-system-metrics      | REDIS_EXPORTER_INCL_SYSTEM_METRICS         | Whether to include system metrics like `total_system_memory_bytes`, defaults to false.
 ping-on-connect             | REDIS_EXPORTER_PING_ON_CONNECT             | Whether to ping the redis instance after connecting and record the duration as a metric, defaults to false.
 is-tile38                   | REDIS_EXPORTER_IS_TILE38                   | Whether to scrape Tile38 specific metrics, defaults to false.
+is-cluster                  | REDIS_EXPORTER_IS_CLUSTER                  | Whether this is a redis cluster (Enable this if you need to fetch key level data on a Redis Cluster).
 export-client-list          | REDIS_EXPORTER_EXPORT_CLIENT_LIST          | Whether to scrape Client List specific metrics, defaults to false.
 export-client-port          | REDIS_EXPORTER_EXPORT_CLIENT_PORT          | Whether to include the client's port when exporting the client list. Warning: including the port increases the number of metrics generated and will make your Prometheus server take up more memory
 skip-tls-verification       | REDIS_EXPORTER_SKIP_TLS_VERIFICATION       | Whether to to skip TLS verification

exporter/exporter.go

@@ -66,6 +66,7 @@ type Options struct {
 	SkipTLSVerification   bool
 	SetClientName         bool
 	IsTile38              bool
+	IsCluster             bool
 	ExportClientList      bool
 	ExportClientsInclPort bool
 	ConnectionTimeouts    time.Duration
@@ -559,7 +560,18 @@ func (e *Exporter) scrapeRedisHost(ch chan<- prometheus.Metric) error {
 
 	e.extractLatencyMetrics(ch, c)
 
-	e.extractCheckKeyMetrics(ch, c)
+	if e.options.IsCluster {
+		clusterClient, err := e.connectToRedisCluster()
+		if err != nil {
+			log.Errorf("Couldn't connect to redis cluster")
+			return err
+		}
+		defer clusterClient.Close()
+
+		e.extractCheckKeyMetrics(ch, clusterClient)
+	} else {
+		e.extractCheckKeyMetrics(ch, c)
+	}
 
 	e.extractSlowLogMetrics(ch, c)
 

exporter/exporter_test.go

@@ -59,40 +59,30 @@ func getTestExporterWithOptions(opt Options) *Exporter {
 	return e
 }
 
-func setupDBKeys(t *testing.T, uri string) error {
-	c, err := redis.DialURL(uri)
-	if err != nil {
-		t.Errorf("couldn't setup redis for uri %s, err: %s ", uri, err)
-		return err
-	}
-	defer c.Close()
-
+func setupKeys(t *testing.T, c redis.Conn, dbNumStr string) error {
 	if _, err := c.Do("SELECT", dbNumStr); err != nil {
 		log.Printf("setupDBKeys() - couldn't setup redis, err: %s ", err)
 		// not failing on this one - cluster doesn't allow for SELECT so we log and ignore the error
 	}
 
 	for _, key := range keys {
-		_, err = c.Do("SET", key, TestValue)
-		if err != nil {
+		if _, err := c.Do("SET", key, TestValue); err != nil {
 			t.Errorf("couldn't setup redis, err: %s ", err)
 			return err
 		}
 	}
 
 	// setting to expire in 300 seconds, should be plenty for a test run
 	for _, key := range keysExpiring {
-		_, err = c.Do("SETEX", key, "300", TestValue)
-		if err != nil {
+		if _, err := c.Do("SETEX", key, "300", TestValue); err != nil {
 			t.Errorf("couldn't setup redis, err: %s ", err)
 			return err
 		}
 	}
 
 	for _, key := range listKeys {
 		for _, val := range keys {
-			_, err = c.Do("LPUSH", key, val)
-			if err != nil {
+			if _, err := c.Do("LPUSH", key, val); err != nil {
 				t.Errorf("couldn't setup redis, err: %s ", err)
 				return err
 			}
@@ -112,20 +102,10 @@ func setupDBKeys(t *testing.T, uri string) error {
 	c.Do("XREADGROUP", "GROUP", "test_group_1", "test_consumer_2", "COUNT", "1", "STREAMS", TestStreamName, ">")
 	c.Do("XREADGROUP", "GROUP", "test_group_2", "test_consumer_1", "COUNT", "1", "STREAMS", TestStreamName, "0")
 
-	time.Sleep(time.Millisecond * 50)
-
 	return nil
 }
 
-func deleteKeysFromDB(t *testing.T, addr string) error {
-
-	c, err := redis.DialURL(addr)
-	if err != nil {
-		t.Errorf("couldn't setup redis, err: %s ", err)
-		return err
-	}
-	defer c.Close()
-
+func deleteKeys(c redis.Conn, dbNumStr string) {
 	if _, err := c.Do("SELECT", dbNumStr); err != nil {
 		log.Printf("deleteKeysFromDB() - couldn't setup redis, err: %s ", err)
 		// not failing on this one - cluster doesn't allow for SELECT so we log and ignore the error
@@ -145,6 +125,71 @@ func deleteKeysFromDB(t *testing.T, addr string) error {
 
 	c.Do("DEL", TestSetName)
 	c.Do("DEL", TestStreamName)
+}
+
+func setupDBKeys(t *testing.T, uri string) error {
+	c, err := redis.DialURL(uri)
+	if err != nil {
+		t.Errorf("couldn't setup redis for uri %s, err: %s ", uri, err)
+		return err
+	}
+	defer c.Close()
+
+	err = setupKeys(t, c, dbNumStr)
+	if err != nil {
+		t.Errorf("couldn't setup redis, err: %s ", err)
+		return err
+	}
+
+	time.Sleep(time.Millisecond * 50)
+
+	return nil
+}
+
+func setupDBKeysCluster(t *testing.T, uri string) error {
+	e := Exporter{redisAddr: uri}
+	c, err := e.connectToRedisCluster()
+	if err != nil {
+		return err
+	}
+
+	defer c.Close()
+
+	err = setupKeys(t, c, "0")
+	if err != nil {
+		t.Errorf("couldn't setup redis, err: %s ", err)
+		return err
+	}
+
+	time.Sleep(time.Millisecond * 50)
+
+	return nil
+}
+
+func deleteKeysFromDB(t *testing.T, addr string) error {
+	c, err := redis.DialURL(addr)
+	if err != nil {
+		t.Errorf("couldn't setup redis, err: %s ", err)
+		return err
+	}
+	defer c.Close()
+
+	deleteKeys(c, dbNumStr)
+
+	return nil
+}
+
+func deleteKeysFromDBCluster(t *testing.T, addr string) error {
+	e := Exporter{redisAddr: addr}
+	c, err := e.connectToRedisCluster()
+	if err != nil {
+		return err
+	}
+
+	defer c.Close()
+
+	deleteKeys(c, "0")
+
 	return nil
 }
 

exporter/keys.go

@@ -95,9 +95,14 @@ func (e *Exporter) extractCheckKeyMetrics(ch chan<- prometheus.Metric, c redis.C
 
 	log.Debugf("allKeys: %#v", allKeys)
 	for _, k := range allKeys {
-		if _, err := doRedisCmd(c, "SELECT", k.db); err != nil {
-			log.Errorf("Couldn't select database %#v when getting key info.", k.db)
-			continue
+		if e.options.IsCluster {
+			//Cluster mode only has one db
+			k.db = "0"
+		} else {
+			if _, err := doRedisCmd(c, "SELECT", k.db); err != nil {
+				log.Errorf("Couldn't select database %#v when getting key info.", k.db)
+				continue
+			}
 		}
 
 		dbLabel := "db" + k.db

exporter/keys_test.go

@@ -52,6 +52,39 @@ func TestKeyValuesAndSizes(t *testing.T) {
 	}
 }
 
+func TestClusterKeyValuesAndSizes(t *testing.T) {
+	e, _ := NewRedisExporter(
+		os.Getenv("TEST_REDIS_CLUSTER_MASTER_URI"),
+		Options{Namespace: "test", CheckSingleKeys: dbNumStrFull + "=" + url.QueryEscape(keys[0]), IsCluster: true},
+	)
+
+	uri := os.Getenv("TEST_REDIS_CLUSTER_MASTER_URI")
+
+	setupDBKeysCluster(t, uri)
+	defer deleteKeysFromDBCluster(t, uri)
+
+	chM := make(chan prometheus.Metric)
+	go func() {
+		e.Collect(chM)
+		close(chM)
+	}()
+
+	want := map[string]bool{"test_key_size": false, "test_key_value": false}
+
+	for m := range chM {
+		for k := range want {
+			if strings.Contains(m.Desc().String(), k) {
+				want[k] = true
+			}
+		}
+	}
+	for k, found := range want {
+		if !found {
+			t.Errorf("didn't find %s", k)
+		}
+	}
+}
+
 func TestParseKeyArg(t *testing.T) {
 	for _, test := range []struct {
 		name          string

exporter/redis.go

@@ -1,13 +1,16 @@
 package exporter
 
 import (
+	"net/url"
 	"strings"
+	"time"
 
 	"github.com/gomodule/redigo/redis"
+	"github.com/mna/redisc"
 	log "github.com/sirupsen/logrus"
 )
 
-func (e *Exporter) connectToRedis() (redis.Conn, error) {
+func (e *Exporter) configureOptions(uri string) ([]redis.DialOption, error) {
 	tlsConfig, err := e.CreateClientTLSConfig()
 	if err != nil {
 		return nil, err
@@ -28,13 +31,22 @@ func (e *Exporter) connectToRedis() (redis.Conn, error) {
 		options = append(options, redis.DialPassword(e.options.Password))
 	}
 
+	if e.options.PasswordMap[uri] != "" {
+		options = append(options, redis.DialPassword(e.options.PasswordMap[uri]))
+	}
+
+	return options, nil
+}
+
+func (e *Exporter) connectToRedis() (redis.Conn, error) {
 	uri := e.redisAddr
 	if !strings.Contains(uri, "://") {
 		uri = "redis://" + uri
 	}
 
-	if e.options.PasswordMap[uri] != "" {
-		options = append(options, redis.DialPassword(e.options.PasswordMap[uri]))
+	options, err := e.configureOptions(uri)
+	if err != nil {
+		return nil, err
 	}
 
 	log.Debugf("Trying DialURL(): %s", uri)
@@ -52,6 +64,50 @@ func (e *Exporter) connectToRedis() (redis.Conn, error) {
 	return c, err
 }
 
+func (e *Exporter) connectToRedisCluster() (redis.Conn, error) {
+	uri := e.redisAddr
+	if strings.Contains(uri, "://") {
+		url, _ := url.Parse(uri)
+		if url.Port() == "" {
+			uri = url.Host + ":6379"
+		} else {
+			uri = url.Host
+		}
+	} else {
+		if frags := strings.Split(uri, ":"); len(frags) != 2 {
+			uri = uri + ":6379"
+		}
+	}
+
+	options, err := e.configureOptions(uri)
+	if err != nil {
+		return nil, err
+	}
+
+	log.Debugf("Creating cluster object")
+	cluster := redisc.Cluster{
+		StartupNodes: []string{uri},
+		DialOptions:  options,
+	}
+	log.Debugf("Running refresh on cluster object")
+	if err := cluster.Refresh(); err != nil {
+		log.Errorf("Cluster refresh failed: %v", err)
+	}
+
+	log.Debugf("Creating redis connection object")
+	conn, err := cluster.Dial()
+	if err != nil {
+		log.Errorf("Dial failed: %v", err)
+	}
+
+	c, err := redisc.RetryConn(conn, 10, 100*time.Millisecond)
+	if err != nil {
+		log.Errorf("RetryConn failed: %v", err)
+	}
+
+	return c, err
+}
+
 func doRedisCmd(c redis.Conn, cmd string, args ...interface{}) (interface{}, error) {
 	log.Debugf("c.Do() - running command: %s %s", cmd, args)
 	res, err := c.Do(cmd, args...)

go.mod

@@ -5,6 +5,7 @@ go 1.13
 require (
 	github.com/gomodule/redigo v1.8.5
 	github.com/prometheus/client_golang v1.11.0
+	github.com/mna/redisc v1.1.7
 	github.com/prometheus/client_model v0.2.0
 	github.com/sirupsen/logrus v1.8.1
 )

go.sum

@@ -57,6 +57,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
+github.com/mna/redisc v1.1.7 h1:FdmtJsfTjoIjNXiQf4ozgNjuE+zxWH+fJSe+I/dD4vc=
+github.com/mna/redisc v1.1.7/go.mod h1:GXeOb7zyYKiT+K8MKdIiJvuv7MfhDoQGcuzfiJQmqQI=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=