Release connections in metricbeat redis module (elastic#12914)

Release pooled connections in metricbeat redis module when they are not
needed anymore so they can be reused.

(cherry picked from commit 129d2ee)

CHANGELOG.next.asciidoc

@@ -145,6 +145,8 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Reuse connections in PostgreSQL metricsets. {issue}12504[12504] {pull}12603[12603]
 - PdhExpandWildCardPathW will not expand counter paths in 32 bit windows systems, workaround will use a different function.{issue}12590[12590]{pull}12622[12622]
 - In the elasticsearch/node_stats metricset, if xpack is enabled, make parsing of ES node load average optional as ES on Windows doesn't report load average. {pull}12866[12866]
+- Ramdisk is not filtered out when collecting disk performance counters in diskio metricset {issue}12814[12814] {pull}12829[12829]
+- Fix connections leak in redis module {pull}12914[12914]
 - Fix wrong uptime reporting by system/uptime metricset under Windows. {pull}12915[12915]
 
 *Packetbeat*

metricbeat/module/redis/info/info.go

@@ -52,8 +52,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 
 // Fetch fetches metrics from Redis by issuing the INFO command.
 func (m *MetricSet) Fetch(r mb.ReporterV2) error {
+	conn := m.Connection()
+	defer conn.Close()
+
 	// Fetch default INFO.
-	info, err := redis.FetchRedisInfo("default", m.Connection())
+	info, err := redis.FetchRedisInfo("default", conn)
 	if err != nil {
 		return errors.Wrap(err, "failed to fetch redis info")
 	}

metricbeat/module/redis/key/key.go

@@ -72,6 +72,8 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 // Fetch fetches information from Redis keys
 func (m *MetricSet) Fetch(r mb.ReporterV2) error {
 	conn := m.Connection()
+	defer conn.Close()
+
 	for _, p := range m.patterns {
 		if err := redis.Select(conn, p.Keyspace); err != nil {
 			msg := errors.Wrapf(err, "Failed to select keyspace %d", p.Keyspace)

metricbeat/module/redis/keyspace/keyspace.go

@@ -50,8 +50,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 
 // Fetch fetches metrics from Redis by issuing the INFO command.
 func (m *MetricSet) Fetch(r mb.ReporterV2) error {
+	conn := m.Connection()
+	defer conn.Close()
+
 	// Fetch default INFO.
-	info, err := redis.FetchRedisInfo("keyspace", m.Connection())
+	info, err := redis.FetchRedisInfo("keyspace", conn)
 	if err != nil {
 		return errors.Wrap(err, "Failed to fetch redis info for keyspaces")
 	}