Metricbeat redis refactors

metricbeat/module/redis/info/data.go

@@ -18,9 +18,9 @@
 package info
 
 import (
-	"github.com/elastic/beats/libbeat/common"
 	s "github.com/elastic/beats/libbeat/common/schema"
 	c "github.com/elastic/beats/libbeat/common/schema/mapstrstr"
+	"github.com/elastic/beats/metricbeat/mb"
 )
 
 var (
@@ -237,12 +237,14 @@ var (
 )
 
 // Map data to MapStr
-func eventMapping(info map[string]string) common.MapStr {
+func eventMapping(r mb.ReporterV2, info map[string]string) {
 	// Full mapping from info
 	source := map[string]interface{}{}
 	for key, val := range info {
 		source[key] = val
 	}
 	data, _ := schema.Apply(source)
-	return data
+	r.Event(mb.Event{
+		MetricSetFields: data,
+	})
 }

metricbeat/module/redis/info/info.go

@@ -19,15 +19,13 @@ package info
 
 import (
 	"strconv"
-	"time"
 
-	"github.com/elastic/beats/libbeat/common"
+	"github.com/pkg/errors"
+
 	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/mb/parse"
 	"github.com/elastic/beats/metricbeat/module/redis"
-
-	rd "github.com/garyburd/redigo/redis"
 )
 
 var (
@@ -43,41 +41,25 @@ func init() {
 
 // MetricSet for fetching Redis server information and statistics.
 type MetricSet struct {
-	mb.BaseMetricSet
-	pool *rd.Pool
+	*redis.MetricSet
 }
 
 // New creates new instance of MetricSet
 func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
-	// Unpack additional configuration options.
-	config := struct {
-		IdleTimeout time.Duration `config:"idle_timeout"`
-		Network     string        `config:"network"`
-		MaxConn     int           `config:"maxconn" validate:"min=1"`
-		Password    string        `config:"password"`
-	}{
-		Network:  "tcp",
-		MaxConn:  10,
-		Password: "",
-	}
-	err := base.Module().UnpackConfig(&config)
+	ms, err := redis.NewMetricSet(base)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrap(err, "failed to create 'info' metricset")
 	}
-
-	return &MetricSet{
-		BaseMetricSet: base,
-		pool: redis.CreatePool(base.Host(), config.Password, config.Network,
-			config.MaxConn, config.IdleTimeout, base.Module().Config().Timeout),
-	}, nil
+	return &MetricSet{ms}, nil
 }
 
 // Fetch fetches metrics from Redis by issuing the INFO command.
-func (m *MetricSet) Fetch() (common.MapStr, error) {
+func (m *MetricSet) Fetch(r mb.ReporterV2) {
 	// Fetch default INFO.
-	info, err := redis.FetchRedisInfo("default", m.pool.Get())
+	info, err := redis.FetchRedisInfo("default", m.Connection())
 	if err != nil {
-		return nil, err
+		logp.Err("Failed to fetch redis info: %s", err)
+		return
 	}
 
 	// In 5.0 some fields are renamed, maintain both names, old ones will be deprecated
@@ -95,12 +77,13 @@ func (m *MetricSet) Fetch() (common.MapStr, error) {
 		}
 	}
 
-	slowLogLength, err := redis.FetchSlowLogLength(m.pool.Get())
+	slowLogLength, err := redis.FetchSlowLogLength(m.Connection())
 	if err != nil {
-		return nil, err
+		logp.Err("Failed to fetch slow log length: %s", err)
+		return
 	}
 	info["slowlog_len"] = strconv.FormatInt(slowLogLength, 10)
 
 	debugf("Redis INFO from %s: %+v", m.Host(), info)
-	return eventMapping(info), nil
+	eventMapping(r, info)
 }

metricbeat/module/redis/info/info_integration_test.go

@@ -27,26 +27,22 @@ import (
 	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
 	"github.com/elastic/beats/metricbeat/module/redis"
 
-	rd "github.com/garyburd/redigo/redis"
 	"github.com/stretchr/testify/assert"
 )
 
-const (
-	password = "foobared"
-)
-
 var redisHost = redis.GetRedisEnvHost() + ":" + redis.GetRedisEnvPort()
 
 func TestFetch(t *testing.T) {
 	compose.EnsureUp(t, "redis")
 
-	f := mbtest.NewEventFetcher(t, getConfig(""))
-	event, err := f.Fetch()
+	ms := mbtest.NewReportingMetricSetV2(t, getConfig())
+	events, err := mbtest.ReportingFetchV2(ms)
 	if err != nil {
 		t.Fatal("fetch", err)
 	}
+	event := events[0].MetricSetFields
 
-	t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event)
+	t.Logf("%s/%s event: %+v", ms.Module().Name(), ms.Name(), event)
 
 	// Check fields
 	assert.Equal(t, 9, len(event))
@@ -57,96 +53,17 @@ func TestFetch(t *testing.T) {
 func TestData(t *testing.T) {
 	compose.EnsureUp(t, "redis")
 
-	f := mbtest.NewEventFetcher(t, getConfig(""))
-
-	err := mbtest.WriteEvent(f, t)
+	ms := mbtest.NewReportingMetricSetV2(t, getConfig())
+	err := mbtest.WriteEventsReporterV2(ms, t, "")
 	if err != nil {
 		t.Fatal("write", err)
 	}
 }
 
-func TestPasswords(t *testing.T) {
-	compose.EnsureUp(t, "redis")
-
-	// Add password and ensure it gets reset
-	defer func() {
-		err := resetPassword(redisHost, password)
-		if err != nil {
-			t.Fatal("resetting password", err)
-		}
-	}()
-
-	err := addPassword(redisHost, password)
-	if err != nil {
-		t.Fatal("adding password", err)
-	}
-
-	// Test Fetch metrics with missing password
-	f := mbtest.NewEventFetcher(t, getConfig(""))
-	_, err = f.Fetch()
-	if assert.Error(t, err, "missing password") {
-		assert.Contains(t, err, "NOAUTH Authentication required.")
-	}
-
-	// Config redis and metricset with an invalid password
-	f = mbtest.NewEventFetcher(t, getConfig("blah"))
-	_, err = f.Fetch()
-	if assert.Error(t, err, "invalid password") {
-		assert.Contains(t, err, "ERR invalid password")
-	}
-
-	// Config redis and metricset with a valid password
-	f = mbtest.NewEventFetcher(t, getConfig(password))
-	_, err = f.Fetch()
-	assert.NoError(t, err, "valid password")
-}
-
-// addPassword will add a password to redis.
-func addPassword(host, pass string) error {
-	c, err := rd.Dial("tcp", host)
-	if err != nil {
-		return err
-	}
-	defer c.Close()
-
-	_, err = c.Do("CONFIG", "SET", "requirepass", pass)
-	return err
-}
-
-// resetPassword changes the password to the redis DB.
-func resetPassword(host, currentPass string) error {
-	c, err := rd.Dial("tcp", host)
-	if err != nil {
-		return err
-	}
-	defer c.Close()
-
-	_, err = c.Do("AUTH", currentPass)
-	if err != nil {
-		return err
-	}
-
-	_, err = c.Do("CONFIG", "SET", "requirepass", "")
-	return err
-}
-
-// writeToRedis will write to the default DB 0.
-func writeToRedis(host string) error {
-	c, err := rd.Dial("tcp", host)
-	if err != nil {
-		return err
-	}
-	defer c.Close()
-
-	_, err = c.Do("SET", "foo", "bar")
-	return err
-}
-
-func getConfig(password string) map[string]interface{} {
+func getConfig() map[string]interface{} {
 	return map[string]interface{}{
 		"module":     "redis",
 		"metricsets": []string{"info"},
 		"hosts":      []string{redisHost},
-		"password":   password,
 	}
 }

metricbeat/module/redis/key/data.go

@@ -24,14 +24,14 @@ import (
 	"github.com/elastic/beats/metricbeat/mb"
 )
 
-func eventMapping(r mb.ReporterV2, keyspace uint, info map[string]interface{}) {
+func eventMapping(keyspace uint, info map[string]interface{}) mb.Event {
 	info["id"] = fmt.Sprintf("%d:%s", keyspace, info["name"])
-	r.Event(mb.Event{
+	return mb.Event{
 		MetricSetFields: info,
 		ModuleFields: common.MapStr{
 			"keyspace": common.MapStr{
 				"id": fmt.Sprintf("db%d", keyspace),
 			},
 		},
-	})
+	}
 }

metricbeat/module/redis/key/key.go

@@ -18,14 +18,12 @@
 package key
 
 import (
-	"time"
+	"github.com/pkg/errors"
 
 	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/mb/parse"
 	"github.com/elastic/beats/metricbeat/module/redis"
-
-	rd "github.com/garyburd/redigo/redis"
 )
 
 var (
@@ -40,8 +38,7 @@ func init() {
 
 // MetricSet for fetching Redis server information and statistics.
 type MetricSet struct {
-	mb.BaseMetricSet
-	pool     *rd.Pool
+	*redis.MetricSet
 	patterns []KeyPattern
 }
 
@@ -54,34 +51,28 @@ type KeyPattern struct {
 
 // New creates new instance of MetricSet
 func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
-	// Unpack additional configuration options.
 	config := struct {
-		IdleTimeout time.Duration `config:"idle_timeout"`
-		Network     string        `config:"network"`
-		MaxConn     int           `config:"maxconn" validate:"min=1"`
-		Password    string        `config:"password"`
-		Patterns    []KeyPattern  `config:"key.patterns" validate:"nonzero,required"`
-	}{
-		Network:  "tcp",
-		MaxConn:  10,
-		Password: "",
-	}
+		Patterns []KeyPattern `config:"key.patterns" validate:"nonzero,required"`
+	}{}
 	err := base.Module().UnpackConfig(&config)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrap(err, "failed to read configuration for 'key' metricset")
+	}
+
+	ms, err := redis.NewMetricSet(base)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create 'key' metricset")
 	}
 
 	return &MetricSet{
-		BaseMetricSet: base,
-		pool: redis.CreatePool(base.Host(), config.Password, config.Network,
-			config.MaxConn, config.IdleTimeout, base.Module().Config().Timeout),
-		patterns: config.Patterns,
+		MetricSet: ms,
+		patterns:  config.Patterns,
 	}, nil
 }
 
 // Fetch fetches information from Redis keys
 func (m *MetricSet) Fetch(r mb.ReporterV2) {
-	conn := m.pool.Get()
+	conn := m.Connection()
 	for _, p := range m.patterns {
 		if err := redis.Select(conn, p.Keyspace); err != nil {
 			logp.Err("Failed to select keyspace %d: %s", p.Keyspace, err)
@@ -90,7 +81,7 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) {
 
 		keys, err := redis.FetchKeys(conn, p.Pattern, p.Limit)
 		if err != nil {
-			logp.Err("Failed to fetch list of keys in keyspace %d with pattern '%s': %s", p.Keyspace, p.Pattern, err)
+			logp.Err("Failed to list keys in keyspace %d with pattern '%s': %s", p.Keyspace, p.Pattern, err)
 			continue
 		}
 		if p.Limit > 0 && len(keys) > int(p.Limit) {
@@ -104,12 +95,11 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) {
 				logp.Err("Failed to fetch key info for key %s in keyspace %d", key, p.Keyspace)
 				continue
 			}
-			eventMapping(r, p.Keyspace, keyInfo)
+			event := eventMapping(p.Keyspace, keyInfo)
+			if !r.Event(event) {
+				debugf("Failed to report event, interrupting Fetch")
+				return
+			}
 		}
 	}
 }
-
-// Close connections
-func (m *MetricSet) Close() error {
-	return m.pool.Close()
-}

metricbeat/module/redis/keyspace/data.go

@@ -23,18 +23,22 @@ import (
 	"github.com/elastic/beats/libbeat/common"
 	s "github.com/elastic/beats/libbeat/common/schema"
 	c "github.com/elastic/beats/libbeat/common/schema/mapstrstr"
+	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/module/redis"
 )
 
 // Map data to MapStr
-func eventsMapping(info map[string]string) []common.MapStr {
-	events := []common.MapStr{}
+func eventsMapping(r mb.ReporterV2, info map[string]string) {
 	for key, space := range getKeyspaceStats(info) {
 		space["id"] = key
-		events = append(events, space)
+		event := mb.Event{
+			MetricSetFields: space,
+		}
+		if !r.Event(event) {
+			debugf("Failed to report event, interrupting Fetch")
+			return
+		}
 	}
-
-	return events
 }
 
 func getKeyspaceStats(info map[string]string) map[string]common.MapStr {